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

feat(sonarr_database_tool): add tvdb and tmdb fucntions

Browse Source 
add functions from tmdb and tvdb script
rename match_series file to sonarr_database_tool
This commit is contained in:
Lucas Vivo committed 2026-10-27 19:53:41 +01:00
1 parent 07229a4435
commit 9929dcbd8c
2 files changed
+488 -310
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

media_editing/match_series_to_tvdb.py
-310
View File
@@ -1,310 +0,0 @@
#!/usr/bin/env python3
import argparse
import json
import logging
import re
from pathlib import Path
import requests
weight_jaccard = 0.5
weight_fuzzy = 1 - weight_jaccard
MATCH_THRESHOLD = 0.85
from serverlibrary import *
logging.setLoggerClass(CustomLogger)
logger: CustomLogger = logging.getLogger(__name__)
def normalize(text: str) -> str:
text = text.lower()
text = re.sub(r"\.[a-z0-9]{2,4}$", "", text)
text = re.sub(
r"\b(1080p|720p|2160p|x264|x265|h264|h265|bluray|web-dl|webrip|dvdrip)\b",
"",
text,
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()
def get_all_episodes_by_tvdbid(tvdb_id: int):
series = tvdb_get(
f"/series/{tvdb_id}/extended",
{"language": "deu"},
)["data"]
episodes = []
for season in series.get("seasons", []):
logger.info(f"Requesting Season: {season.get('number')}")
season_data = {}
try:
season_data = tvdb_get(
f"/seasons/{season['id']}/extended",
{"language": "deu"},
)["data"]
except requests.exceptions.RequestException:
logger.error(
"Failed to fetch season %s",
season["id"],
)
for ep in season_data.get("episodes", []):
episodes.append(
{
"id": ep["id"],
"season": ep["seasonNumber"],
"episode": ep["number"],
"title": ep["name"],
}
)
return series, episodes
def parse_series_name(series_name: str):
"""
Universe (1987) [66757]
-> ("Universe", 1987, 66757)
Universe (1987)
-> ("Universe", 1987, None)
Universe
-> ("Universe", None, None)
"""
tmdb_id = None
year = None
id_match = re.search(r"\[(\d+)]", series_name)
if id_match:
tmdb_id = int(id_match.group(1))
year_match = re.search(r"\((\d{4})\)", series_name)
if year_match:
year = int(year_match.group(1))
title = re.sub(r"\[\d+]", "", series_name)
title = re.sub(r"\(\d{4}\)", "", title)
title = title.strip()
return title, year, tmdb_id
def get_tvdbid_by_name(series_name: str):
title, year, tmdb_id = parse_series_name(series_name)
params = {
"query": title,
"type": "series",
"language": "deu",
}
if year:
params["year"] = str(year)
results = tvdb_get(
"/search",
params,
)["data"]
if not results:
raise RuntimeError(f"Series not found: {series_name}")
# Filter by year if available
if tmdb_id is not None:
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:
filtered.append(result)
if filtered:
results = filtered
return results[0]
def best_episode_match(filename, episodes):
search_words = words(normalize(filename))
logger.debug(f"Search Words: {search_words}")
best_score = 0.0
best_episode = None
for episode in episodes:
if not episode["title"] or episode["title"] == "Unkown":
continue
candidate_words = words(normalize(episode["title"]))
fuzzy = score_match_fuzzy(
search_words,
candidate_words,
)
jaccard = score_match_jaccard(
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,
jaccard,
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
return best_episode, best_score
def process_series(series_path: Path, dry_run: bool = False):
series_name = series_path.name
logger.debug(f"Pricess series with name : {series_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"])
series_data, 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}_series_data.json", "w", encoding="utf-8") as f:
json.dump(series_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 from TVDB: {media_data['name']}, ID: {tvdb_id}, Year: {media_data.get('year')}"
)
extra_episodes_path = series_path / "Extra Episodes"
for file in extra_episodes_path.iterdir():
if not file.is_file():
continue
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}: {normalize(episode.get('title'))}"
)
continue
safe_title = re.sub(
r'[<>:"/\\|?*]',
"",
episode["title"],
)
new_name = (
f"{media_data.get('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
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:
logger.info(f"{file.name} -> {target}, score={score:.3f}")
safe_rename(file, target)
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',
)
parser.add_argument(
"-ll",
"--log-level",
type=str.upper,
default="INFO",
choices=["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
help="Set log level",
)
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)
input_path = Path(args.path).expanduser().resolve()
if not input_path.is_dir():
parser.error(f"Directory not found: {input_path}")
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):
logger.info(f"Process series: {input_path.name}")
process_series(input_path, dry_run)
# 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)
if __name__ == "__main__":
main()
media_editing/sonarr_database_tool.py
Executable
+488
View File
@@ -0,0 +1,488 @@
#!/usr/bin/env python3
import argparse
import json
import logging
import re
from pathlib import Path
import requests
weight_jaccard = 0.5
weight_fuzzy = 1 - weight_jaccard
MATCH_THRESHOLD = 0.85
from serverlibrary import *
logging.setLoggerClass(CustomLogger)
logger: CustomLogger = logging.getLogger(__name__)
def normalize(text: str, title: str | None = None, year: int | str | None = None) -> str:
text = text.lower()
# remove series/movie title
if title:
text = re.sub(re.escape(title.lower()), " ", text)
# remove release year
if year:
text = re.sub(rf"\b{re.escape(str(year))}\b", " ", text)
# remove file extension
text = re.sub(r"\.[a-z0-9]{2,4}$", " ", text)
# remove text inside square brackets
text = re.sub(r"\[[^\]]*\]", " ", text)
# remove episode numbers (S01E01)
text = re.sub(r"\bs\d+e\d+\b", " ", text, flags=re.IGNORECASE)
# remove common release tags
text = re.sub(
r"\b(?:1080p|720p|2160p|x264|x265|h264|h265|bluray|web[- ]?dl|webrip|dvdrip)\b",
" ",
text,
flags=re.IGNORECASE,
)
# normalize separators
text = re.sub(r"[._-]", " ", text)
# German normalization
text = text.replace("ß", "ss")
# "(123 456)" -> "(123)"
text = re.sub(r"\((\d+)\s+\d+\)", r"(\1)", text)
# "(123___456)" -> "(123)"
text = re.sub(r"\((\d+)_+\d+\)", r"(\1)", text)
# remove remaining punctuation
text = re.sub(r"[^\w\s]", " ", text)
# collapse whitespace
text = re.sub(r"\s+", " ", text).strip()
return text
def parse_series_name(series_name: str):
"""
Universe (1987) [66757]
-> ("Universe", 1987, 66757)
Universe (1987)
-> ("Universe", 1987, None)
Universe
-> ("Universe", None, None)
"""
tmdb_id = None
year = None
id_match = re.search(r"\[(\d+)]", series_name)
if id_match:
tmdb_id = int(id_match.group(1))
year_match = re.search(r"\((\d{4})\)", series_name)
if year_match:
year = int(year_match.group(1))
title = re.sub(r"\[\d+]", "", series_name)
title = re.sub(r"\(\d{4}\)", "", title)
title = title.strip()
return title, year, tmdb_id
def find_by_tvdb_id(tvdb_id:str, language="deu"):
"""
Try to determine whether a TMDB ID is a TV show or movie.
Returns:
("tv", data)
("movie", data)
(None, None)
"""
return tvdb_get(
f"/series/{tvdb_id}/extended",
{"language": language},
)["data"]
def find_by_tmdb_id(tmdb_id:str, language="de-DE", tmdb_type="tv"):
"""
Try to determine whether a TMDB ID is a TV show or movie.
Returns:
("tv", data)
("movie", data)
(None, None)
"""
return tmdb_get(
f"/{type}/{tmdb_id}",
{
"language": language,
},
)
def get_all_episodes_by_tvdbid(tvdb_id: str, language: str = "deu"):
series_data = find_by_tvdb_id(tvdb_id)
episodes_data = []
for season in series_data.get("seasons", []):
logger.info(f"Requesting Season: {season.get('number')}")
season_data = {}
try:
season_data = tvdb_get(
f"/seasons/{season['id']}/extended",
{"language":language},
)["data"]
except requests.exceptions.RequestException:
logger.error(
"Failed to fetch season %s",
season["id"],
)
for episode in season_data.get("episodes", []):
episodes_data.append(
{
"id": episode["id"],
"season": episode["seasonNumber"],
"episode": episode["number"],
"title": episode["name"],
}
)
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(
f"/tv/{tmdb_id}/season/{season_number}",
{
"language": language,
},
)
series_data = find_by_tmdb_id(tmdb_id)
episodes_data = []
for season in series_data.get("seasons", []):
season_number = season["season_number"]
season_data = get_season(series_data.get("id"), season_number)
for episode in season_data.get("episodes", []):
episodes_data.append(
{
"id": episode["id"],
"season": season_number,
"episode": episode["episode_number"],
"title": episode["name"],
}
)
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)
params = {
"query": title,
"type": "series",
"language": language,
}
if year:
params["year"] = str(year)
results = tvdb_get(
"/search",
params,
)["data"]
if not results:
raise RuntimeError(f"Series not found: {series_name}")
# Filter by year if available
if tmdb_id is not None:
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:
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"):
"""
Search TMDB by title.
media_type:
tv | movie
"""
title, year, tmdb_id = parse_series_name(series_name)
params = {
"query": title,
"language": language,
}
if year:
if media_type == "tv":
params["first_air_date_year"] = year
else:
params["year"] = year
return tmdb_get(
f"/search/{media_type}",
params,
)
def best_episode_title_match(filename, episodes):
search_words = words(normalize(filename))
logger.debug(f"Search Words: {search_words}")
best_score = 0.0
best_episode = None
for episode in episodes:
if not episode["title"] or episode["title"] == "Unkown":
continue
candidate_words = words(normalize(episode["title"]))
fuzzy = score_match_fuzzy(
search_words,
candidate_words,
)
jaccard = score_match_jaccard(
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,
jaccard,
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
return best_episode, best_score
def translate_series_folder(episodes, series_path, file_title: str, series_title: str,args):
series = series_path.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():
continue
match_SE_name = re.search(r"S(\d{2})E(\d{2})", file.name, re.I)
if not match_SE_name:
logger.error(f"No season/episode number found: {file.name}")
continue
season = int(match_SE_name.group(1))
episode = int(match_SE_name.group(2))
episode_number = episode_map.get((season, episode))
if not episode_number:
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*(\[[^\]]+\].*)?$",
file.stem,
re.I,
)
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
quality = (
episode_season_number_match.group(3) or ""
) # [WEBRip-1080p][AAC 5.1][h265]
new_name = (
f"{prefix} - {episode_number['title']}"
f"{(' ' + quality) if quality else ''}"
f"{file.suffix}"
)
logger.trace(f"{file.name} -> {new_name}")
file_list.append(file)
new_file_list.append(new_name)
logger.trace(f"{series} -> {series_title}")
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:
has_changes = True
break
if not has_changes:
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)
logger.info("All files renamed")
return
def reference_extra_episodes(episodes, series_path: Path, series_title: str, args):
extra_episodes_path = series_path / "Extra Episodes"
for file in extra_episodes_path.iterdir():
if not file.is_file():
continue
filename = re.sub(series_title, "", file.stem)
episode, score = best_episode_title_match(
filename,
episodes,
)
if episode is None:
continue
if score < MATCH_THRESHOLD:
logger.warning(
f"LOW CONFIDENCE: {filename} {score:.2f}: {normalize(episode.get('title'))}"
)
continue
safe_title = re.sub(
r'[<>:"/\\|?*]',
"",
episode["title"],
)
new_name = (
f"{series_title}"
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
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 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:
episodes = json.load(f)
with open(series_path.parent / 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:
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:
json.dump(series_data, f, ensure_ascii=False, indent=2)
with open(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)
logger.info(
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)
translate_series_folder(episodes,series_path,media_data["name"],args)
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',
)
parser.add_argument(
"-ll",
"--log-level",
type=str.upper,
default="INFO",
choices=["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
help="Set log level",
)
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)
input_path = Path(args.path).expanduser().resolve()
if not input_path.is_dir():
parser.error(f"Directory not found: {input_path}")
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):
logger.info(f"Process series: {input_path.name}")
process_series(input_path, args)
# 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, args)
if __name__ == "__main__":
main()