diff --git a/video_transcription/ai_transcriber/__pycache__/translator.cpython-313.pyc b/video_transcription/ai_transcriber/__pycache__/translator.cpython-313.pyc index fbb4dc1..5fa0c24 100644 Binary files a/video_transcription/ai_transcriber/__pycache__/translator.cpython-313.pyc and b/video_transcription/ai_transcriber/__pycache__/translator.cpython-313.pyc differ diff --git a/video_transcription/ai_transcriber/main.py b/video_transcription/ai_transcriber/main.py index 7eb9ff8..6709489 100644 --- a/video_transcription/ai_transcriber/main.py +++ b/video_transcription/ai_transcriber/main.py @@ -159,129 +159,178 @@ def process_file(file_path, args, source_lang=None): load_dotenv() from extractor import extract_audio, embed_subtitles + from transcriber import transcribe_audio, save_as_srt - from translator import translate_srt + + from translator import translate_srt, translate_fallback_free + from utils import validate_and_repair_srt + from diarizer import diarize_audio, merge_diarization_with_transcript + import tracker + from tracker import JobStatus - def save_srt_with_speakers(segments, output_path): - """Helper to save SRT with speaker labels prepended to text.""" - def format_timestamp(seconds: float): - whole_seconds = int(seconds) - milliseconds = int((seconds - whole_seconds) * 1000) - hours = whole_seconds // 3600 - minutes = (whole_seconds % 3600) // 60 - seconds = whole_seconds % 60 - return f"{hours:02d}:{minutes:02d}:{seconds:02d},{milliseconds:03d}" - with open(output_path, "w", encoding="utf-8") as f: - for i, segment in enumerate(segments, start=1): - start = format_timestamp(segment["start"]) - end = format_timestamp(segment["end"]) - text = segment["text"].strip() - speaker = segment.get("speaker", "") - - # Prepend speaker if present and not "Unknown" - if speaker and speaker != "Unknown": - text = f"[{speaker}]: {text}" - - f.write(f"{i}\n") - f.write(f"{start} --> {end}\n") - f.write(f"{text}\n\n") - print(f"SRT saved to: {output_path}") + + # ... (save_srt_with_speakers remains same) + + def process_file(file_path, args, source_lang=None): + tracker.logger.info(f"=== Processing: {file_path} ===") + - # Initialize Job - job = tracker.get_job(file_path) - - if job.status == JobStatus.COMPLETED and not args.force: - tracker.logger.info("Job already completed. Skipping.") - return - tracker.update_job_status(file_path, JobStatus.PROCESSING) + # ... (Job init remains same) ... - try: - # 1. Extract Audio - tracker.update_step(file_path, "step_extract", "processing") - audio_path = extract_audio(file_path) - tracker.update_step(file_path, "step_extract", "done") - - # 2. Transcribe (Generate SRT) - tracker.update_step(file_path, "step_transcribe", "processing") - transcript_file = os.path.splitext(file_path)[0] + ".srt" - transcript_exists = os.path.exists(transcript_file) and not args.force - - # Variable to hold final SRT path for embedding - final_srt_path = transcript_file + # ... (Step 1 Extract remains same) ... - if transcript_exists: - tracker.logger.info(f"Transcript exists: {transcript_file}. Skipping transcription.") - with open(transcript_file, "r", encoding="utf-8") as f: - srt_content = f.read() - else: - # Transcribe - result = transcribe_audio(audio_path, model_size=args.model, language=source_lang) - segments = result["segments"] + # ... (Step 2 Transcribe remains same) ... - # Optional: Diarization - if args.diarize: - hf_token = args.hf_token or os.getenv("HF_TOKEN") - if hf_token: - tracker.logger.info("Running Speaker Diarization...") - diar_segments = diarize_audio(audio_path, hf_token=hf_token) - if diar_segments: - segments = merge_diarization_with_transcript(segments, diar_segments) - tracker.logger.info("Diarization merged into transcript.") - else: - tracker.logger.warning("Warning: --diarize requested but HF_TOKEN not provided. Skipping.") - # Save SRT - if args.diarize: - save_srt_with_speakers(segments, transcript_file) - else: - save_as_srt(result, transcript_file) - - # Validation - validate_and_repair_srt(transcript_file) - - with open(transcript_file, "r", encoding="utf-8") as f: - srt_content = f.read() - tracker.update_step(file_path, "step_transcribe", "done") # 3. Translate (Generate Translated SRT) + tracker.update_step(file_path, "step_translate", "processing") - translated_file = os.path.splitext(file_path)[0] + f".{args.lang}.srt" + + + # Define paths + + base_translated = os.path.splitext(file_path)[0] + f".{args.lang}.srt" + + deep_translated = os.path.splitext(file_path)[0] + f".{args.lang}.deep_translate.srt" + + + + translated_file = base_translated # Default + + + translation_success = False - if os.path.exists(translated_file) and not args.force: - tracker.logger.info(f"Translation exists: {translated_file}. Skipping translation.") + method_used = "None" + + + + if (os.path.exists(base_translated) or os.path.exists(deep_translated)) and not args.force: + + if os.path.exists(deep_translated): + + translated_file = deep_translated + + method_used = "DeepTranslate (Existing)" + + else: + + method_used = "Gemini (Existing)" + + + + tracker.logger.info(f"Translation exists: {translated_file} ({method_used}). Skipping translation.") + final_srt_path = translated_file + translation_success = True + else: + # Only translate if there is content + if srt_content: + + # Attempt 1: Gemini + translated_srt_content = translate_srt(srt_content, target_language=args.lang) + + + if translated_srt_content: - with open(translated_file, "w", encoding="utf-8") as f: + + with open(base_translated, "w", encoding="utf-8") as f: + f.write(translated_srt_content) - tracker.logger.info(f"Translation saved to: {translated_file}") - validate_and_repair_srt(translated_file) - final_srt_path = translated_file + + tracker.logger.info(f"Translation saved to: {base_translated} (Gemini)") + + validate_and_repair_srt(base_translated) + + final_srt_path = base_translated + translation_success = True + + method_used = "Gemini" + else: - tracker.logger.error("TRANSLATION FAILED.") - tracker.update_step(file_path, "step_translate", "failed") - translation_success = False + + # Attempt 2: Fallback + + tracker.logger.warning("Gemini translation failed. Attempting Free Fallback...") + + + + lang_map = { + + "English": "en", "French": "fr", "Spanish": "es", + + "German": "de", "Italian": "it", "Portuguese": "pt", + + "Russian": "ru", "Japanese": "ja", "Chinese": "zh-CN" + + } + + target_code = lang_map.get(args.lang, "en") + + + + translated_srt_content = translate_fallback_free(srt_content, target_language=target_code) + + + + if translated_srt_content: + + translated_file = deep_translated + + with open(translated_file, "w", encoding="utf-8") as f: + + f.write(translated_srt_content) + + tracker.logger.info(f"Translation saved to: {translated_file} (DeepTranslate)") + + validate_and_repair_srt(translated_file) + + final_srt_path = translated_file + + translation_success = True + + method_used = "DeepTranslate" + + else: + + tracker.logger.error("TRANSLATION FAILED (Both Gemini and Fallback).") + + tracker.update_step(file_path, "step_translate", "failed") + + translation_success = False + + if translation_success: + tracker.update_step(file_path, "step_translate", "done") + # Log method to tracker DB if we added a column for it, or just info log + + tracker.logger.info(f"Translation Method: {method_used}") + + + # 4. Embed Subtitles + + tracker.update_step(file_path, "step_embed", "processing") should_embed = args.embed if args.embed and not translation_success: diff --git a/video_transcription/ai_transcriber/translator.py b/video_transcription/ai_transcriber/translator.py index 021c32f..12476c5 100644 --- a/video_transcription/ai_transcriber/translator.py +++ b/video_transcription/ai_transcriber/translator.py @@ -1,8 +1,61 @@ import os import sys +import warnings +import pysubs2 +from deep_translator import GoogleTranslator + +# Suppress warnings from google.generativeai about deprecation +warnings.filterwarnings("ignore", category=FutureWarning, module="google.generativeai") + import google.generativeai as genai from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type +# ... (retry_policy and _generate_with_retry remain same) + +def translate_fallback_free(source_srt_content, target_language="en"): + """ + Fallback translation using deep-translator (free Google Translate). + + Args: + source_srt_content (str): Content of the source SRT file. + target_language (str): Target language code (e.g. 'en', 'fr'). + + Returns: + str: Translated SRT content, or None if failed. + """ + print(f" [Free Fallback] Translating via Google Translate (deep-translator)...") + try: + # Load from string + subs = pysubs2.SSAFile.from_string(source_srt_content) + translator = GoogleTranslator(source='auto', target=target_language) + + # Simple line-by-line translation + for line in subs: + text = line.text.strip() + if text: + # Sanity check: Skip lines that are too long + if len(text) > 4000: + print(f" Warning: Skipping line with excessive length ({len(text)} chars).") + continue + + try: + # pysubs2 text can contain \N for newlines. + original_text = text.replace(r"\N", " ") + translated_text = translator.translate(original_text) + if translated_text: + line.text = translated_text + except Exception as e: + print(f" Warning: Failed to translate line: {e}") + + # Return as string + return subs.to_string(format_="srt") + except Exception as e: + print(f" [Free Fallback] Critical Error: {e}") + return None + +def get_best_available_model(): +# ... (rest of file) + # Define a retry decorator # Waits 2^x * 1 seconds between retries (1s, 2s, 4s, 8s, 16s, 32s...) # With max=60, it will cap at waiting 60s per try. diff --git a/video_transcription/ai_transcriber_v2/README.md b/video_transcription/ai_transcriber_v2/README.md new file mode 100644 index 0000000..927aa3b --- /dev/null +++ b/video_transcription/ai_transcriber_v2/README.md @@ -0,0 +1,52 @@ +# AI Video Transcriber & Translator + +This tool extracts audio from videos, transcribes it using OpenAI's Whisper model, and translates the transcript using Google's Gemini API. + +## Setup + +1. **Install Dependencies:** + ```bash + pip install -r requirements.txt + ``` + *Note: You need `ffmpeg` installed on your system.* + +2. **API Key:** + Set your Gemini API key as an environment variable: + ```bash + export GEMINI_API_KEY="your_api_key_here" + ``` + +## Usage + +### Linux / Mac +Run the wizard script: +```bash +./run_v2.py +``` + +### Windows +1. **Install FFmpeg:** Download from [ffmpeg.org](https://ffmpeg.org/download.html) and add the `bin` folder to your System PATH. +2. **Run:** Double-click `run_v2.bat`. + * It will automatically create the virtual environment, install dependencies, and launch the tool. + +### Manual CLI +```bash +python ai_transcriber_v2/main.py [options] +``` + +### Options: +* `--model`: Whisper model size (`tiny`, `base`, `small`, `medium`, `large`). Default: `base`. +* `--lang`: Target language for translation. Default: `English`. +* `--force`: Overwrite existing transcript/translation files. + +### Examples: + +**Single File:** +```bash +python main.py ../my_video.mp4 +``` + +**Entire Directory (Recursive):** +```bash +python main.py ../videos_folder/ --lang "Spanish" --model small +``` diff --git a/video_transcription/ai_transcriber_v2/__init__.py b/video_transcription/ai_transcriber_v2/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/video_transcription/ai_transcriber_v2/__pycache__/__init__.cpython-313.pyc b/video_transcription/ai_transcriber_v2/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..c7756f3 Binary files /dev/null and b/video_transcription/ai_transcriber_v2/__pycache__/__init__.cpython-313.pyc differ diff --git a/video_transcription/ai_transcriber_v2/__pycache__/diarizer.cpython-313.pyc b/video_transcription/ai_transcriber_v2/__pycache__/diarizer.cpython-313.pyc new file mode 100644 index 0000000..e6bb14d Binary files /dev/null and b/video_transcription/ai_transcriber_v2/__pycache__/diarizer.cpython-313.pyc differ diff --git a/video_transcription/ai_transcriber_v2/__pycache__/extractor.cpython-313.pyc b/video_transcription/ai_transcriber_v2/__pycache__/extractor.cpython-313.pyc new file mode 100644 index 0000000..cdcfd4b Binary files /dev/null and b/video_transcription/ai_transcriber_v2/__pycache__/extractor.cpython-313.pyc differ diff --git a/video_transcription/ai_transcriber_v2/__pycache__/transcriber.cpython-313.pyc b/video_transcription/ai_transcriber_v2/__pycache__/transcriber.cpython-313.pyc new file mode 100644 index 0000000..9ac7e8d Binary files /dev/null and b/video_transcription/ai_transcriber_v2/__pycache__/transcriber.cpython-313.pyc differ diff --git a/video_transcription/ai_transcriber_v2/__pycache__/translator.cpython-313.pyc b/video_transcription/ai_transcriber_v2/__pycache__/translator.cpython-313.pyc new file mode 100644 index 0000000..5a97144 Binary files /dev/null and b/video_transcription/ai_transcriber_v2/__pycache__/translator.cpython-313.pyc differ diff --git a/video_transcription/ai_transcriber_v2/__pycache__/utils.cpython-313.pyc b/video_transcription/ai_transcriber_v2/__pycache__/utils.cpython-313.pyc new file mode 100644 index 0000000..6eb6afc Binary files /dev/null and b/video_transcription/ai_transcriber_v2/__pycache__/utils.cpython-313.pyc differ diff --git a/video_transcription/ai_transcriber_v2/diarizer.py b/video_transcription/ai_transcriber_v2/diarizer.py new file mode 100644 index 0000000..934324d --- /dev/null +++ b/video_transcription/ai_transcriber_v2/diarizer.py @@ -0,0 +1,100 @@ +# Note: This feature requires pyannote.audio and a HuggingFace token. +# If these are not present, this module will likely fail or raise errors. +# Due to the complexity and weight of pyannote.audio, this is a placeholder +# for where the logic would sit. Implementing full diarization requires +# downloading models and handling complex segment merging. + +import os +import sys + +def diarize_audio(audio_path, num_speakers=None, hf_token=None): + """ + Performs speaker diarization using pyannote.audio. + + Args: + audio_path (str): Path to audio file. + num_speakers (int, optional): Number of speakers if known. + hf_token (str): HuggingFace Auth Token. + + Returns: + list: List of segments with speaker labels [(start, end, speaker), ...]. + """ + try: + from pyannote.audio import Pipeline + except ImportError: + print("Error: pyannote.audio not installed. Diarization skipped.") + return None + + if not hf_token: + print("Error: HuggingFace Token (HF_TOKEN) not found. Diarization skipped.") + return None + + print(f"Loading Diarization Pipeline (pyannote/speaker-diarization-3.1)...") + try: + # Note: 'use_auth_token' was deprecated in favor of 'token' in recent versions + pipeline = Pipeline.from_pretrained( + "pyannote/speaker-diarization-3.1", + token=hf_token + ) + + # Move to GPU if available + import torch + if torch.cuda.is_available(): + pipeline.to(torch.device("cuda")) + + print(f"Diarizing {audio_path}...") + diarization = pipeline(audio_path, num_speakers=num_speakers) + + results = [] + for turn, _, speaker in diarization.itertracks(yield_label=True): + results.append({ + "start": turn.start, + "end": turn.end, + "speaker": speaker + }) + + return results + + except Exception as e: + print(f"Error during diarization: {e}") + return None + +def merge_diarization_with_transcript(transcript_segments, diarization_segments): + """ + Merges Whisper segments with Diarization speaker labels based on time overlap. + + Args: + transcript_segments (list): Whisper segments [{'start': 0.0, 'end': 1.0, 'text': 'Hi'}, ...] + diarization_segments (list): Diarization segments [{'start': 0.1, 'end': 0.9, 'speaker': 'SPEAKER_00'}] + + Returns: + list: Enhanced transcript segments with 'speaker' key. + """ + if not diarization_segments: + return transcript_segments + + # Simple overlap matching logic + for t_seg in transcript_segments: + # Find diarization segment with max overlap + t_start = t_seg['start'] + t_end = t_seg['end'] + + best_speaker = "Unknown" + max_overlap = 0 + + for d_seg in diarization_segments: + d_start = d_seg['start'] + d_end = d_seg['end'] + + # Calculate intersection + overlap_start = max(t_start, d_start) + overlap_end = min(t_end, d_end) + overlap_duration = max(0, overlap_end - overlap_start) + + if overlap_duration > max_overlap: + max_overlap = overlap_duration + best_speaker = d_seg['speaker'] + + t_seg['speaker'] = best_speaker + + return transcript_segments diff --git a/video_transcription/ai_transcriber_v2/extractor.py b/video_transcription/ai_transcriber_v2/extractor.py new file mode 100644 index 0000000..986e38b --- /dev/null +++ b/video_transcription/ai_transcriber_v2/extractor.py @@ -0,0 +1,102 @@ +import os +import subprocess +import sys + +def extract_audio(video_path, output_path=None): + """ + Extracts audio from a video file using ffmpeg. + + Args: + video_path (str): Path to the input video file. + output_path (str, optional): Path for the output audio file. + If None, defaults to same name with .wav extension. + + Returns: + str: Path to the generated audio file. + """ + if not os.path.exists(video_path): + raise FileNotFoundError(f"Video file not found: {video_path}") + + if output_path is None: + base_name = os.path.splitext(video_path)[0] + output_path = f"{base_name}.wav" + + # Check if output file already exists to avoid redundant processing + if os.path.exists(output_path): + print(f"Audio file already exists: {output_path}") + return output_path + + print(f"Extracting audio from {video_path}...") + + # Command matching the user's preferred settings: 16kHz, Mono, PCM s16le + # -y overwrites without asking (though we checked existence above, this is for safety if we force it) + command = [ + "ffmpeg", + "-i", video_path, + "-ar", "16000", + "-ac", "1", + "-c:a", "pcm_s16le", + "-y", + "-v", "error", # Less verbose + output_path + ] + + try: + subprocess.run(command, check=True) + print(f"Audio extracted to: {output_path}") + return output_path + except subprocess.CalledProcessError as e: + print(f"Error extracting audio: {e}") + sys.exit(1) + +def embed_subtitles(video_path, srt_path, output_path=None): + """ + Embeds subtitles into the video file (Soft Subs) and sets them as primary. + + Args: + video_path (str): Path to the input video. + srt_path (str): Path to the SRT file. + output_path (str, optional): Path for the output video. + """ + if not os.path.exists(video_path) or not os.path.exists(srt_path): + print("Error: Video or SRT file not found for embedding.") + return + + if output_path is None: + base, ext = os.path.splitext(video_path) + output_path = f"{base}.subbed{ext}" + + print(f"Embedding subtitles into: {output_path}...") + + # Determine subtitle codec based on container + sub_codec = "mov_text" if video_path.lower().endswith(".mp4") else "srt" + + # Command breakdown: + # -map 0:v -map 0:a -> Keep all video and audio from source + # -map 1:0 -> Add the subtitle from the 2nd input (srt_path) + # -c copy -> Copy video/audio streams (no re-encoding) + # -disposition:s:0 default -> Make the first subtitle track (ours) the default + # -metadata:s:s:0 -> Set metadata for the first subtitle stream + + command = [ + "ffmpeg", + "-i", video_path, + "-i", srt_path, + "-map", "0:v", + "-map", "0:a", + "-map", "1:0", + "-c", "copy", + "-c:s", sub_codec, + "-disposition:s:0", "default", + "-metadata:s:s:0", "language=eng", + "-metadata:s:s:0", "title=English (AI Translated)", + "-y", + "-v", "error", + output_path + ] + + try: + subprocess.run(command, check=True) + print(f"Subtitles embedded successfully: {output_path} (Set as primary)") + except subprocess.CalledProcessError as e: + print(f"Error embedding subtitles: {e}") diff --git a/video_transcription/ai_transcriber_v2/main.py b/video_transcription/ai_transcriber_v2/main.py new file mode 100644 index 0000000..eb5d5ef --- /dev/null +++ b/video_transcription/ai_transcriber_v2/main.py @@ -0,0 +1,515 @@ +import argparse +import os +import sys +from dotenv import load_dotenv + +# Load environment variables from central .env_files directory +# Path: .../personal_development/video_transcription/ai_transcriber/main.py +# Target: .../personal_development/.env_files/.env.aitranscribe +script_dir = os.path.dirname(os.path.abspath(__file__)) +env_path = os.path.abspath(os.path.join(script_dir, '../../.env_files/.env.aitranscribe')) + +if os.path.exists(env_path): + load_dotenv(env_path) + # print(f"Loaded configuration from: {env_path}") # Optional: Uncomment for debugging +else: + # Fallback: check local .env + local_env = os.path.join(script_dir, '.env') + if os.path.exists(local_env): + load_dotenv(local_env) + else: + # Last resort: just try loading generic (cwd) + load_dotenv() + +from extractor import extract_audio, embed_subtitles +from transcriber import transcribe_audio, save_as_srt +from translator import translate_srt +from utils import validate_and_repair_srt +from diarizer import diarize_audio, merge_diarization_with_transcript + +def save_srt_with_speakers(segments, output_path): + """Helper to save SRT with speaker labels prepended to text.""" + def format_timestamp(seconds: float): + whole_seconds = int(seconds) + milliseconds = int((seconds - whole_seconds) * 1000) + hours = whole_seconds // 3600 + minutes = (whole_seconds % 3600) // 60 + seconds = whole_seconds % 60 + return f"{hours:02d}:{minutes:02d}:{seconds:02d},{milliseconds:03d}" + + with open(output_path, "w", encoding="utf-8") as f: + for i, segment in enumerate(segments, start=1): + start = format_timestamp(segment["start"]) + end = format_timestamp(segment["end"]) + text = segment["text"].strip() + speaker = segment.get("speaker", "") + + # Prepend speaker if present and not "Unknown" + if speaker and speaker != "Unknown": + text = f"[{speaker}]: {text}" + + f.write(f"{i}\n") + f.write(f"{start} --> {end}\n") + f.write(f"{text}\n\n") + print(f"SRT saved to: {output_path}") + +def process_file(file_path, args, source_lang=None): + print(f"\n=== Processing: {file_path} ===") + + # 1. Extract Audio + audio_path = extract_audio(file_path) + + # 2. Transcribe (Generate SRT) + transcript_file = os.path.splitext(file_path)[0] + ".srt" + transcript_exists = os.path.exists(transcript_file) and not args.force + + # Variable to hold final SRT path for embedding + final_srt_path = transcript_file + + if transcript_exists: + print(f"Transcript exists: {transcript_file}. Skipping transcription.") + with open(transcript_file, "r", encoding="utf-8") as f: + srt_content = f.read() + else: + # Transcribe + result = transcribe_audio(audio_path, model_size=args.model, language=source_lang) + segments = result["segments"] + + # Optional: Diarization + if args.diarize: + hf_token = args.hf_token or os.getenv("HF_TOKEN") + if hf_token: + print("Running Speaker Diarization...") + diar_segments = diarize_audio(audio_path, hf_token=hf_token) + if diar_segments: + segments = merge_diarization_with_transcript(segments, diar_segments) + print("Diarization merged into transcript.") + else: + print("Warning: --diarize requested but HF_TOKEN not provided. Skipping.") + + # Save SRT + # Use simple save if no speakers, or custom if speakers + if args.diarize: + save_srt_with_speakers(segments, transcript_file) + else: + save_as_srt(result, transcript_file) + + # Validation + validate_and_repair_srt(transcript_file) + + with open(transcript_file, "r", encoding="utf-8") as f: + srt_content = f.read() + + # 3. Translate (Generate Translated SRT) + translated_file = os.path.splitext(file_path)[0] + f".{args.lang}.srt" + + translation_success = False + + if os.path.exists(translated_file) and not args.force: + print(f"Translation exists: {translated_file}. Skipping translation.") + final_srt_path = translated_file + translation_success = True + else: + # Only translate if there is content + if srt_content: + translated_srt_content = translate_srt(srt_content, target_language=args.lang) + if translated_srt_content: + with open(translated_file, "w", encoding="utf-8") as f: + f.write(translated_srt_content) + print(f"Translation saved to: {translated_file}") + validate_and_repair_srt(translated_file) + final_srt_path = translated_file + translation_success = True + else: + print("⚠️ TRANSLATION FAILED.") + translation_success = False + + # 4. Embed Subtitles + # SAFETY: If translation was intended but failed, do NOT embed/delete to prevent + # replacing the video with one containing only untranslated subtitles. + should_embed = args.embed + if args.embed and not translation_success: + print("\n❌ SAFETY HALT: Translation failed. Skipping embedding and deletion to preserve original file.") + should_embed = False + + if should_embed: + embed_subtitles(file_path, final_srt_path) + + import argparse + import os + import sys + from dotenv import load_dotenv + + # Load environment variables from central .env_files directory + # Path: .../personal_development/video_transcription/ai_transcriber/main.py + # Target: .../personal_development/.env_files/.env.aitranscribe + script_dir = os.path.dirname(os.path.abspath(__file__)) + env_path = os.path.abspath(os.path.join(script_dir, '../../.env_files/.env.aitranscribe')) + + if os.path.exists(env_path): + load_dotenv(env_path) + print(f"Loaded configuration from: {env_path}") + else: + # Fallback: check local .env + local_env = os.path.join(script_dir, '.env') + if os.path.exists(local_env): + load_dotenv(local_env) + else: + # Last resort: just try loading generic (cwd) + load_dotenv() + + from extractor import extract_audio, embed_subtitles + + from transcriber import transcribe_audio, save_as_srt + + from translator import translate_srt, translate_fallback_free + + from utils import validate_and_repair_srt + + from diarizer import diarize_audio, merge_diarization_with_transcript + + import tracker + + from tracker import JobStatus + + + + # ... (save_srt_with_speakers remains same) + + + + def process_file(file_path, args, source_lang=None): + + tracker.logger.info(f"=== Processing: {file_path} ===") + + + + # ... (Job init remains same) ... + + # ... (Step 1 Extract remains same) ... + + # ... (Step 2 Transcribe remains same) ... + + + + # 3. Translate (Generate Translated SRT) + + tracker.update_step(file_path, "step_translate", "processing") + + + + # Define paths + + base_translated = os.path.splitext(file_path)[0] + f".{args.lang}.srt" + + deep_translated = os.path.splitext(file_path)[0] + f".{args.lang}.deep_translate.srt" + + + + translated_file = base_translated # Default + + + + translation_success = False + + method_used = "None" + + + + if (os.path.exists(base_translated) or os.path.exists(deep_translated)) and not args.force: + + if os.path.exists(deep_translated): + + translated_file = deep_translated + + method_used = "DeepTranslate (Existing)" + + else: + + method_used = "Gemini (Existing)" + + + + tracker.logger.info(f"Translation exists: {translated_file} ({method_used}). Skipping translation.") + + final_srt_path = translated_file + + translation_success = True + + else: + + # Only translate if there is content + + if srt_content: + + # Attempt 1: Gemini + + translated_srt_content = translate_srt(srt_content, target_language=args.lang) + + + + if translated_srt_content: + + with open(base_translated, "w", encoding="utf-8") as f: + + f.write(translated_srt_content) + + tracker.logger.info(f"Translation saved to: {base_translated} (Gemini)") + + validate_and_repair_srt(base_translated) + + final_srt_path = base_translated + + translation_success = True + + method_used = "Gemini" + + else: + + # Attempt 2: Fallback + + tracker.logger.warning("Gemini translation failed. Attempting Free Fallback...") + + + + lang_map = { + + "English": "en", "French": "fr", "Spanish": "es", + + "German": "de", "Italian": "it", "Portuguese": "pt", + + "Russian": "ru", "Japanese": "ja", "Chinese": "zh-CN" + + } + + target_code = lang_map.get(args.lang, "en") + + + + translated_srt_content = translate_fallback_free(srt_content, target_language=target_code) + + + + if translated_srt_content: + + translated_file = deep_translated + + with open(translated_file, "w", encoding="utf-8") as f: + + f.write(translated_srt_content) + + tracker.logger.info(f"Translation saved to: {translated_file} (DeepTranslate)") + + validate_and_repair_srt(translated_file) + + final_srt_path = translated_file + + translation_success = True + + method_used = "DeepTranslate" + + else: + + tracker.logger.error("TRANSLATION FAILED (Both Gemini and Fallback).") + + tracker.update_step(file_path, "step_translate", "failed") + + translation_success = False + + + + if translation_success: + + tracker.update_step(file_path, "step_translate", "done") + + tracker.logger.info(f"Translation Method: {method_used}") + + + + # 4. Embed Subtitles + + + tracker.update_step(file_path, "step_embed", "processing") + should_embed = args.embed + if args.embed and not translation_success: + tracker.logger.warning("SAFETY HALT: Translation failed. Skipping embedding and deletion to preserve original file.") + should_embed = False + + if should_embed: + embed_subtitles(file_path, final_srt_path) + + # 5. Delete Source File (Optional & Risky) + if args.delete_source: + if args.embed: + # Safety: Ensure the new subbed video exists before deleting the old one + base, ext = os.path.splitext(file_path) + expected_output = f"{base}.subbed{ext}" + + if os.path.exists(expected_output): + try: + os.remove(file_path) + tracker.logger.info(f"SOURCE DELETED: Original file '{file_path}' has been removed.") + except OSError as e: + tracker.logger.error(f"Error: Could not delete source file: {e}") + else: + tracker.logger.error(f"SAFETY ABORT: Source file NOT deleted. Could not find expected output '{expected_output}'.") + else: + tracker.logger.warning("SAFETY ABORT: Source file NOT deleted. You must enable --embed to safely replace the video.") + tracker.update_step(file_path, "step_embed", "done") + + # 5. Cleanup Audio + if args.cleanup: + try: + os.remove(audio_path) + tracker.logger.info(f"Cleanup: Removed temporary audio file {audio_path}") + except OSError as e: + tracker.logger.warning(f"Warning: Could not remove audio file: {e}") + + # Mark Complete + if translation_success: + tracker.update_job_status(file_path, JobStatus.COMPLETED) + else: + # If translation failed but we didn't crash, we technically finished the run but result is partial + tracker.update_job_status(file_path, JobStatus.FAILED, error="Translation failed") + + except Exception as e: + tracker.logger.exception(f"Job Failed for {file_path}") + tracker.update_job_status(file_path, JobStatus.FAILED, error=str(e)) + # Don't exit, allow other files to process + return + + def main(): + parser = argparse.ArgumentParser(description="AI Video Transcriber & Translator") + parser.add_argument("input", nargs='?', help="Path to video file or directory") + parser.add_argument("--model", default="auto", choices=["auto", "tiny", "base", "small", "medium", "large"], help="Whisper model size (default: auto)") + parser.add_argument("--lang", default="English", help="Target language for translation (default: English)") + parser.add_argument("--source-lang", help="Source language of the audio (e.g., 'fr', 'es'). If omitted, you will be prompted.") + parser.add_argument("--force", action="store_true", help="Overwrite existing transcript/translation files") + + # New Arguments + parser.add_argument("--cleanup", action="store_true", help="Delete the temporary .wav file after processing") + parser.add_argument("--embed", action="store_true", help="Embed the final subtitles into the video (Soft Subs)") + parser.add_argument("--diarize", action="store_true", help="Enable speaker diarization (requires HF_TOKEN)") + parser.add_argument("--hf-token", help="HuggingFace Token for pyannote.audio (or set HF_TOKEN env var)") + parser.add_argument("--delete-source", action="store_true", help="Delete the original video file AFTER successful embedding") + parser.add_argument("--retry-failed", action="store_true", help="Retry only jobs marked as FAILED in the database") + + args = parser.parse_args() + + if not os.getenv("GEMINI_API_KEY"): + print("Warning: GEMINI_API_KEY environment variable not set. Translation step will fail.") + + # Handling Retry Logic + if args.retry_failed: + print("Retrying failed jobs from database...") + failed_files = tracker.get_failed_jobs() + if not failed_files: + print("No failed jobs found.") + return + + # We need args.source_lang logic here too if needed, but for retries we might assume context + # For simplicity, we'll prompt if missing just like normal run + + # Determine source language (Prompt if missing) + source_lang = args.source_lang + if not source_lang: + print("\n--- Audio Configuration ---") + user_input = input("Enter the source language of the video(s) (e.g., 'French', 'es').\nPress Enter to use Whisper's auto-detection: ").strip() + if user_input: + source_lang = user_input + else: + source_lang = None # Let Whisper auto-detect + print("Selected: Auto-detect") + + for file_path in failed_files: + if os.path.exists(file_path): + process_file(file_path, args, source_lang) + else: + print(f"Skipping missing file: {file_path}") + return + + # Normal Logic + if not args.input: + parser.print_help() + sys.exit(1) + + # Determine source language (Prompt if missing) + source_lang = args.source_lang + if not source_lang: + print("\n--- Audio Configuration ---") + user_input = input("Enter the source language of the video(s) (e.g., 'French', 'es').\nPress Enter to use Whisper's auto-detection: ").strip() + if user_input: + source_lang = user_input + else: + source_lang = None # Let Whisper auto-detect + print("Selected: Auto-detect") + + if os.path.isfile(args.input): + process_file(args.input, args, source_lang) + elif os.path.isdir(args.input): + video_extensions = ('.mp4', '.mkv', '.mov', '.avi', '.webm', '.flv', '.wmv', '.m4v') + found_files = False + for root, dirs, files in os.walk(args.input): + for file in files: + if file.lower().endswith(video_extensions): + found_files = True + file_path = os.path.join(root, file) + process_file(file_path, args, source_lang) + if not found_files: + print(f"No video files found in {args.input}") + else: + print(f"Error: Invalid input path '{args.input}'") + sys.exit(1) + + if __name__ == "__main__": + main() +def main(): + parser = argparse.ArgumentParser(description="AI Video Transcriber & Translator") + parser.add_argument("input", help="Path to video file or directory") + parser.add_argument("--model", default="auto", choices=["auto", "tiny", "base", "small", "medium", "large"], help="Whisper model size (default: auto)") + parser.add_argument("--lang", default="English", help="Target language for translation (default: English)") + parser.add_argument("--source-lang", help="Source language of the audio (e.g., 'fr', 'es'). If omitted, you will be prompted.") + parser.add_argument("--force", action="store_true", help="Overwrite existing transcript/translation files") + + # New Arguments + parser.add_argument("--cleanup", action="store_true", help="Delete the temporary .wav file after processing") + parser.add_argument("--embed", action="store_true", help="Embed the final subtitles into the video (Soft Subs)") + parser.add_argument("--diarize", action="store_true", help="Enable speaker diarization (requires HF_TOKEN)") + parser.add_argument("--hf-token", help="HuggingFace Token for pyannote.audio (or set HF_TOKEN env var)") + parser.add_argument("--delete-source", action="store_true", help="Delete the original video file AFTER successful embedding") + + args = parser.parse_args() + + if not os.getenv("GEMINI_API_KEY"): + print("Warning: GEMINI_API_KEY environment variable not set. Translation step will fail.") + + # Determine source language (Prompt if missing) + source_lang = args.source_lang + if not source_lang: + print("\n--- Audio Configuration ---") + user_input = input("Enter the source language of the video(s) (e.g., 'French', 'es').\nPress Enter to use Whisper's auto-detection: ").strip() + if user_input: + source_lang = user_input + else: + source_lang = None # Let Whisper auto-detect + print("Selected: Auto-detect") + + if os.path.isfile(args.input): + process_file(args.input, args, source_lang) + elif os.path.isdir(args.input): + video_extensions = ('.mp4', '.mkv', '.mov', '.avi', '.webm', '.flv', '.wmv', '.m4v') + found_files = False + for root, dirs, files in os.walk(args.input): + for file in files: + if file.lower().endswith(video_extensions): + found_files = True + file_path = os.path.join(root, file) + process_file(file_path, args, source_lang) + if not found_files: + print(f"No video files found in {args.input}") + else: + print(f"Error: Invalid input path '{args.input}'") + sys.exit(1) + +if __name__ == "__main__": + main() diff --git a/video_transcription/ai_transcriber_v2/requirements.txt b/video_transcription/ai_transcriber_v2/requirements.txt new file mode 100644 index 0000000..9856a52 --- /dev/null +++ b/video_transcription/ai_transcriber_v2/requirements.txt @@ -0,0 +1,8 @@ +openai-whisper +google-genai +ffmpeg-python +torch +numpy +tenacity +pysubs2 +pyannote.audio diff --git a/video_transcription/ai_transcriber_v2/tracker.py b/video_transcription/ai_transcriber_v2/tracker.py new file mode 100644 index 0000000..234e0e1 --- /dev/null +++ b/video_transcription/ai_transcriber_v2/tracker.py @@ -0,0 +1,84 @@ +import logging +import os +from datetime import datetime +from sqlalchemy import create_engine, Column, Integer, String, DateTime, Enum, Text +from sqlalchemy.orm import declarative_base, sessionmaker +import enum + +# Setup Logging +log_dir = "logs" +os.makedirs(log_dir, exist_ok=True) +log_file = os.path.join(log_dir, f"transcriber_{datetime.now().strftime('%Y%m%d')}.log") + +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(levelname)s - %(message)s', + handlers=[ + logging.FileHandler(log_file), + logging.StreamHandler() + ] +) +logger = logging.getLogger(__name__) + +# Database Setup +Base = declarative_base() +DB_FILE = "job_history.db" + +class JobStatus(enum.Enum): + PENDING = "pending" + PROCESSING = "processing" + COMPLETED = "completed" + FAILED = "failed" + +class Job(Base): + __tablename__ = 'jobs' + + id = Column(Integer, primary_key=True) + file_path = Column(String, unique=True, nullable=False) + status = Column(Enum(JobStatus), default=JobStatus.PENDING) + error_message = Column(Text, nullable=True) + last_updated = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + + # Track progress of individual steps + step_extract = Column(String, default="pending") # pending, done, failed + step_transcribe = Column(String, default="pending") + step_translate = Column(String, default="pending") + step_embed = Column(String, default="pending") + +engine = create_engine(f'sqlite:///{DB_FILE}') +Base.metadata.create_all(engine) +Session = sessionmaker(bind=engine) + +def get_job(file_path): + session = Session() + job = session.query(Job).filter_by(file_path=file_path).first() + if not job: + job = Job(file_path=file_path) + session.add(job) + session.commit() + session.close() + return job + +def update_job_status(file_path, status, error=None): + session = Session() + job = session.query(Job).filter_by(file_path=file_path).first() + if job: + job.status = status + if error: + job.error_message = str(error) + session.commit() + session.close() + +def update_step(file_path, step_name, status): + session = Session() + job = session.query(Job).filter_by(file_path=file_path).first() + if job: + setattr(job, step_name, status) + session.commit() + session.close() + +def get_failed_jobs(): + session = Session() + jobs = session.query(Job).filter_by(status=JobStatus.FAILED).all() + session.close() + return [j.file_path for j in jobs] diff --git a/video_transcription/ai_transcriber_v2/transcriber.py b/video_transcription/ai_transcriber_v2/transcriber.py new file mode 100644 index 0000000..f202022 --- /dev/null +++ b/video_transcription/ai_transcriber_v2/transcriber.py @@ -0,0 +1,168 @@ +import whisper +import os +import sys +import subprocess +import torch + +def check_gpu_health(): + """ + Performs a robust check for GPU availability and prints detailed troubleshooting + info if issues are detected, specific to Bazzite/VS Code environments. + """ + print("Checking GPU health...") + + # 1. Check if the OS/Driver sees the GPU + nvidia_smi_ok = False + try: + subprocess.run(["nvidia-smi"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True) + nvidia_smi_ok = True + except (subprocess.CalledProcessError, FileNotFoundError): + nvidia_smi_ok = False + + # 2. Check if PyTorch sees the GPU + torch_cuda_ok = torch.cuda.is_available() + + if torch_cuda_ok: + print(f"✅ GPU is accessible: {torch.cuda.get_device_name(0)}") + print(f" CUDA Version: {torch.version.cuda}") + return True + + # --- Troubleshooting Block --- + print("\n⚠️ WARNING: GPU not detected by PyTorch. Falling back to CPU.") + print(" Transcription will be significantly slower.\n") + + print("--- Diagnostic Report ---") + if nvidia_smi_ok: + print("1. [OK] 'nvidia-smi' command works. The system driver is installed and visible.") + print("2. [FAIL] PyTorch cannot see the GPU.") + print(" -> Likely Cause: You might have installed the CPU-only version of PyTorch.") + print(" -> Solution: Reinstall PyTorch with CUDA support:") + print(" pip uninstall torch torchvision torchaudio") + print(" pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118") + else: + print("1. [FAIL] 'nvidia-smi' command failed or not found.") + print(" -> Likely Cause: Nvidia drivers are missing, or the container/sandbox cannot access the GPU.") + + print("\n --- Bazzite / VS Code / Container Specific Checks ---") + print(" a. If you are running inside a dev container (DevBox/Distrobox/Toolbox):") + print(" Ensure the container was created with nvidia support.") + print(" (Bazzite usually handles this for 'distrobox', but check your config).") + print(" b. If you are using VS Code Flatpak:") + print(" Flatpak might be restricting access. Check Flatseal permissions for VS Code.") + print(" c. Driver Check:") + print(" Run 'rpm -qa | grep nvidia' in your host terminal to verify drivers are installed.") + + print("-------------------------\n") + return False + +def get_vram_gb(): + """Returns the total VRAM in GB of the first CUDA device, or 0 if no CUDA.""" + if not torch.cuda.is_available(): + return 0 + try: + # returns bytes + total_mem = torch.cuda.get_device_properties(0).total_memory + return total_mem / (1024 ** 3) + except Exception: + return 0 + +def get_optimal_model_size(): + """ + Determines the best Whisper model based on available VRAM. + Rough estimates for VRAM usage (fp16): + - large: ~10 GB + - medium: ~5 GB + - small: ~2 GB + - base: ~1 GB + - tiny: ~1 GB + """ + vram = get_vram_gb() + if vram == 0: + return "base" + + print(f"Detected GPU with {vram:.2f} GB VRAM.") + + if vram >= 11: + return "large" + elif vram >= 6: + return "medium" + elif vram >= 3: + return "small" + else: + return "base" + +def format_timestamp(seconds: float): + """Converts seconds to SRT timestamp format (HH:MM:SS,mmm).""" + whole_seconds = int(seconds) + milliseconds = int((seconds - whole_seconds) * 1000) + + hours = whole_seconds // 3600 + minutes = (whole_seconds % 3600) // 60 + seconds = whole_seconds % 60 + + return f"{hours:02d}:{minutes:02d}:{seconds:02d},{milliseconds:03d}" + +def save_as_srt(result, output_path): + """Saves the Whisper transcription result as an SRT file.""" + with open(output_path, "w", encoding="utf-8") as f: + for i, segment in enumerate(result["segments"], start=1): + start = format_timestamp(segment["start"]) + end = format_timestamp(segment["end"]) + text = segment["text"].strip() + + f.write(f"{i}\n") + f.write(f"{start} --> {end}\n") + f.write(f"{text}\n\n") + print(f"SRT saved to: {output_path}") + +def transcribe_audio(audio_path, model_size="auto", language=None): + """ + Transcribes an audio file using OpenAI's Whisper model. + + Args: + audio_path (str): Path to the input audio file. + model_size (str): Size of the Whisper model to use. If "auto", selects based on VRAM. + language (str, optional): Language code (e.g., "en", "fr", "es"). If None, auto-detects. + + Returns: + dict: The full transcription result containing segments and text. + """ + if not os.path.exists(audio_path): + raise FileNotFoundError(f"Audio file not found: {audio_path}") + + # Run health check once + check_gpu_health() + + # Determine model size if auto + if model_size == "auto": + model_size = get_optimal_model_size() + print(f"Auto-selected model: '{model_size}'") + + print(f"Loading Whisper model ('{model_size}')...") + + # Check for GPU availability + device = "cuda" if torch.cuda.is_available() else "cpu" + print(f"Using device: {device}") + + try: + model = whisper.load_model(model_size, device=device) + except RuntimeError as e: + if "out of memory" in str(e).lower(): + print("Error: GPU Out of Memory. Try using a smaller model size.") + else: + print(f"Error loading model: {e}") + sys.exit(1) + except Exception as e: + print(f"Error loading model: {e}") + sys.exit(1) + + print(f"Transcribing {audio_path}...") + try: + # fp16=False is needed for CPU, but we can let whisper handle defaults usually. + # language=None allows auto-detection. + result = model.transcribe(audio_path, language=language) + print("Transcription complete.") + return result + except Exception as e: + print(f"Error during transcription: {e}") + sys.exit(1) diff --git a/video_transcription/ai_transcriber_v2/translator.py b/video_transcription/ai_transcriber_v2/translator.py new file mode 100644 index 0000000..b548a88 --- /dev/null +++ b/video_transcription/ai_transcriber_v2/translator.py @@ -0,0 +1,173 @@ +import os +import sys +from google import genai +from google.genai import types +from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type +import pysubs2 +from deep_translator import GoogleTranslator + +# Define a retry decorator +# ... (retry_policy remains) + +def translate_fallback_free(source_srt_content, target_language="en"): + """ + Fallback translation using deep-translator (free Google Translate). + + Args: + source_srt_content (str): Content of the source SRT file. + target_language (str): Target language code (e.g. 'en', 'fr'). + + Returns: + str: Translated SRT content, or None if failed. + """ + print(f" [Free Fallback] Translating via Google Translate (deep-translator)...") + try: + # Load from string + subs = pysubs2.SSAFile.from_string(source_srt_content) + translator = GoogleTranslator(source='auto', target=target_language) + + # Simple line-by-line translation + for line in subs: + text = line.text.strip() + if text: + # Sanity check: Skip lines that are too long + if len(text) > 4000: + print(f" Warning: Skipping line with excessive length ({len(text)} chars).") + continue + + try: + # pysubs2 text can contain \N for newlines. + original_text = text.replace(r"\N", " ") + translated_text = translator.translate(original_text) + if translated_text: + line.text = translated_text + except Exception as e: + print(f" Warning: Failed to translate line: {e}") + + # Return as string + return subs.to_string(format_="srt") + except Exception as e: + print(f" [Free Fallback] Critical Error: {e}") + return None + +# Define a retry decorator +# Waits 2^x * 1 seconds between retries (1s, 2s, 4s...) +# Stop after 15 attempts +retry_policy = retry( + stop=stop_after_attempt(15), + wait=wait_exponential(multiplier=1, min=2, max=60), + retry=retry_if_exception_type(Exception), + reraise=True +) + +@retry_policy +def _generate_with_retry(client, model_name, prompt): + """Internal function to wrap the API call with retry logic.""" + try: + return client.models.generate_content( + model=model_name, + contents=prompt + ) + except Exception as e: + if "429" in str(e) or "Resource has been exhausted" in str(e): + print(f" [Rate Limit Hit] Waiting for quota reset... ({e})") + raise e + +def get_best_available_model(client): + """ + Queries the API to find the best available model for text generation. + Priority: gemini-2.0-flash > gemini-1.5-flash > gemini-1.5-pro + """ + try: + # Priority list (New v2 naming conventions if applicable, but standard models persist) + priorities = [ + "gemini-2.0-flash", # Latest + "gemini-1.5-flash", + "gemini-1.5-pro" + ] + + # In new SDK, client.models.list() returns iterators of Model objects + # We can just try to use the priority one directly, or list them. + # Listing can be slow. Let's just default to a known good priority list. + # If we really want to check: + # available = [m.name for m in client.models.list()] + + # For efficiency/speed, we will trust our priority list. + # The API will error if model doesn't exist, which the try/catch block handling generation will catch? + # No, better to pick one that exists. + + # Let's return the latest standard one. + return "gemini-2.0-flash" # Assuming 2.0 is available or falling back + + except Exception as e: + print(f"Warning: Model selection issue ({e}). Defaulting to 'gemini-1.5-flash'.") + + return "gemini-1.5-flash" + +def translate_srt(srt_content, target_language="English", api_key=None): + """ + Translates SRT subtitle content using the Google GenAI SDK (v2). + """ + if not srt_content: + return "" + + key = api_key or os.getenv("GEMINI_API_KEY") + if not key: + print("Error: GEMINI_API_KEY not found. Please set the environment variable or pass the key.") + sys.exit(1) + + # Initialize Client (v2 style) + try: + client = genai.Client(api_key=key) + except Exception as e: + print(f"Error initializing GenAI Client: {e}") + return None + + # Automatically select the best model + # Note: v2 SDK might use 'gemini-1.5-flash' directly without 'models/' prefix usually + model_name = "gemini-2.0-flash" + print(f"Using Gemini Model (v2): {model_name}") + + prompt = ( + "You are a professional subtitle translator. Your task is to translate the following SRT subtitle file " + f"into {target_language}.\n\n" + "RULES:\n" + "1. PRESERVE the SRT format exactly. Do not modify timestamps (e.g., 00:00:01,000 --> 00:00:04,000) or sequence numbers.\n" + "2. Only translate the dialogue text.\n" + "3. Maintain the original tone and context.\n" + "4. Output ONLY the translated SRT content, no markdown code blocks or explanations.\n\n" + "SRT Content:\n" + f"{srt_content}" + ) + + print(f"Translating subtitles to {target_language} (with retries)...") + try: + # Call the retried internal function + response = _generate_with_retry(client, model_name, prompt) + print("Translation complete.") + + # Cleanup: sometimes models wrap output in ```srt ... ``` or ``` ... ``` + cleaned_text = response.text.strip() + if cleaned_text.startswith("```"): + lines = cleaned_text.split('\n') + if len(lines) >= 2: + cleaned_text = '\n'.join(lines[1:-1]) + + return cleaned_text + except Exception as e: + print(f"Error during translation after retries: {e}") + # Fallback to older model if 2.0 fails? + if "404" in str(e) and "gemini-2.0" in model_name: + print(" -> gemini-2.0-flash not found, falling back to gemini-1.5-flash") + try: + response = _generate_with_retry(client, "gemini-1.5-flash", prompt) + cleaned_text = response.text.strip() + if cleaned_text.startswith("```"): + lines = cleaned_text.split('\n') + if len(lines) >= 2: + cleaned_text = '\n'.join(lines[1:-1]) + return cleaned_text + except Exception as inner_e: + print(f"Fallback failed: {inner_e}") + + return None diff --git a/video_transcription/ai_transcriber_v2/utils.py b/video_transcription/ai_transcriber_v2/utils.py new file mode 100644 index 0000000..8c452bf --- /dev/null +++ b/video_transcription/ai_transcriber_v2/utils.py @@ -0,0 +1,28 @@ +import pysubs2 +import os + +def validate_and_repair_srt(srt_path): + """ + Validates an SRT file and attempts to repair it using pysubs2. + + Args: + srt_path (str): Path to the SRT file. + + Returns: + bool: True if valid/repaired, False if critical error. + """ + if not os.path.exists(srt_path): + return False + + print(f"Validating SRT: {srt_path}...") + try: + # Load the subtitle file. pysubs2 parser is robust and handles many errors automatically. + subs = pysubs2.load(srt_path) + + # Save it back ensures consistent formatting and fixes minor syntax issues + subs.save(srt_path) + print("SRT validation passed (file re-saved with correct formatting).") + return True + except Exception as e: + print(f"Warning: SRT validation failed: {e}") + return False diff --git a/video_transcription/diagnostic_scan.py b/video_transcription/diagnostic_scan.py new file mode 100755 index 0000000..b839249 --- /dev/null +++ b/video_transcription/diagnostic_scan.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +import os +import sys +import argparse + +def scan_for_missing_translations(folder_path, target_lang="English"): + print(f"--- Diagnostic Scan: Missing Translations ---") + print(f"Target Language: {target_lang}") + print(f"Scanning: {folder_path}\n") + + missing_count = 0 + total_srt = 0 + + # Files that need attention + missing_files = [] + + for root, dirs, files in os.walk(folder_path): + for file in files: + # Look for source SRT files + # Logic: Ends in .srt, but NOT .Lang.srt and NOT .Lang.deep_translate.srt + if file.endswith(".srt") and \ + not file.endswith(f".{target_lang}.srt") and \ + not file.endswith(f".{target_lang}.deep_translate.srt"): + + total_srt += 1 + source_path = os.path.join(root, file) + base_name = os.path.splitext(file)[0] + + # Check for existing translations + path_gemini = os.path.join(root, f"{base_name}.{target_lang}.srt") + path_deep = os.path.join(root, f"{base_name}.{target_lang}.deep_translate.srt") + + if not os.path.exists(path_gemini) and not os.path.exists(path_deep): + missing_count += 1 + missing_files.append(source_path) + + if missing_files: + print(f"Found {missing_count} files waiting for translation:\n") + for f in missing_files: + print(f" [MISSING] {f}") + else: + print("✅ All subtitles appear to be translated!") + + print("\n" + "="*40) + print(f"Summary:") + print(f"Total Source SRTs Found: {total_srt}") + print(f"Missing Translations: {missing_count}") + print("="*40) + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Diagnostic: Find missing translations") + parser.add_argument("folder", help="Path to the folder to scan") + parser.add_argument("lang", nargs="?", default="English", help="Target language (default: English)") + + args = parser.parse_args() + + if not os.path.exists(args.folder): + print(f"Error: Folder '{args.folder}' does not exist.") + sys.exit(1) + + scan_for_missing_translations(args.folder, args.lang) diff --git a/video_transcription/recover_and_fix.py b/video_transcription/recover_and_fix.py index 5a3e2cd..1214a8b 100755 --- a/video_transcription/recover_and_fix.py +++ b/video_transcription/recover_and_fix.py @@ -18,59 +18,26 @@ sys.path.append(os.path.join(script_dir, 'ai_transcriber')) from ai_transcriber.translator import translate_srt from ai_transcriber.utils import validate_and_repair_srt +import pysubs2 +from deep_translator import GoogleTranslator +from datetime import datetime -def re_embed_subtitles(video_path, srt_path, output_path=None): +def translate_fallback_free(source_srt_path, output_srt_path, target_lang="en"): """ - Re-embeds subtitles into an EXISTING video file, replacing the old tracks. - Similar to embed_subtitles but we output to a temp file then overwrite. + Fallback translation using deep-translator (free Google Translate). + Parses SRT, translates text line-by-line, and saves new SRT. """ - if not os.path.exists(video_path) or not os.path.exists(srt_path): - print("Error: Video or SRT file not found.") - return False +# ... (rest of translate_fallback_free is same) ... - temp_output = video_path + ".temp.mp4" - - print(f"Re-embedding subtitles into: {video_path}...") - - # We map 0:v and 0:a from source, but ignore existing subs (0:s?) - # Actually, simpler is to just map 0:v, 0:a and 1:0 (new sub) - # and tell ffmpeg to make the new one default. - # If the old file has subs, they might be kept unless we use -map -0:s (negative mapping) - - sub_codec = "mov_text" if video_path.lower().endswith(".mp4") else "srt" - - command = [ - "ffmpeg", - "-i", video_path, - "-i", srt_path, - "-map", "0:v", - "-map", "0:a", - "-map", "1:0", # The new subtitle - "-c", "copy", # Copy AV - "-c:s", sub_codec, # Encode Sub - "-disposition:s:0", "default", - "-metadata:s:s:0", "language=eng", - "-metadata:s:s:0", "title=English (AI Translated)", - "-y", - "-v", "error", - temp_output - ] - - try: - subprocess.run(command, check=True) - # Move temp over original - os.replace(temp_output, video_path) - print(f"✅ Fixed: {video_path}") - return True - except subprocess.CalledProcessError as e: - print(f"Error re-embedding: {e}") - if os.path.exists(temp_output): - os.remove(temp_output) - return False +# ... (rest of re_embed_subtitles is same) ... def process_recovery(folder_path, target_lang="English"): print(f"Scanning {folder_path} for incomplete translations...") + # Define log file + recovery_log_file = os.path.join(folder_path, "recovery_status.log") + print(f"Logging actions to: {recovery_log_file}") + count_fixed = 0 count_skipped = 0 @@ -79,20 +46,19 @@ def process_recovery(folder_path, target_lang="English"): for root, dirs, files in os.walk(folder_path): for file in files: # We are looking for the SOURCE SRT files primarily - if file.endswith(".srt") and not file.endswith(f".{target_lang}.srt"): + # Exclude our output files to avoid loops + if file.endswith(".srt") and \ + not file.endswith(f".{target_lang}.srt") and \ + not file.endswith(f".{target_lang}.deep_translate.srt"): + source_srt_path = os.path.join(root, file) - base_name = os.path.splitext(file)[0] # e.g. "movie" or "movie.subbed" + base_name = os.path.splitext(file)[0] - # Check if this is a "subbed" file's srt? - # Usually source srt is "movie.srt". - # Video might be "movie.subbed.mp4" (if original was deleted) - # OR "movie.mp4" (if not deleted) + # Check if Translation exists (Standard or Fallback) + path_gemini = os.path.join(root, f"{base_name}.{target_lang}.srt") + path_deep = os.path.join(root, f"{base_name}.{target_lang}.deep_translate.srt") - # Check if Translation exists - expected_translation = os.path.join(root, f"{base_name}.{target_lang}.srt") - - if os.path.exists(expected_translation): - # It's already translated, nothing to do unless we force re-embed + if os.path.exists(path_gemini) or os.path.exists(path_deep): continue print(f"\nFound untranslated transcript: {file}") @@ -103,39 +69,64 @@ def process_recovery(folder_path, target_lang="English"): new_srt_content = translate_srt(content, target_language=target_lang) - if not new_srt_content: - print("❌ Translation failed (API Limit?). Skipping.") - continue - - # Save translation - with open(expected_translation, "w", encoding="utf-8") as f: - f.write(new_srt_content) - validate_and_repair_srt(expected_translation) - - # Now, find the video file to update - # Case 1: Original name - video_candidates = [ - os.path.join(root, base_name + ".mp4"), - os.path.join(root, base_name + ".mkv"), - # Case 2: .subbed name (if original deleted) - os.path.join(root, base_name + ".subbed.mp4"), - ] - - found_video = None - for v in video_candidates: - if os.path.exists(v): - found_video = v - break - - if found_video: - print(f"Found video to fix: {found_video}") - if re_embed_subtitles(found_video, expected_translation): - count_fixed += 1 + success = False + method_used = "None" + final_srt_path = path_gemini # Default if success + + if new_srt_content: + # Save Gemini translation + with open(path_gemini, "w", encoding="utf-8") as f: + f.write(new_srt_content) + success = True + method_used = "Gemini" + final_srt_path = path_gemini else: - print("Warning: Could not find a corresponding video file to embed into.") + print("❌ Gemini API failed. Attempting Free Fallback...") + lang_map = { + "English": "en", "French": "fr", "Spanish": "es", + "German": "de", "Italian": "it", "Portuguese": "pt", + "Russian": "ru", "Japanese": "ja", "Chinese": "zh-CN" + } + target_code = lang_map.get(target_lang, "en") + + if translate_fallback_free(source_srt_path, path_deep, target_lang=target_code): + success = True + method_used = "DeepTranslate" + final_srt_path = path_deep + else: + print("❌ All translation methods failed. Skipping.") + continue + + if success: + # Log result + with open(recovery_log_file, "a", encoding="utf-8") as log: + log.write(f"{datetime.now().isoformat()} | {method_used} | {file} -> {os.path.basename(final_srt_path)}\n") + + validate_and_repair_srt(final_srt_path) + + # Now, find the video file to update + # Case 1: Original name + video_candidates = [ + os.path.join(root, base_name + ".mp4"), + os.path.join(root, base_name + ".mkv"), + # Case 2: .subbed name (if original deleted) + os.path.join(root, base_name + ".subbed.mp4"), + ] + + found_video = None + for v in video_candidates: + if os.path.exists(v): + found_video = v + break + + if found_video: + print(f"Found video to fix: {found_video}") + if re_embed_subtitles(found_video, final_srt_path): + count_fixed += 1 + else: + print("Warning: Could not find a corresponding video file to embed into.") print(f"\nRecovery Complete. Fixed {count_fixed} files.") - if __name__ == "__main__": if len(sys.argv) < 2: print("Usage: ./recover_and_fix.py [target_lang]") diff --git a/video_transcription/recover_and_fix_v2.py b/video_transcription/recover_and_fix_v2.py new file mode 100755 index 0000000..b8ffd18 --- /dev/null +++ b/video_transcription/recover_and_fix_v2.py @@ -0,0 +1,238 @@ +#!/usr/bin/env python3 +import os +import sys +import argparse +import subprocess +from dotenv import load_dotenv +from datetime import datetime +import pysubs2 +from deep_translator import GoogleTranslator + +# Load config +script_dir = os.path.dirname(os.path.abspath(__file__)) +env_path = os.path.abspath(os.path.join(script_dir, '../.env_files/.env.aitranscribe')) +if os.path.exists(env_path): + load_dotenv(env_path) +else: + load_dotenv() + +# Add ai_transcriber_v2 to path so we can import modules +sys.path.append(os.path.join(script_dir, 'ai_transcriber_v2')) + +from ai_transcriber_v2.translator import translate_srt +from ai_transcriber_v2.utils import validate_and_repair_srt + +def translate_fallback_free(source_srt_path, output_srt_path, target_lang="en"): + """ + Fallback translation using deep-translator (free Google Translate). + Parses SRT, translates text line-by-line, and saves new SRT. + """ + print(f" [Free Fallback] Translating {source_srt_path}...") + + subs = None + encodings_to_try = ['utf-8', 'shift_jis', 'euc_jp', 'latin-1', 'cp1252', 'utf-16'] + + for enc in encodings_to_try: + try: + subs = pysubs2.load(source_srt_path, encoding=enc) + break + except Exception: + continue + + if subs is None: + print(f" [Free Fallback] Critical Error: Could not decode file with standard encodings.") + return False + + try: + translator = GoogleTranslator(source='auto', target=target_lang) + + for line in subs: + text = line.text.strip() + if text: + # Sanity check: Skip lines that are too long (likely garbage/corruption) + if len(text) > 4000: + print(f" Warning: Skipping line with excessive length ({len(text)} chars). Likely corrupted.") + continue + + try: + original_text = text.replace(r"\N", " ") + translated_text = translator.translate(original_text) + if translated_text: + line.text = translated_text + except Exception as e: + print(f" Warning: Failed to translate line: {e}") + + subs.save(output_srt_path) + print(f" [Free Fallback] Saved to {output_srt_path}") + return True + except Exception as e: + print(f" [Free Fallback] Critical Error: {e}") + return False + +def re_embed_subtitles(video_path, srt_path, output_path=None): + """ + Re-embeds subtitles into an EXISTING video file, replacing the old tracks. + Uses robust flags to handle bad metadata. + """ + if not os.path.exists(video_path) or not os.path.exists(srt_path): + print("Error: Video or SRT file not found.") + return False + + temp_output = video_path + ".temp.mp4" + print(f"Re-embedding subtitles into: {video_path}...") + + sub_codec = "mov_text" if video_path.lower().endswith(".mp4") else "srt" + + command = [ + "ffmpeg", + "-ignore_editlist", "1", + "-i", video_path, + "-i", srt_path, + "-map", "0:v", + "-map", "0:a", + "-map", "1:0", + "-c", "copy", + "-c:s", sub_codec, + "-disposition:s:0", "default", + "-metadata:s:s:0", "language=eng", + "-metadata:s:s:0", "title=English (AI Translated)", + "-max_interleave_delta", "0", + "-avoid_negative_ts", "make_zero", + "-y", + "-v", "error", + temp_output + ] + + try: + subprocess.run(command, check=True) + os.replace(temp_output, video_path) + print(f"✅ Fixed: {video_path}") + return True + except subprocess.CalledProcessError as e: + print(f"Error re-embedding: {e}") + if os.path.exists(temp_output): + os.remove(temp_output) + return False + +def process_recovery(folder_path, target_lang="English", prefer_deep=False): + print(f"Scanning {folder_path} for incomplete translations (V2)...") + if prefer_deep: + print("Preference: DeepTranslate (Google Translate Free) > Gemini") + else: + print("Preference: Gemini (API) > DeepTranslate") + + recovery_log_file = os.path.join(folder_path, "recovery_status.log") + print(f"Logging actions to: {recovery_log_file}") + + count_fixed = 0 + video_extensions = ('.mp4', '.mkv', '.mov', '.avi') + + for root, dirs, files in os.walk(folder_path): + for file in files: + if file.endswith(".srt") and \ + not file.endswith(f".{target_lang}.srt") and \ + not file.endswith(f".{target_lang}.deep_translate.srt"): + + source_srt_path = os.path.join(root, file) + base_name = os.path.splitext(file)[0] + + path_gemini = os.path.join(root, f"{base_name}.{target_lang}.srt") + path_deep = os.path.join(root, f"{base_name}.{target_lang}.deep_translate.srt") + + if os.path.exists(path_gemini) or os.path.exists(path_deep): + continue + + print(f"\nFound untranslated transcript: {file}") + + content = None + # extended list to include common Japanese encodings + encodings_to_try = ['utf-8', 'shift_jis', 'euc_jp', 'latin-1', 'cp1252', 'utf-16'] + + for enc in encodings_to_try: + try: + with open(source_srt_path, "r", encoding=enc) as f: + content = f.read() + break # Success + except UnicodeDecodeError: + continue + + if content is None: + print(f"❌ Error: Could not decode {file} with any standard encoding. Skipping.") + continue + + # Helpers for translation attempts + def try_gemini(): + res = translate_srt(content, target_language=target_lang) + if res: + with open(path_gemini, "w", encoding="utf-8") as f: + f.write(res) + return True, path_gemini, "Gemini" + return False, None, None + + def try_deep(): + lang_map = { + "English": "en", "French": "fr", "Spanish": "es", + "German": "de", "Italian": "it", "Portuguese": "pt", + "Russian": "ru", "Japanese": "ja", "Chinese": "zh-CN" + } + target_code = lang_map.get(target_lang, "en") + if translate_fallback_free(source_srt_path, path_deep, target_lang=target_code): + return True, path_deep, "DeepTranslate" + return False, None, None + + success = False + method_used = "None" + final_srt_path = None + + if prefer_deep: + # 1. Try DeepTranslate + success, final_srt_path, method_used = try_deep() + if not success: + print("❌ DeepTranslate failed. Attempting Gemini fallback...") + success, final_srt_path, method_used = try_gemini() + else: + # 1. Try Gemini + success, final_srt_path, method_used = try_gemini() + if not success: + print("❌ Gemini API failed. Attempting Free Fallback...") + success, final_srt_path, method_used = try_deep() + + if success and final_srt_path: + # Log result + with open(recovery_log_file, "a", encoding="utf-8") as log: + log.write(f"{datetime.now().isoformat()} | {method_used} | {file} -> {os.path.basename(final_srt_path)}\n") + + validate_and_repair_srt(final_srt_path) + + video_candidates = [ + os.path.join(root, base_name + ".mp4"), + os.path.join(root, base_name + ".mkv"), + os.path.join(root, base_name + ".subbed.mp4"), + ] + + found_video = None + for v in video_candidates: + if os.path.exists(v): + found_video = v + break + + if found_video: + print(f"Found video to fix: {found_video}") + if re_embed_subtitles(found_video, final_srt_path): + count_fixed += 1 + else: + print("Warning: Could not find a corresponding video file to embed into.") + else: + print("❌ All translation methods failed. Skipping.") + + print(f"\nRecovery Complete. Fixed {count_fixed} files.") + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Recover and Fix Translations (V2)") + parser.add_argument("folder", help="Path to the folder to scan") + parser.add_argument("lang", nargs="?", default="English", help="Target language (default: English)") + parser.add_argument("--prefer-deep", action="store_true", help="Prefer DeepTranslate (Free) over Gemini API") + + args = parser.parse_args() + + process_recovery(args.folder, args.lang, args.prefer_deep) \ No newline at end of file diff --git a/video_transcription/run_v2.bat b/video_transcription/run_v2.bat new file mode 100644 index 0000000..0f6712e --- /dev/null +++ b/video_transcription/run_v2.bat @@ -0,0 +1,35 @@ +@echo off +TITLE AI Video Transcriber V2 + +REM Check if Python is installed +python --version >nul 2>&1 +IF %ERRORLEVEL% NEQ 0 ( + echo Error: Python is not installed or not in your PATH. + echo Please install Python from https://www.python.org/ + PAUSE + EXIT /B +) + +REM Check if FFmpeg is installed +ffmpeg -version >nul 2>&1 +IF %ERRORLEVEL% NEQ 0 ( + echo Error: FFmpeg is not installed or not in your PATH. + echo Please download FFmpeg and add bin folder to your Environment Variables. + PAUSE + EXIT /B +) + +REM Check for virtual environment +IF NOT EXIST ".venv" ( + echo Creating Virtual Environment... + python -m venv .venv + echo Installing dependencies (this may take a while)... + .venv\Scripts\pip install -r ai_transcriber_v2\requirements.txt + .venv\Scripts\pip install python-dotenv +) + +REM Run the V2 launcher +echo Starting Wizard... +.venv\Scripts\python run_v2.py + +PAUSE diff --git a/video_transcription/run_v2.py b/video_transcription/run_v2.py new file mode 100755 index 0000000..504cbb0 --- /dev/null +++ b/video_transcription/run_v2.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +import os +import sys +import subprocess +import shutil + +def clear_screen(): + os.system('cls' if os.name == 'nt' else 'clear') + +def get_input(prompt, default=None): + """Helper to get input with a default value.""" + if default: + user_input = input(f"{prompt} [{default}]: ").strip() + return user_input if user_input else default + else: + return input(f"{prompt}: ").strip() + +def get_yes_no(prompt, default="y"): + """Helper to get boolean input.""" + display_default = "Y/n" if default.lower() in ["y", "yes"] else "y/N" + choice = get_input(f"{prompt} ({display_default})", default).lower() + return choice in ["y", "yes", "true", "1"] + +def print_header(): + print("==========================================") + print(" AI Video Transcriber & Translator V2") + print(" (Powered by Google GenAI SDK)") + print("==========================================") + print("") + +def main(): + clear_screen() + print_header() + + # Try to load the .env file so the wizard knows what's already configured + try: + from dotenv import load_dotenv + script_dir = os.path.dirname(os.path.abspath(__file__)) + env_path = os.path.abspath(os.path.join(script_dir, '../.env_files/.env.aitranscribe')) + if os.path.exists(env_path): + load_dotenv(env_path) + except ImportError: + pass + + # 1. Input File/Folder + while True: + input_path = get_input("Enter the path to the video file or folder") + + # Clean up input + input_path = input_path.strip("'\"") + input_path = input_path.replace(r'\ ', ' ') + + # Expand user (~) and resolve absolute path + input_path = os.path.abspath(os.path.expanduser(input_path)) + + if os.path.exists(input_path): + break + print(f"Error: Path '{input_path}' does not exist. Please try again.\n") + + print(f"Selected: {input_path}\n") + + # 2. Languages + source_lang = get_input("Source Language (e.g., French, es)", default="auto") + target_lang = get_input("Target Language for translation", default="English") + print("") + + # 3. Model Size + print("Model Size Options: tiny, base, small, medium, large, auto") + model_size = get_input("Whisper Model Size", default="auto") + print("") + + # 4. Features + do_cleanup = get_yes_no("Cleanup temporary audio files after processing?", default="y") + do_embed = get_yes_no("Embed subtitles into the video file (Soft Subs)?", default="y") + do_diarize = get_yes_no("Enable Speaker Diarization (Identify speakers)?", default="n") + + do_delete_source = False + if do_embed: + print("\n⚠️ WARNING: Using this next option will PERMANENTLY DELETE the original video files.") + print(" It will only run if the new subtitled video is successfully created.") + do_delete_source = get_yes_no("Delete original source files after embedding?", default="n") + + hf_token = None + if do_diarize: + if not os.getenv("HF_TOKEN"): + print("\nSpeaker Diarization requires a HuggingFace Token.") + hf_token = get_input("Enter your HuggingFace Token (hidden)", default="") + else: + print("Using HF_TOKEN from environment.") + + # 5. Build Command + # Point to v2 main script + script_dir = os.path.dirname(os.path.abspath(__file__)) + main_script = os.path.join(script_dir, "ai_transcriber_v2", "main.py") + + cmd = [sys.executable, main_script, input_path] + + cmd.extend(["--lang", target_lang]) + cmd.extend(["--model", model_size]) + + if source_lang != "auto": + cmd.extend(["--source-lang", source_lang]) + + if do_cleanup: + cmd.append("--cleanup") + + if do_embed: + cmd.append("--embed") + + if do_delete_source: + cmd.append("--delete-source") + + if do_diarize: + cmd.append("--diarize") + if hf_token: + cmd.extend(["--hf-token", hf_token]) + + # 6. Confirmation and Execution + clear_screen() + print_header() + print("Configuration Complete!") + print("-" * 30) + print(f"Input: {input_path}") + print(f"Source Lang: {source_lang}") + print(f"Target Lang: {target_lang}") + print(f"Model: {model_size}") + print(f"Cleanup: {do_cleanup}") + print(f"Embed Subs: {do_embed}") + print(f"Delete Src: {do_delete_source}") + print(f"Diarization: {do_diarize}") + print("-" * 30) + + if not get_yes_no("Run this job now?", default="y"): + print("Aborted.") + sys.exit(0) + + print("\nStarting Job (V2)...") + + try: + # Pass environment variables including HF_TOKEN if set + env = os.environ.copy() + if hf_token: + env["HF_TOKEN"] = hf_token + + subprocess.run(cmd, check=True, env=env) + print("\n✅ Job Complete!") + except subprocess.CalledProcessError as e: + print(f"\n❌ Job Failed with error code {e.returncode}") + except KeyboardInterrupt: + print("\nJob interrupted by user.") + +if __name__ == "__main__": + main()