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

feat(sonarr_database_tool): working translation and extra_episodes functions

Browse Source 
get last update time of tvdb/tmdb
get_tvdb_by_tmdb_id
add the same dict keys for both databases
lookup episodes in german for foreign language media
This commit is contained in:
lucas authored and Lucas Vivo committed 2026-10-29 09:02:30 +01:00
1 parent 73b870b142
commit 2d97877d00
1 file changed
+382 -100
Download Patch File Download Diff File Expand all files Collapse all files
media_editing/sonarr_database_tool.py
+382 -100
View File
@@ -1,23 +1,39 @@
#!/usr/bin/env python3
#!/home/lucas/scripts/.venv/bin/python3
import argparse
import json
import logging
import re
from datetime import datetime, timedelta
import sys
from pathlib import Path
import filecmp
import requests
weight_jaccard = 0.5
weight_fuzzy = 1 - weight_jaccard
MATCH_THRESHOLD = 0.85
from serverlibrary import *
from serverlibrary import (
VIDEO_SUFFIXES,
CustomLogger,
combined_score,
safe_rename,
score_match_fuzzy,
score_match_jaccard,
score_match_recall_jaccard,
setup_logging,
tmdb_get,
tvdb_get,
words,
)
logging.setLoggerClass(CustomLogger)
logger: CustomLogger = logging.getLogger(__name__)
def normalize(text: str, title: str | None = None, year: int | str | None = None) -> str:
def normalize(
text: str, title: str | None = None, year: int | str | None = None
) -> str:
text = text.lower()
# remove series/movie title
if title:
@@ -53,7 +69,15 @@ def normalize(text: str, title: str | None = None, year: int | str | None = None
return text
def parse_series_name(series_name: str):
def files_equal(file1: Path, file2: Path) -> bool:
if filecmp.cmp(file1, file2, shallow=False):
return True
logger.debug(f"Files {file1.name} and {file2.name} are not equal")
return False
def parse_series_name(series_name: str) -> tuple[str, str, str]:
"""
Universe (1987) [66757]
-> ("Universe", 1987, 66757)
@@ -63,16 +87,16 @@ def parse_series_name(series_name: str):
-> ("Universe", None, None)
"""
tmdb_id = None
year = None
tmdb_id = ""
year = ""
id_match = re.search(r"\[(\d+)]", series_name)
if id_match:
tmdb_id = int(id_match.group(1))
tmdb_id = id_match.group(1)
year_match = re.search(r"\((\d{4})\)", series_name)
if year_match:
year = int(year_match.group(1))
year = year_match.group(1)
title = re.sub(r"\[\d+]", "", series_name)
title = re.sub(r"\(\d{4}\)", "", title)
@@ -80,7 +104,8 @@ def parse_series_name(series_name: str):
return title, year, tmdb_id
def find_by_tvdb_id(tvdb_id:str, language="deu"):
def find_by_tvdb_id(tvdb_id: str, language="deu"):
"""
Try to determine whether a TMDB ID is a TV show or movie.
@@ -93,9 +118,28 @@ def find_by_tvdb_id(tvdb_id:str, language="deu"):
return tvdb_get(
f"/series/{tvdb_id}/extended",
{"language": language},
)["data"]
)["data"]
def find_by_tmdb_id(tmdb_id:str, language="de-DE", tmdb_type="tv"):
def get_tmdb_last_updated(tmdb_id: str) -> str:
latest = None
data = tmdb_get(f"/tv/{tmdb_id}/changes")
for change in data["changes"]:
for item in change["items"]:
ts = datetime.strptime(
item["time"],
"%Y-%m-%d %H:%M:%S UTC",
)
if latest is None or ts > latest:
latest = ts
if latest is None:
logger.debug("Use default lastChange of 1 week")
latest = datetime.now() - timedelta(days=7)
return latest.strftime("%Y-%m-%d %H:%M:%S")
def find_by_tmdb_id(tmdb_id: str, language="de-DE", type="tv"):
"""
Try to determine whether a TMDB ID is a TV show or movie.
@@ -104,14 +148,25 @@ def find_by_tmdb_id(tmdb_id:str, language="de-DE", tmdb_type="tv"):
("movie", data)
(None, None)
"""
return tmdb_get(
logger.trace(f"Find entry to id {tmdb_id}")
result = tmdb_get(
f"/{type}/{tmdb_id}",
{
"language": language,
},
)
result["type"] = type
result["year"] = (
f"{(result.get('first_air_date') or result.get('release_date') or '')[:4]}"
)
result["tmdb_id"] = result["id"]
result["tvdb_id"] = None
result["origin_country"] = result["origin_country"][0]
result["name"] = result.get("name") or result.get("title")
return result
def get_all_episodes_by_tvdbid(tvdb_id: str, language: str = "deu"):
series_data = find_by_tvdb_id(tvdb_id)
@@ -123,7 +178,7 @@ def get_all_episodes_by_tvdbid(tvdb_id: str, language: str = "deu"):
try:
season_data = tvdb_get(
f"/seasons/{season['id']}/extended",
{"language":language},
{"language": language},
)["data"]
except requests.exceptions.RequestException:
logger.error(
@@ -131,17 +186,30 @@ def get_all_episodes_by_tvdbid(tvdb_id: str, language: str = "deu"):
season["id"],
)
for episode in season_data.get("episodes", []):
if language in episode["nameTranslations"] and (
not episode["name"] or series_data["originalLanguage"] != language
):
episode["name"] = tvdb_get(
f"/episodes/{episode['id']}/translations/{language}"
)["data"]["name"]
episodes_data.append(
{
"id": episode["id"],
"season": episode["seasonNumber"],
"episode": episode["number"],
"title": episode["name"],
"name": episode["name"],
}
)
series_data["number_of_seasons"] = len(series_data.get("seasons"))
series_data["number_of_episodes"] = len(episodes_data)
network = series_data.get("originalNetwork") or series_data.get("latestNetwork")
if isinstance(network, dict):
network = network.get("name")
series_data["network"] = network
return series_data, episodes_data
def get_all_episodes_by_tmdbid(tmdb_id, language: str = "de-DE"):
def get_season(tmdb_id, season_number):
return tmdb_get(
@@ -150,6 +218,7 @@ def get_all_episodes_by_tmdbid(tmdb_id, language: str = "de-DE"):
"language": language,
},
)
series_data = find_by_tmdb_id(tmdb_id)
episodes_data = []
@@ -163,15 +232,74 @@ def get_all_episodes_by_tmdbid(tmdb_id, language: str = "de-DE"):
"id": episode["id"],
"season": season_number,
"episode": episode["episode_number"],
"title": episode["name"],
"name": episode["name"],
}
)
network = series_data["networks"]
if isinstance(network, list):
network = network[0].get("name")
series_data["network"] = network
return series_data, episodes_data
def get_tvdbid_by_name(series_name: str,language:str="deu"):
title, year, tmdb_id = parse_series_name(series_name)
def get_tvdbid_by_tmdb(series_path: Path, language: str = "deu"):
title, year, tmdb_id = parse_series_name(series_path.name)
if tmdb_id:
results = tvdb_get(
f"/search/remoteid/{tmdb_id}",
{"language": language},
)["data"]
else:
return get_tvdbid_by_name(series_path)
result = None
for res in results:
if "series" not in res:
continue
if normalize(title) in res["series"].get("name", "").casefold() and str(
res["series"].get("year", "")
) == str(year):
result = res["series"]
logger.info(result)
if not result:
logger.error(f"Results: {results}")
raise Exception(f"Couldnt find results for Series {title} ")
result["tvdb_id"] = result["id"]
result["type"] = "tv"
if result.get("translations"):
result["original_name"] = result.get("translations").get(
result.get("primary_language")
)
result["tmdb_id"] = tmdb_id
result["origin_country"] = result.get("originalCountry")
status = result.get("status")
result["name"] = result.get("name") or result.get("title")
if isinstance(status, dict):
result["status"] = status.get("name")
return result
def databes_entry_is_newer(last_updated: str, file: Path) -> bool:
tvdb_time = datetime.strptime(
last_updated,
"%Y-%m-%d %H:%M:%S",
)
file_time = datetime.fromtimestamp(file.stat().st_mtime)
if tvdb_time > file_time:
logger.debug(
f"Database entry is newer than {file.name}: {file_time} > {tvdb_time}"
)
return tvdb_time > file_time
def get_tvdbid_by_name(series_path: Path, language: str = "deu"):
title, year, tmdb_id = parse_series_name(series_path.name)
logger.trace(f"Find series {title} by tvdb name search")
params = {
"query": title,
"type": "series",
@@ -187,50 +315,120 @@ def get_tvdbid_by_name(series_name: str,language:str="deu"):
)["data"]
if not results:
raise RuntimeError(f"Series not found: {series_name}")
raise RuntimeError(f"Series not found: {series_path}")
# Filter by year if available
if tmdb_id is not None:
filtered = []
filtered = []
for result in results:
result_tmdb_id = result.get("remote_ids")
for id in result_tmdb_id:
if id.get("sourceName") == "TheMovieDB.com":
result_tmdb_id = id.get("id")
if result_tmdb_id and int(result_tmdb_id) == tmdb_id:
for result in results:
result_tmdb_id = result.get("remote_ids")
result["tmdb_id"] = None
if not result_tmdb_id:
continue
for id in result_tmdb_id:
if id.get("sourceName") == "TheMovieDB.com":
result["tmdb_id"] = id.get("id")
if tmdb_id and result["tmdb_id"] and result["tmdb_id"] == tmdb_id:
filtered.append(result)
if filtered:
results = filtered
results[0]["id"] = results[0]["tvdb_id"]
return results[0]
def get_tmdbid_by_name(series_name, media_type="tv", language="de-DE"):
result = results[0]
result["lastUpdated"] = tvdb_get(f"/series/{result['tvdb_id']}")["data"][
"lastUpdated"
]
result["id"] = result["tvdb_id"]
result["type"] = "tv"
if result.get("translations"):
result["original_name"] = result.get("translations").get(
result.get("primary_language")
)
result["origin_country"] = ""
if result.get("country"):
result["origin_country"] = result.get("country")[0]
status = result.get("status")
result["name"] = result.get("name") or result.get("title")
if isinstance(status, dict):
result["status"] = status.get("name")
if result["primary_language"] != language:
result["name"] = result["translations"][language]
result["overview"] = result["overviews"][language]
return result
def get_tmdbid_by_name(series_path: Path, type="tv", language="de-DE"):
"""
Search TMDB by title.
media_type:
tv | movie
"""
title, year, tmdb_id = parse_series_name(series_name)
title, year, _ = parse_series_name(series_path.name)
logger.trace(f"Find series {title} by tmdb name search")
params = {
"query": title,
"language": language,
"language": "en-EN",
}
if year:
if media_type == "tv":
if type == "tv":
params["first_air_date_year"] = year
else:
params["year"] = year
return tmdb_get(
f"/search/{media_type}",
results = tmdb_get(
f"/search/{type}",
params,
)
if not results:
raise RuntimeError(f"Series not found: {series_path}")
result = results["results"][0]
result["type"] = type
result["year"] = (
f"{(result.get('first_air_date') or result.get('release_date') or '')[:4]}"
)
result["tvdb_id"] = None
result["origin_country"] = result["origin_country"][0]
result["name"] = result.get("name") or result.get("title")
return result
def print_results(
media_data: dict, series_data: dict, episodes: list, extended: bool = False
):
logger.info(f"TVDB ID : {media_data.get('tvdb_id')}")
logger.info(f"TMDB ID : {media_data.get('tmdb_id')}")
logger.info(f"Title {media_data.get('name')}")
logger.info(
f"Original : {media_data.get('original_name') or media_data.get('original_title')}"
)
logger.info(f"Country : {media_data['origin_country']}")
logger.info(f"Network : {series_data['network']}")
if extended:
logger.info(f"Year : {media_data['year']}")
logger.info(f"Rating : {media_data.get('vote_average')}")
logger.info(f"Votes : {media_data.get('vote_count')}")
logger.info(f"Status : {media_data.get('status')}")
logger.info(f"Overview : {media_data.get('overview')}")
genres = ", ".join(g["name"] for g in series_data.get("genres", []))
logger.info(f"Genres : {genres}")
logger.info(
f"Poster : https://image.tmdb.org/t/p/original{media_data.get('poster_path', '')}"
)
logger.info(
f"Backdrop : https://image.tmdb.org/t/p/original{media_data.get('backdrop_path', '')}"
)
if media_data.get("type") == "tv":
logger.info(f"Seasons : {series_data.get('number_of_seasons')}")
logger.info(f"Episodes : {series_data.get('number_of_episodes')}")
def best_episode_title_match(filename, episodes):
search_words = words(normalize(filename))
@@ -239,9 +437,9 @@ def best_episode_title_match(filename, episodes):
best_episode = None
for episode in episodes:
if not episode["title"] or episode["title"] == "Unkown":
if not episode["name"] or episode["name"] == "Unkown":
continue
candidate_words = words(normalize(episode["title"]))
candidate_words = words(normalize(episode["name"]))
fuzzy = score_match_fuzzy(
search_words,
candidate_words,
@@ -279,15 +477,22 @@ def best_episode_title_match(filename, episodes):
return best_episode, best_score
def translate_series_folder(episodes, series_path, file_title: str, series_title: str,args):
def translate_series_folder(episodes, series_path, media_data: dict, args):
series = series_path.name
logger.info(f"Execute translate_series for {media_data['name']}")
episode_map = {(ep["season"], ep["episode"]): ep for ep in episodes}
file_list = []
new_file_list = []
for file in series_path.rglob("*"):
if file.is_dir():
if (
not file.is_file()
or file.suffix.lower() not in VIDEO_SUFFIXES
or "Season" not in file.parent.name
):
continue
match_SE_name = re.search(r"S(\d{2})E(\d{2})", file.name, re.I)
logger.trace(f"Translate file: {file.name}")
match_SE_name = re.search(r"S(\d{2}|\d{4})E(\d{2,3})", file.name, re.IGNORECASE)
if not match_SE_name:
logger.error(f"No season/episode number found: {file.name}")
continue
@@ -300,35 +505,40 @@ def translate_series_folder(episodes, series_path, file_title: str, series_title
logger.warning(f"Episode not found: S{season:02d}E{episode:02d}")
continue
episode_season_number_match = re.match(
r"^(.*?S\d{2}E\d{2})\s*-\s*(.*?)\s*(\[[^\]]+\].*)?$",
r"^(.*?S(\d{2}|\d{4})E\d{2,3})\s*-\s*.*?\s*((?:\[[^\]]+\])+)?$",
file.stem,
re.I,
re.IGNORECASE,
)
if not episode_season_number_match:
logger.warning(f"S{season:02d}E{episode:02d} not found")
continue
prefix = episode_season_number_match.group(1).replace(
file_title, series_title
) # Andor (2022) - S01E07
prefix = f"{media_data['name']} ({media_data['year']}) - S{season:02d}E{episode:02d}" # Andor (2022) - S01E07
quality = (
episode_season_number_match.group(3) or ""
) # [WEBRip-1080p][AAC 5.1][h265]
title = episode_number["name"].replace("/", "_")
if title == "_":
title = "slash"
new_name = (
f"{prefix} - {episode_number['title']}"
f"{(' ' + quality) if quality else ''}"
f"{file.suffix}"
f"{prefix} - {title}{(' ' + quality) if quality else ''}{file.suffix}"
)
logger.trace(f"Prefix: {prefix}, Title: {title}, Quality: {quality}")
logger.trace(f"{file.name} -> {new_name}")
file_list.append(file)
new_file_list.append(new_name)
logger.trace(f"{series} -> {series_title}")
new_folder_name = (
f"{media_data['name']} ({media_data['year']}) [{media_data['tmdb_id']}]"
)
if args.translate_series_folder:
logger.trace(f"{series} -> {new_folder_name}")
file_list.append(series_path)
new_file_list.append(new_folder_name)
else:
logger.info(f"Dont translate Series folder: {series}")
file_list.append(series_path)
new_file_list.append(series_title)
has_changes = False
for file, new_file in zip(file_list, new_file_list):
if file.name != new_file:
@@ -338,21 +548,23 @@ def translate_series_folder(episodes, series_path, file_title: str, series_title
logger.info("No changes necessary.")
return
if args.dry_run:
return
for file, new_file in zip(file_list, new_file_list):
safe_rename(file, new_file)
if args.dry_run:
logger.debug(f"{file.name} -> {new_file}")
continue
safe_rename(file, new_file, logger=logger)
logger.info("All files renamed")
return
def reference_extra_episodes(episodes, series_path: Path, series_title: str, args):
def reference_extra_episodes(episodes, series_path: Path, media_data: dict, args):
logger.info(f"Execute reference_extra_episodes for {media_data['name']}")
extra_episodes_path = series_path / "Extra Episodes"
for file in extra_episodes_path.iterdir():
if not file.is_file():
for file in extra_episodes_path.glob("*"):
if not file.is_file() or file.suffix.lower() not in VIDEO_SUFFIXES:
continue
filename = re.sub(series_title, "", file.stem)
logger.trace(f"Reference file: {file.name}")
filename = re.sub(media_data["name"], "", file.stem)
episode, score = best_episode_title_match(
filename,
episodes,
@@ -362,80 +574,132 @@ def reference_extra_episodes(episodes, series_path: Path, series_title: str, arg
if score < MATCH_THRESHOLD:
logger.warning(
f"LOW CONFIDENCE: {filename} {score:.2f}: {normalize(episode.get('title'))}"
f"LOW CONFIDENCE: {filename} {score:.2f}: {normalize(episode.get('name'))}"
)
continue
safe_title = re.sub(
r'[<>:"/\\|?*]',
"",
episode["title"],
episode["name"],
)
new_name = (
f"{series_title}"
f"{media_data['name']}"
f" - S{episode['season']:02d}"
f"E{episode['episode']:02d}"
f" - {safe_title}"
f"{file.suffix}"
)
target = file.parent / f"Season {episode['season']:02d}" / new_name
target = file.parent.parent / f"Season {episode['season']:02d}" / new_name
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 args.dry_run:
logger.info(f"{file.name} -> {target}, score={score:.3f}")
safe_rename(file, target)
def get_series_data(series_path: Path, database:str="tvdb"):
series_name, year, tmdb_id = parse_series_name(series_path.name)
if args.dry_run:
logger.debug(f"{file.name} -> {target}")
continue
# logger.info(f"{file.name} -> {target}, score={score:.3f}")
safe_rename(file, target, logger=logger)
if Path(series_path.parent / f"{series_name}_episodes.json").is_file()\
and Path(series_path.parent / f"{series_name}_{database}_media_data.json").is_file()\
and Path(series_path.parent / f"{series_name}_{database}_series_data.json").is_file():
with open(series_path.parent / f"{series_name}_{database}_episodes.json", encoding="utf-8") as f:
def get_series_data(series_path: Path, database: str = "tvdb"):
series_name, _, tmdb_id = parse_series_name(series_path.name)
if database == "tvdb":
# if tmdb_id:
# media_data = get_tvdbid_by_tmdb(series_path)
# else:
media_data = get_tvdbid_by_name(series_path)
elif database == "tmdb":
if tmdb_id:
media_data = find_by_tmdb_id(tmdb_id)
else:
media_data = get_tmdbid_by_name(series_path)
media_data["lastUpdated"] = get_tmdb_last_updated(tmdb_id)
else:
raise ValueError(f"Wrong database name: {database}")
buffer_file = series_path / f"__{series_name}_{database}_media_data.json"
with open(
buffer_file,
"w",
encoding="utf-8",
) as f:
json.dump(media_data, f, ensure_ascii=False, indent=2)
if (
(series_path / f"{series_name}_{database}_media_data.json").is_file()
and files_equal(
series_path / f"__{series_name}_{database}_media_data.json",
series_path / f"{series_name}_{database}_media_data.json",
)
and not databes_entry_is_newer(
media_data["lastUpdated"],
series_path / f"{series_name}_{database}_media_data.json",
)
):
buffer_file.unlink()
logger.info("Load from json:")
with open(
series_path / f"{series_name}_{database}_episodes.json",
encoding="utf-8",
) as f:
episodes = json.load(f)
with open(series_path.parent / f"{series_name}_{database}_media_data.json", encoding="utf-8") as f:
with open(
series_path / f"{series_name}_{database}_media_data.json",
encoding="utf-8",
) as f:
media_data = json.load(f)
with open(series_path.parent / f"{series_name}_{database}_series_data.json", encoding="utf-8") as f:
with open(
series_path / f"{series_name}_{database}_series_data.json",
encoding="utf-8",
) as f:
series_data = json.load(f)
else:
if database == "tvdb":
logger.info("Query from TVDB")
media_data = get_tvdbid_by_name(series_name)
series_data, episodes = get_all_episodes_by_tvdbid(media_data["id"])
elif database == "tmdb":
logger.info("Query from TMDB")
media_data = get_tmdbid_by_name(series_name)
series_data, episodes = get_all_episodes_by_tmdbid(media_data["id"])
logger.info("Dump to json")
with open(f"{series_name}_{database}_media_data.json", "w", encoding="utf-8") as f:
json.dump(media_data, f, ensure_ascii=False, indent=2)
with open(f"{series_name}_{database}_series_data.json", "w", encoding="utf-8") as f:
safe_rename(
buffer_file,
Path(f"{series_name}_{database}_media_data.json"),
overwrite=True,
logger=logger,
)
with open(
series_path / f"{series_name}_{database}_series_data.json",
"w",
encoding="utf-8",
) as f:
json.dump(series_data, f, ensure_ascii=False, indent=2)
with open(f"{series_name}_{database}_episodes.json", "w", encoding="utf-8") as f:
with open(
series_path / f"{series_name}_{database}_episodes.json",
"w",
encoding="utf-8",
) as f:
json.dump(episodes, f, ensure_ascii=False, indent=2)
return series_data, episodes, media_data
def process_series(series_path: Path, args):
series_name = series_path.name
logger.debug(f"Pricess series with name : {series_name}")
series_data, episodes, media_data = get_series_data(series_path)
series_data, episodes, media_data = get_series_data(series_path, args.database)
logger.info(
f"Found series name from TVDB: {media_data['name']}, ID: {media_data["id"]}, Year: {media_data.get('year')}"
f"Found series name from TVDB: {media_data['name']}, ID: {media_data['id']}, Year: {media_data.get('year')}"
)
reference_extra_episodes(episodes,series_path,media_data["name"],args)
print_results(media_data, series_data, episodes, extended=args.extended)
reference_extra_episodes(episodes, series_path, media_data, args)
translate_series_folder(episodes,series_path,media_data["name"],args)
translate_series_folder(episodes, series_path, media_data, args)
def main():
@@ -453,17 +717,34 @@ def main():
choices=["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
help="Set log level",
)
parser.add_argument(
"-db",
"--database",
type=str,
default="tvdb",
help="Set which database to use for queries. dafault: TvDB",
)
parser.add_argument(
"-lsd--leave-series-dir",
dest="translate_series_folder",
action="store_false",
help="Dont Translate Series Folder",
)
parser.add_argument(
"-e",
"--extended",
action="store_true",
help="Print Extended Infos",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Dry run only",
)
args = parser.parse_args()
dry_run = args.dry_run
setup_logging(args.log_level)
setup_logging(args.log_level, "sonarr_translation_tool")
input_path = Path(args.path).expanduser().resolve()
@@ -473,7 +754,7 @@ def main():
logger.info(f"Pricess folder: {input_path}")
# Input is already a series folder
if re.match(r"^(.*?)\s*\((\d{4})\)\s*\[(\d+)]$", input_path.name):
if re.match(r"^(.*?)\s*\((\d{4})\)(?:\s*\[(\d*)])?$", input_path.name):
logger.info(f"Process series: {input_path.name}")
process_series(input_path, args)
@@ -482,6 +763,7 @@ def main():
for series_path in sorted(input_path.iterdir()):
if series_path.is_dir():
process_series(series_path, args)
sys.exit(0)
if __name__ == "__main__":