remove tmdb_get from sonar translation script
This commit is contained in:
1 parent
93a464f49d
commit
f54919554b
1 file changed
+107
-87
@@ -1,48 +1,29 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import re
|
||||
import sys
|
||||
import requests
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from difflib import SequenceMatcher
|
||||
import logging
|
||||
|
||||
weight_jaccard = 0.6
|
||||
weight_fuzzy = 1 - weight_jaccard
|
||||
|
||||
from serverlibrary import (words, word_similarity, score_match_jaccard, score_match_fuzzy,
|
||||
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)
|
||||
logger: CustomLogger = logging.getLogger(__name__) # type: ignore[assignment]
|
||||
|
||||
|
||||
TMDB_TOKEN = "eyJhbGciOiJIUzI1NiJ9.eyJhdWQiOiJlOTIwMzJlMDk1NmEzMWEyZjQ4MDA4NGI0MTg4OTIyZCIsIm5iZiI6MTc5MDQzMjc5NS42NjcsInN1YiI6IjZhYjdkNjFiZDY5NmI1ZGFjNGM5NDRkMyIsInNjb3BlcyI6WyJhcGlfcmVhZCJdLCJ2ZXJzaW9uIjoxfQ.IjddKtQ5X62A63kYslH24Pxg_TFCzu-QpGak_Pemgr8"
|
||||
|
||||
if not TMDB_TOKEN:
|
||||
print("ERROR: TMDB_API_TOKEN is not set.")
|
||||
print()
|
||||
print("Example:")
|
||||
print(" export TMDB_API_TOKEN='your_token'")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
BASE_URL = "https://api.themoviedb.org/3"
|
||||
|
||||
HEADERS = {
|
||||
"Authorization": f"Bearer {TMDB_TOKEN}",
|
||||
"accept": "application/json",
|
||||
}
|
||||
|
||||
|
||||
def tmdb_get(endpoint, params=None):
|
||||
"""Perform a GET request against TMDb."""
|
||||
url = BASE_URL + endpoint
|
||||
print(f"Set of Tmdb query: {url}")
|
||||
|
||||
response = requests.get(
|
||||
url,
|
||||
headers=HEADERS,
|
||||
params=params,
|
||||
timeout=15,
|
||||
)
|
||||
|
||||
response.raise_for_status()
|
||||
|
||||
return response.json()
|
||||
|
||||
def find_by_title(title, media_type="tv", year=None, language="de-DE"):
|
||||
def find_tmdbid_by_title(title, media_type="tv", year=None, language="de-DE"):
|
||||
"""
|
||||
Search TMDB by title.
|
||||
|
||||
@@ -103,7 +84,6 @@ def get_tv_details(tmdb_id , language="de-DE"):
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def get_movie_details(tmdb_id, language="de-DE"):
|
||||
"""
|
||||
Retrieve movie details in German.
|
||||
@@ -124,7 +104,7 @@ def get_season(tmdb_id, season_number):
|
||||
},
|
||||
)
|
||||
|
||||
def get_episode_list(show,series_path: Path):
|
||||
def get_episode_list(show,series_path: Path, get_all: bool = False):
|
||||
#_, show = find_by_tmdb_id(tmdb_id)
|
||||
file_seasons = sorted({
|
||||
int(m.group(1))
|
||||
@@ -138,7 +118,7 @@ def get_episode_list(show,series_path: Path):
|
||||
season_number = season["season_number"]
|
||||
|
||||
# skip specials if desired
|
||||
if season_number == 0 or season_number not in file_seasons:
|
||||
if not get_all and (season_number == 0 or season_number not in file_seasons):
|
||||
continue
|
||||
|
||||
season_data = get_season(show.get("id"), season_number)
|
||||
@@ -159,26 +139,26 @@ def get_episode_list(show,series_path: Path):
|
||||
def print_results(result):
|
||||
media_type, data = result
|
||||
|
||||
print(f"Type : {media_type}")
|
||||
print(f"TMDB ID : {data.get('id')}")
|
||||
print(f"Title : {data.get('name') or data.get('title')}")
|
||||
print(f"Original : {data.get('original_name') or data.get('original_title')}")
|
||||
print(f"Year : {(data.get('first_air_date') or data.get('release_date') or '')[:4]}")
|
||||
print(f"Rating : {data.get('vote_average')}")
|
||||
print(f"Votes : {data.get('vote_count')}")
|
||||
print(f"Status : {data.get('status')}")
|
||||
print(f"Tagline : {data.get('tagline')}")
|
||||
print(f"Overview : {data.get('overview')}")
|
||||
logger.info(f"Type : {media_type}")
|
||||
logger.info(f"TMDB ID : {data.get('id')}")
|
||||
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(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"Votes : {data.get('vote_count')}")
|
||||
logger.info(f"Status : {data.get('status')}")
|
||||
logger.info(f"Tagline : {data.get('tagline')}")
|
||||
logger.info(f"Overview : {data.get('overview')}")
|
||||
|
||||
genres = ", ".join(g["name"] for g in data.get("genres", []))
|
||||
print(f"Genres : {genres}")
|
||||
logger.info(f"Genres : {genres}")
|
||||
|
||||
print(f"Poster : https://image.tmdb.org/t/p/original{data.get('poster_path', '')}")
|
||||
print(f"Backdrop : https://image.tmdb.org/t/p/original{data.get('backdrop_path', '')}")
|
||||
logger.info(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":
|
||||
print(f"Seasons : {data.get('number_of_seasons')}")
|
||||
print(f"Episodes : {data.get('number_of_episodes')}")
|
||||
logger.info(f"Seasons : {data.get('number_of_seasons')}")
|
||||
logger.info(f"Episodes : {data.get('number_of_episodes')}")
|
||||
|
||||
def normalize(text, title, year):
|
||||
text = text.lower()
|
||||
@@ -209,20 +189,18 @@ def best_episode_match(filename, episodes,title,year):
|
||||
best = None
|
||||
best_score = 0
|
||||
|
||||
for ep in episodes:
|
||||
score = SequenceMatcher(
|
||||
None,
|
||||
filename_norm,
|
||||
normalize(ep["title"], title, year)
|
||||
).ratio()
|
||||
for episode in episodes:
|
||||
score_fuzzy = score_match_fuzzy(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)
|
||||
|
||||
if score > best_score:
|
||||
best_score = score
|
||||
best = ep
|
||||
best = episode
|
||||
|
||||
return best, best_score
|
||||
|
||||
def rename_media(media_data,episodes, series_path,file_title):
|
||||
def rename_media(episodes, series_path,file_title,tmdb_title):
|
||||
series=series_path.name
|
||||
episode_map = {
|
||||
(ep["season"], ep["episode"]): ep
|
||||
@@ -236,7 +214,7 @@ def rename_media(media_data,episodes, series_path,file_title):
|
||||
match_season_folder = re.search(r"Season (\d{2})", file.name, re.I)
|
||||
if match_season_folder:
|
||||
continue # witchout bothering anyone
|
||||
print(f"No season/episode number found: {file.name}")
|
||||
logger.error(f"No season/episode number found: {file.name}")
|
||||
continue
|
||||
|
||||
season = int(match_SE_name.group(1))
|
||||
@@ -245,7 +223,7 @@ def rename_media(media_data,episodes, series_path,file_title):
|
||||
episode_number = episode_map.get((season, episode))
|
||||
|
||||
if not episode_number:
|
||||
print(f"Episode not found: S{season:02d}E{episode:02d}")
|
||||
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*(\[[^\]]+\].*)?$",
|
||||
@@ -253,10 +231,9 @@ def rename_media(media_data,episodes, series_path,file_title):
|
||||
re.I,
|
||||
)
|
||||
if not episode_season_number_match:
|
||||
print(f"S{season:02d}E{episode:02d} not found")
|
||||
logger.warning(f"S{season:02d}E{episode:02d} not found")
|
||||
continue
|
||||
prefix = episode_season_number_match.group(1).replace(file_title,(media_data.get('name')
|
||||
or media_data.get('title'))) # Andor (2022) - S01E07
|
||||
prefix = episode_season_number_match.group(1).replace(file_title,tmdb_title) # Andor (2022) - S01E07
|
||||
quality = episode_season_number_match.group(3) or "" # [WEBRip-1080p][AAC 5.1][h265]
|
||||
|
||||
new_name = (
|
||||
@@ -265,43 +242,40 @@ def rename_media(media_data,episodes, series_path,file_title):
|
||||
f"{file.suffix}"
|
||||
)
|
||||
|
||||
print(f"{file.name} -> {new_name}")
|
||||
logger.info(f"{file.name} -> {new_name}")
|
||||
|
||||
file_list.append(file)
|
||||
new_file_list.append(new_name)
|
||||
|
||||
title=(f"{(media_data.get('name') or media_data.get('title'))} "
|
||||
f"({(media_data.get('first_air_date') or media_data.get('release_date') or '')[:4]}) "
|
||||
f"[{media_data.get('id')}]")
|
||||
print(f"{series} -> {title}")
|
||||
logger.info(f"{series} -> {tmdb_title}")
|
||||
|
||||
file_list.append(series_path)
|
||||
new_file_list.append(title)
|
||||
new_file_list.append(tmdb_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:
|
||||
print("No changes necessary.")
|
||||
logger.info("No changes necessary.")
|
||||
return
|
||||
answer = input("Rename files? [y/N]: ").strip().lower()
|
||||
if answer in ("y", "yes"):
|
||||
print("Proceeding...")
|
||||
logger.info("Proceeding...")
|
||||
for file, new_file in zip(file_list,new_file_list):
|
||||
file.rename(file.with_name(new_file))
|
||||
print("All files renamed")
|
||||
logger.info("All files renamed")
|
||||
return
|
||||
print("Cancelled.")
|
||||
logger.warning("Cancelled.")
|
||||
return
|
||||
|
||||
def process_series(series_path: Path):
|
||||
def process_series(series_path: Path,skip_rename: bool = False):
|
||||
series = series_path.name
|
||||
|
||||
m = re.match(r"^(.*?)\s*\((\d{4})\)\s*\[(\d+)\]$", series)
|
||||
|
||||
if not m:
|
||||
print(f"Skipping invalid folder: {series}")
|
||||
logger.warning(f"Skipping invalid folder: {series}")
|
||||
return
|
||||
|
||||
file_title = m.group(1).strip()
|
||||
@@ -309,31 +283,56 @@ def process_series(series_path: Path):
|
||||
file_tmdb_id = int(m.group(3))
|
||||
|
||||
if file_tmdb_id == 0:
|
||||
print("TMDB ID required")
|
||||
logger.error("TMDB ID required")
|
||||
return
|
||||
|
||||
try:
|
||||
data = find_by_tmdb_id(file_tmdb_id)
|
||||
|
||||
if not data:
|
||||
print("TMDB lookup failed")
|
||||
logger.error("TMDB lookup failed")
|
||||
return
|
||||
|
||||
media_type, media_data = data
|
||||
|
||||
#print_results(data)
|
||||
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]}"
|
||||
new_series_folder_name = (f"{tmdb_title} "
|
||||
f"({tmdb_year}) "
|
||||
f"[{media_data.get('id')}]")
|
||||
|
||||
episodes = get_episode_list(media_data,series_path)
|
||||
#print_results(data)
|
||||
get_all=False
|
||||
extra_episodes_path = series_path / "Extra Episodes"
|
||||
if extra_episodes_path.is_dir():
|
||||
get_all = True
|
||||
|
||||
episodes = get_episode_list(media_data,series_path, get_all=get_all)
|
||||
|
||||
process_extra_episodes(episodes, extra_episodes_path, tmdb_title,tmdb_year)
|
||||
|
||||
if skip_rename:
|
||||
return
|
||||
|
||||
rename_media(
|
||||
media_data,
|
||||
episodes,
|
||||
series_path,
|
||||
file_title
|
||||
file_title,new_series_folder_name
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
print(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():
|
||||
|
||||
if not (file.suffix == ".mkv" or file.suffix == ".mp4"):
|
||||
continue
|
||||
|
||||
best_matc, best_score = best_episode_match(file,episode_list,tmdb_title,tmdb_year)
|
||||
|
||||
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
@@ -342,8 +341,29 @@ def main():
|
||||
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",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
LOG_LEVELS = {
|
||||
"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)
|
||||
|
||||
if not input_path.is_dir():
|
||||
|
||||
Reference in new issue
Block a user