# 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.row_factory = sqlite3.Row 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, label_logic="OR", ): """ Retourne une DataFrame filtrée selon les critères fournis. label_logic: "OR" (au moins un label) ou "AND" (tous les labels). """ label_names = label_names or [] params = [] base_query = """ SELECT DISTINCT v.* FROM videos v WHERE 1=1 """ # 🔖 Filtres par label if label_names: if label_logic == "AND": # Toutes les étiquettes doivent être présentes placeholders = ",".join("?" * len(label_names)) base_query += f""" AND v.file_name IN ( SELECT vl.video_file_name FROM video_labels vl JOIN labels l ON l.id = vl.label_id WHERE l.name IN ({placeholders}) GROUP BY vl.video_file_name HAVING COUNT(DISTINCT l.name) = {len(label_names)} ) """ params.extend(label_names) else: # OR : au moins une étiquette correspond placeholders = ",".join("?" * len(label_names)) base_query += f""" AND v.file_name IN ( SELECT vl.video_file_name FROM video_labels vl JOIN labels l ON l.id = vl.label_id WHERE l.name IN ({placeholders}) ) """ params.extend(label_names) # 📆 Jour de la semaine if day_of_week: base_query += " AND v.day_of_week = ?" params.append(day_of_week) # 🗺️ Mot-clé d'adresse (et exclusion unknown) if address_keyword: base_query += " AND v.address NOT LIKE '%unknown%' AND v.address LIKE ?" params.append(f"%{address_keyword}%") # 📅 Plage de dates if start_date: base_query += " AND v.record_datetime >= ?" params.append(start_date) if end_date: base_query += " AND v.record_datetime <= ?" params.append(end_date) # 💪 Niveau de difficulté if difficulty and difficulty != "Tous": base_query += " AND v.difficulty_level = ?" params.append(difficulty) # 🔽 Tri base_query += " ORDER BY v.record_datetime DESC" with get_conn() as conn: return pd.read_sql_query(base_query, conn, params=params) def get_video_playlists(file_name): with get_conn() as conn: query = """ SELECT p.name FROM playlists p JOIN video_playlists vp ON vp.playlist_id = p.id WHERE vp.video_file_name = ? """ return [row[0] for row in conn.execute(query, (file_name,))] def get_videos_in_playlist(playlist_id): with get_conn() as conn: query = """ SELECT v.* FROM videos v JOIN video_playlists vp ON vp.video_file_name = v.file_name WHERE vp.playlist_id = ? ORDER BY vp.position """ return conn.execute(query, (playlist_id,)).fetchall() def add_video_to_playlist(playlist_id, file_name): with get_conn() as conn: conn.execute(""" INSERT OR IGNORE INTO video_playlists (video_file_name, playlist_id, position) VALUES (?, ?, COALESCE((SELECT MAX(position)+1 FROM video_playlists WHERE playlist_id=?), 0)) """, (file_name, playlist_id, playlist_id)) conn.commit() def remove_video_from_playlist(playlist_id, file_name): with get_conn() as conn: conn.execute("DELETE FROM video_playlists WHERE playlist_id=? AND video_file_name=?", (playlist_id, file_name)) conn.commit()