291 lines
10 KiB
Python
291 lines
10 KiB
Python
# 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() |