Spaces:
Sleeping
Sleeping
| # 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() |