database.py

import sqlite3
import pandas as pd
from config import DB_PATH


def get_conn():
    return sqlite3.connect(DB_PATH, check_same_thread=False)


def init_db():
    conn = get_conn()
    cur = conn.cursor()

    cur.execute("""
        CREATE TABLE IF NOT EXISTS analysis_results (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL,
            file_name TEXT NOT NULL,
            file_type TEXT NOT NULL,
            label TEXT NOT NULL,
            score REAL NOT NULL,
            saved_to TEXT NOT NULL,
            source TEXT DEFAULT ''
        )
    """)

    conn.commit()
    conn.close()


def insert_result(timestamp, file_name, file_type, label, score, saved_to, source=""):
    conn = get_conn()
    cur = conn.cursor()

    cur.execute("""
        INSERT INTO analysis_results
        (timestamp, file_name, file_type, label, score, saved_to, source)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    """, (
        str(timestamp),
        str(file_name),
        str(file_type),
        str(label),
        float(score),
        str(saved_to),
        str(source)
    ))

    conn.commit()
    conn.close()


def read_results_df():
    conn = get_conn()

    try:
        df = pd.read_sql_query("""
            SELECT
                timestamp,
                file_name,
                file_type,
                label,
                score,
                saved_to,
                source
            FROM analysis_results
            ORDER BY id DESC
        """, conn)

        return df

    except Exception:
        return pd.DataFrame(columns=[
            "timestamp",
            "file_name",
            "file_type",
            "label",
            "score",
            "saved_to",
            "source"
        ])

    finally:
        conn.close()


def delete_all_results():
    conn = get_conn()
    cur = conn.cursor()

    cur.execute("DELETE FROM analysis_results")

    conn.commit()
    conn.close()


def export_results_csv(csv_path="results_export.csv"):
    df = read_results_df()
    df.to_csv(csv_path, index=False, encoding="utf-8-sig")
    return csv_path


def get_stats():
    conn = get_conn()
    cur = conn.cursor()

    stats = {
        "total": 0,
        "fake_count": 0,
        "real_count": 0,
        "image_count": 0,
        "audio_count": 0,
        "video_count": 0,
        "avg_score": 0.0
    }

    try:
        cur.execute("SELECT COUNT(*) FROM analysis_results")
        stats["total"] = cur.fetchone()[0] or 0

        cur.execute("""
            SELECT COUNT(*)
            FROM analysis_results
            WHERE lower(label) LIKE '%fake%'
               OR lower(label) LIKE '%deepfake%'
               OR lower(label) LIKE '%synthetic%'
               OR lower(label) LIKE '%suspicious%'
        """)
        stats["fake_count"] = cur.fetchone()[0] or 0

        stats["real_count"] = stats["total"] - stats["fake_count"]

        cur.execute("""
            SELECT COUNT(*)
            FROM analysis_results
            WHERE lower(file_type) = 'image'
        """)
        stats["image_count"] = cur.fetchone()[0] or 0

        cur.execute("""
            SELECT COUNT(*)
            FROM analysis_results
            WHERE lower(file_type) = 'audio'
        """)
        stats["audio_count"] = cur.fetchone()[0] or 0

        cur.execute("""
            SELECT COUNT(*)
            FROM analysis_results
            WHERE lower(file_type) = 'video'
        """)
        stats["video_count"] = cur.fetchone()[0] or 0

        cur.execute("""
            SELECT AVG(score)
            FROM analysis_results
        """)
        avg_score = cur.fetchone()[0]
        stats["avg_score"] = round(float(avg_score), 4) if avg_score is not None else 0.0

        return stats

    except Exception:
        return stats

    finally:
        conn.close()