lucas/Server-scripts
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

match series to tvdb episodes py

Browse Source 
adapt serverlibrary
This commit is contained in:
lucas committed 2026-10-24 15:19:17 +02:00
1 parent 104b912404
commit d241017c2a
3 files changed
+181 -114
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

media_editing/match_series_to_tvdb.py
Regular → Executable
+76 -42
View File
@@ -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()
main()
serverlibrary/__init__.py
+14 -11
View File
@@ -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,
)
serverlibrary/core.py
+91 -61
View File
@@ -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()
return response.json()