115 lines
3.3 KiB
Python
115 lines
3.3 KiB
Python
import sqlite3
|
||
from pathlib import Path
|
||
import pandas as pd
|
||
|
||
DB_PATH = Path.home() / "Documents/.DanceVideos/db.sqlite"
|
||
|
||
# --- Connexion SQLite avec timeout ---
|
||
def get_conn():
|
||
"""
|
||
Retourne une connexion SQLite avec timeout de 30s.
|
||
Utiliser un context manager pour garantir la fermeture.
|
||
"""
|
||
return sqlite3.connect(DB_PATH, timeout=30, check_same_thread=False)
|
||
|
||
# --- Vidéos ---
|
||
def load_videos():
|
||
"""
|
||
Retourne toutes les vidéos dans un DataFrame pandas.
|
||
"""
|
||
with get_conn() as conn:
|
||
return pd.read_sql_query("SELECT * FROM videos ORDER BY record_datetime DESC", conn)
|
||
|
||
# --- Labels ---
|
||
def load_labels():
|
||
"""
|
||
Retourne tous les labels existants dans une liste.
|
||
"""
|
||
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):
|
||
"""
|
||
Crée tous les labels de la liste s'ils n'existent pas.
|
||
"""
|
||
if not label_names:
|
||
return
|
||
|
||
with get_conn() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.executemany(
|
||
"INSERT OR IGNORE INTO labels (name) VALUES (?)",
|
||
[(name,) for name in label_names]
|
||
)
|
||
conn.commit()
|
||
|
||
def get_label_ids(label_names):
|
||
"""
|
||
Retourne un dictionnaire {label_name: label_id}.
|
||
"""
|
||
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
|
||
|
||
# --- Vidéo / Labels ---
|
||
def load_video_labels(video_file_name):
|
||
"""
|
||
Retourne la liste des labels associés à une vidéo.
|
||
"""
|
||
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, (video_file_name,))]
|
||
|
||
def save_video_labels(video_file_name, label_names):
|
||
"""
|
||
Associe une vidéo à plusieurs labels.
|
||
Supprime d'abord les labels précédents pour cette vidéo.
|
||
"""
|
||
if label_names is None:
|
||
label_names = []
|
||
|
||
# 1️⃣ Créer tous les labels
|
||
create_labels(label_names)
|
||
|
||
# 2️⃣ Récupérer les IDs
|
||
label_ids = get_label_ids(label_names)
|
||
|
||
# 3️⃣ Associer la vidéo aux labels dans une seule transaction
|
||
with get_conn() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
# 🔹 Supprimer les anciennes associations
|
||
cursor.execute(
|
||
"DELETE FROM video_labels WHERE video_file_name = ?",
|
||
(video_file_name,)
|
||
)
|
||
|
||
# 🔹 Ajouter les nouvelles associations
|
||
for lid in label_ids.values():
|
||
cursor.execute("""
|
||
INSERT INTO video_labels (video_file_name, label_id)
|
||
VALUES (?, ?)
|
||
""", (video_file_name, lid))
|
||
|
||
conn.commit()
|
||
cursor = conn.cursor()
|
||
for lid in label_ids.values():
|
||
cursor.execute("""
|
||
INSERT OR REPLACE INTO video_labels (video_file_name, label_id)
|
||
VALUES (?, ?)
|
||
""", (video_file_name, lid))
|
||
conn.commit()
|