from __future__ import annotations import argparse import json import pathlib import sqlite3 from typing import Any from moviebot.config import load_yaml from moviebot.crawlers.common import normalize_api_movie from moviebot.models import NormalizedMovie from moviebot.nfo_writer import NfoWriter from moviebot.strm_writer import StrmWriter from moviebot.utils import library_name, movie_folder_name def db_to_file_path(db_path: str | None, root: pathlib.Path, mount: str, local_root: pathlib.Path) -> pathlib.Path | None: if not db_path: return None normalized = str(db_path).replace("\\", "/") prefix = f"/{mount.strip('/')}/" if normalized.startswith(prefix): return (local_root / normalized[len(prefix) :]).resolve() return (root / normalized.lstrip("/")).resolve() def remove_file(path: pathlib.Path | None, allowed_root: pathlib.Path) -> None: if not path: return try: path.relative_to(allowed_root.resolve()) except ValueError: return if path.exists() and path.is_file(): path.unlink() def row_to_movie(row: sqlite3.Row) -> NormalizedMovie: return NormalizedMovie( source=row["source"] or "", source_slug=row["source_slug"] or "", title=row["title"] or "", original_title=row["original_title"] or "", year=row["year"] or "", media_type=row["media_type"] or "movie", tmdb_id=row["tmdb_id"] or "", imdb_id=row["imdb_id"] or "", poster_url=row["poster_url"] or "", backdrop_url=row["backdrop_url"] or "", runtime=row["runtime"] or "", content=row["content"] or "", genres=json.loads(row["genres_json"] or "[]"), countries=json.loads(row["countries_json"] or "[]"), actors=json.loads(row["actors_json"] or "[]"), directors=json.loads(row["directors_json"] or "[]"), ) def upsert_repaired_movie(conn: sqlite3.Connection, movie: NormalizedMovie) -> int: conn.execute( """ INSERT INTO movies ( merge_key, tmdb_id, media_type, title, original_title, year, imdb_id, poster_url, backdrop_url, runtime, content, genres_json, countries_json, actors_json, directors_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(merge_key) DO UPDATE SET tmdb_id=excluded.tmdb_id, media_type=excluded.media_type, title=excluded.title, original_title=excluded.original_title, year=excluded.year, imdb_id=COALESCE(NULLIF(excluded.imdb_id, ''), movies.imdb_id), poster_url=excluded.poster_url, backdrop_url=excluded.backdrop_url, runtime=excluded.runtime, content=excluded.content, genres_json=excluded.genres_json, countries_json=excluded.countries_json, actors_json=excluded.actors_json, directors_json=excluded.directors_json, updated_at=CURRENT_TIMESTAMP """, ( movie.merge_key, movie.tmdb_id, movie.media_type, movie.title, movie.original_title, movie.year, movie.imdb_id, movie.poster_url, movie.backdrop_url, movie.runtime, movie.content, json.dumps(movie.genres, ensure_ascii=False), json.dumps(movie.countries, ensure_ascii=False), json.dumps(movie.actors, ensure_ascii=False), json.dumps(movie.directors, ensure_ascii=False), ), ) row = conn.execute("SELECT id FROM movies WHERE merge_key = ?", (movie.merge_key,)).fetchone() return int(row["id"]) def repair_slug(config_path: str, source: str, slug: str) -> dict[str, Any]: app_config = load_yaml(config_path) root = pathlib.Path.cwd().resolve() library_root = (root / app_config["paths"]["libraries"]).resolve() db_mounts = app_config.get("db_mounts", {}) strm_mount = db_mounts.get("libraries", "/strm") playlist_mount = db_mounts.get("playlists", "/m3u8") strm_writer = StrmWriter( app_config["paths"]["libraries"], app_config["playlist_server"]["public_url"], strm_mount, ) nfo_writer = NfoWriter(app_config["paths"]["libraries"], strm_mount) conn = sqlite3.connect(app_config["database"]["path"]) conn.row_factory = sqlite3.Row source_row = conn.execute( """ SELECT ms.id AS source_id, ms.source, ms.source_slug, ms.raw_json, ms.movie_id, m.title, m.original_title, m.year, m.media_type, m.tmdb_id, m.imdb_id, m.poster_url, m.backdrop_url, m.runtime, m.content, m.genres_json, m.countries_json, m.actors_json, m.directors_json FROM movie_sources ms JOIN movies m ON m.id = ms.movie_id WHERE ms.source = ? AND ms.source_slug = ? """, (source, slug), ).fetchone() if not source_row: conn.close() raise ValueError(f"Slug not found: source={source} slug={slug}") old_movie_id = int(source_row["movie_id"]) old_movie = row_to_movie(source_row) payload = json.loads(source_row["raw_json"] or "{}") repaired = normalize_api_movie(source, slug, payload) target_movie_id = upsert_repaired_movie(conn, repaired) conn.execute("UPDATE movie_sources SET movie_id = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?", (target_movie_id, source_row["source_id"])) old_episode_rows = conn.execute("SELECT * FROM episodes WHERE movie_id = ? AND source = ?", (old_movie_id, source)).fetchall() old_folder = library_root / library_name(old_movie.media_type) / movie_folder_name(old_movie.title, old_movie.year, old_movie.tmdb_id) remove_file(old_folder / "tvshow.nfo", library_root) remove_file(old_folder / "movie.nfo", library_root) for old_episode in old_episode_rows: old_strm = db_to_file_path(old_episode["strm_path"], root, strm_mount, library_root) remove_file(old_strm, library_root) remove_file(old_strm.with_suffix(".nfo") if old_strm else None, library_root) server_ids: dict[str, int] = {} for server in repaired.servers: conn.execute( """ INSERT INTO servers (movie_id, source, name, priority, raw_json) VALUES (?, ?, ?, ?, ?) ON CONFLICT(movie_id, source, name) DO UPDATE SET priority=excluded.priority, raw_json=excluded.raw_json, updated_at=CURRENT_TIMESTAMP """, (target_movie_id, source, server.name, server.priority, json.dumps(server.raw, ensure_ascii=False)), ) server_row = conn.execute( "SELECT id FROM servers WHERE movie_id = ? AND source = ? AND name = ?", (target_movie_id, source, server.name), ).fetchone() server_ids[server.name] = int(server_row["id"]) fixed_episodes = 0 for server in repaired.servers: for episode in server.episodes: existing = conn.execute( """ SELECT * FROM episodes WHERE movie_id = ? AND source = ? AND origin_m3u8_url = ? ORDER BY id LIMIT 1 """, (old_movie_id, source, episode.m3u8_url), ).fetchone() if existing: episode_id = int(existing["id"]) access_key = existing["access_key"] cleaned_path = existing["cleaned_m3u8_path"] else: episode_id = 0 access_key = "" cleaned_path = None if not cleaned_path: cleaned_path = "" result = strm_writer.write(repaired, episode.filename, cleaned_path, access_key, episode.season, episode.episode) conn.execute( """ UPDATE episodes SET movie_id = ?, server_id = ?, source_episode_id = ?, title = ?, filename = ?, season = ?, episode = ?, strm_path = ?, raw_json = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( target_movie_id, server_ids[server.name], episode.source_episode_id, episode.title, episode.filename, episode.season, episode.episode, result.db_path, json.dumps(episode.raw, ensure_ascii=False), episode_id, ), ) nfo_writer.write_for_strm(repaired, result.file_path) fixed_episodes += 1 if old_movie_id != target_movie_id: conn.execute("DELETE FROM servers WHERE movie_id = ? AND source = ?", (old_movie_id, source)) if not conn.execute("SELECT 1 FROM movie_sources WHERE movie_id = ? LIMIT 1", (old_movie_id,)).fetchone(): conn.execute("DELETE FROM movies WHERE id = ?", (old_movie_id,)) current = old_folder while current != library_root and current.exists(): try: current.rmdir() except OSError: break current = current.parent conn.commit() conn.close() return { "old_movie_id": old_movie_id, "old_media_type": old_movie.media_type, "target_movie_id": target_movie_id, "new_media_type": repaired.media_type, "fixed_episodes": fixed_episodes, } def main() -> None: parser = argparse.ArgumentParser(description="Repair one slug from raw DB data.") parser.add_argument("--config", default="app.yaml") parser.add_argument("--source", required=True) parser.add_argument("--slug", required=True) args = parser.parse_args() result = repair_slug(args.config, args.source, args.slug) for key, value in result.items(): print(f"{key}: {value}") if __name__ == "__main__": main()