DanceVideos/app/db.py

# db.py (modifié)
import sqlite3
from pathlib import Path
import pandas as pd

DB_PATH = Path.home() / "Documents/.DanceVideos/db.sqlite"

def get_conn():
    conn = sqlite3.connect(DB_PATH, timeout=30, check_same_thread=False)
    conn.execute("PRAGMA foreign_keys = ON;")
    return conn


def load_videos():
    with get_conn() as conn:
        return pd.read_sql_query("SELECT * FROM videos ORDER BY record_datetime DESC", conn)

def load_labels():
    with get_conn() as conn:
        df = pd.read_sql_query("SELECT name FROM labels ORDER BY name", conn)
        return df["name"].tolist()

def create_labels(label_names):
    if not label_names:
        return
    with get_conn() as conn:
        conn.executemany(
            "INSERT OR IGNORE INTO labels (name) VALUES (?)",
            [(name,) for name in label_names]
        )
        conn.commit()

def get_label_ids(label_names):
    label_ids = {}
    with get_conn() as conn:
        cursor = conn.cursor()
        for name in label_names:
            cursor.execute("SELECT id FROM labels WHERE name=?", (name,))
            row = cursor.fetchone()
            if row:
                label_ids[name] = row[0]
    return label_ids

def load_video_labels(file_name):
    with get_conn() as conn:
        cursor = conn.cursor()
        query = """
            SELECT l.name
            FROM labels l
            JOIN video_labels vl ON l.id = vl.label_id
            WHERE vl.video_file_name = ?
        """
        return [row[0] for row in cursor.execute(query, (file_name,))]

def save_video_labels(file_name, label_names):
    if label_names is None:
        label_names = []
    create_labels(label_names)
    label_ids = get_label_ids(label_names)
    with get_conn() as conn:
        cursor = conn.cursor()
        cursor.execute("DELETE FROM video_labels WHERE video_file_name = ?", (file_name,))
        for lid in label_ids.values():
            cursor.execute(
                "INSERT OR REPLACE INTO video_labels (video_file_name, label_id) VALUES (?, ?)",
                (file_name, lid),
            )
        cursor.execute("""
            DELETE FROM labels
            WHERE id NOT IN (SELECT DISTINCT label_id FROM video_labels)
        """)
        conn.commit()

def update_video_difficulty(file_name, level):
    with get_conn() as conn:
        conn.execute("UPDATE videos SET difficulty_level = ? WHERE file_name = ?", (level, file_name))
        conn.commit()

def get_unique_days():
    """Retourne la liste unique des jours de la semaine enregistrés dans la base."""
    with get_conn() as conn:
        df = pd.read_sql_query("SELECT DISTINCT day_of_week FROM videos WHERE day_of_week IS NOT NULL ORDER BY day_of_week", conn)
        return [d for d in df["day_of_week"].dropna().tolist() if d.strip()]

def get_unique_difficulties():
    """Retourne la liste des niveaux de difficulté existants."""
    with get_conn() as conn:
        df = pd.read_sql_query("SELECT DISTINCT difficulty_level FROM videos WHERE difficulty_level IS NOT NULL ORDER BY difficulty_level", conn)
        return [d for d in df["difficulty_level"].dropna().tolist() if d.strip()]

def get_unique_addresses():
    """Retourne les adresses connues (exclut 'unknown')."""
    with get_conn() as conn:
        df = pd.read_sql_query("SELECT DISTINCT address FROM videos WHERE address NOT LIKE '%unknown%' ORDER BY address", conn)
        return [a for a in df["address"].dropna().tolist() if a.strip()]


def search_videos(
    label_names=None,
    day_of_week=None,
    address_keyword=None,
    start_date=None,
    end_date=None,
    difficulty=None
):
    """
    Retourne une DataFrame filtrée selon les critères fournis.
    """
    label_names = label_names or []

    query = """
        SELECT DISTINCT v.*
        FROM videos v
        LEFT JOIN video_labels vl ON vl.video_file_name = v.file_name
        LEFT JOIN labels l ON l.id = vl.label_id
        WHERE 1=1
    """
    params = []

    # 🔖 Filtres par label (tous doivent être présents)
    if label_names:
        placeholders = ",".join("?" for _ in label_names)
        query += f" AND l.name IN ({placeholders})"
        params += label_names

    # 📆 Jour de la semaine
    if day_of_week:
        query += " AND v.day_of_week = ?"
        params.append(day_of_week)

    # 🗺️ Mot-clé d'adresse (et exclusion unknown)
    if address_keyword:
        query += " AND v.address NOT LIKE '%unknown%' AND v.address LIKE ?"
        params.append(f"%{address_keyword}%")

    # 📅 Filtre par date
    if start_date:
        query += " AND v.record_datetime >= ?"
        params.append(start_date)
    if end_date:
        query += " AND v.record_datetime <= ?"
        params.append(end_date)

    # 💪 Niveau de difficulté
    if difficulty and difficulty != "Tous":
        query += " AND v.difficulty_level = ?"
        params.append(difficulty)

    query += " ORDER BY v.record_datetime DESC"

    with get_conn() as conn:
        return pd.read_sql_query(query, conn, params=params)