Spaces:
Sleeping
Sleeping
| """ | |
| database.py | |
| ----------- | |
| Tiny SQLite persistence layer for detection history. | |
| Table `detections`: | |
| id, filename, file_type, prediction, confidence, | |
| frames_analyzed, processing_time, created_at | |
| """ | |
| import os | |
| import sqlite3 | |
| from contextlib import closing | |
| from datetime import datetime, timezone | |
| BASE_DIR = os.path.dirname(os.path.abspath(__file__)) | |
| DB_PATH = os.environ.get("DEEPFAKE_DB_PATH", os.path.join(BASE_DIR, "deepfake.db")) | |
| SCHEMA = """ | |
| CREATE TABLE IF NOT EXISTS detections ( | |
| id INTEGER PRIMARY KEY AUTOINCREMENT, | |
| filename TEXT NOT NULL, | |
| file_type TEXT NOT NULL CHECK (file_type IN ('image', 'video')), | |
| prediction TEXT NOT NULL CHECK (prediction IN ('REAL', 'FAKE', 'UNKNOWN')), | |
| confidence REAL NOT NULL DEFAULT 0, | |
| frames_analyzed INTEGER NOT NULL DEFAULT 0, | |
| processing_time REAL NOT NULL DEFAULT 0, | |
| created_at TEXT NOT NULL | |
| ); | |
| CREATE INDEX IF NOT EXISTS idx_detections_created | |
| ON detections (created_at DESC); | |
| """ | |
| def _utcnow(): | |
| return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") | |
| def get_connection(): | |
| conn = sqlite3.connect(DB_PATH) | |
| conn.row_factory = sqlite3.Row | |
| conn.execute("PRAGMA journal_mode=WAL;") | |
| return conn | |
| def init_db(): | |
| with closing(get_connection()) as conn: | |
| with conn: | |
| conn.executescript(SCHEMA) | |
| def insert_detection(filename, file_type, prediction, confidence, | |
| frames_analyzed=0, processing_time=0.0): | |
| with closing(get_connection()) as conn: | |
| with conn: | |
| cur = conn.execute( | |
| """INSERT INTO detections | |
| (filename, file_type, prediction, confidence, | |
| frames_analyzed, processing_time, created_at) | |
| VALUES (?, ?, ?, ?, ?, ?, ?)""", | |
| (filename, file_type, prediction, float(confidence), | |
| int(frames_analyzed), float(processing_time), _utcnow()), | |
| ) | |
| return cur.lastrowid | |
| def _to_dict(row): | |
| return dict(row) if row is not None else None | |
| def get_detection(det_id): | |
| with closing(get_connection()) as conn: | |
| row = conn.execute( | |
| "SELECT * FROM detections WHERE id = ?", (det_id,) | |
| ).fetchone() | |
| return _to_dict(row) | |
| def get_history(limit=50): | |
| with closing(get_connection()) as conn: | |
| rows = conn.execute( | |
| "SELECT * FROM detections ORDER BY id DESC LIMIT ?", (int(limit),) | |
| ).fetchall() | |
| return [_to_dict(r) for r in rows] | |
| def get_stats(): | |
| with closing(get_connection()) as conn: | |
| total = conn.execute("SELECT COUNT(*) AS c FROM detections").fetchone()["c"] | |
| real_count = conn.execute( | |
| "SELECT COUNT(*) AS c FROM detections WHERE prediction = 'REAL'" | |
| ).fetchone()["c"] | |
| fake_count = conn.execute( | |
| "SELECT COUNT(*) AS c FROM detections WHERE prediction = 'FAKE'" | |
| ).fetchone()["c"] | |
| unknown_count = conn.execute( | |
| "SELECT COUNT(*) AS c FROM detections WHERE prediction = 'UNKNOWN'" | |
| ).fetchone()["c"] | |
| avg_conf = conn.execute( | |
| "SELECT AVG(confidence) AS a FROM detections " | |
| "WHERE prediction IN ('REAL','FAKE')" | |
| ).fetchone()["a"] | |
| recent = get_history(limit=5) | |
| return { | |
| "total": total, | |
| "real_count": real_count, | |
| "fake_count": fake_count, | |
| "unknown_count": unknown_count, | |
| "avg_confidence": round(avg_conf, 1) if avg_conf is not None else 0.0, | |
| "recent": recent, | |
| } | |
| def clear_history(): | |
| with closing(get_connection()) as conn: | |
| with conn: | |
| cur = conn.execute("DELETE FROM detections") | |
| return cur.rowcount | |
| if __name__ == "__main__": | |
| init_db() | |
| print("Database initialized at:", DB_PATH) | |