File size: 4,050 Bytes
48c7fed | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | from sqlalchemy import create_engine, Column, String, Float, DateTime, Integer, JSON, Text, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
import json
import os
Base = declarative_base()
class AnalysisResult(Base):
"""Database model for storing analysis results"""
__tablename__ = "analysis_results"
id = Column(Integer, primary_key=True)
# File information
file_id = Column(String(50), unique=True, nullable=False, index=True)
filename = Column(String(255), nullable=False)
file_format = Column(String(10), nullable=False) # pdf, docx, txt, raw
file_size = Column(Integer, nullable=True) # in bytes
# Upload information
upload_timestamp = Column(DateTime, nullable=False, default=datetime.utcnow, index=True)
user_id = Column(String(100), nullable=True) # For future auth integration
# Text analysis
text_preview = Column(String(500), nullable=True) # First 500 chars
text_length = Column(Integer, nullable=False) # Total character count
word_count = Column(Integer, nullable=False)
# Overall detection results
overall_ai_score = Column(Float, nullable=False) # 0-1
overall_confidence = Column(String(20), nullable=False) # very_low, low, medium, high, very_high
# Detector results (JSON)
detector_results = Column(JSON, nullable=False) # Full detector results
# Status
analysis_status = Column(String(20), default="completed") # completed, failed, pending
error_message = Column(Text, nullable=True)
# Report storage
report_html_path = Column(String(500), nullable=True)
# Metadata
notes = Column(Text, nullable=True)
is_flagged = Column(Boolean, default=False) # For manual flagging
def to_dict(self):
"""Convert to dictionary"""
return {
"id": self.id,
"file_id": self.file_id,
"filename": self.filename,
"file_format": self.file_format,
"file_size": self.file_size,
"upload_timestamp": self.upload_timestamp.isoformat() if self.upload_timestamp else None,
"text_length": self.text_length,
"word_count": self.word_count,
"overall_ai_score": self.overall_ai_score,
"overall_ai_score_percentage": f"{self.overall_ai_score * 100:.1f}%",
"overall_confidence": self.overall_confidence,
"detector_results": self.detector_results,
"analysis_status": self.analysis_status,
"error_message": self.error_message,
"is_flagged": self.is_flagged,
"notes": self.notes,
}
@staticmethod
def get_status_label(score: float) -> str:
"""Get human-readable status based on AI score"""
if score < 0.3:
return "Likely Human"
elif score < 0.6:
return "Suspicious"
else:
return "Likely AI"
class Session:
"""Database session manager"""
_engine = None
_SessionLocal = None
@classmethod
def init(cls):
"""Initialize database connection"""
# Get DATABASE_URL from environment or use SQLite default
database_url = os.getenv('DATABASE_URL', 'sqlite:///./analysis_results.db')
cls._engine = create_engine(
database_url,
connect_args={"check_same_thread": False} if "sqlite" in database_url else {}
)
# Create tables
Base.metadata.create_all(cls._engine)
cls._SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=cls._engine)
@classmethod
def get_session(cls):
"""Get a new database session"""
if cls._SessionLocal is None:
cls.init()
return cls._SessionLocal()
@classmethod
def close(cls):
"""Close database connection"""
if cls._engine:
cls._engine.dispose()
|