Spaces:
Sleeping
Sleeping
File size: 3,866 Bytes
2641a98 | 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 116 117 118 119 120 121 122 123 124 | """
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)
|