david/personal_development
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

syncing scripts

Browse Source
This commit is contained in:
david
2026-01-11 15:10:56 -05:00
parent 362c06dab4
commit 203fc69393
24 changed files with 1982 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video_transcription/ai_transcriber/__pycache__/translator.cpython-313.pyc
BIN
View File
Binary file not shown.
video_transcription/ai_transcriber/main.py
+135 -86
View File
@@ -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:
video_transcription/ai_transcriber/translator.py
+53
View File
@@ -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.