diff --git a/media_editing/match_series_to_tvdb.py b/media_editing/match_series_to_tvdb.py old mode 100644 new mode 100755 index afb007b..8c924da --- a/media_editing/match_series_to_tvdb.py +++ b/media_editing/match_series_to_tvdb.py @@ -1,25 +1,33 @@ #!/usr/bin/env python3 -import re -import json import argparse +import json import logging +import re +import sys from pathlib import Path -from typing import Any weight_jaccard = 0.5 weight_fuzzy = 1 - weight_jaccard -from serverlibrary import (words, word_similarity, score_match_jaccard, score_match_fuzzy, - combined_score, tmdb_get, tvdb_get) +from serverlibrary import ( + combined_score, + score_match_fuzzy, + score_match_jaccard, + score_match_recall_jaccard, + tvdb_get, + words, +) MATCH_THRESHOLD = 0.85 + class CustomLogger(logging.Logger): def trace(self, msg: str, *args, **kwargs) -> None: if self.isEnabledFor(TRACE): self._log(TRACE, msg, args, **kwargs) + TRACE = 5 logging.setLoggerClass(CustomLogger) @@ -34,12 +42,14 @@ def normalize(text: str) -> str: r"\b(1080p|720p|2160p|x264|x265|h264|h265|bluray|web-dl|webrip|dvdrip)\b", "", text, - flags=re.I, + flags=re.IGNORECASE, ) text = re.sub(r"[._-]", " ", text) text = re.sub(r"ß", "ss", text) text = re.sub(r"\s+", " ", text) + text = re.sub(r"\((\d+)\s+\d+\)", r"(\1)", text) + text = re.sub(r"\((\d+)_+\d+\)", r"(\1)", text) return text.strip() @@ -70,6 +80,7 @@ def get_all_episodes_by_tvdbid(tvdb_id: int): return episodes + def parse_series_name(series_name: str): """ Universe (1987) [66757] @@ -97,7 +108,8 @@ def parse_series_name(series_name: str): return title, year, tmdb_id -def get_tvdbid_by_name(series_name: str) -> tuple[int, Any]: + +def get_tvdbid_by_name(series_name: str): title, year, tmdb_id = parse_series_name(series_name) params = { @@ -133,20 +145,19 @@ def get_tvdbid_by_name(series_name: str) -> tuple[int, Any]: if filtered: results = filtered - return int(results[0]["tvdb_id"]), results[0] + return results[0] + def best_episode_match(filename, episodes): - search_words = words(filename) - + search_words = words(normalize(filename)) + logger.debug(f"Search Words: {search_words}") best_score = 0.0 best_episode = None for episode in episodes: - logger.trace(f" Title: {episode["title"]}") if not episode["title"] or episode["title"] == "Unkown": continue - candidate_words = words(episode["title"]) - + candidate_words = words(normalize(episode["title"])) fuzzy = score_match_fuzzy( search_words, candidate_words, @@ -155,6 +166,14 @@ def best_episode_match(filename, episodes): search_words, candidate_words, ) + recall_jaccard1 = score_match_recall_jaccard( + search_words, + candidate_words, + ) + recall_jaccard2 = score_match_recall_jaccard( + candidate_words, + search_words, + ) score = combined_score( fuzzy, @@ -162,8 +181,15 @@ def best_episode_match(filename, episodes): weight_fuzzy=weight_fuzzy, weight_jaccard=weight_jaccard, ) + score = max(score, recall_jaccard1, recall_jaccard2) + + if "hyänen" in search_words and score > 0.6: + logger.warning( + f"SEARCH WORDS: {search_words}, CANDIDATE WORDS: {candidate_words}" + ) if score > best_score: + logger.trace(f"{search_words}:\t\t{candidate_words},\t\t{score}") best_score = score best_episode = episode @@ -173,45 +199,55 @@ def best_episode_match(filename, episodes): def process_series(series_path: Path, dry_run: bool = False): series_name = series_path.name logger.debug(f"Pricess series with name : {series_name}") - #tvdb_id, media_data = get_tvdbid_by_name(series_name) - # episodes = get_all_episodes_by_tvdbid(tvdb_id) - #with open("media_data.json", "w", encoding="utf-8") as f: - # json.dump(media_data, f, ensure_ascii=False, indent=2) - - with open("media_data.json", encoding="utf-8") as f: - media_data = json.load(f) - with open("episodes.json", encoding="utf-8") as f: - episodes = json.load(f) - - - logger.info(f"Found series name {media_data["name"]}") + if Path(f"{series_name}_episodes.json").is_file(): + logger.info("load json files") + with open(f"{series_name}_media_data.json", encoding="utf-8") as f: + media_data = json.load(f) + with open(f"{series_name}_episodes.json", encoding="utf-8") as f: + episodes = json.load(f) + tvdb_id = int(media_data["tvdb_id"]) + for ep in episodes: + logger.debug(f"{ep.get('title')}") + else: + logger.info("QUery from TVDB") + media_data = get_tvdbid_by_name(series_name) + tvdb_id = int(media_data["tvdb_id"]) + episodes = get_all_episodes_by_tvdbid(tvdb_id) + logger.info("Dump to json") + with open(f"{series_name}_media_data.json", "w", encoding="utf-8") as f: + json.dump(media_data, f, ensure_ascii=False, indent=2) + with open(f"{series_name}_episodes.json", "w", encoding="utf-8") as f: + json.dump(episodes, f, ensure_ascii=False, indent=2) + logger.info(f"Found series name {media_data['name']}") extra_episodes_path = series_path / "Extra Episodes" for file in extra_episodes_path.iterdir(): if not file.is_file(): continue - filename = normalize(re.sub( media_data["name"],"", file.stem)) + filename = re.sub(media_data["name"], "", file.stem) + episode, score = best_episode_match( filename, episodes, ) - if episode is None: continue if score < MATCH_THRESHOLD: - logger.warning(f"LOW CONFIDENCE: {filename} {score:.2f}: {episode.get("title")}") + logger.warning( + f"LOW CONFIDENCE: {filename} {score:.2f}: {normalize(episode.get('title'))}" + ) continue safe_title = re.sub( r'[<>:"/\\|?*]', - '', + "", episode["title"], ) new_name = ( - f"{media_data.get("name")}" + f"{media_data.get('name')}" f" - S{episode['season']:02d}" f"E{episode['episode']:02d}" f" - {safe_title}" @@ -219,12 +255,10 @@ def process_series(series_path: Path, dry_run: bool = False): ) target = file.with_name(new_name) - - logger.info( - f"{file.name}" - f" -> {new_name}, " - f" score={score:.3f}" - ) + if score < 1: + logger.debug(f"{file.name} -> {new_name}, score={score:.3f}") + elif score == 1: + logger.trace(f"{file.name} -> {new_name}, score={score:.3f}") if not dry_run: file.rename(target) @@ -234,7 +268,7 @@ def main(): parser = argparse.ArgumentParser() parser.add_argument( "path", - help='Either a series folder "Maya the Bee (1975) [10620]" or a root folder containing such folders' + help='Either a series folder "Maya the Bee (1975) [10620]" or a root folder containing such folders', ) parser.add_argument( @@ -268,7 +302,7 @@ def main(): format="%(asctime)s %(levelname)s %(message)s", ) - input_path = Path(args.path) + input_path = Path(args.path).expanduser().resolve() if not input_path.is_dir(): parser.error(f"Directory not found: {input_path}") @@ -278,14 +312,14 @@ def main(): # Input is already a series folder if re.match(r"^(.*?)\s*\((\d{4})\)\s*\[(\d+)]$", input_path.name): logger.info(f"Process series: {input_path.name}") - process_series(input_path,dry_run) + process_series(input_path, dry_run) - # Input is a root folder containing series folders + # Input is a root folder containing series folders else: for series_path in sorted(input_path.iterdir()): if series_path.is_dir(): - process_series(series_path,dry_run) + process_series(series_path, dry_run) if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/serverlibrary/__init__.py b/serverlibrary/__init__.py index ca3807f..7f32da0 100755 --- a/serverlibrary/__init__.py +++ b/serverlibrary/__init__.py @@ -1,11 +1,14 @@ -from .core import safe_rename -from .core import safe_delete -from .core import get_audio_tracks -from .core import get_subtitle_tracks -from .core import words -from .core import word_similarity -from .core import score_match_jaccard -from .core import score_match_fuzzy -from .core import combined_score -from .core import tmdb_get -from .core import tvdb_get +from .core import ( + combined_score, + get_audio_tracks, + get_subtitle_tracks, + safe_delete, + safe_rename, + score_match_fuzzy, + score_match_jaccard, + score_match_recall_jaccard, + tmdb_get, + tvdb_get, + word_similarity, + words, +) diff --git a/serverlibrary/core.py b/serverlibrary/core.py index 98d8d5a..912cf8a 100755 --- a/serverlibrary/core.py +++ b/serverlibrary/core.py @@ -1,11 +1,11 @@ - -from pathlib import Path -import shutil -import subprocess import json import re -import requests +import shutil +import subprocess from difflib import SequenceMatcher +from pathlib import Path + +import requests # ----------------------------- @@ -17,19 +17,26 @@ def safe_delete(path: Path): else: path.unlink() -def safe_rename(src: Path, dst: Path, dry: bool=False, confirm: bool=False, - silent=False,overwrite=False) -> (str, bool): + +def safe_rename( + src: Path, + dst: Path, + dry: bool = False, + confirm: bool = False, + silent=False, + overwrite=False, +) -> tuple[str, bool]: + """ + Safely rename a file or directory without overwriting existing files. + Automatically appends _1, _2, ... if needed. """ - Safely rename a file or directory without overwriting existing files. - Automatically appends _1, _2, ... if needed. - """ src = Path(src) dst = Path(dst) # nothing to do if src.resolve() == dst.resolve(): - return + return "", False directory = dst.parent base = dst.stem @@ -42,7 +49,7 @@ def safe_rename(src: Path, dst: Path, dry: bool=False, confirm: bool=False, while final_dst.exists(): final_dst = directory / f"{base}_{counter}{ext}" counter += 1 - ret=f"{src} → {dst}" + ret = f"{src} → {dst}" if not silent: print(ret) @@ -56,11 +63,14 @@ def safe_rename(src: Path, dst: Path, dry: bool=False, confirm: bool=False, if dst.exists() and not overwrite: print("File already exists...") - return "",False + return "", False src.rename(dst) return ret, True -def get_audio_tracks( source: str, title: int | None = None, handbrake: bool = False) -> list[dict]: + +def get_audio_tracks( + source: str, title: int | None = None, handbrake: bool = False +) -> list[dict]: """ Return audio track information from a media file or DVD title. @@ -103,8 +113,10 @@ def get_audio_tracks( source: str, title: int | None = None, handbrake: bool = F result = subprocess.run( [ "HandBrakeCLI", - "-i", source, - "-t", str(title), + "-i", + source, + "-t", + str(title), "--scan", ], stdout=subprocess.PIPE, @@ -113,22 +125,24 @@ def get_audio_tracks( source: str, title: int | None = None, handbrake: bool = F check=True, ) - pattern = re.compile( - r"\+\s+(\d+),\s+(.+?)\s+\((?:AC3|AAC|MP3|E-AC-3|DTS)" - ) + pattern = re.compile(r"\+\s+(\d+),\s+(.+?)\s+\((?:AC3|AAC|MP3|E-AC-3|DTS)") for match in pattern.finditer(result.stdout): - tracks.append({ - "number": int(match.group(1)), - "description": match.group(2).strip(), - }) + tracks.append( + { + "number": int(match.group(1)), + "description": match.group(2).strip(), + } + ) else: result = subprocess.run( [ "ffprobe", - "-v", "quiet", - "-print_format", "json", + "-v", + "quiet", + "-print_format", + "json", "-show_streams", source, ], @@ -143,31 +157,33 @@ def get_audio_tracks( source: str, title: int | None = None, handbrake: bool = F if stream.get("codec_type") != "audio": continue - tracks.append({ - "number": stream["index"], - "description": ( - stream.get("tags", {}).get("title") - or stream.get("tags", {}).get("language") - or f"Track {stream['index']}" - ), - "primary": stream.get("codec_type") == "audio" and stream.get("disposition", {}).get("default") == 1 - }) + tracks.append( + { + "number": stream["index"], + "description": ( + stream.get("tags", {}).get("title") + or stream.get("tags", {}).get("language") + or f"Track {stream['index']}" + ), + "primary": stream.get("codec_type") == "audio" + and stream.get("disposition", {}).get("default") == 1, + } + ) if not tracks: - tracks.append({ - "number": 0, - "description": "", - "primary": False - }) + tracks.append({"number": 0, "description": "", "primary": False}) return tracks + def get_subtitle_tracks(source): result = subprocess.run( [ "ffprobe", - "-v", "quiet", - "-print_format", "json", + "-v", + "quiet", + "-print_format", + "json", "-show_streams", source, ], @@ -184,36 +200,48 @@ def get_subtitle_tracks(source): if stream.get("codec_type") != "subtitle": continue - tracks.append({ - "number": stream["index"], - "description": ( - stream.get("tags", {}).get("title") - or stream.get("tags", {}).get("language") - or f"Subtitle {stream['index']}" - ), - "primary": ( - stream.get("disposition", {}).get("default") == 1 - ), - }) + tracks.append( + { + "number": stream["index"], + "description": ( + stream.get("tags", {}).get("title") + or stream.get("tags", {}).get("language") + or f"Subtitle {stream['index']}" + ), + "primary": (stream.get("disposition", {}).get("default") == 1), + } + ) if not tracks: - tracks.append({ - "number": 0, - "description": "", - "primary": False, - }) + tracks.append( + { + "number": 0, + "description": "", + "primary": False, + } + ) return tracks def words(text): - return set(filter(None, re.split(r"[^a-z0-9]+", text.lower()))) + return set(re.findall(r"[^\W_]+", text.lower())) def word_similarity(a: str, b: str) -> float: return SequenceMatcher(None, a, b).ratio() +def score_match_recall_jaccard( + search_words: set[str], candidate_words: set[str] +) -> float: + if not search_words: + return 0.0 + + intersection = search_words & candidate_words + return len(intersection) / len(search_words) + + def score_match_jaccard(search_words, candidate_words): intersection = search_words & candidate_words union = search_words | candidate_words @@ -239,8 +267,9 @@ def score_match_fuzzy(search_words, candidate_words): return total / len(search_words) -def combined_score(score_fuzzy: float, score_jaccard: float, - weight_fuzzy: float, weight_jaccard: float) -> float: +def combined_score( + score_fuzzy: float, score_jaccard: float, weight_fuzzy: float, weight_jaccard: float +) -> float: return score_fuzzy * weight_fuzzy + score_jaccard * weight_jaccard @@ -278,6 +307,7 @@ def tmdb_get(endpoint, params=None): return response.json() + def tvdb_get(endpoint, params=None): """Perform a GET request against TVDB.""" BASE_URL = "https://api4.thetvdb.com/v4" @@ -293,4 +323,4 @@ def tvdb_get(endpoint, params=None): response.raise_for_status() - return response.json() \ No newline at end of file + return response.json()