from __future__ import annotations import json import logging import re import sqlite3 from pathlib import Path from typing import Any from .config import load_yaml from .http import HttpClient from .models import NormalizedMovie from .nfo_writer import NfoWriter from .strm_writer import StrmWriter from .tmdb import TmdbResolver class SlugRemapper: def __init__(self, app_config: dict[str, Any], http: HttpClient): self.app_config = app_config self.db_path = app_config["database"]["path"] self.http = http self.tmdb = TmdbResolver(app_config.get("tmdb", {}), http) db_mounts = app_config.get("db_mounts", {}) self.strm_writer = StrmWriter( app_config["paths"]["libraries"], app_config["playlist_server"]["public_url"], db_mounts.get("libraries", "/strm"), ) self.nfo_writer = NfoWriter(app_config["paths"]["libraries"], db_mounts.get("libraries", "/strm")) self.log = logging.getLogger(self.__class__.__name__) def run(self, movie_map_path: str, tvshow_map_path: str, only_slug: str | None = None) -> None: maps = self._load_maps(movie_map_path, "movie") + self._load_maps(tvshow_map_path, "tv") with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") sources = conn.execute( """ SELECT ms.id AS source_id, ms.source, ms.source_slug, ms.movie_id, ms.raw_json, 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.movie_id IS NOT NULL ORDER BY ms.source, ms.source_slug """ ).fetchall() for source in sources: if only_slug and source["source_slug"] != only_slug: continue mapping = self._match_mapping(source["source"], source["source_slug"], maps) if not mapping: continue movie = self._movie_from_row(source) self.tmdb.apply_mapping(movie, mapping) target_id = self._upsert_target_movie(conn, movie) self._move_source_to_target( conn, int(source["movie_id"]), target_id, int(source["source_id"]), source["source"], source["raw_json"], mapping, ) canonical = self._movie_from_movie_id(conn, target_id) self._rewrite_strm_files(conn, target_id, canonical, source["source"]) self.log.info( "Remapped slug source=%s slug=%s old_movie_id=%s target_movie_id=%s tmdb_id=%s", source["source"], source["source_slug"], source["movie_id"], target_id, canonical.tmdb_id, ) conn.commit() def _load_maps(self, path: str, media_type: str) -> list[dict[str, Any]]: if not path or not Path(path).exists(): return [] data = load_yaml(path) entries = data.get("maps", data if isinstance(data, list) else []) result = [] for entry in entries or []: item = dict(entry) item.setdefault("media_type", media_type) result.append(item) return result def _match_mapping(self, source: str, slug: str, maps: list[dict[str, Any]]) -> dict[str, Any] | None: for entry in maps: entry_source = str(entry.get("source") or "").lower() if entry_source and entry_source != source.lower(): continue if entry.get("slug") and str(entry["slug"]).lower() == slug.lower(): return entry if entry.get("slug_pattern"): match = re.match(str(entry["slug_pattern"]), slug, flags=re.IGNORECASE) if match: item = dict(entry) season_group = item.get("season_group") if season_group: value = match.group(int(season_group)) if value: item["season"] = int(value) return item return None def _movie_from_row(self, row: sqlite3.Row) -> NormalizedMovie: return NormalizedMovie( source=row["source"], source_slug=row["source_slug"], title=row["title"] or row["source_slug"], 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 _movie_from_movie_id(self, conn: sqlite3.Connection, movie_id: int) -> NormalizedMovie: row = conn.execute( """ SELECT ms.source, ms.source_slug, m.* FROM movies m LEFT JOIN movie_sources ms ON ms.movie_id = m.id WHERE m.id = ? ORDER BY ms.id LIMIT 1 """, (movie_id,), ).fetchone() return self._movie_from_row(row) def _upsert_target_movie(self, 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=COALESCE(NULLIF(movies.countries_json, '[]'), excluded.countries_json), actors_json=COALESCE(NULLIF(movies.actors_json, '[]'), excluded.actors_json), directors_json=COALESCE(NULLIF(movies.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 _move_source_to_target( self, conn: sqlite3.Connection, old_movie_id: int, target_movie_id: int, source_id: int, source: str, raw_json: str, mapping: dict[str, Any], ) -> None: conn.execute("UPDATE movie_sources SET movie_id = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?", (target_movie_id, source_id)) server_rows = conn.execute("SELECT * FROM servers WHERE movie_id = ? AND source = ?", (old_movie_id, source)).fetchall() server_map: dict[int, int] = {} for server in server_rows: existing = conn.execute( "SELECT id FROM servers WHERE movie_id = ? AND source = ? AND name = ?", (target_movie_id, source, server["name"]), ).fetchone() if existing: target_server_id = int(existing["id"]) else: conn.execute( """ INSERT INTO servers (movie_id, source, name, priority, raw_json) VALUES (?, ?, ?, ?, ?) """, (target_movie_id, source, server["name"], server["priority"], server["raw_json"]), ) target_server_id = int(conn.execute("SELECT last_insert_rowid()").fetchone()[0]) server_map[int(server["id"])] = target_server_id season = mapping.get("season") m3u8_urls = self._source_m3u8_urls(raw_json) if m3u8_urls: placeholders = ",".join("?" for _ in m3u8_urls) episode_rows = conn.execute( f""" SELECT * FROM episodes WHERE movie_id = ? AND source = ? AND origin_m3u8_url IN ({placeholders}) """, (old_movie_id, source, *sorted(m3u8_urls)), ).fetchall() else: episode_rows = [] self.log.warning("No m3u8 URLs found for remap source=%s source_id=%s", source, source_id) for episode in episode_rows: target_server_id = server_map.get(int(episode["server_id"] or 0)) self._move_episode_to_target(conn, episode, target_movie_id, target_server_id, int(season) if season is not None else None) if old_movie_id != target_movie_id: remaining = conn.execute("SELECT COUNT(*) FROM movie_sources WHERE movie_id = ?", (old_movie_id,)).fetchone()[0] if remaining == 0: conn.execute("DELETE FROM servers WHERE movie_id = ?", (old_movie_id,)) conn.execute("DELETE FROM movies WHERE id = ?", (old_movie_id,)) def _move_episode_to_target( self, conn: sqlite3.Connection, episode: sqlite3.Row, target_movie_id: int, target_server_id: int | None, season: int | None, ) -> None: final_season = season if season is not None else int(episode["season"]) existing = conn.execute( """ SELECT * FROM episodes WHERE id != ? AND movie_id = ? AND source = ? AND filename = ? AND origin_m3u8_url = ? LIMIT 1 """, ( episode["id"], target_movie_id, episode["source"], episode["filename"], episode["origin_m3u8_url"], ), ).fetchone() if not existing: conn.execute( """ UPDATE episodes SET movie_id = ?, server_id = ?, season = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, (target_movie_id, target_server_id, final_season, episode["id"]), ) return crawl_status = self._merged_status(existing["crawl_status"], episode["crawl_status"]) last_error = None if crawl_status == "success" else (existing["last_error"] or episode["last_error"]) raw_json = existing["raw_json"] if existing["raw_json"] and existing["raw_json"] != "{}" else episode["raw_json"] conn.execute( """ UPDATE episodes SET server_id = COALESCE(?, server_id), source_episode_id = ?, title = ?, season = ?, episode = ?, cleaned_m3u8_path = ?, strm_path = ?, crawl_status = ?, retry_count = MAX(retry_count, ?), last_error = ?, raw_json = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( target_server_id, existing["source_episode_id"] or episode["source_episode_id"], existing["title"] or episode["title"], final_season, int(existing["episode"] or episode["episode"]), existing["cleaned_m3u8_path"] or episode["cleaned_m3u8_path"], existing["strm_path"] or episode["strm_path"], crawl_status, int(episode["retry_count"] or 0), last_error, raw_json, existing["id"], ), ) conn.execute("DELETE FROM episodes WHERE id = ?", (episode["id"],)) @staticmethod def _merged_status(left: str | None, right: str | None) -> str: statuses = {left or "", right or ""} if "success" in statuses: return "success" if "pending" in statuses: return "pending" if "failed" in statuses: return "failed" return left or right or "pending" def _source_m3u8_urls(self, raw_json: str) -> set[str]: try: raw = json.loads(raw_json or "{}") except json.JSONDecodeError: return set() urls: set[str] = set() def walk(value: Any) -> None: if isinstance(value, dict): link = value.get("link_m3u8") if isinstance(link, str) and link.startswith(("http://", "https://")): urls.add(link.strip()) for child in value.values(): walk(child) elif isinstance(value, list): for child in value: walk(child) walk(raw) return urls def _rewrite_strm_files(self, conn: sqlite3.Connection, movie_id: int, movie: NormalizedMovie, source: str) -> None: rows = conn.execute( """ SELECT id, filename, cleaned_m3u8_path, access_key, season, episode FROM episodes WHERE movie_id = ? AND source = ? AND crawl_status = 'success' AND cleaned_m3u8_path IS NOT NULL ORDER BY season, episode, filename """, (movie_id, source), ).fetchall() wrote_tv_nfo = False for row in rows: result = self.strm_writer.write( movie, row["filename"], row["cleaned_m3u8_path"], row["access_key"], int(row["season"]), int(row["episode"]), ) conn.execute("UPDATE episodes SET strm_path = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?", (result.db_path, row["id"])) if movie.media_type == "movie": self.nfo_writer.write_for_strm(movie, result.file_path) else: wrote_tv_nfo = True if wrote_tv_nfo: self.nfo_writer.write(movie)