# 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 delete_video(file_name): with get_conn() as conn: conn.execute("DELETE FROM videos WHERE file_name = ?", (file_name,)) conn.commit() 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 update_video_alias(file_name, alias): with get_conn() as conn: conn.execute("UPDATE videos SET alias = ? WHERE file_name = ?", (alias, 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", include_playlists=None, exclude_playlists=None, logic="OR", # logique entre playlists incluses **kwargs, ): """ Retourne une DataFrame filtrée selon les critères fournis. label_logic: "OR" (au moins un label) ou "AND" (tous les labels) logic: "OR" ou "AND" pour la combinaison de playlists incluses """ label_names = label_names or [] include_playlists = include_playlists or [] exclude_playlists = exclude_playlists or [] params = [] base_query = """ SELECT DISTINCT v.* FROM videos v WHERE 1=1 """ # 🔖 Filtres par label if label_names: if label_logic == "AND": 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: 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) # 🎵 Filtres par playlists incluses if include_playlists: placeholders = ",".join("?" * len(include_playlists)) if logic == "AND": # Vidéos présentes dans TOUTES les playlists base_query += f""" AND v.file_name IN ( SELECT vp.video_file_name FROM video_playlists vp JOIN playlists p ON p.id = vp.playlist_id WHERE p.id IN ({placeholders}) GROUP BY vp.video_file_name HAVING COUNT(DISTINCT p.id) = {len(include_playlists)} ) """ params.extend(include_playlists) else: # Vidéos présentes dans AU MOINS une playlist base_query += f""" AND v.file_name IN ( SELECT vp.video_file_name FROM video_playlists vp JOIN playlists p ON p.id = vp.playlist_id WHERE p.id IN ({placeholders}) ) """ params.extend(include_playlists) # ❌ Filtres par playlists exclues if exclude_playlists: placeholders = ",".join("?" * len(exclude_playlists)) base_query += f""" AND v.file_name NOT IN ( SELECT vp.video_file_name FROM video_playlists vp JOIN playlists p ON p.id = vp.playlist_id WHERE p.id IN ({placeholders}) ) """ params.extend(exclude_playlists) # 📆 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_video_file_names_in_playlist(playlist_id): """Retourne un set de file_name pour une playlist donnée.""" with get_conn() as conn: rows = conn.execute( # "SELECT video_file_name FROM video_playlists WHERE playlist_id = ?", "SELECT video_file_name FROM playlist_videos WHERE playlist_id = ?", (playlist_id,) ).fetchall() names = [] for r in rows: if hasattr(r, "keys"): names.append(r["video_file_name"]) else: names.append(r[0]) return set(names) def get_videos_in_playlist(playlist_id): """Retourne une liste d'objets Video complets pour une playlist donnée.""" import pandas as pd from models import Video with get_conn() as conn: df = pd.read_sql_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 ASC """, conn, params=(playlist_id,)) return [Video(**row) for _, row in df.iterrows()] def add_video_to_playlist(playlist_id, file_name): print(dict(playlist_id=playlist_id, video_file_name=file_name),flush=True,) 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()