Blame for database.py

# database.py - Database handler untuk user dan history

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, Text, Boolean, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from datetime import datetime
import os

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    email = Column(String(100), unique=True, nullable=False)
    username = Column(String(50), unique=True, nullable=False)
    password_hash = Column(String(200), nullable=False)
    full_name = Column(String(100))
    is_verified = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    last_login = Column(DateTime)
    
    detections = relationship('DetectionHistory', back_populates='user', cascade='all, delete-orphan')

class DetectionHistory(Base):
    __tablename__ = 'detection_history'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'), nullable=False)
    text_preview = Column(String(200))
    text_full = Column(Text, nullable=True)
    file_name = Column(String(200), nullable=True)
    file_type = Column(String(50), nullable=True)
    ai_probability = Column(Float)
    human_probability = Column(Float)
    confidence = Column(Float)
    is_ai = Column(Integer)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    user = relationship('User', back_populates='detections')

class Database:
    def __init__(self):
        self.engine = create_engine('sqlite:///detections.db', connect_args={'check_same_thread': False})
        Base.metadata.create_all(self.engine)
        self.Session = sessionmaker(bind=self.engine)
    
    # ============ USER METHODS ============
    
    def get_user_by_id(self, user_id):
        session = self.Session()
        try:
            return session.query(User).filter(User.id == user_id).first()
        finally:
            session.close()
    
    def get_user_by_email(self, email):
        session = self.Session()
        try:
            return session.query(User).filter(User.email == email).first()
        finally:
            session.close()
    
    def get_user_by_username(self, username):
        session = self.Session()
        try:
            return session.query(User).filter(User.username == username).first()
        finally:
            session.close()
    
    def create_user(self, username, email, password_hash, full_name=None):
        session = self.Session()
        try:
            user = User(
                username=username,
                email=email,
                password_hash=password_hash,
                full_name=full_name if full_name else username,
                is_verified=True
            )
            session.add(user)
            session.commit()
            session.refresh(user)
            return user
        except Exception as e:
            session.rollback()
            print(f"Error creating user: {e}")
            return None
        finally:
            session.close()
    
    def update_user(self, user_id, **kwargs):
        session = self.Session()
        try:
            user = session.query(User).filter(User.id == user_id).first()
            if user:
                for key, value in kwargs.items():
                    if hasattr(user, key):
                        setattr(user, key, value)
                session.commit()
                return True
            return False
        finally:
            session.close()
    
    # ============ DETECTION METHODS ============
    
    def add_detection(self, user_id, text, result, file_name=None, file_type=None):
        session = self.Session()
        try:
            history = DetectionHistory(
                user_id=user_id,
                text_preview=text[:150] + "..." if len(text) > 150 else text,
                text_full=text[:5000],
                file_name=file_name,
                file_type=file_type,
                ai_probability=result['ai_probability'],
                human_probability=result['human_probability'],
                confidence=result['confidence'],
                is_ai=1 if result['is_ai'] else 0
            )
            session.add(history)
            session.commit()
            return history.id
        except Exception as e:
            print(f"Error saving detection: {e}")
            session.rollback()
            return None
        finally:
            session.close()
    
    def get_user_history(self, user_id, limit=50):
        session = self.Session()
        try:
            history = session.query(DetectionHistory).filter(
                DetectionHistory.user_id == user_id
            ).order_by(
                DetectionHistory.created_at.desc()
            ).limit(limit).all()
            return history
        finally:
            session.close()
    
    def get_user_stats(self, user_id):
        session = self.Session()
        try:
            total = session.query(DetectionHistory).filter(DetectionHistory.user_id == user_id).count()
            ai_count = session.query(DetectionHistory).filter(
                DetectionHistory.user_id == user_id,
                DetectionHistory.is_ai == 1
            ).count()
            human_count = session.query(DetectionHistory).filter(
                DetectionHistory.user_id == user_id,
                DetectionHistory.is_ai == 0
            ).count()
            
            return {
                'total': total,
                'ai_count': ai_count,
                'human_count': human_count
            }
        finally:
            session.close()