Spaces:
Running
Running
| # database.py - COMPLETE VERSION | |
| 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=True) | |
| phone = Column(String(20), unique=True, nullable=True) | |
| 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=False) | |
| verification_code = Column(String(10)) | |
| verification_code_expires = Column(DateTime) | |
| reset_token = Column(String(100), nullable=True) | |
| reset_token_expires = Column(DateTime, nullable=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=True) | |
| 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) | |
| is_pinned = Column(Boolean, default=False) | |
| created_at = Column(DateTime, default=datetime.utcnow) | |
| user = relationship('User', back_populates='detections') | |
| class Database: | |
| def __init__(self): | |
| # Cek apakah di production (Railway / Hugging Face Spaces) | |
| DATABASE_URL = os.environ.get('DATABASE_URL') | |
| if DATABASE_URL: | |
| # Gunakan PostgreSQL | |
| self.engine = create_engine(DATABASE_URL) | |
| else: | |
| # Gunakan SQLite untuk development | |
| 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) | |
| def get_session(self): | |
| return self.Session() | |
| # ============ 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): | |
| if not email: | |
| return None | |
| session = self.Session() | |
| try: | |
| return session.query(User).filter(User.email == email).first() | |
| finally: | |
| session.close() | |
| def get_user_by_phone(self, phone): | |
| if not phone: | |
| return None | |
| session = self.Session() | |
| try: | |
| return session.query(User).filter(User.phone == phone).first() | |
| finally: | |
| session.close() | |
| def get_user_by_username(self, username): | |
| if not username: | |
| return None | |
| session = self.Session() | |
| try: | |
| return session.query(User).filter(User.username == username).first() | |
| finally: | |
| session.close() | |
| def create_user(self, username, password_hash, email=None, phone=None, full_name=None, is_verified=False): | |
| session = self.Session() | |
| try: | |
| user = User( | |
| username=username, | |
| password_hash=password_hash, | |
| email=email if email else None, | |
| phone=phone if phone else None, | |
| full_name=full_name, | |
| is_verified=is_verified | |
| ) | |
| 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 | |
| except Exception as e: | |
| session.rollback() | |
| print(f"Error updating user: {e}") | |
| return False | |
| finally: | |
| session.close() | |
| # ============ VERIFICATION METHODS ============ | |
| def set_verification_code(self, user_id, code, expires): | |
| session = self.Session() | |
| try: | |
| user = session.query(User).filter(User.id == user_id).first() | |
| if user: | |
| user.verification_code = code | |
| user.verification_code_expires = expires | |
| session.commit() | |
| return True | |
| return False | |
| finally: | |
| session.close() | |
| def verify_user(self, user_id, code): | |
| session = self.Session() | |
| try: | |
| user = session.query(User).filter(User.id == user_id).first() | |
| if user and user.verification_code == code: | |
| if user.verification_code_expires and user.verification_code_expires > datetime.utcnow(): | |
| user.is_verified = True | |
| user.verification_code = None | |
| user.verification_code_expires = None | |
| session.commit() | |
| return True | |
| return False | |
| finally: | |
| session.close() | |
| # ============ DETECTION METHODS ============ | |
| def add_detection(self, text, result, user_id=None, file_name=None, file_type=None): | |
| session = self.Session() | |
| try: | |
| history = DetectionHistory( | |
| user_id=user_id, | |
| text_preview=text[:200] + "..." if len(text) > 200 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.is_pinned.desc(), | |
| DetectionHistory.created_at.desc() | |
| ).limit(limit).all() | |
| return history | |
| finally: | |
| session.close() | |
| def get_all_history(self, limit=50): | |
| session = self.Session() | |
| try: | |
| history = session.query(DetectionHistory).order_by( | |
| DetectionHistory.created_at.desc() | |
| ).limit(limit).all() | |
| return history | |
| finally: | |
| session.close() | |
| def delete_detection(self, detection_id, user_id): | |
| session = self.Session() | |
| try: | |
| detection = session.query(DetectionHistory).filter( | |
| DetectionHistory.id == detection_id, | |
| DetectionHistory.user_id == user_id | |
| ).first() | |
| if detection: | |
| session.delete(detection) | |
| session.commit() | |
| return True | |
| return False | |
| finally: | |
| session.close() | |
| def toggle_pin(self, detection_id, user_id): | |
| session = self.Session() | |
| try: | |
| detection = session.query(DetectionHistory).filter( | |
| DetectionHistory.id == detection_id, | |
| DetectionHistory.user_id == user_id | |
| ).first() | |
| if detection: | |
| detection.is_pinned = not detection.is_pinned | |
| session.commit() | |
| return detection.is_pinned | |
| return None | |
| 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() | |
| avg_confidence = session.query(DetectionHistory.confidence).filter( | |
| DetectionHistory.user_id == user_id | |
| ).all() | |
| avg_conf = sum(c[0] for c in avg_confidence) / len(avg_confidence) if avg_confidence else 0 | |
| return { | |
| 'total': total, | |
| 'ai_count': ai_count, | |
| 'human_count': human_count, | |
| 'avg_confidence': round(avg_conf, 2) | |
| } | |
| finally: | |
| session.close() | |
| def clear_user_history(self, user_id): | |
| session = self.Session() | |
| try: | |
| session.query(DetectionHistory).filter(DetectionHistory.user_id == user_id).delete() | |
| session.commit() | |
| return True | |
| except: | |
| session.rollback() | |
| return False | |
| finally: | |
| session.close() |