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

fix(sonar_translation): add new logging

Browse Source
This commit is contained in:
lucas committed 2026-10-27 16:38:50 +01:00
1 parent cabf1a21a8
commit 2a48f50717
1 file changed
+81 -80
Download Patch File Download Diff File Expand all files Collapse all files
media_editing/sonar_translation.py
+81 -80
View File
@@ -1,4 +1,4 @@
#!/usr/bin/env python3 #!/home/lucas/scripts/.venv/bin/python
import re import re
import requests import requests
@@ -9,15 +9,7 @@ import logging
weight_jaccard = 0.6 weight_jaccard = 0.6
weight_fuzzy = 1 - weight_jaccard weight_fuzzy = 1 - weight_jaccard
from serverlibrary import (words, word_similarity, score_match_jaccard, score_match_fuzzy, from serverlibrary import *
combined_score, tmdb_get, tvdb_get)
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) logging.setLoggerClass(CustomLogger)
logger: CustomLogger = logging.getLogger(__name__) # type: ignore[assignment] logger: CustomLogger = logging.getLogger(__name__) # type: ignore[assignment]
@@ -72,7 +64,8 @@ def find_by_tmdb_id(tmdb_id, language="de-DE"):
return None, None return None, None
def get_tv_details(tmdb_id , language="de-DE"):
def get_tv_details(tmdb_id, language="de-DE"):
""" """
Retrieve TV show details in German. Retrieve TV show details in German.
""" """
@@ -84,6 +77,7 @@ def get_tv_details(tmdb_id , language="de-DE"):
}, },
) )
def get_movie_details(tmdb_id, language="de-DE"): def get_movie_details(tmdb_id, language="de-DE"):
""" """
Retrieve movie details in German. Retrieve movie details in German.
@@ -96,6 +90,7 @@ def get_movie_details(tmdb_id, language="de-DE"):
}, },
) )
def get_season(tmdb_id, season_number): def get_season(tmdb_id, season_number):
return tmdb_get( return tmdb_get(
f"/tv/{tmdb_id}/season/{season_number}", f"/tv/{tmdb_id}/season/{season_number}",
@@ -104,14 +99,16 @@ def get_season(tmdb_id, season_number):
}, },
) )
def get_episode_list(show,series_path: Path, get_all: bool = False):
#_, show = find_by_tmdb_id(tmdb_id) def get_episode_list(show, series_path: Path, get_all: bool = False):
file_seasons = sorted({ # _, show = find_by_tmdb_id(tmdb_id)
int(m.group(1)) file_seasons = sorted(
for d in series_path.iterdir() {
if d.is_dir() int(m.group(1))
and (m := re.match(r"^Season\s+0*(\d+)$", d.name, re.I)) for d in series_path.iterdir()
}) if d.is_dir() and (m := re.match(r"^Season\s+0*(\d+)$", d.name, re.I))
}
)
episodes = [] episodes = []
for season in show.get("seasons", []): for season in show.get("seasons", []):
@@ -124,14 +121,16 @@ def get_episode_list(show,series_path: Path, get_all: bool = False):
season_data = get_season(show.get("id"), season_number) season_data = get_season(show.get("id"), season_number)
for episode in season_data.get("episodes", []): for episode in season_data.get("episodes", []):
episodes.append({ episodes.append(
"season": season_number, {
"episode": episode["episode_number"], "season": season_number,
"title": episode["name"], "episode": episode["episode_number"],
"air_date": episode["air_date"], "title": episode["name"],
"runtime": episode.get("runtime"), "air_date": episode["air_date"],
"tmdb_id": episode["id"], "runtime": episode.get("runtime"),
}) "tmdb_id": episode["id"],
}
)
return episodes return episodes
@@ -142,8 +141,12 @@ def print_results(result):
logger.info(f"Type : {media_type}") logger.info(f"Type : {media_type}")
logger.info(f"TMDB ID : {data.get('id')}") logger.info(f"TMDB ID : {data.get('id')}")
logger.info(f"Title : {data.get('name') or data.get('title')}") logger.info(f"Title : {data.get('name') or data.get('title')}")
logger.info(f"Original : {data.get('original_name') or data.get('original_title')}") logger.info(
logger.info(f"Year : {(data.get('first_air_date') or data.get('release_date') or '')[:4]}") f"Original : {data.get('original_name') or data.get('original_title')}"
)
logger.info(
f"Year : {(data.get('first_air_date') or data.get('release_date') or '')[:4]}"
)
logger.info(f"Rating : {data.get('vote_average')}") logger.info(f"Rating : {data.get('vote_average')}")
logger.info(f"Votes : {data.get('vote_count')}") logger.info(f"Votes : {data.get('vote_count')}")
logger.info(f"Status : {data.get('status')}") logger.info(f"Status : {data.get('status')}")
@@ -153,13 +156,18 @@ def print_results(result):
genres = ", ".join(g["name"] for g in data.get("genres", [])) genres = ", ".join(g["name"] for g in data.get("genres", []))
logger.info(f"Genres : {genres}") logger.info(f"Genres : {genres}")
logger.info(f"Poster : https://image.tmdb.org/t/p/original{data.get('poster_path', '')}") logger.info(
logger.info(f"Backdrop : https://image.tmdb.org/t/p/original{data.get('backdrop_path', '')}") f"Poster : https://image.tmdb.org/t/p/original{data.get('poster_path', '')}"
)
logger.info(
f"Backdrop : https://image.tmdb.org/t/p/original{data.get('backdrop_path', '')}"
)
if media_type == "tv": if media_type == "tv":
logger.info(f"Seasons : {data.get('number_of_seasons')}") logger.info(f"Seasons : {data.get('number_of_seasons')}")
logger.info(f"Episodes : {data.get('number_of_episodes')}") logger.info(f"Episodes : {data.get('number_of_episodes')}")
def normalize(text, title, year): def normalize(text, title, year):
text = text.lower() text = text.lower()
@@ -183,15 +191,20 @@ def normalize(text, title, year):
return text return text
def best_episode_match(filename, episodes,title,year):
def best_episode_match(filename, episodes, title, year):
filename_norm = normalize(Path(filename).stem, title, year) filename_norm = normalize(Path(filename).stem, title, year)
best = None best = None
best_score = 0 best_score = 0
for episode in episodes: for episode in episodes:
score_fuzzy = score_match_fuzzy(filename_norm, normalize(episode["title"], title, year)) score_fuzzy = score_match_fuzzy(
score_jaccard = score_match_jaccard(filename_norm, normalize(episode["title"], title, year)) filename_norm, normalize(episode["title"], title, year)
)
score_jaccard = score_match_jaccard(
filename_norm, normalize(episode["title"], title, year)
)
score = combined_score(score_fuzzy, score_jaccard, weight_fuzzy, weight_jaccard) score = combined_score(score_fuzzy, score_jaccard, weight_fuzzy, weight_jaccard)
if score > best_score: if score > best_score:
@@ -200,12 +213,10 @@ def best_episode_match(filename, episodes,title,year):
return best, best_score return best, best_score
def rename_media(episodes, series_path,file_title,tmdb_title):
series=series_path.name def rename_media(episodes, series_path, file_title, tmdb_title):
episode_map = { series = series_path.name
(ep["season"], ep["episode"]): ep episode_map = {(ep["season"], ep["episode"]): ep for ep in episodes}
for ep in episodes
}
file_list = [] file_list = []
new_file_list = [] new_file_list = []
for file in series_path.rglob("*"): for file in series_path.rglob("*"):
@@ -213,12 +224,12 @@ def rename_media(episodes, series_path,file_title,tmdb_title):
if not match_SE_name: if not match_SE_name:
match_season_folder = re.search(r"Season (\d{2})", file.name, re.I) match_season_folder = re.search(r"Season (\d{2})", file.name, re.I)
if match_season_folder: if match_season_folder:
continue # witchout bothering anyone continue # witchout bothering anyone
logger.error(f"No season/episode number found: {file.name}") logger.error(f"No season/episode number found: {file.name}")
continue continue
season = int(match_SE_name.group(1)) season = int(match_SE_name.group(1))
episode= int(match_SE_name.group(2)) episode = int(match_SE_name.group(2))
episode_number = episode_map.get((season, episode)) episode_number = episode_map.get((season, episode))
@@ -233,8 +244,12 @@ def rename_media(episodes, series_path,file_title,tmdb_title):
if not episode_season_number_match: if not episode_season_number_match:
logger.warning(f"S{season:02d}E{episode:02d} not found") logger.warning(f"S{season:02d}E{episode:02d} not found")
continue continue
prefix = episode_season_number_match.group(1).replace(file_title,tmdb_title) # Andor (2022) - S01E07 prefix = episode_season_number_match.group(1).replace(
quality = episode_season_number_match.group(3) or "" # [WEBRip-1080p][AAC 5.1][h265] file_title, tmdb_title
) # Andor (2022) - S01E07
quality = (
episode_season_number_match.group(3) or ""
) # [WEBRip-1080p][AAC 5.1][h265]
new_name = ( new_name = (
f"{prefix} - {episode_number['title']}" f"{prefix} - {episode_number['title']}"
@@ -262,14 +277,15 @@ def rename_media(episodes, series_path,file_title,tmdb_title):
answer = input("Rename files? [y/N]: ").strip().lower() answer = input("Rename files? [y/N]: ").strip().lower()
if answer in ("y", "yes"): if answer in ("y", "yes"):
logger.info("Proceeding...") logger.info("Proceeding...")
for file, new_file in zip(file_list,new_file_list): for file, new_file in zip(file_list, new_file_list):
file.rename(file.with_name(new_file)) file.rename(file.with_name(new_file))
logger.info("All files renamed") logger.info("All files renamed")
return return
logger.warning("Cancelled.") logger.warning("Cancelled.")
return return
def process_series(series_path: Path,skip_rename: bool = False):
def process_series(series_path: Path, skip_rename: bool = False):
series = series_path.name series = series_path.name
m = re.match(r"^(.*?)\s*\((\d{4})\)\s*\[(\d+)\]$", series) m = re.match(r"^(.*?)\s*\((\d{4})\)\s*\[(\d+)\]$", series)
@@ -279,7 +295,7 @@ def process_series(series_path: Path,skip_rename: bool = False):
return return
file_title = m.group(1).strip() file_title = m.group(1).strip()
#file_year = int(m.group(2)) # file_year = int(m.group(2))
file_tmdb_id = int(m.group(3)) file_tmdb_id = int(m.group(3))
if file_tmdb_id == 0: if file_tmdb_id == 0:
@@ -297,48 +313,44 @@ def process_series(series_path: Path,skip_rename: bool = False):
tmdb_title = f"{(media_data.get('name') or media_data.get('title'))}" tmdb_title = f"{(media_data.get('name') or media_data.get('title'))}"
tmdb_year = f"{(media_data.get('first_air_date') or media_data.get('release_date') or '')[:4]}" tmdb_year = f"{(media_data.get('first_air_date') or media_data.get('release_date') or '')[:4]}"
new_series_folder_name = (f"{tmdb_title} " new_series_folder_name = f"{tmdb_title} ({tmdb_year}) [{media_data.get('id')}]"
f"({tmdb_year}) "
f"[{media_data.get('id')}]")
#print_results(data) # print_results(data)
get_all=False get_all = False
extra_episodes_path = series_path / "Extra Episodes" extra_episodes_path = series_path / "Extra Episodes"
if extra_episodes_path.is_dir(): if extra_episodes_path.is_dir():
get_all = True get_all = True
episodes = get_episode_list(media_data,series_path, get_all=get_all) episodes = get_episode_list(media_data, series_path, get_all=get_all)
process_extra_episodes(episodes, extra_episodes_path, tmdb_title,tmdb_year) process_extra_episodes(episodes, extra_episodes_path, tmdb_title, tmdb_year)
if skip_rename: if skip_rename:
return return
rename_media( rename_media(episodes, series_path, file_title, new_series_folder_name)
episodes,
series_path,
file_title,new_series_folder_name
)
except Exception as e: except Exception as e:
logger.error(f"Error processing {series}: {e}") logger.error(f"Error processing {series}: {e}")
def process_extra_episodes(episode_list: list,extra_episodes_path: Path,tmdb_title,tmdb_year):
for file in extra_episodes_path.iterdir():
def process_extra_episodes(
episode_list: list, extra_episodes_path: Path, tmdb_title, tmdb_year
):
for file in extra_episodes_path.iterdir():
if not (file.suffix == ".mkv" or file.suffix == ".mp4"): if not (file.suffix == ".mkv" or file.suffix == ".mp4"):
continue continue
best_matc, best_score = best_episode_match(file,episode_list,tmdb_title,tmdb_year) best_matc, best_score = best_episode_match(
file, episode_list, tmdb_title, tmdb_year
)
def main(): def main():
parser = argparse.ArgumentParser() parser = argparse.ArgumentParser()
parser.add_argument( parser.add_argument(
"path", "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( parser.add_argument(
@@ -351,19 +363,7 @@ def main():
) )
args = parser.parse_args() args = parser.parse_args()
LOG_LEVELS = { setup_logging(args.log_level, "sonar_translation")
"TRACE": TRACE,
"DEBUG": logging.DEBUG,
"INFO": logging.INFO,
"WARNING": logging.WARNING,
"ERROR": logging.ERROR,
"CRITICAL": logging.CRITICAL,
}
logging.basicConfig(
level=LOG_LEVELS[args.log_level.upper()],
format="%(asctime)s %(levelname)s %(message)s",
)
input_path = Path(args.path) input_path = Path(args.path)
if not input_path.is_dir(): if not input_path.is_dir():
@@ -379,5 +379,6 @@ def main():
if series_path.is_dir(): if series_path.is_dir():
process_series(series_path) process_series(series_path)
if __name__ == "__main__": if __name__ == "__main__":
main() main()