lucas
6d229c59f7
add more translations add setup logging function for file and stdout logger combine translation functions: translate_audio_tracks, translate_audio_descriptions formatting changes fix move_override function
385 lines
10 KiB
Python
Executable File
385 lines
10 KiB
Python
Executable File
import json
|
|
import re
|
|
import sys
|
|
import shutil
|
|
import subprocess
|
|
from difflib import SequenceMatcher
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
import requests
|
|
import logging
|
|
|
|
|
|
# -----------------------------
|
|
# Logger
|
|
# -----------------------------
|
|
TRACE = 5
|
|
|
|
red = "\033[91m"
|
|
green = "\033[92m"
|
|
yellow = "\033[93m"
|
|
blue = "\033[94m"
|
|
reset = "\033[0m"
|
|
|
|
|
|
class CustomLogger(logging.Logger):
|
|
def trace(self, msg: str, *args, **kwargs) -> None:
|
|
if self.isEnabledFor(TRACE):
|
|
self._log(TRACE, msg, args, **kwargs)
|
|
|
|
|
|
LOG_LEVELS = {
|
|
"TRACE": TRACE,
|
|
"DEBUG": logging.DEBUG,
|
|
"INFO": logging.INFO,
|
|
"WARNING": logging.WARNING,
|
|
"ERROR": logging.ERROR,
|
|
"CRITICAL": logging.CRITICAL,
|
|
}
|
|
|
|
|
|
def setup_logging(log_level: str = "INFO", logfile: str = "logfile") -> None:
|
|
formatter = logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s")
|
|
|
|
stdout_handler = logging.StreamHandler(sys.stdout)
|
|
stdout_handler.setFormatter(formatter)
|
|
stdout_handler.setLevel(LOG_LEVELS[log_level.upper()])
|
|
|
|
ext = Path(logfile).suffix
|
|
if not ext:
|
|
ext = ".log"
|
|
logfile_data = (
|
|
Path("/home/lucas/logs/") / f"{logfile}_{datetime.now():%Y%m%d_%H%M%S}{ext}"
|
|
)
|
|
|
|
file_handler = logging.FileHandler(logfile_data, encoding="utf-8")
|
|
file_handler.setFormatter(formatter)
|
|
file_handler.setLevel(TRACE)
|
|
|
|
root_logger = logging.getLogger()
|
|
root_logger.setLevel(TRACE)
|
|
root_logger.handlers.clear() # avoid duplicate logging
|
|
|
|
root_logger.addHandler(stdout_handler)
|
|
root_logger.addHandler(file_handler)
|
|
root_logger.info(f"Set logger to {log_level}, Logfile: {logfile_data}")
|
|
|
|
|
|
# -----------------------------
|
|
# Helpers
|
|
# -----------------------------
|
|
def safe_delete(path: Path):
|
|
if path.is_dir():
|
|
shutil.rmtree(path)
|
|
else:
|
|
path.unlink()
|
|
|
|
|
|
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.
|
|
"""
|
|
|
|
src = Path(src)
|
|
dst = Path(dst)
|
|
|
|
# nothing to do
|
|
if src.resolve() == dst.resolve():
|
|
return "", False
|
|
|
|
directory = dst.parent
|
|
base = dst.stem
|
|
ext = dst.suffix
|
|
|
|
final_dst = dst
|
|
counter = 1
|
|
|
|
# find free name
|
|
while final_dst.exists():
|
|
final_dst = directory / f"{base}_{counter}{ext}"
|
|
counter += 1
|
|
ret = f"{src} → {dst}"
|
|
if not silent:
|
|
print(ret)
|
|
|
|
if dry:
|
|
return ret, True
|
|
|
|
if confirm:
|
|
input("Press Enter to confirm...")
|
|
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
if dst.exists() and not overwrite:
|
|
print("File already exists...")
|
|
return "", False
|
|
src.rename(dst)
|
|
return ret, True
|
|
|
|
|
|
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.
|
|
|
|
When ``handbrake`` is True, audio tracks are detected using
|
|
HandBrakeCLI's scan output. This is useful for DVDs and other
|
|
optical media sources where ffprobe may provide incomplete track
|
|
information.
|
|
|
|
When ``handbrake`` is False, ffprobe is used to inspect the media
|
|
file and return all audio streams.
|
|
|
|
Args:
|
|
source:
|
|
Path to the media file or DVD device/directory.
|
|
|
|
title:
|
|
DVD title number to scan when using HandBrakeCLI.
|
|
Ignored when using ffprobe.
|
|
|
|
handbrake:
|
|
If True, use HandBrakeCLI for track detection.
|
|
If False, use ffprobe.
|
|
|
|
Returns:
|
|
List of dictionaries containing:
|
|
|
|
- number: audio track number/index
|
|
- description: language, title, or track description
|
|
|
|
Example:
|
|
get_audio_tracks("movie.mkv")
|
|
[{'number': 1, 'description': 'eng'}]
|
|
|
|
get_audio_tracks("/dev/dvd", title=1, handbrake=True)
|
|
[{'number': 1, 'description': 'Deutsch'}]
|
|
"""
|
|
tracks: list[dict] = []
|
|
|
|
if handbrake:
|
|
result = subprocess.run(
|
|
[
|
|
"HandBrakeCLI",
|
|
"-i",
|
|
source,
|
|
"-t",
|
|
str(title),
|
|
"--scan",
|
|
],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
|
|
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(),
|
|
}
|
|
)
|
|
|
|
else:
|
|
result = subprocess.run(
|
|
[
|
|
"ffprobe",
|
|
"-v",
|
|
"quiet",
|
|
"-print_format",
|
|
"json",
|
|
"-show_streams",
|
|
source,
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
|
|
info = json.loads(result.stdout)
|
|
|
|
for stream in info["streams"]:
|
|
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,
|
|
}
|
|
)
|
|
|
|
if not tracks:
|
|
tracks.append({"number": 0, "description": "", "primary": False})
|
|
|
|
return tracks
|
|
|
|
|
|
def get_subtitle_tracks(source):
|
|
result = subprocess.run(
|
|
[
|
|
"ffprobe",
|
|
"-v",
|
|
"quiet",
|
|
"-print_format",
|
|
"json",
|
|
"-show_streams",
|
|
source,
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
|
|
info = json.loads(result.stdout)
|
|
|
|
tracks = []
|
|
|
|
for stream in info["streams"]:
|
|
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),
|
|
}
|
|
)
|
|
|
|
if not tracks:
|
|
tracks.append(
|
|
{
|
|
"number": 0,
|
|
"description": "",
|
|
"primary": False,
|
|
}
|
|
)
|
|
|
|
return tracks
|
|
|
|
|
|
def words(text):
|
|
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
|
|
|
|
if not union:
|
|
return 0.0
|
|
|
|
return len(intersection) / len(union)
|
|
|
|
|
|
def score_match_fuzzy(search_words, candidate_words):
|
|
total = 0.0
|
|
|
|
for sw in search_words:
|
|
best = 0.0
|
|
|
|
for cw in candidate_words:
|
|
best = max(best, word_similarity(sw, cw))
|
|
|
|
total += best
|
|
|
|
# logger.trace(f"Total: {total}/{len(search_words)}")
|
|
return total / len(search_words)
|
|
|
|
|
|
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
|
|
|
|
|
|
TMDB_TOKEN = "eyJhbGciOiJIUzI1NiJ9.eyJhdWQiOiJlOTIwMzJlMDk1NmEzMWEyZjQ4MDA4NGI0MTg4OTIyZCIsIm5iZiI6MTc5MDQzMjc5NS42NjcsInN1YiI6IjZhYjdkNjFiZDY5NmI1ZGFjNGM5NDRkMyIsInNjb3BlcyI6WyJhcGlfcmVhZCJdLCJ2ZXJzaW9uIjoxfQ.IjddKtQ5X62A63kYslH24Pxg_TFCzu-QpGak_Pemgr8"
|
|
TVDB_TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJhZ2UiOiIiLCJhcGlrZXkiOiI3NDBmYTBlOS1jODVmLTRmZGEtOWI1NS0wNzI0Y2E5NGRmN2QiLCJjb21tdW5pdHlfc3VwcG9ydGVkIjpmYWxzZSwiZXhwIjoxNzk1NDA5NDY4LCJnZW5kZXIiOiIiLCJoaXRzX3Blcl9kYXkiOjEwMDAwMDAwMCwiaGl0c19wZXJfbW9udGgiOjEwMDAwMDAwMCwiaWQiOiIzMTA0ODU5IiwiaXNfbW9kIjpmYWxzZSwiaXNfc3lzdGVtX2tleSI6ZmFsc2UsImlzX3RydXN0ZWQiOmZhbHNlLCJwaW4iOm51bGwsInJvbGVzIjpbXSwidGVuYW50IjoidHZkYiIsInV1aWQiOiIifQ.b8hNBL9KKrvmQatgAv_APgoriIocuZsZsKOLkk3tKzI08-xsqa0IILJy6bo6LYalIOU2swbFbGmrwEPeSEhsQfCZbUdPFwe6zi-7lfiynK66Uk6XdrL1wNsjzXz-Au55R8hMUIwwx2oAcQAVtIj7R8y70lh0N8BIaJ4BF0Mi0LL26-lLz-UjPQ1tWuqG8uvvLUodd2dvdvAkjcBaYXC5VFu5vEVkJOeX-OIAU0JTN3eY4XsLklTEXOd9SDhVBWV1fifWSMeWn3-o51_aQ75RkNMJa4f2VxCT9zoxXTja-ko9ewSvh14JBSzOWS7RJZV_l6r3l9xCiyXx_i996lH2xQp-Ouua3VlKuwbyetepkiy4yaAigANclpI-ByGiOsdd4XirWnwZPE6v6cIrEhlPRtUiCYTJJrkXKvv1TtBHUNGx1Zzu7ZVORIZ3SXy48VdHCugsFyrmP_y-MGkwE_U211mgVQT-cIiSPEYkchwTL66xLZ8DI9-lwzbHCOxJKOJ2ipUk9QkpEaYmhRPZ1RS6ZhKF1tVhD5-SVNKdRSBbfn6iN7ZLVghSOaM7DFzDGJSDTuF-DQDphQ1pkT_VWQlFjD2Ch8dFHDSJ7BRleLptbkCTmV8mG-UVixGCqVnPhl2YjmY9Vh3QzTqQ-WXRytdqQnvQr3c3HqoZmhdMYtzTCbE"
|
|
|
|
HEADERS_TMDB = {
|
|
"Authorization": f"Bearer {TMDB_TOKEN}",
|
|
"accept": "application/json",
|
|
}
|
|
|
|
|
|
HEADERS_TVDB = {
|
|
"Authorization": f"Bearer {TVDB_TOKEN}",
|
|
"accept": "application/json",
|
|
}
|
|
|
|
|
|
def tmdb_get(endpoint, params=None):
|
|
"""Perform a GET request against TMDb."""
|
|
|
|
BASE_URL = "https://api.themoviedb.org/3"
|
|
|
|
url = BASE_URL + endpoint
|
|
print(f"Set of Tmdb query: {url}")
|
|
|
|
response = requests.get(
|
|
url,
|
|
headers=HEADERS_TMDB,
|
|
params=params,
|
|
timeout=15,
|
|
)
|
|
|
|
response.raise_for_status()
|
|
|
|
return response.json()
|
|
|
|
|
|
def tvdb_get(endpoint, params=None):
|
|
"""Perform a GET request against TVDB."""
|
|
BASE_URL = "https://api4.thetvdb.com/v4"
|
|
url = BASE_URL + endpoint
|
|
print(f"Set of TVDB query: {url}")
|
|
|
|
response = requests.get(
|
|
url,
|
|
headers=HEADERS_TVDB,
|
|
params=params,
|
|
timeout=15,
|
|
)
|
|
|
|
response.raise_for_status()
|
|
|
|
return response.json()
|