hbs2/test
0
1subgen_version = '2026.06.3'2 3"""4ENVIRONMENT VARIABLES DOCUMENTATION5 6This application supports both new standardized environment variable names and legacy names for backwards compatibility. The new names follow a consistent naming convention: 7 8STANDARDIZED NAMING CONVENTION:9- Use UPPERCASE with underscores for separation10- Group related variables with consistent prefixes: 11 * PLEX_* for Plex server integration12 * JELLYFIN_* for Jellyfin server integration13 * PROCESS_* for media processing triggers14 * SKIP_* for all skip conditions15 * SUBTITLE_* for subtitle-related settings16 * WHISPER_* for Whisper model settings17 * TRANSCRIBE_* for transcription settings18 19BACKWARDS COMPATIBILITY: 20Legacy environment variable names are still supported. If both new and old names are set,21the new standardized name takes precedence. 22 23NEW NAME → OLD NAME (for backwards compatibility):24- PLEX_TOKEN → PLEXTOKEN25- PLEX_SERVER → PLEXSERVER26- JELLYFIN_TOKEN → JELLYFINTOKEN27- JELLYFIN_SERVER → JELLYFINSERVER28- PROCESS_ADDED_MEDIA → PROCADDEDMEDIA29- PROCESS_MEDIA_ON_PLAY → PROCMEDIAONPLAY30- SUBTITLE_LANGUAGE_NAME → NAMESUBLANG31- WEBHOOK_PORT → WEBHOOKPORT32- SKIP_IF_EXTERNAL_SUBTITLES_EXIST → SKIPIFEXTERNALSUB33- SKIP_IF_TARGET_SUBTITLES_EXIST → SKIP_IF_TO_TRANSCRIBE_SUB_ALREADY_EXIST34- SKIP_IF_INTERNAL_SUBTITLES_LANGUAGE → SKIPIFINTERNALSUBLANG35- SKIP_SUBTITLE_LANGUAGES → SKIP_LANG_CODES36- SKIP_IF_AUDIO_LANGUAGES → SKIP_IF_AUDIO_TRACK_IS37- SKIP_ONLY_SUBGEN_SUBTITLES → ONLY_SKIP_IF_SUBGEN_SUBTITLE38- SKIP_IF_NO_LANGUAGE_BUT_SUBTITLES_EXIST → SKIP_IF_LANGUAGE_IS_NOT_SET_BUT_SUBTITLES_EXIST39 40MIGRATION GUIDE:41Users can gradually migrate to the new names. Both will work simultaneously during the42transition period. The old names may be deprecated in future versions. 43"""44 45import ast46import asyncio47import ctypes48import ctypes.util49import gc50import hashlib51import json52import logging53import os54import queue55import subprocess56import sys57import threading58import time59import xml.etree.ElementTree as ET60from contextlib import asynccontextmanager61from datetime import datetime62from threading import Event, Lock, Timer63from typing import List, Union64 65import av66import faster_whisper67import ffmpeg68import numpy as np69import requests70import stable_whisper71import torch72from fastapi import Body, FastAPI, File, Form, Header, Query, Request, UploadFile73from fastapi.responses import StreamingResponse74from stable_whisper import Segment75from watchdog.events import FileSystemEventHandler76from watchdog.observers.polling import PollingObserver as Observer77 78from language_code import LanguageCode79 80 81def convert_to_bool(in_bool):82 # Convert the input to string and lower case, then check against true values83 return str(in_bool).lower() in ('true', 'on', '1', 'y', 'yes')84 85def get_env_with_fallback(new_name: str, old_name: str, default_value=None, convert_func=None):86 """87 Get environment variable with backwards compatibility fallback.88 89 Args:90 new_name: The new standardized environment variable name91 old_name: The legacy environment variable name for backwards compatibility92 default_value: Default value if neither variable is set93 convert_func: Optional function to convert the value (e.g., convert_to_bool, int)94 95 Returns:96 The environment variable value, converted if convert_func is provided97 """98 # Try new name first, then fall back to old name99 value = os.getenv(new_name) or os.getenv(old_name)100 101 if value is None:102 value = default_value103 104 # Apply conversion function if provided105 if convert_func and value is not None:106 return convert_func(value)107 108 return value109 110# Server Integration - with backwards compatibility111plextoken = get_env_with_fallback('PLEX_TOKEN', 'PLEXTOKEN', 'token here')112plexserver = get_env_with_fallback('PLEX_SERVER', 'PLEXSERVER', 'http://192.168.1.111:32400')113jellyfintoken = get_env_with_fallback('JELLYFIN_TOKEN', 'JELLYFINTOKEN', 'token here')114jellyfinserver = get_env_with_fallback('JELLYFIN_SERVER', 'JELLYFINSERVER', 'http://192.168.1.111:8096')115 116# Whisper Configuration117whisper_model = os.getenv('WHISPER_MODEL', 'distil-small.en')118whisper_threads = int(os.getenv('WHISPER_THREADS', 15))119concurrent_transcriptions = int(os.getenv('CONCURRENT_TRANSCRIPTIONS', 1))120transcribe_device = os.getenv('TRANSCRIBE_DEVICE', 'gpu')121 122# Processing Control - with backwards compatibility123procaddedmedia = get_env_with_fallback('PROCESS_ADDED_MEDIA', 'PROCADDEDMEDIA', True, convert_to_bool)124procmediaonplay = get_env_with_fallback('PROCESS_MEDIA_ON_PLAY', 'PROCMEDIAONPLAY', True, convert_to_bool)125 126# Subtitle Configuration - with backwards compatibility127subtitle_language_name = get_env_with_fallback('SUBTITLE_LANGUAGE_NAME', 'NAMESUBLANG', '')128 129# System Configuration - with backwards compatibility130webhookport = get_env_with_fallback('WEBHOOK_PORT', 'WEBHOOKPORT', 9000, int)131word_level_highlight = convert_to_bool(os.getenv('WORD_LEVEL_HIGHLIGHT', False))132debug = convert_to_bool(os.getenv('DEBUG', True))133use_path_mapping = convert_to_bool(os.getenv('USE_PATH_MAPPING', False))134path_mapping_from = os.getenv('PATH_MAPPING_FROM', r'/tv')135path_mapping_to = os.getenv('PATH_MAPPING_TO', r'/Volumes/TV')136model_location = os.getenv('MODEL_PATH', './models')137monitor = convert_to_bool(os.getenv('MONITOR', False))138transcribe_folders = os.getenv('TRANSCRIBE_FOLDERS', '')139transcribe_or_translate = os.getenv('TRANSCRIBE_OR_TRANSLATE', 'transcribe').lower()140clear_vram_on_complete = convert_to_bool(os.getenv('CLEAR_VRAM_ON_COMPLETE', True))141compute_type = os.getenv('COMPUTE_TYPE', 'auto')142append = convert_to_bool(os.getenv('APPEND', False))143reload_script_on_change = convert_to_bool(os.getenv('RELOAD_SCRIPT_ON_CHANGE', False))144lrc_for_audio_files = convert_to_bool(os.getenv('LRC_FOR_AUDIO_FILES', True))145custom_regroup = os.getenv('CUSTOM_REGROUP', 'cm_sl=84_sl=42++++++1')146detect_language_length = int(os.getenv('DETECT_LANGUAGE_LENGTH', 30))147detect_language_offset = int(os.getenv('DETECT_LANGUAGE_OFFSET', 0))148model_cleanup_delay = int(os.getenv('MODEL_CLEANUP_DELAY', 30))149asr_timeout = int(os.getenv('ASR_TIMEOUT', 18000))150webhook_url_completed = os.getenv('WEBHOOK_URL_COMPLETED', '')151 152# Skip Configuration - with backwards compatibility153skip_if_external_sub_exists = get_env_with_fallback('SKIP_IF_EXTERNAL_SUBTITLES_EXIST', 'SKIPIFEXTERNALSUB', False, convert_to_bool)154skip_if_target_subtitle_exists = get_env_with_fallback('SKIP_IF_TARGET_SUBTITLES_EXIST', 'SKIP_IF_TO_TRANSCRIBE_SUB_ALREADY_EXIST', True, convert_to_bool)155skip_if_internal_sub_language = LanguageCode.from_string(get_env_with_fallback('SKIP_IF_INTERNAL_SUBTITLES_LANGUAGE', 'SKIPIFINTERNALSUBLANG', ''))156ignore_forced_subtitles = convert_to_bool(os.getenv('IGNORE_FORCED_SUBTITLES', True))157plex_queue_next_episode = convert_to_bool(os.getenv('PLEX_QUEUE_NEXT_EPISODE', False))158plex_queue_season = convert_to_bool(os.getenv('PLEX_QUEUE_SEASON', False))159plex_queue_series = convert_to_bool(os.getenv('PLEX_QUEUE_SERIES', False))160# Language and Skip Configuration - with backwards compatibility161skip_subtitle_languages = ([LanguageCode.from_string(code) for code in get_env_with_fallback('SKIP_SUBTITLE_LANGUAGES', 'SKIP_LANG_CODES', '').split("|")]162 if get_env_with_fallback('SKIP_SUBTITLE_LANGUAGES', 'SKIP_LANG_CODES')163 else[]164)165force_detected_language_to = LanguageCode.from_string(os.getenv('FORCE_DETECTED_LANGUAGE_TO', ''))166preferred_audio_languages =[167 LanguageCode.from_string(code) 168 for code in os.getenv('PREFERRED_AUDIO_LANGUAGES', 'eng').split("|")169] # in order of preference170limit_to_preferred_audio_languages = convert_to_bool(os.getenv('LIMIT_TO_PREFERRED_AUDIO_LANGUAGE', False))171skip_audio_languages = ([LanguageCode.from_string(code) for code in get_env_with_fallback('SKIP_IF_AUDIO_LANGUAGES', 'SKIP_IF_AUDIO_TRACK_IS', '').split("|")]172 if get_env_with_fallback('SKIP_IF_AUDIO_LANGUAGES', 'SKIP_IF_AUDIO_TRACK_IS')173 else[]174)175 176# Additional Subtitle Configuration - with backwards compatibility177subtitle_language_naming_type = os.getenv('SUBTITLE_LANGUAGE_NAMING_TYPE', 'ISO_639_2_B')178only_match_subgen_subtitles = get_env_with_fallback('SKIP_ONLY_SUBGEN_SUBTITLES', 'ONLY_SKIP_IF_SUBGEN_SUBTITLE', False, convert_to_bool)179skip_unknown_language = convert_to_bool(os.getenv('SKIP_UNKNOWN_LANGUAGE', False))180skip_if_no_audio_language_but_subtitles_exist = get_env_with_fallback('SKIP_IF_NO_LANGUAGE_BUT_SUBTITLES_EXIST', 'SKIP_IF_LANGUAGE_IS_NOT_SET_BUT_SUBTITLES_EXIST', False, convert_to_bool)181ignore_forced_subtitles = convert_to_bool(os.getenv('IGNORE_FORCED_SUBTITLES', True))182should_whisper_detect_audio_language = convert_to_bool(os.getenv('SHOULD_WHISPER_DETECT_AUDIO_LANGUAGE', False))183show_in_subname_subgen = convert_to_bool(os.getenv('SHOW_IN_SUBNAME_SUBGEN', True))184show_in_subname_model = convert_to_bool(os.getenv('SHOW_IN_SUBNAME_MODEL', True))185 186# Advanced Configuration187try:188 kwargs = ast.literal_eval(os.getenv('SUBGEN_KWARGS', '{}') or '{}')189except ValueError:190 kwargs = {}191 logging.info("kwargs (SUBGEN_KWARGS) is an invalid dictionary, defaulting to empty '{}'")192 193if transcribe_device == "gpu":194 transcribe_device = "cuda"195 196VIDEO_EXTENSIONS = (197 ".mp4", ".mkv", ".avi", ".mov", ".wmv", ".flv", ".webm", ".mpg", ".mpeg", 198 ".3gp", ".ogv", ".vob", ".rm", ".rmvb", ".ts", ".m4v", ".f4v", ".svq3", 199 ".asf", ".m2ts", ".divx", ".xvid"200)201 202AUDIO_EXTENSIONS = (203 ".mp3", ".wav", ".aac", ".flac", ".ogg", ".wma", ".alac", ".m4a", ".opus", 204 ".aiff", ".aif", ".pcm", ".ra", ".ram", ".mid", ".midi", ".ape", ".wv", 205 ".amr", ".vox", ".tak", ".spx", ".m4b", ".mka"206)207 208@asynccontextmanager209async def lifespan(app: FastAPI):210 if transcribe_folders:211 threading.Thread(target=transcribe_existing, args=(transcribe_folders,), daemon=True).start()212 yield213 214app = FastAPI(lifespan=lifespan)215 216model = None217model_cleanup_timer = None218model_cleanup_lock = Lock()219 220# Locks to ensure thread-safety during concurrent AI operations221model_load_lock = Lock()222active_direct_tasks = 0223active_direct_tasks_lock = Lock()224 225in_docker = os.path.exists('/.dockerenv')226docker_status = "Docker" if in_docker else "Standalone"227 228# ============================================================================229# TASK RESULT STORAGE (for blocking endpoints)230# ============================================================================231 232class TaskResult:233 """Stores the result of a queued task for blocking retrieval"""234 def __init__(self):235 self.result = None236 self.error = None237 self.done = Event()238 239 def set_result(self, result):240 self.result = result241 self.done.set()242 243 def set_error(self, error):244 self.error = error245 self.done.set()246 247 def wait(self, timeout=None):248 """Block until result is ready. Returns True if completed, False if timeout."""249 return self.done.wait(timeout)250 251# Dictionary to store task results keyed by task_id252# Entries are cleaned up in /asr endpoint finally block to prevent unbounded growth253task_results = {}254task_results_lock = Lock()255 256# ============================================================================257# HASH GENERATION FOR DEDUPLICATION258# ============================================================================259 260def generate_audio_hash(audio_content: bytes, task: str = None, language: str = None) -> str:261 """262 Generate a deterministic hash from audio content and optional parameters. 263 264 Same audio + same task + same language = always same hash. 265 This ensures duplicate requests are caught by the queue. 266 267 Args:268 audio_content: Raw audio bytes from uploaded file269 task: Optional task type ('transcribe' or 'translate')270 language: Optional target language code271 272 Returns: 273 SHA256 hash (first 16 chars for brevity in logs)274 """275 hash_input = audio_content276 277 # Include task and language for fine-grained deduplication278 if task:279 hash_input += task.encode('utf-8')280 if language:281 hash_input += language.encode('utf-8')282 283 full_hash = hashlib.sha256(hash_input).hexdigest()284 return full_hash[:16] # Use first 16 chars for shorter IDs in logs285 286# ============================================================================287# REFACTORED DEDUPLICATED QUEUE WITH BETTER TRACKING288# ============================================================================289 290class DeduplicatedQueue(queue.PriorityQueue):291 """Queue that prevents duplicates, handles priority, and tracks status."""292 def __init__(self):293 super().__init__()294 self._queued = set() # Tracks task IDs waiting in queue295 self._processing = set() # Tracks task IDs currently being handled296 self._lock = Lock()297 298 def put(self, item, block=True, timeout=None):299 with self._lock:300 task_id = item["path"]301 if task_id not in self._queued and task_id not in self._processing:302 # Priority: 0 (Detect), 1 (ASR), 2 (Transcribe)303 task_type = item.get("type", "transcribe")304 priority = 0 if task_type == "detect_language" else (1 if task_type == "asr" else 2)305 306 # PriorityQueue requires a tuple: (priority, tie_breaker, item)307 super().put((priority, time.time(), item), block, timeout)308 self._queued.add(task_id)309 return True310 return False311 312 def get(self, block=True, timeout=None):313 # PriorityQueue returns the tuple, we want just the item314 priority, timestamp, item = super().get(block, timeout)315 with self._lock:316 task_id = item["path"]317 self._queued.discard(task_id)318 self._processing.add(task_id)319 return item320 321 def mark_done(self, item):322 with self._lock:323 task_id = item["path"]324 self._processing.discard(task_id)325 326 def is_idle(self):327 with self._lock:328 return self.empty() and len(self._processing) == 0329 330 def is_active(self, task_id):331 """Checks if a task_id is currently queued or processing."""332 with self._lock:333 return task_id in self._queued or task_id in self._processing334 335 def get_queued_tasks(self):336 with self._lock:337 return list(self._queued)338 339 def get_processing_tasks(self):340 with self._lock:341 return list(self._processing)342 343# Start queue344task_queue = DeduplicatedQueue()345 346# ============================================================================347# TRANSCRIPTION WORKER348# ============================================================================349 350def transcription_worker():351 """Main worker thread with centralized logging and status tracking."""352 while True:353 task = None354 next_task = None355 try:356 task = task_queue.get(block=True, timeout=1)357 task_type = task.get("type", "transcribe")358 path = task.get("path", "unknown")359 display_name = os.path.basename(path) if ("/" in str(path) or "\\" in str(path)) else path360 361 # Status for START log362 proc_count = len(task_queue.get_processing_tasks())363 queue_count = len(task_queue.get_queued_tasks())364 logging.info(f"WORKER START :[{task_type.upper():<10}] {display_name:^40} | Jobs: {proc_count} processing, {queue_count} queued")365 366 start_time = time.time()367 if task_type == "detect_language": 368 if "audio_content" in task: 369 detect_language_from_upload(task)370 else: 371 # Capture the transcription task to queue later372 next_task = detect_language_task(task['path'], original_task_data=task)373 elif task_type == "asr":374 asr_task_worker(task)375 else: # transcribe376 gen_subtitles(task['path'], task['transcribe_or_translate'], task['force_language'], audio_tracks=task.get('audio_tracks'))377 378 # --- METADATA REFRESH LOGIC ---379 if 'plex_item_id' in task:380 try:381 logging.info(f"Refreshing Plex Metadata for item {task['plex_item_id']}")382 refresh_plex_metadata(task['plex_item_id'], task['plex_server'], task['plex_token'])383 except Exception as e:384 logging.error(f"Failed to refresh Plex metadata: {e}")385 386 if 'jellyfin_item_id' in task:387 try:388 logging.info(f"Refreshing Jellyfin Metadata for item {task['jellyfin_item_id']}")389 refresh_jellyfin_metadata(task['jellyfin_item_id'], task['jellyfin_server'], task['jellyfin_token'])390 except Exception as e:391 logging.error(f"Failed to refresh Jellyfin metadata: {e}")392 # ------------------------------393 394 # Status for FINISH log395 elapsed = time.time() - start_time396 m, s = divmod(int(elapsed), 60)397 remaining_queued = len(task_queue.get_queued_tasks())398 logging.info(f"WORKER FINISH: [{task_type.upper():<10}] {display_name:^40} in {m}m {s}s | Remaining: {remaining_queued} queued")399 400 except queue.Empty:401 continue402 except Exception as e:403 logging.error(f"Error processing task: {e}", exc_info=True)404 finally:405 if task:406 task_queue.task_done()407 task_queue.mark_done(task)408 409 # Now that the detect task is removed from processing, it's safe to queue the transcription410 if next_task:411 if task_queue.put(next_task):412 logging.debug(f"Queued transcription for detected language: {next_task['path']}")413 else:414 logging.debug(f"Transcription already queued/processing for: {next_task['path']}")415 416 delete_model()417 418# Create worker threads419for _ in range(concurrent_transcriptions):420 threading.Thread(target=transcription_worker, daemon=True).start()421 422# Define a filter class to hide common logging we don't want to see423class MultiplePatternsFilter(logging.Filter):424 def filter(self, record):425 # Define the patterns to search for426 patterns =[427 "Compression ratio threshold is not met",428 "Processing segment at",429 "Log probability threshold is",430 "Reset prompt",431 "Attempting to release",432 "released on ",433 "Attempting to acquire",434 "acquired on",435 "header parsing failed",436 "timescale not set",437 "misdetection possible",438 "srt was added",439 "doesn't have any audio to transcribe",440 "Calling on_"441 ]442 # Return False if any of the patterns are found, True otherwise443 return not any(pattern in record.getMessage() for pattern in patterns)444 445# Configure logging446if debug:447 level = logging.DEBUG448else:449 level = logging.INFO450 451logging.basicConfig(452 stream=sys.stderr, 453 level=level, 454 format="%(asctime)s %(levelname)s: %(message)s",455 datefmt="%Y-%m-%d %H:%M:%S" # This removes the ,123 part456)457 458# Get the root logger459logger = logging.getLogger()460logger.setLevel(level) # Set the logger level461 462for handler in logger.handlers:463 handler.addFilter(MultiplePatternsFilter())464 465logging.getLogger("multipart").setLevel(logging.WARNING)466logging.getLogger("urllib3").setLevel(logging.WARNING)467logging.getLogger("watchfiles").setLevel(logging.WARNING)468logging.getLogger("asyncio").setLevel(logging.WARNING)469logging.getLogger("httpcore").setLevel(logging.WARNING)470logging.getLogger("httpx").setLevel(logging.WARNING)471logging.getLogger("huggingface_hub").setLevel(logging.WARNING)472 473 474class ProgressHandler:475 def __init__(self, filename):476 self.filename = filename477 self.start_time = time.time()478 self.last_print_time = 0479 self.interval = 5480 481 @staticmethod482 def _fmt_t(seconds):483 """Format seconds as [H:]MM:SS without milliseconds."""484 m, s = divmod(int(seconds), 60)485 h, m = divmod(m, 60)486 if h > 0:487 return f"{h}:{m:02d}:{s:02d}"488 return f"{m:02d}:{s:02d}"489 490 def __call__(self, seek, total):491 if docker_status == 'Docker' or debug:492 current_time = time.time()493 if self.last_print_time == 0 or (current_time - self.last_print_time) >= self.interval:494 self.last_print_time = current_time495 496 pct = int((seek / total) * 100) if total > 0 else 0497 elapsed = current_time - self.start_time498 speed = seek / elapsed if elapsed > 0 else 0499 eta = (total - seek) / speed if speed > 0 else 0500 501 proc = len(task_queue.get_processing_tasks())502 queued = len(task_queue.get_queued_tasks())503 504 clean_name = (self.filename[:37] + '..') if len(self.filename) > 40 else self.filename505 506 logging.info(507 f"[ {clean_name:<40}] {pct:>3}% | "508 f"{int(seek):>5}/{int(total):<5}s "509 f"[{self._fmt_t(elapsed):>5}<{self._fmt_t(eta):>5}, {speed:>5.2f}s/s] | "510 f"Jobs: {proc} processing, {queued} queued"511 )512 513TIME_OFFSET = 5514 515def appendLine(result):516 if append:517 lastSegment = result.segments[-1]518 date_time_str = datetime.now().strftime("%d %b %Y - %H:%M:%S")519 appended_text = f"Transcribed by whisperAI with faster-whisper ({whisper_model}) on {date_time_str}"520 521 # Create a new segment with the updated information522 newSegment = Segment(523 start=lastSegment.start + TIME_OFFSET,524 end=lastSegment.end + TIME_OFFSET,525 text=appended_text,526 words=[], # Empty list for words527 id=lastSegment.id + 1528 )529 530 # Append the new segment to the result's segments531 result.segments.append(newSegment)532 533@app.get("/plex")534@app.get("/webhook")535@app.get("/jellyfin")536@app.get("/asr")537@app.get("/emby")538@app.get("/detect-language")539@app.get("/tautulli")540def handle_get_request(request: Request):541 return {"You accessed this request incorrectly via a GET request. See https://github.com/McCloudS/subgen for proper configuration"}542 543@app.get("/")544def webui():545 return {"The webui for configuration was removed on 1 October 2024, please configure via environment variables or in your Docker settings. "}546 547@app.get("/status")548def status():549 return {"version": f"Subgen {subgen_version}, stable-ts {stable_whisper.__version__}, faster-whisper {faster_whisper.__version__} ({docker_status})"}550 551@app.post("/tautulli")552def receive_tautulli_webhook(553 source: Union[str, None] = Header(None),554 event: str = Body(None),555 file: str = Body(None),556):557 if source == "Tautulli":558 logging.debug(f"Tautulli event detected is: {event}")559 if((event == "added" and procaddedmedia) or (event == "played" and procmediaonplay)):560 fullpath = file561 logging.debug(f"Full file path: {fullpath}")562 563 gen_subtitles_queue(path_mapping(fullpath), transcribe_or_translate)564 else:565 return {566 "message": "This doesn't appear to be a properly configured Tautulli webhook, please review the instructions again!"}567 568 return ""569 570@app.post("/plex")571def receive_plex_webhook(572 user_agent: Union[str] = Header(None),573 payload: Union[str] = Form(),574):575 try:576 plex_json = json.loads(payload)577 if "PlexMediaServer" not in user_agent:578 return {"message": "This doesn't appear to be a properly configured Plex webhook, please review the instructions again"}579 580 event = plex_json["event"]581 logging.debug(f"Plex event detected is: {event}")582 583 if (event == "library.new" and procaddedmedia) or (event == "media.play" and procmediaonplay):584 rating_key = plex_json['Metadata']['ratingKey']585 fullpath = get_plex_file_name(rating_key, plexserver, plextoken)586 logging.debug(f"Full file path: {fullpath}")587 588 # Queue the current item with its specific ID for refreshing589 gen_subtitles_queue(590 path_mapping(fullpath), 591 transcribe_or_translate, 592 plex_item_id=rating_key, 593 plex_server=plexserver, 594 plex_token=plextoken595 )596 597 # Note: refresh_plex_metadata is removed here; it is now handled by the worker thread.598 599 if plex_queue_next_episode:600 next_key = get_next_plex_episode(plex_json['Metadata']['ratingKey'], stay_in_season=False)601 if next_key:602 next_file = get_plex_file_name(next_key, plexserver, plextoken)603 gen_subtitles_queue(604 path_mapping(next_file), 605 transcribe_or_translate,606 plex_item_id=next_key, # Pass the NEXT ID so it refreshes when done607 plex_server=plexserver,608 plex_token=plextoken609 )610 611 if plex_queue_series or plex_queue_season:612 current_rating_key = plex_json['Metadata']['ratingKey']613 stay_in_season = plex_queue_season # Determine if we're staying in the season or not614 615 while current_rating_key is not None:616 try:617 # Queue the current episode618 file_path = path_mapping(get_plex_file_name(current_rating_key, plexserver, plextoken))619 620 gen_subtitles_queue(621 file_path, 622 transcribe_or_translate,623 plex_item_id=current_rating_key, # Pass the specific loop ID for refreshing624 plex_server=plexserver,625 plex_token=plextoken626 )627 628 logging.debug(f"Queued episode with ratingKey {current_rating_key}")629 630 # Get the next episode631 next_episode_rating_key = get_next_plex_episode(current_rating_key, stay_in_season=stay_in_season)632 if next_episode_rating_key is None:633 break # Exit the loop if no next episode634 current_rating_key = next_episode_rating_key635 636 except Exception as e:637 logging.error(f"Error processing episode with ratingKey {current_rating_key} or reached end of series: {e}")638 break # Stop processing on error639 640 logging.info("All episodes in the series (or season) have been queued.")641 642 except Exception as e:643 logging.error(f"Failed to process Plex webhook: {e}")644 645 return ""646 647@app.post("/jellyfin")648def receive_jellyfin_webhook(649 user_agent: str = Header(None),650 NotificationType: str = Body(None),651 file: str = Body(None),652 ItemId: str = Body(None),653):654 if "Jellyfin-Server" in user_agent:655 logging.debug(f"Jellyfin event detected is: {NotificationType}")656 logging.debug(f"itemid is: {ItemId}")657 658 if (NotificationType == "ItemAdded" and procaddedmedia) or (NotificationType == "PlaybackStart" and procmediaonplay):659 fullpath = get_jellyfin_file_name(ItemId, jellyfinserver, jellyfintoken)660 logging.debug(f"Full file path: {fullpath}")661 662 # Queue item with Jellyfin metadata ID for delayed refresh663 gen_subtitles_queue(664 path_mapping(fullpath), 665 transcribe_or_translate,666 jellyfin_item_id=ItemId,667 jellyfin_server=jellyfinserver,668 jellyfin_token=jellyfintoken669 )670 671 # Note: refresh_jellyfin_metadata removed here; handled by worker.672 else:673 return {674 "message": "This doesn't appear to be a properly configured Jellyfin webhook, please review the instructions again!"}675 676 return ""677 678@app.post("/emby")679def receive_emby_webhook(680 user_agent: Union[str, None] = Header(None),681 data: Union[str, None] = Form(None),682):683 if not data:684 return ""685 686 data_dict = json.loads(data)687 event = data_dict['Event']688 logging.debug("Emby event detected is: " + event)689 690 # Check if it's a notification test event691 if event == "system.notificationtest":692 logging.info("Emby test message received!")693 return {"message": "Notification test received successfully!"}694 695 if (event == "library.new" and procaddedmedia) or (event == "playback.start" and procmediaonplay):696 fullpath = data_dict['Item']['Path']697 logging.debug(f"Full file path: {fullpath}")698 gen_subtitles_queue(path_mapping(fullpath), transcribe_or_translate)699 700 return ""701 702@app.post("/batch")703def batch(704 directory: str = Query(...),705 forceLanguage: Union[str, None] = Query(default=None)706):707 transcribe_existing(directory, LanguageCode.from_string(forceLanguage))708 709# ============================================================================710# REFACTORED /ASR ENDPOINT WITH HASH-BASED DEDUPLICATION AND BLOCKING711# ============================================================================712 713@app.post("/asr")714async def asr(715 task: Union[str, None] = Query(default="transcribe", enum=["transcribe", "translate"]),716 language: Union[str, None] = Query(default=None),717 video_file: Union[str, None] = Query(default=None),718 initial_prompt: Union[str, None] = Query(default=None),719 audio_file: UploadFile = File(...),720 encode: bool = Query(default=True, description="Encode audio first through ffmpeg"),721 output: Union[str, None] = Query(default="srt", enum=["txt", "vtt", "srt", "tsv", "json"]),722 word_timestamps: bool = Query(default=False, description="Word-level timestamps"),723):724 """725 ASR endpoint that uses audio content hash for deduplication. 726 BLOCKS until processing is complete, then returns the result.727 728 If identical audio + task + language is already being processed,729 waits for that task to complete and returns the same result.730 """731 task_id = None732 733 try:734 logging.info(735 f"ASR {task.capitalize()} received for file '{video_file}'" 736 if video_file 737 else f"ASR {task.capitalize()} received"738 )739 740 # Read audio file content into memory741 file_content = await audio_file.read()742 743 if not file_content:744 await audio_file.close()745 return {746 "status": "error",747 "message": "Audio file is empty"748 }749 750 # Generate deterministic hash from audio (and optionally task/language)751 audio_hash = generate_audio_hash(file_content, task, language)752 753 # FIX: Use video file path if available to match TRANSCRIBE tasks754 if video_file:755 task_id = path_mapping(video_file)756 logging.debug(f"Using mapped video file path as task ID for ASR request: {task_id}")757 else:758 task_id = f"asr-{audio_hash}"759 logging.debug(f"Generated audio hash: {audio_hash} for ASR request")760 761 # Handle forced language762 final_language = language763 if force_detected_language_to: 764 final_language = force_detected_language_to.to_iso_639_1()765 logging.info(f"Forcing detected language to {force_detected_language_to}")766 767 # Create result container for this task768 with task_results_lock:769 if task_id not in task_results:770 task_results[task_id] = TaskResult()771 task_result = task_results[task_id]772 773 # Queue the ASR task774 asr_task_data = {775 'path': task_id, # DeduplicatedQueue uses this for dedup776 'type': 'asr',777 'task': task,778 'language': final_language,779 'video_file': video_file,780 'initial_prompt': initial_prompt,781 'audio_content': file_content,782 'encode': encode,783 'output': output,784 'word_timestamps': word_timestamps,785 'result_container': task_result,786 }787 788 # Try to queue (returns False if already queued/processing)789 if task_queue.put(asr_task_data):790 logging.info(f"ASR task {task_id} queued")791 else:792 logging.info(f"ASR task {task_id} already queued/processing - waiting for result")793 794 # EVENT LOOP BLOCK FIX: Use asyncio.to_thread so FastAPI can still respond to /status795 if await asyncio.to_thread(task_result.wait, asr_timeout):796 if task_result.error:797 logging.error(f"ASR task {task_id} failed: {task_result.error}")798 return {799 "status": "error",800 "task_id": task_id,801 "message": f"ASR processing failed: {task_result.error}"802 }803 else: 804 logging.info(f"ASR task {task_id} completed")805 return StreamingResponse(806 iter(task_result.result),807 media_type="text/plain",808 headers={'Source': f'{task.capitalize()}d using stable-ts from Subgen!'}809 )810 else:811 logging.error(f"ASR task {task_id} timed out")812 return {813 "status": "timeout",814 "task_id": task_id,815 "message": f"ASR processing timed out after {asr_timeout} seconds"816 }817 818 except Exception as e: 819 logging.error(f"Error in ASR endpoint: {e}", exc_info=True)820 return {"status": "error", "message": f"Error: {str(e)}"}821 finally:822 await audio_file.close()823 # Clean up task_results entry after task completes824 with task_results_lock:825 if task_id in task_results:826 del task_results[task_id]827 logging.debug(f"Cleaned up task_results entry for {task_id}")828 829# ============================================================================830# ASR WORKER FUNCTION831# ============================================================================832 833def get_audio_start_time(video_path: str) -> float:834 """835 Use ffprobe to detect the audio stream start_time offset from a video file.836 837 Some containers (especially Amazon WEB-DL) have audio streams that start838 later than the video stream. Bazarr compensates with adelay silence padding,839 but Whisper ignores digital silence, causing all timestamps to be early by840 the start_time offset.841 842 Returns the audio start_time in seconds, or 0.0 if not detectable.843 """844 if not video_path or not os.path.isfile(video_path):845 return 0.0846 847 try:848 result = subprocess.run(['ffprobe', '-v', 'error', '-select_streams', 'a:0',849 '-show_entries', 'stream=start_time',850 '-of', 'json', video_path],851 capture_output=True, text=True, timeout=10852 )853 if result.returncode != 0:854 return 0.0855 856 data = json.loads(result.stdout)857 streams = data.get('streams', [])858 if streams:859 start_time = float(streams[0].get('start_time', 0))860 if start_time > 0.1: # only apply for significant offsets861 logging.info(f"Detected audio start_time offset: {start_time:.3f}s for {os.path.basename(video_path)}")862 return start_time863 except (subprocess.TimeoutExpired, json.JSONDecodeError, ValueError, OSError) as e:864 logging.debug(f"Could not detect audio start_time for {video_path}: {e}")865 866 return 0.0867 868 869def apply_timestamp_offset(result, offset: float) -> None:870 """871 Shift all segment and word timestamps forward by the given offset.872 873 This compensates for audio start_time offsets in containers where the874 audio stream starts later than the video stream. Whisper produces875 timestamps relative to the audio stream start, but subtitles need876 to be aligned to the video/container timeline.877 878 Note: Segment.start/end are properties that delegate to the first/last879 word timestamps, so we only need to shift word timestamps to avoid880 double-application. For segments without words, we shift _default_start/end.881 """882 if offset <= 0:883 return884 885 for segment in result.segments:886 if hasattr(segment, 'words') and segment.words:887 for word in segment.words:888 word.start += offset889 word.end += offset890 else:891 # Segments without words use _default_start/_default_end892 if hasattr(segment, '_default_start'):893 segment._default_start += offset894 if hasattr(segment, '_default_end'):895 segment._default_end += offset896 897 logging.info(f"Applied +{offset:.3f}s timestamp offset to {len(result.segments)} segments")898 899 900def asr_task_worker(task_data: dict) -> None:901 """902 Worker function that processes ASR tasks from the queue. 903 Called by transcription_worker when task type is 'asr'.904 """905 result = None906 task_id = task_data.get('path', 'unknown')907 result_container = task_data.get('result_container')908 909 try:910 task = task_data['task']911 language = task_data['language']912 video_file = task_data.get('video_file')913 _initial_prompt = task_data.get('initial_prompt')914 file_content = task_data['audio_content']915 encode = task_data['encode']916 917 start_model()918 919 args = {}920 display_name = os.path.basename(video_file) if video_file else task_id921 args['progress_callback'] = ProgressHandler(display_name)922 923 # Handle audio encoding924 if encode:925 args['audio'] = file_content926 else:927 args['audio'] = np.frombuffer(file_content, np.int16).flatten().astype(np.float32) / 32768.0928 args['input_sr'] = 16000929 930 if custom_regroup and custom_regroup.lower() != 'default':931 args['regroup'] = custom_regroup932 933 args.update(kwargs)934 935 # Detect audio start_time offset from source file (if accessible)936 audio_offset = get_audio_start_time(video_file) if video_file else 0.0937 938 # Perform transcription939 result = model.transcribe(task=task, language=language, **args, verbose=None)940 941 # Apply audio start_time offset to compensate for container timing942 # Whisper ignores silence padding (adelay) from Bazarr, so timestamps943 # are relative to audio stream start, not container start944 if audio_offset > 0:945 apply_timestamp_offset(result, audio_offset)946 947 appendLine(result)948 949 # Set result for blocking endpoint950 if result_container:951 result_container.set_result(result.to_srt_vtt(filepath=None, word_level=word_level_highlight))952 953 except Exception as e:954 logging.error(f"Error processing ASR (ID: {task_id}): {e}", exc_info=True)955 if result_container: 956 result_container.set_error(str(e))957 958 finally:959 delete_model()960 961async def get_audio_chunk(audio_file, offset=detect_language_offset, length=detect_language_length, sample_rate=16000, audio_format=np.int16):962 """963 Extract a chunk of audio from a file, starting at the given offset and of the given length.964 965 :param audio_file: The audio file (UploadFile or file-like object).966 :param offset: The offset in seconds to start the extraction.967 :param length: The length in seconds for the chunk to be extracted.968 :param sample_rate: The sample rate of the audio (default 16000).969 :param audio_format: The audio format to interpret (default int16, 2 bytes per sample).970 971 :return: A numpy array containing the extracted audio chunk.972 """973 974 # Number of bytes per sample (for int16, 2 bytes per sample)975 bytes_per_sample = np.dtype(audio_format).itemsize976 977 # Calculate the start byte based on offset and sample rate978 start_byte = offset * sample_rate * bytes_per_sample979 980 # Calculate the length in bytes based on the length in seconds981 length_in_bytes = length * sample_rate * bytes_per_sample982 983 # Seek to the start position (this assumes the audio_file is a file-like object)984 await audio_file.seek(start_byte)985 986 # Read the required chunk of audio (length_in_bytes)987 chunk = await audio_file.read(length_in_bytes)988 989 # Convert the chunk into a numpy array (normalized to float32)990 audio_data = np.frombuffer(chunk, dtype=audio_format).flatten().astype(np.float32) / 32768.0991 992 return audio_data993 994# ============================================================================995# REFACTORED /DETECT-LANGUAGE ENDPOINT WITH HASH-BASED DEDUPLICATION AND BLOCKING996# ============================================================================997 998@app.post("/detect-language")999async def detect_language(1000 audio_file: UploadFile = File(...),1001 encode: bool = Query(default=True),1002 video_file: Union[str, None] = Query(default=None),1003 detect_lang_length: int = Query(default=detect_language_length),1004 detect_lang_offset: int = Query(default=detect_language_offset)1005):1006 global active_direct_tasks1007 1008 if force_detected_language_to: 1009 await audio_file.close()1010 return {"detected_language": force_detected_language_to.to_name(), "language_code": force_detected_language_to.to_iso_639_1()}1011 1012 task_started = False1013 try:1014 file_content = await audio_file.read()1015 if not file_content:1016 return {"detected_language": "Unknown", "language_code": "und", "status": "error"}1017 1018 logging.info("Immediate language detection (Queue Bypass)" + (f" for {video_file}" if video_file else ""))1019 1020 # Track that we are directly using the model outside the queue1021 with active_direct_tasks_lock:1022 active_direct_tasks += 11023 task_started = True1024 1025 # --- RUN IMMEDIATELY ---1026 # EVENT LOOP BLOCK FIX: Offload heavy ops to background thread1027 await asyncio.to_thread(start_model)1028 1029 if encode:1030 audio_bytes = await asyncio.to_thread(1031 extract_audio_segment_from_content, 1032 file_content, 1033 detect_lang_offset, 1034 detect_lang_length1035 )1036 audio_data = np.frombuffer(audio_bytes, np.int16).flatten().astype(np.float32) / 32768.01037 else:1038 audio_data = await get_audio_chunk(audio_file, detect_lang_offset, detect_lang_length)1039 1040 # Offload the heavy AI inference to a background thread1041 result = await asyncio.to_thread(model.transcribe, audio_data, input_sr=16000, verbose=False)1042 1043 detected = LanguageCode.from_string(result.language)1044 1045 logging.info(f"Detect Language Result: {detected.to_name()} ({detected.to_iso_639_1()})")1046 1047 return {1048 "detected_language": detected.to_name(),1049 "language_code": detected.to_iso_639_1()1050 }1051 1052 except Exception as e: 1053 logging.error(f"Error in API detect-language: {e}", exc_info=True)1054 return {"detected_language": "Unknown", "language_code": "und", "status": "error"}1055 finally: 1056 await audio_file.close()1057 # Decrement counter so delete_model() knows we are done1058 if task_started:1059 with active_direct_tasks_lock:1060 active_direct_tasks -= 11061 delete_model() # Schedules VRAM cleanup if system is idle1062 1063# ============================================================================1064# DETECT LANGUAGE WORKER FOR UPLOADED AUDIO1065# ============================================================================1066 1067def detect_language_from_upload(task_data: dict) -> None:1068 """1069 Worker function that processes detect-language tasks from uploaded audio. 1070 Sets the result in the result_container when complete.1071 """1072 detected_language = LanguageCode.NONE1073 task_id = task_data.get('path', 'unknown')1074 result_container = task_data.get('result_container')1075 1076 try:1077 video_file = task_data.get('video_file')1078 file_content = task_data['audio_content']1079 encode = task_data['encode']1080 detect_lang_length = task_data['detect_lang_length']1081 detect_lang_offset = task_data['detect_lang_offset']1082 1083 logging.info(1084 f"Detecting language for '{video_file}' ({detect_lang_length}s starting at {detect_lang_offset}s) - ID: {task_id}"1085 if video_file1086 else f"Detecting language ({detect_lang_length}s starting at {detect_lang_offset}s) - ID: {task_id}"1087 )1088 1089 start_model()1090 1091 args = {}1092 args['progress_callback'] = None1093 1094 # Handle audio extraction1095 if encode:1096 audio_bytes = extract_audio_segment_from_content(1097 file_content, 1098 detect_lang_offset, 1099 detect_lang_length1100 )1101 args['audio'] = audio_bytes1102 args['input_sr'] = 160001103 else:1104 args['audio'] = np.frombuffer(file_content, np.int16).flatten().astype(np.float32) / 32768.01105 args['input_sr'] = 160001106 1107 args.update(kwargs)1108 args['verbose'] = False # Hide the confusing progress bar1109 1110 result = model.transcribe(**args)1111 detected_language = LanguageCode.from_string(result.language)1112 language_code = detected_language.to_iso_639_1()1113 1114 logging.info(f"Detected language: {detected_language.to_name()} ({language_code}) - ID: {task_id}")1115 1116 # Set the result for the blocking endpoint1117 if result_container:1118 result_container.set_result({1119 "detected_language": detected_language.to_name(),1120 "language_code": language_code1121 })1122 1123 except Exception as e:1124 logging.error(1125 f"Error detecting language (ID: {task_id}) for '{task_data.get('video_file')}': {e}"1126 if task_data.get('video_file')1127 else f"Error detecting language (ID: {task_id}): {e}",1128 exc_info=True1129 )1130 if result_container: 1131 result_container.set_error(str(e))1132 1133 finally:1134 delete_model()1135 1136# ============================================================================1137# HELPER: Extract audio segment from in-memory content1138# ============================================================================1139 1140def extract_audio_segment_from_content(audio_content: bytes, start_time: int, duration: int) -> bytes:1141 """1142 Extract a segment of audio from in-memory content using FFmpeg.1143 1144 Args:1145 audio_content: Raw audio bytes1146 start_time: Start time in seconds1147 duration: Duration in seconds1148 1149 Returns:1150 Audio bytes of the extracted segment1151 """1152 try: 1153 logging.info(f"Extracting audio segment: start_time={start_time}s, duration={duration}s")1154 1155 out, _ = (1156 ffmpeg1157 .input('pipe:0', ss=start_time, t=duration)1158 .output('pipe:1', format='wav', acodec='pcm_s16le', ar=16000)1159 .run(input=audio_content, capture_stdout=True, capture_stderr=True)1160 )1161 1162 if not out:1163 raise ValueError("FFmpeg output is empty")1164 1165 return out1166 1167 except ffmpeg.Error as e:1168 logging.error(f"FFmpeg error: {e.stderr.decode()}")1169 return audio_content # Fallback to original if extraction fails1170 except Exception as e:1171 logging.error(f"Error extracting audio segment: {str(e)}")1172 return audio_content # Fallback to original1173 1174def detect_language_task(path, original_task_data=None):1175 """1176 Worker function that detects language for a local file.1177 Returns the task data to be queued for transcription.1178 """1179 detected_language = LanguageCode.NONE1180 1181 try:1182 logging.info(1183 f"Detecting language of file: {path} "1184 f"({detect_language_length}s starting at {detect_language_offset}s)"1185 )1186 1187 start_model()1188 1189 audio_segment = extract_audio_segment_to_memory(1190 path, 1191 detect_language_offset, 1192 int(detect_language_length)1193 )1194 1195 # FIX: Hide confusing progress bar and use from_string for ISO codes1196 result = model.transcribe(audio_segment, verbose=False)1197 detected_language = LanguageCode.from_string(result.language)1198 1199 logging.info(f"Detected language: {detected_language.to_name()}")1200