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)