def _compute_bpp_proxy(ctx: EncodeContext) -> float:
"""
Proxy estremamente semplificato del "bitrate per pixel".
Restituisce kbps per megapixel circa (solo per logging/euristica).
"""
pixels = max(1, ctx.width * ctx.height)
mpix = pixels / 1_000_000.0
return ctx.source_video_kbps_est / mpix if mpix > 0 else ctx.source_video_kbps_est
def classify_complexity_heuristic(ctx: EncodeContext) -> Complexity:
"""
Classificazione grossolana LOW/MID/HIGH basata sul "kbps per megapixel".
DA SOSTITUIRE/RAFFINARE con SI/TI + VMAF nei passi successivi.
"""
bpp_proxy = _compute_bpp_proxy(ctx)
# Soglie iniziali, da ritarare:
# - sotto ~700 kb/s per MPix → LOW
# - tra 700 e 1300 → MID
# - sopra 1300 → HIGH
if bpp_proxy < 700:
comp = Complexity.LOW
elif bpp_proxy < 1300:
comp = Complexity.MID
else:
comp = Complexity.HIGH
logger.info(
"[smart] complessità euristica: bpp_proxy≈%.1f kb/s/MPix → %s",
bpp_proxy, comp.value
)
return comp
@dataclass
class SmartPlan:
complexity: Complexity
vf_height: int
v_kbps: int
floor_kbps: int
max_video_kbps_by_limit: int
limit_total_kbps: int
target_audio_kbps_total: int
def smart_pick_bucket_and_bitrate(
ctx: EncodeContext,
limit_bytes: int,
n_audio_eff: int,
complexity: Complexity,
) -> SmartPlan:
"""
Sceglie bucket (vf_height) e v_kbps "smart" rispettando:
- limite di dimensione (limit_bytes),
- numero di tracce audio,
- floors SMART_FLOORS_KBPS[bucket][complexity],
- MAX_TARGET_HEIGHT,
- non superare il bitrate video stimato della sorgente.
"""
if ctx.duration_s <= 0:
raise ValueError("Durata non valida in smart_pick_bucket_and_bitrate")
# Stima kb/s consentiti dal limite.
limit_total_kbps = int((limit_bytes * 8) / ctx.duration_s / 1000.0)
target_audio_kbps_total = max(0, n_audio_eff * AUDIO_KBPS_PER_TRACK)
max_video_kbps_by_limit = max(
MIN_VIDEO_KBPS,
limit_total_kbps - target_audio_kbps_total
)
logger.info(
"[smart] limite bytes=%s (%d B) → limit_total_kbps≈%d, "
"audio_kbps_total_target=%d, max_v_by_limit≈%d",
_fmt_bytes(limit_bytes),
limit_bytes,
limit_total_kbps,
target_audio_kbps_total,
max_video_kbps_by_limit,
)
# Nessun senso aumentare oltre il sorgente.
max_video_kbps_global = min(max_video_kbps_by_limit, ctx.source_video_kbps_est)
# ---- Candidati bucket (risoluzioni) ----
# Niente upscale: non superiamo ctx.height, e rispettiamo MAX_TARGET_HEIGHT.
max_h = min(ctx.height, MAX_TARGET_HEIGHT)
candidate_heights = sorted(
[h for h in SMART_FLOORS_KBPS.keys() if h <= max_h]
)
if not candidate_heights:
# Se il video è addirittura più piccolo del bucket minimo, usiamo il più basso
# (la tua logica di scale/no-scale si valuterà poi in ffmpeg).
candidate_heights = [min(SMART_FLOORS_KBPS.keys())]
logger.info(
"[smart] candidate buckets (<=%d): %s",
max_h, ", ".join(str(h) for h in candidate_heights)
)
chosen = None
# Proviamo dal bucket più alto verso il basso.
for h in sorted(candidate_heights, reverse=True):
floors = SMART_FLOORS_KBPS[h]
floor_kbps = floors[complexity.value]
# v_kbps massimo che possiamo permetterci su questo bucket.
v_kbps_candidate = max_video_kbps_global
logger.info(
"[smart] bucket %d: floor=%d kb/s (complexity=%s), "
"max_v_global=%d kb/s",
h, floor_kbps, complexity.value, v_kbps_candidate
)
if v_kbps_candidate < floor_kbps:
logger.info(
"[smart] bucket %d scartato: v_kbps_candidate=%d < floor=%d",
h, v_kbps_candidate, floor_kbps
)
continue
# Se arriviamo qui, questo bucket è fattibile.
chosen = SmartPlan(
complexity=complexity,
vf_height=h,
v_kbps=v_kbps_candidate,
floor_kbps=floor_kbps,
max_video_kbps_by_limit=max_video_kbps_by_limit,
limit_total_kbps=limit_total_kbps,
target_audio_kbps_total=target_audio_kbps_total,
)
logger.info(
"[smart] bucket %d fattibile: scelgo v_kbps=%d (≥ floor=%d).",
h, v_kbps_candidate, floor_kbps
)
break
if chosen is None:
# Nessun bucket compatibile col limite → al chiamante la decisione di split.
msg = (
"Nessun bucket compatibile con il limite: "
f"max_video_kbps_global={max_video_kbps_global}, "
f"min_floor={min(f[complexity.value] for f in SMART_FLOORS_KBPS.values())}"
)
logger.info("[smart] %s", msg)
raise EncodeNotFeasible(msg)
return chosen
def smart_plan_for_size(
ctx: EncodeContext,
limit_bytes: int,
n_audio_eff: int,
) -> SmartPlan:
"""
Wrapper: classifica la complessità e poi sceglie bucket+bitrate.
(Per ora usa solo l'euristica bpp_proxy; in un secondo step si
potranno aggiungere SI/TI + VMAF e ricalibrare complexity.)
"""
logger.info(
"[smart] calcolo piano smart per durata=%s (%.0fs), "
"limit_bytes=%s (%d B), n_audio_eff=%d",
_fmt_hms(ctx.duration_s),
ctx.duration_s,
_fmt_bytes(limit_bytes),
limit_bytes,
n_audio_eff,
)
complexity = classify_complexity_heuristic(ctx)
plan = smart_pick_bucket_and_bitrate(
ctx=ctx,
limit_bytes=limit_bytes,
n_audio_eff=n_audio_eff,
complexity=complexity,
)
logger.info(
"[smart] piano scelto: complexity=%s, vf_h=%d, v_kbps=%d, "
"floor=%d, limit_total_kbps=%d, audio_kbps_total=%d",
plan.complexity.value,
plan.vf_height,
plan.v_kbps,
plan.floor_kbps,
plan.limit_total_kbps,
plan.target_audio_kbps_total,
)
return plan
def smart_check_file(file_path: str, chat_id: int, item_id: int, app) -> dict:
"""
Variante 'smart' di check_file che:
- Se originale <= limite: original_ok (come check_file).
- Altrimenti:
- decide quali audio tenere (loggando),
- estrae EncodeContext,
- calcola un piano video 'smart' (complessità, bucket, v_kbps),
- NON lancia encode, ma ritorna il piano.
Serve per testare/loggare il comportamento prima di integrare l'encode reale.
"""
try:
src = Path(file_path)
if not src.exists():
msg = "❌ File non trovato"
try:
send_telegram_message(chat_id, msg)
except Exception:
pass
return {"status": "error", "msg": msg, "path": None}
orig_size = src.stat().st_size
# safe_limit_bytes è globale (come nella tua check_file)
if orig_size <= safe_limit_bytes:
logger.info(
"[smart_check] originale <= limite: size=%s (%d B) <= limit=%s (%d B) — skip encode.",
_fmt_bytes(orig_size),
orig_size,
_fmt_bytes(safe_limit_bytes),
safe_limit_bytes,
)
return {
"status": "original_ok",
"path": str(src),
"size": orig_size,
}
# ---- app / profilo lingua (come in check_file) ----
if app == "sonarr":
api_key = sonarr_api
host = sonarr_host
port = sonarr_port
elif app == "radarr":
api_key = radarr_api
host = radarr_host
port = radarr_port
else:
raise ValueError("app must be 'radarr' or 'sonarr'")
profile_language = get_profile_language_iso639_2(
app, api_key, host, port, item_id
)
# ---- Analisi tracce audio + decisione keep/remove ----
langs = _analyze_audio_languages_with_docker(src)
filter_enabled, keep_relidx, notify_msg, summary_str = _decide_audio_selection(
profile_language, langs
)
logger.info("[smart_audio] Tracce: %s", summary_str)
if notify_msg:
try:
send_telegram_message(
owner_chatid,
f"ℹ️ {src.name}: {notify_msg}"
)
except Exception:
pass
ctx = extract_encode_context(str(src))
# Log keep/remove
if filter_enabled:
keep_1based = [i + 1 for i in sorted(keep_relidx)]
all_1based = [int(t.get("track")) for t in langs]
remove_1based = [
t for t in all_1based if t not in keep_1based
]
logger.info(
"[smart_audio] selection: keep=%s (%d/%d), remove=%s",
", ".join(f"#{i}" for i in keep_1based) if keep_1based else "(none)",
len(keep_1based),
len(langs),
", ".join(f"#{i}" for i in remove_1based)
if remove_1based else "(none)",
)
else:
logger.info(
"[smart_audio] selection: keep=all (%d/%d), remove=(none)",
ctx.n_audio,
ctx.n_audio,
)
dur_txt = _fmt_hms(ctx.duration_s)
# Log sorgente
try:
send_telegram_message(
chat_id,
"[SMART] Analisi compressione GPU (HEVC NVENC)… (solo test piano)"
)
a_src_est = max(0, ctx.source_total_kbps_est - ctx.source_video_kbps_est)
logger.info(
"[smart_src] name=%s, dur=%s (%.0fs), size=%s (%d B), "
"limit=%s (%d B), app=%s, profile_lang=%s",
src.name,
dur_txt,
ctx.duration_s,
_fmt_bytes(orig_size),
orig_size,
_fmt_bytes(safe_limit_bytes),
safe_limit_bytes,
app,
profile_language or "n/a",
)
logger.info(
"[smart_src] video: %dx%d, %s-bit, v_src≈%d kb/s",
ctx.width,
ctx.height,
("10" if ctx.is_10bit else "8"),
ctx.source_video_kbps_est,
)
logger.info(
"[smart_src] audio: tracks=%d, a_src_est≈%d kb/s",
ctx.n_audio,
a_src_est,
)
logger.info(
"[smart_src] container: total_src≈%d kb/s, dur=%s (%.0fs), "
"size=%s (%d B), ext=%s",
ctx.source_total_kbps_est,
dur_txt,
ctx.duration_s,
_fmt_bytes(orig_size),
orig_size,
ctx.ext_low,
)
except Exception:
pass
# ---- Calcolo piano smart (NIENTE ENCODE) ----
limit_try1 = int(safe_limit_bytes * (1.0 - FUDGE_VBR_FIRST_TRY))
n_audio_eff = len(keep_relidx) if filter_enabled else ctx.n_audio
logger.info(
"[smart_plan] uso limit_try1=%s (%d B, fudge=%.2f%%), n_audio_eff=%d",
_fmt_bytes(limit_try1),
limit_try1,
FUDGE_VBR_FIRST_TRY * 100.0,
n_audio_eff,
)
try:
smart_plan = smart_plan_for_size(
ctx=ctx,
limit_bytes=limit_try1,
n_audio_eff=n_audio_eff,
)
except EncodeNotFeasible as e:
# Pianificazione non possibile → segnaliamo che servirà split
logger.info(
"[smart_plan] Encode non fattibile con i floors smart: %s", e
)
try:
send_telegram_message(
chat_id,
f"❌ [SMART] Impossibile rispettare i floor minimi con il limite: {e}\n"
"Suggerito split/gestione a monte."
)
except Exception:
pass
return {
"status": "too_big",
"path": str(src),
"size": orig_size,
"msg": "smart_encode_not_feasible",
}
# --- TEST REALI SOLO PIANO (senza telegram pian, senza smart_final) ---
if SMART_TEST_ENABLED:
logger.info("[smart_test] abilitato: avvio test reali (mini-encode) sul piano smart")
try:
smart_run_tests(
src=src,
ctx=ctx,
smart_plan=smart_plan,
keep_relidx=keep_relidx,
filter_enabled=filter_enabled,
)
except Exception as te:
logger.exception("[smart_test] errore durante i test smart: %s", te)
logger.info("[smart_test] abilitato: test completati, ritorno solo il piano (nessun messaggio Telegram piano / smart_final).")
# Qui NON proseguiamo con i messaggi Telegram e il log [smart_final]
return {
"status": "smart_plan_test",
"path": str(src),
"size": orig_size,
"plan": {
"complexity": smart_plan.complexity.value,
"vf_height": smart_plan.vf_height,
"v_kbps": smart_plan.v_kbps,
"floor_kbps": smart_plan.floor_kbps,
"audio_kbps_total": smart_plan.target_audio_kbps_total,
"limit_total_kbps": smart_plan.limit_total_kbps,
},
}
# ----------------------------------------------------------------------
# Riepilogo finale del piano (solo log, nessun encode)
try:
send_telegram_message(
chat_id,
"ℹ️ [SMART] Piano stimato:\n"
f"- complessità: {smart_plan.complexity.value}\n"
f"- vf_height: {smart_plan.vf_height}\n"
f"- v_kbps: ~{smart_plan.v_kbps} kb/s\n"
f"- audio totale: ~{smart_plan.target_audio_kbps_total} kb/s\n"
f"- totale piano: ~{smart_plan.limit_total_kbps} kb/s (limite fudgiato)"
)
except Exception:
pass
logger.info(
"[smart_final] piano: complexity=%s, vf_h=%d, v_kbps=%d, "
"floor=%d, audio_kbps_total=%d, limit_total_kbps≈%d, "
"safe_limit_bytes=%d (%s)",
smart_plan.complexity.value,
smart_plan.vf_height,
smart_plan.v_kbps,
smart_plan.floor_kbps,
smart_plan.target_audio_kbps_total,
smart_plan.limit_total_kbps,
safe_limit_bytes,
_fmt_bytes(safe_limit_bytes),
)
# Ritorniamo SOLO il piano (nessun file creato/ritoccato)
return {
"status": "smart_plan",
"path": str(src),
"size": orig_size,
"plan": {
"complexity": smart_plan.complexity.value,
"vf_height": smart_plan.vf_height,
"v_kbps": smart_plan.v_kbps,
"floor_kbps": smart_plan.floor_kbps,
"audio_kbps_total": smart_plan.target_audio_kbps_total,
"limit_total_kbps": smart_plan.limit_total_kbps,
},
}
except Exception as e:
logger.exception("[smart_check_file] eccezione: %s", e)
try:
send_telegram_message(chat_id, f"❌ Errore smart_check_file: {e}")
except Exception:
pass
return {"status": "error", "path": None, "msg": str(e)}
def _smart_make_temp_dir(src: Path) -> Path:
"""
Crea una cartella temporanea nello stesso percorso del file sorgente.
Es: /path/file.mkv -> /path/.mtb_smarttmp_file
Se esiste già, la cancella e la ricrea.
"""
base_dir = src.parent
# uso il nome del file senza estensione per distinguere i vari test
temp_name = f".mtb_smarttmp_{src.stem}"
temp_dir = base_dir / temp_name
# se esiste, rimuovo tutto
if temp_dir.exists():
try:
shutil.rmtree(temp_dir)
except Exception as e:
logger.warning("[smart_test] impossibile rimuovere temp_dir esistente %s: %s", temp_dir, e)
try:
temp_dir.mkdir(parents=False, exist_ok=False)
except Exception as e:
logger.warning("[smart_test] impossibile creare temp_dir %s: %s", temp_dir, e)
# provo a usare una sottocartella random dentro base_dir
import uuid
temp_dir = base_dir / f".mtb_smarttmp_{src.stem}_{uuid.uuid4().hex[:8]}"
temp_dir.mkdir(parents=False, exist_ok=False)
logger.info("[smart_test] temp_dir creata: %s", temp_dir)
return temp_dir
def _smart_pick_test_segments(duration_s: float) -> list[tuple[float, float]]:
"""
Ritorna una lista di (t_start, t_dur) in secondi per gli spezzoni di test.
Usa le percentuali SMART_TEST_SEGMENT_ANCHORS (0–100) e SMART_TEST_SEGMENT_SEC.
"""
duration_s = max(1.0, float(duration_s or 0.0))
seg_dur = min(SMART_TEST_SEGMENT_SEC, max(5.0, duration_s / 20.0))
segments: list[tuple[float, float]] = []
for anchor in SMART_TEST_SEGMENT_ANCHORS:
# anchor è percentuale: 10, 50, 90 → 0.1, 0.5, 0.9
t_start = duration_s * (anchor / 100.0)
t_start = max(0.0, min(duration_s - seg_dur, t_start))
t_start = round(t_start, 3)
segments.append((t_start, round(seg_dur, 3)))
# Elimina duplicati grossolani (se il video è corto)
unique_segments: list[tuple[float, float]] = []
seen = set()
for t_start, t_dur in segments:
key = int(t_start)
if key in seen:
continue
seen.add(key)
unique_segments.append((t_start, t_dur))
logger.info("[smart_test] segments scelti (t_start, t_dur): %s", unique_segments)
return unique_segments
def _nvenc_cmd_segment(
file_path: str,
out_path: str,
vkbps: int,
vf_h: int,
is_10bit: bool, # non usato qui ma tenuto per coerenza
n_audio: int,
ext_low: str,
t_start: float,
t_dur: float,
audio_map_relidx: list[int] | None = None,
force_stereo: bool = FORCE_STEREO,
use_dplii_matrix: bool = True,
keep_subtitles: bool = True,
) -> list[str]:
"""
Come _nvenc_cmd, ma limitato a uno spezzone [t_start, t_start+t_dur].
Usa HEVC NVENC e lo stesso mapping audio/sottotitoli della produzione.
"""
cmd = _docker_ffmpeg_args(use_gpu=True) + [
"-y", "-nostdin", "-hide_banner", "-loglevel", "error",
"-hwaccel", "cuda",
"-ss", str(t_start),
"-i", file_path,
"-t", str(t_dur),
"-vf", f"scale=-2:{vf_h}:force_original_aspect_ratio=decrease",
]
# --- Mapping ---
# Video principale
cmd += ["-map", "0:v:0"]
# Audio: o filtro esplicito o tutte
if audio_map_relidx is not None:
cmd += ["-map", "-0:a"] # droppa tutte le audio
for idx in sorted(audio_map_relidx):
cmd += ["-map", f"0:a:{idx}"]
eff_n_audio = len(audio_map_relidx)
else:
cmd += ["-map", "0:a?"]
eff_n_audio = n_audio
# Sottotitoli (se richiesto)
if keep_subtitles:
cmd += ["-map", "0:s?"]
eff_n_subs = 1
else:
eff_n_subs = 0
# --- Video (stessa logica della produzione) ---
cmd += [
"-c:v", "hevc_nvenc",
"-rc", "vbr_hq", "-multipass", "fullres",
"-b:v", f"{int(vkbps)}k",
"-maxrate", f"{int(vkbps * 1.08)}k",
"-bufsize", f"{int(vkbps * 2.0)}k",
"-rc-lookahead", "32",
"-spatial_aq", "1", "-temporal_aq", "1", "-aq-strength", "8",
"-bf", "3", "-b_ref_mode", "middle", "-g", "240",
]
# --- Audio ---
if eff_n_audio > 0:
cmd += ["-c:a", "libfdk_aac"]
cmd += ["-b:a", f"{AUDIO_KBPS_PER_TRACK}k"]
for i in range(eff_n_audio):
cmd += [f"-b:a:{i}", f"{AUDIO_KBPS_PER_TRACK}k"]
if force_stereo:
cmd += [f"-ac:a:{i}", "2"]
if use_dplii_matrix:
cmd += [f"-filter:a:{i}", "aresample=matrix_encoding=dplii"]
cmd += [f"-metadata:s:a:{i}", f"title=AAC LC 2.0 {AUDIO_KBPS_PER_TRACK} kbps"]
cmd += ["-disposition:a:0", "default"]
for i in range(1, eff_n_audio):
cmd += [f"-disposition:a:{i}", "0"]
else:
cmd += ["-an"]
# --- Sottotitoli ---
if keep_subtitles and eff_n_subs:
cmd += ["-c:s", "copy"]
cmd += ["-disposition:s", "0"]
# --- Metadati/Capitoli del container ---
cmd += ["-map_metadata", "-1"]
# Output
cmd.append(out_path)
return cmd
def smart_run_tests(
src: Path,
ctx: EncodeContext,
smart_plan: SmartPlan,
keep_relidx: list[int],
filter_enabled: bool,
) -> None:
"""
Esegue mini-encode di test in una cartella temporanea accanto al file.
Usa il piano smart (vf_height, v_kbps) e la stessa selezione audio.
Alla fine cancella la cartella temporanea.
"""
temp_dir = _smart_make_temp_dir(src)
segments = _smart_pick_test_segments(ctx.duration_s)
# Determina audio_map_relidx da passare a ffmpeg
audio_map_relidx = keep_relidx if filter_enabled and keep_relidx else None
try:
for idx, (t_start, t_dur) in enumerate(segments, start=1):
out_path = temp_dir / f"smarttest_{idx:02d}.mkv"
cmd = _nvenc_cmd_segment(
file_path=str(src),
out_path=str(out_path),
vkbps=int(smart_plan.v_kbps),
vf_h=int(smart_plan.vf_height),
is_10bit=ctx.is_10bit,
n_audio=ctx.n_audio,
ext_low=ctx.ext_low,
t_start=t_start,
t_dur=t_dur,
audio_map_relidx=audio_map_relidx,
)
logger.info(
"[smart_test] sample#%d: t_start=%.2fs, t_dur=%.2fs, "
"vf_h=%d, v_kbps=%d, out=%s",
idx, t_start, t_dur,
smart_plan.vf_height, smart_plan.v_kbps,
out_path,
)
rc, elapsed, stderr = _run(cmd)
if rc != 0:
tail = (stderr[-400:] if stderr else "")
logger.error(
"[smart_test] sample#%d FAILED: rc=%s, elapsed=%.1fs, stderr_tail=%s",
idx, rc, elapsed, tail
)
else:
try:
out_size = os.path.getsize(out_path)
except Exception:
out_size = -1
logger.info(
"[smart_test] sample#%d OK: rc=0, elapsed=%.1fs, size=%s (%d B)",
idx, elapsed, _fmt_bytes(out_size), out_size
)
finally:
# cleanup della cartella temporanea
try:
shutil.rmtree(temp_dir, ignore_errors=True)
logger.info("[smart_test] cleanup temp_dir: %s", temp_dir)
except Exception as e:
logger.warning("[smart_test] errore rimozione temp_dir %s: %s", temp_dir, e)
