# src/tools/audio_utils.py
import re
import os
from dotenv import load_dotenv
import zipfile
import requests
from indic_transliteration.sanscript import transliterate
from indic_transliteration import sanscript
import json
import time
from google.generativeai import GenerativeModel, configure
from google.generativeai.types import GenerationConfig
try:
from google import genai
from google.genai import types
except ImportError:
genai = None
types = None
print("Warning: google-genai package not installed. Gemini TTS will not work. Install with: pip install google-genai")
from src.tools.prompt_utils import get_regional_translation_prompt, get_named_entity_identification_prompt, get_pun_translation_prompt
from typing import Literal
import subprocess
import wave
import io
from pydub import AudioSegment
import tempfile
import shutil
import traceback
# Exports - ensure these names are importable
__all__ = [
"audio_fn_from_string",
"audio_fn",
"pun_audio_fn_from_string",
"trim_audio_to_max_duration",
"pad_audio_to_duration",
"combine_audio_with_video",
"pun_generate_segment_audios",
"pun_check_audio_files",
"pun_get_audio_duration"
]
# Load environment variables and configure Gemini
load_dotenv()
GEMINI_API_KEY = os.environ.get('GEMINI_API_KEY')
# Optionally allow overriding technical terms through env var (comma separated)
_env_terms = os.environ.get("TECHNICAL_TERMS", "")
DEFAULT_TECHNICAL_TERMS = [
"CNN", "RNN", "LSTM", "GRU", "Transformer", "Transformer-based", "BERT", "GPT",
"OpenAI", "Gemini", "ffmpeg", "Playwright", "TensorFlow", "PyTorch", "scikit-learn",
"API", "MSE", "R-squared", "R2", "accuracy", "precision", "recall", "F1", "BLEU",
"AUC", "ROC", "Linear Regression", "logistic regression", "SGD", "Adam", "epoch",
"batch", "GPU", "TPU", "embedding", "token", "tokenizer"
]
TECHNICAL_TERMS = [t.strip() for t in (_env_terms.split(",") if _env_terms else []) if t.strip()] or DEFAULT_TECHNICAL_TERMS
# Configure Gemini API globally if key present
if GEMINI_API_KEY:
try:
configure(api_key=GEMINI_API_KEY)
gemini = GenerativeModel('gemini-2.0-flash')
except Exception:
gemini = None
else:
gemini = None
print('Warning: GEMINI_API_KEY not found in environment variables. Some functions may not work.')
LANGUAGE_TO_SCRIPT_MAP = {
"hindi": sanscript.DEVANAGARI,
"punjabi": sanscript.GURMUKHI,
"gujarati": sanscript.GUJARATI,
"marathi": sanscript.DEVANAGARI,
"kannada": sanscript.KANNADA,
}
# ---------------- Helpers ----------------
def get_temp_dir():
return tempfile.gettempdir()
def pun_remove_special_characters(s: str) -> str:
if s is None:
return s
s = s.replace('`', '').replace('\xa0', '').replace(' ', '')
s = s.replace('\u200b', '') # zero-width space
return s
def _clean_text_for_tts(text: str) -> str:
if text is None:
return ""
text = text.replace('"', '')
text = text.replace("'", '')
text = text.replace('`', '')
text = ' '.join(text.split())
text = text.replace('/', ' or ')
return text.strip()
# Utility: create safe markers for preserving tokens
def _make_marker(term: str) -> str:
# safe marker unlikely to be touched by translation
safe = "__KEEPTERM__" + re.sub(r'\W+', '_', term) + "__"
return safe
def _protect_technical_terms(text: str) -> (str, dict):
"""
Replace occurrences of technical terms with markers, return (protected_text, mapping)
mapping: marker -> original term
"""
mapping = {}
if not text:
return text, mapping
# Sort by length to protect longest terms first
terms_sorted = sorted(TECHNICAL_TERMS, key=lambda x: -len(x))
protected = text
for term in terms_sorted:
# Word-boundary, case-insensitive
# allow terms containing spaces (e.g., "Linear Regression")
pattern = re.compile(re.escape(term), re.IGNORECASE)
def _repl(m):
orig = m.group(0)
marker = _make_marker(term)
mapping[marker] = orig # store original matched case
return marker
protected = pattern.sub(_repl, protected)
return protected, mapping
def _restore_markers(text: str, mapping: dict) -> str:
if not mapping:
return text
restored = text
for marker, orig in mapping.items():
restored = restored.replace(marker, orig)
return restored
# Heuristic to decide whether to keep a token as English / technical
def _looks_like_english_token(token: str) -> bool:
if not token:
return False
# If it's explicitly in technical terms (case-insensitive)
for t in TECHNICAL_TERMS:
if token.lower() == t.lower():
return True
# Mostly ASCII chars
ascii_ratio = sum(1 for ch in token if ord(ch) < 128) / max(1, len(token))
if ascii_ratio > 0.85:
return True
# if contains digits or common code symbols or CamelCase
if re.search(r'[A-Z][a-z]+[A-Z]', token) or re.search(r'[\d\(\)\=\+\-_/\.]', token):
return True
# Contains dot notation or file extensions
if '.' in token and len(token) <= 40:
return True
return False
# --- Keep English named entities as-is for Hinglish ---
def transliterate_english_to_native_script(sentence, target_language):
if sentence is None:
return ""
if isinstance(sentence, str):
# If the token looks like an english/technical token, keep as-is
if _looks_like_english_token(sentence):
return sentence.strip()
ascii_ratio = sum(1 for ch in sentence if ord(ch) < 128) / max(1, len(sentence))
if ascii_ratio > 0.8:
return sentence.strip()
script = LANGUAGE_TO_SCRIPT_MAP.get((target_language or "").lower())
if not script:
return sentence
try:
return transliterate(sentence, sanscript.ITRANS, script)
except Exception:
return sentence
# ---------------- Translation helpers (Gemini) ----------------
def translate_to_regional(target_language, text):
if not GEMINI_API_KEY or not gemini:
print("Warning: Gemini not configured; translate_to_regional returning original text.")
return text
try:
# protect technical terms with markers so translator won't alter them
protected_text, mapping = _protect_technical_terms(text or "")
cleaned_text = re.sub(r'``````', '', protected_text or '', flags=re.DOTALL)
cleaned_text = re.sub(r'```.*?```', '', cleaned_text, flags=re.DOTALL)
prompt = get_regional_translation_prompt(target_language, cleaned_text)
response = gemini.generate_content(
prompt,
generation_config=GenerationConfig(response_mime_type="application/json")
)
raw_text = response.text
# attempt parse JSON
try:
result = json.loads(raw_text)
translated = result.get("translation", text)
# restore protected terms
return _restore_markers(translated, mapping)
except json.JSONDecodeError:
match = re.search(r"\{.*\}", raw_text, re.DOTALL)
if match:
try:
result = json.loads(match.group(0))
translated = result.get("translation", text)
return _restore_markers(translated, mapping)
except Exception:
pass
if (target_language or "").lower() not in ["english", "en"]:
return _restore_markers(raw_text.strip(), mapping)
return text
except Exception as e:
print(f":x: Gemini error: {str(e)}")
try:
if 'raw_text' in locals() and (target_language or "").lower() not in ["english", "en"]:
return _restore_markers(raw_text.strip(), {})
except Exception:
pass
return text
def pun_translate_to_regional(target_language: str, text: str) -> str:
if not GEMINI_API_KEY or not gemini:
print("Warning: Gemini not configured; pun_translate_to_regional returning original text.")
return text
try:
protected_text, mapping = _protect_technical_terms(text or "")
cleaned_text = re.sub(r'```python.*?```', '', protected_text or '', flags=re.DOTALL)
cleaned_text = re.sub(r'```.*?```', '', cleaned_text, flags=re.DOTALL)
prompt = get_pun_translation_prompt(target_language, cleaned_text)
response = gemini.generate_content(prompt)
raw_text = response.text.strip()
match = re.search(r'\{.*\}', raw_text, re.DOTALL)
if match:
try:
result = json.loads(match.group(0))
translated = result.get('translation', text)
return _restore_markers(translated, mapping)
except Exception:
pass
return _restore_markers(raw_text, mapping)
except Exception as e:
print(f':x: Gemini error: {str(e)}')
try:
if 'raw_text' in locals() and (target_language or "").lower() not in ["english", "en"]:
return _restore_markers(raw_text.strip(), {})
except Exception:
pass
return text
# ---------------- Named entity identification ----------------
def indentify_named_entities(input_text):
prompt = get_named_entity_identification_prompt()
if not GEMINI_API_KEY or not gemini:
print('Warning: GEMINI_API_KEY not found. Returning input wrapped in triple quotes.')
return f"'''{input_text}'''"
try:
response = gemini.generate_content(f"{prompt}\n\nText to process:\n{input_text}")
oration = response.text
return oration
except Exception as e:
print(f'Error identifying named entities with Gemini: {e}')
return f"'''{input_text}'''"
def pun_indentify_named_entities(input_text: str) -> str:
prompt = get_named_entity_identification_prompt()
if not GEMINI_API_KEY or not gemini:
print('Warning: GEMINI_API_KEY not found. Returning input wrapped in triple quotes.')
return f"'''{input_text}'''"
try:
response = gemini.generate_content(f"{prompt}\n\nText to process:\n{input_text}")
oration = response.text
return oration
except Exception as e:
print(f'Error identifying named entities with Gemini: {e}')
return f"'''{input_text}'''"
# ---------------- Text -> dictionary parsing (shared) ----------------
def process_text_into_dictionary(input_text):
main_text_match = re.search(r"'''(.*?)'''", input_text, re.DOTALL)
if not main_text_match:
raise ValueError("No text found within triple single quotes.")
main_text = main_text_match.group(1).strip()
pattern = r"((.*?))|([^<]+)"
result = {}
hindi_counter = 1
english_counter = 1
for match in re.finditer(pattern, main_text):
if match.group(2):
result[f"english_{english_counter}"] = match.group(2).strip()
english_counter += 1
elif match.group(3):
hindi_text = match.group(3).strip()
if hindi_text:
result[f"hindi_{hindi_counter}"] = hindi_text
hindi_counter += 1
return result
def pun_process_text_into_dictionary(input_text: str) -> dict:
main_text_match = re.search(r"'''(.*?)'''", input_text, re.DOTALL)
if not main_text_match:
main_text = input_text.strip()
else:
main_text = main_text_match.group(1).strip()
pattern = r'((.*?))|([^<]+)'
result = {}
hindi_counter = 1
english_counter = 1
for match in re.finditer(pattern, main_text):
if match.group(2):
result[f'english_{english_counter}'] = match.group(2).strip()
english_counter += 1
elif match.group(3):
hindi_text = match.group(3).strip()
if hindi_text:
result[f'hindi_{hindi_counter}'] = hindi_text
hindi_counter += 1
return result
# ---------------- Dictionary -> final text (Hinglish behavior) ----------------
def process_dictionary(input_dict, target_language):
outputs = []
for key, value in sorted(input_dict.items(), key=lambda kv: kv[0]):
if key.startswith("hindi") or key.startswith("regional") or key.startswith("marathi") or key.startswith("kannada"):
# Protect technical terms, translate, then restore
translated = translate_to_regional(target_language, value)
# After translation, ensure technical terms remain original if translation altered them
# Best-effort: try restoring markers (translate_to_regional already restores markers)
outputs.append(translated.strip())
elif key.startswith("english"):
# keep english as-is (NER tagged)
outputs.append(value.strip())
else:
outputs.append(value.strip())
return " ".join([o for o in outputs if o])
def pun_process_dictionary(input_dict: dict, target_language: str) -> str:
outputs = []
for key, value in input_dict.items():
if key.startswith('hindi'):
outputs.append(pun_translate_to_regional(text=value, target_language=target_language))
elif key.startswith('english'):
outputs.append(value.strip())
return ' '.join(outputs)
def pun_convert_english_to_devanaigiri_script(sentence):
try:
devanagari_text = transliterate(sentence, sanscript.ITRANS, sanscript.DEVANAGARI)
return devanagari_text
except Exception:
return sentence
# ---------------- TTS: Gemini English + regional path ----------------
def generate_english_tts_with_gemini(text: str, voice_name: str, output_path: str, tts_gender: str = None) -> str:
if not genai or not types:
raise ImportError("google-genai package is required for Gemini TTS. Install with: pip install google-genai")
if not GEMINI_API_KEY:
raise ValueError("GEMINI_API_KEY not found in environment variables")
# Voice mapping: only voices supported by Gemini TTS API
# Based on API test results: charon, fenrir, puck, aoede work
# Removed: titan, lyra, aura, sol (not supported by API)
# Additional supported voices: achernar, achird, algenib, algieba, alnilam, autonoe,
# callirrhoe, despina, enceladus, erinome, gacrux, iapetus, kore, laomedeia, leda,
# orus, pulcherrima, rasalgethi, sadachbia, sadaltager, schedar, sulafat, umbriel,
# vindemiatrix, zephyr, zubenelgenubi
# Map user-friendly voice names to Gemini voice keys (male/female map to same prebuilt voice)
voice_map = {
"achernar": {"male": "achernar", "female": "achernar"},
"achird": {"male": "achird", "female": "achird"},
"algenib": {"male": "algenib", "female": "algenib"},
"algieba": {"male": "algieba", "female": "algieba"},
"alnilam": {"male": "alnilam", "female": "alnilam"},
"aoede": {"male": "aoede", "female": "aoede"},
"autonoe": {"male": "autonoe", "female": "autonoe"},
"callirrhoe": {"male": "callirrhoe", "female": "callirrhoe"},
"charon": {"male": "charon", "female": "charon"},
"despina": {"male": "despina", "female": "despina"},
"enceladus": {"male": "enceladus", "female": "enceladus"},
"erinome": {"male": "erinome", "female": "erinome"},
"fenrir": {"male": "fenrir", "female": "fenrir"},
"gacrux": {"male": "gacrux", "female": "gacrux"},
"iapetus": {"male": "iapetus", "female": "iapetus"},
"kore": {"male": "kore", "female": "kore"},
"laomedeia": {"male": "laomedeia", "female": "laomedeia"},
"leda": {"male": "leda", "female": "leda"},
"orus": {"male": "orus", "female": "orus"},
"puck": {"male": "puck", "female": "puck"},
"pulcherrima": {"male": "pulcherrima", "female": "pulcherrima"},
"rasalgethi": {"male": "rasalgethi", "female": "rasalgethi"},
"sadachbia": {"male": "sadachbia", "female": "sadachbia"},
"sadaltager": {"male": "sadaltager", "female": "sadaltager"},
"schedar": {"male": "schedar", "female": "schedar"},
"sulafat": {"male": "sulafat", "female": "sulafat"},
"umbriel": {"male": "umbriel", "female": "umbriel"},
"vindemiatrix":{"male": "vindemiatrix", "female": "vindemiatrix"},
"zephyr": {"male": "zephyr", "female": "zephyr"},
"zubenelgenubi": {"male": "zubenelgenubi", "female": "zubenelgenubi"},
}
# Allowed voices list based on actual Gemini TTS API supported voices (lowercase)
allowed_voices = {
"achernar", "achird", "algenib", "algieba", "alnilam", "aoede", "autonoe",
"callirrhoe", "charon", "despina", "enceladus", "erinome", "fenrir", "gacrux",
"iapetus", "kore", "laomedeia", "leda", "orus", "puck", "pulcherrima",
"rasalgethi", "sadachbia", "sadaltager", "schedar", "sulafat", "umbriel",
"vindemiatrix", "zephyr", "zubenelgenubi"
}
requested = (voice_name or "").strip()
# Default to 'female' if no gender provided to match wrapper defaults (audio_fn_from_string default is 'female')
gender_key = 'male' if str(tts_gender or '').lower().startswith('m') else 'female'
if not requested:
gemini_voice = 'aoede'
else:
mapping = voice_map.get(requested.lower())
if mapping:
gemini_voice = mapping.get(gender_key)
else:
# If user supplied a direct Gemini voice name, accept it (lowercase)
gemini_voice = requested.lower()
# Validate against allowed voices (case-insensitive); if invalid, fallback to a safe default.
if gemini_voice.lower() not in allowed_voices:
print(f"[WARN] Requested Gemini voice '{gemini_voice}' not in allowed list; falling back to 'aoede'.")
gemini_voice = 'aoede'
else:
# Ensure voice name is lowercase as API expects lowercase
gemini_voice = gemini_voice.lower()
try:
client = genai.Client(api_key=GEMINI_API_KEY)
response = client.models.generate_content(
model='gemini-2.5-flash-preview-tts',
contents=text,
config=types.GenerateContentConfig(
response_modalities=["AUDIO"],
speech_config=types.SpeechConfig(
voice_config=types.VoiceConfig(
prebuilt_voice_config=types.PrebuiltVoiceConfig(
voice_name=gemini_voice.lower() # API expects lowercase
)
)
)
)
)
audio_data = b''
if hasattr(response, 'candidates') and response.candidates:
for candidate in response.candidates:
if hasattr(candidate, 'content') and hasattr(candidate.content, 'parts'):
for part in candidate.content.parts:
if hasattr(part, 'inline_data') and part.inline_data:
if hasattr(part.inline_data, 'data'):
audio_data += part.inline_data.data
elif hasattr(part.inline_data, '_raw_data'):
audio_data += part.inline_data._raw_data
if not audio_data and hasattr(response, 'text'):
import base64
try:
audio_data = base64.b64decode(response.text)
except Exception:
pass
if not audio_data:
raise ValueError('No audio data received from Gemini API. The response format may have changed.')
temp_wav = output_path.replace('.mp3', '_temp.wav')
try:
with wave.open(temp_wav, 'wb') as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(24000)
wav_file.writeframes(audio_data)
except Exception as e:
print(f"[WARN] Could not write temp WAV: {e}")
with open(output_path, 'wb') as f:
f.write(audio_data)
return output_path
try:
subprocess.run([
'ffmpeg', '-y', '-i', temp_wav,
'-acodec', 'libmp3lame', '-ar', '44100', '-ac', '2', '-b:a', '128k',
output_path
], check=True, capture_output=True, text=True)
if os.path.exists(temp_wav):
os.remove(temp_wav)
return output_path
except subprocess.CalledProcessError as e:
print(f'[ERROR] FFmpeg conversion failed: {e.stderr}')
with open(output_path, 'wb') as f:
f.write(audio_data)
return output_path
except FileNotFoundError:
print('[ERROR] ffmpeg not found. Please install ffmpeg and add it to PATH')
with open(output_path, 'wb') as f:
f.write(audio_data)
return output_path
except Exception as e:
traceback.print_exc()
raise
# ---------------- Legacy batching (no-op kept) ----------------
def process_tts_in_batches(text, target_language, gender, final_speech_file_path, char_limit=300):
return None
# ---------------- Higher-level generation helpers (Hinglish by default) ----------------
def generate_tts_static_words(input_text, target_language, gender, speech_file_path):
text = indentify_named_entities(input_text)
dictionary = process_text_into_dictionary(text)
final_text = process_dictionary(dictionary, target_language)
return process_tts_in_batches(final_text, target_language, gender, speech_file_path)
def generate_tts_indian(input_text, target_language, gender, speech_file_path):
regional_text = translate_to_regional(target_language, input_text)
if regional_text is None:
return None
return process_tts_in_batches(regional_text, target_language, gender, speech_file_path)
def generate_tts_hinglish(input_text, target_language, gender, speech_file_path):
return generate_tts_static_words(input_text, target_language, gender, speech_file_path)
def generate_tts_hinglish(input_text, target_language, gender, speech_file_path):
"""Generate TTS for Hinglish (mix of Hindi/regional + English named entities)."""
return generate_tts_static_words(input_text, target_language, gender, speech_file_path)
# ---------------- Main Serverless TTS ----------------
def _generate_speech(input_text, speech_file_path, tts_client, target_language, tts_gender, tts_voice_name, toggle_hinglish):
if not input_text or not input_text.strip():
print(f"[WARN] Input text empty, skipping TTS for {speech_file_path}")
return None
clean_text = _clean_text_for_tts(input_text)
target_language_lower = (target_language or "english").lower()
use_hinglish = str(toggle_hinglish).lower() in ['true', '1', 'yes', 'y']
for attempt in range(1, 4):
try:
if target_language_lower in ["english", "en"]:
result = generate_english_tts_with_gemini(clean_text, tts_voice_name or 'nova', speech_file_path, tts_gender)
if result and os.path.exists(result):
return speech_file_path
else:
if use_hinglish:
text_with_entities = indentify_named_entities(clean_text)
text_dict = process_text_into_dictionary(text_with_entities)
final_text = process_dictionary(text_dict, target_language)
else:
final_text = translate_to_regional(target_language, clean_text)
final_text = final_text or clean_text
result = generate_english_tts_with_gemini(final_text, tts_voice_name or 'nova', speech_file_path, tts_gender)
if result and os.path.exists(result):
return result
except Exception as e:
print(f"[ERROR] TTS attempt {attempt} failed: {e}")
time.sleep(1.5)
raise RuntimeError(f"[FAIL] TTS failed after all attempts for text: {clean_text[:50]}...")
# ---------------- Wrappers (Hinglish default) ----------------
def audio_fn_from_string(input_text, folder_path, target_language='english', tts_gender='female', tts_voice_name='Puck', toggle_hinglish=True, client=None, file_name_prefix="speech"):
folder_path = folder_path if folder_path and os.access(folder_path, os.W_OK) else get_temp_dir()
os.makedirs(folder_path, exist_ok=True)
speech_file_path = os.path.join(folder_path, f"{file_name_prefix}.mp3")
return _generate_speech(input_text, speech_file_path, client, target_language, tts_gender, tts_voice_name, toggle_hinglish)
def audio_fn(text_file_path, target_language='english', tts_gender='female', tts_voice_name='Puck', toggle_hinglish=True, client=None):
with open(text_file_path, "r", encoding="utf-8") as f:
input_text = f.read()
folder_path = os.path.dirname(text_file_path) or get_temp_dir()
base_filename = os.path.splitext(os.path.basename(text_file_path))[0]
speech_file_path = os.path.join(folder_path, f"{base_filename}.mp3")
return _generate_speech(input_text, speech_file_path, client, target_language, tts_gender, tts_voice_name, toggle_hinglish)
# ---------------- pun_ wrappers (consistent behavior) ----------------
def pun_process_dictionary(input_dict: dict, target_language: str) -> str:
outputs = []
for key, value in input_dict.items():
if key.startswith('hindi'):
outputs.append(pun_translate_to_regional(text=value, target_language=target_language))
elif key.startswith('english'):
outputs.append(value.strip())
return ' '.join(outputs)
def pun_audio_fn_from_string(text_input: str, client = None,
target_language: str = 'english', tts_gender: str = 'female', tts_voice_name: str = 'nova',
toggle_hinglish: bool = True, text_source_type: Literal['file', 'string']='file',
file_name_prefix: str='speech', output_dir: str | None = None) -> (str | None):
print('==============================================')
try:
if text_source_type == 'file':
with open(text_input, 'r', encoding='utf-8') as file:
input_text = file.read()
if not input_text.strip():
print(f'Warning: Input text file is empty, skipping TTS for {text_input}')
return None
directory = os.path.dirname(text_input)
base_filename = os.path.splitext(os.path.basename(text_input))[0]
speech_file_path = os.path.join(directory, f'{base_filename}.mp3')
elif text_source_type == 'string':
input_text = text_input
if not output_dir:
raise ValueError("output_dir must be provided when text_source_type is 'string'")
os.makedirs(output_dir, exist_ok=True)
speech_file_path = os.path.join(output_dir, f'{file_name_prefix}.mp3')
else:
print(f"Invalid text_source_type: {text_source_type}")
return None
return _generate_speech(input_text, speech_file_path, None, target_language, tts_gender, tts_voice_name, toggle_hinglish)
except FileNotFoundError:
print(f'Error: Text file not found at {text_input}')
return None
except Exception as e:
print(f'Error generating audio: {e}')
return None
# ---------------- Utilities: PowerPoint audio inspection & trimming ----------------
def pun_check_audio_files(pptx_path, output_dir):
print(f'Checking audio files in: {pptx_path}')
os.makedirs(output_dir, exist_ok=True)
audio_mapping = {}
with zipfile.ZipFile(pptx_path, 'r') as z:
media_files = [f for f in z.namelist() if f.startswith('ppt/media/')]
audio_files = [f for f in media_files if f.endswith('.mp3')]
slide_files = [f for f in z.namelist() if f.startswith('ppt/slides/slide') and f.endswith('.xml')]
slide_files.sort(key=lambda x: int(re.search('slide(\\d+)\\.xml$', x).group(1)))
for audio_file in audio_files:
match = re.search('media(\\d+)\\.mp3$', audio_file)
if match:
slide_num = int(match.group(1))
if slide_num <= len(slide_files):
audio_mapping[slide_num] = audio_file
output_path = os.path.join(output_dir, f'slide_{slide_num}_{os.path.basename(audio_file)}')
with open(output_path, 'wb') as f:
f.write(z.read(audio_file))
return audio_mapping
def pun_get_audio_duration(file_path: str) -> float:
if not os.path.exists(file_path):
return 0.0
try:
audio = AudioSegment.from_file(file_path)
return len(audio) / 1000.0
except Exception:
return 0.0
def trim_audio_to_max_duration(input_path: str, max_seconds: float) -> str:
if not os.path.exists(input_path):
return input_path
try:
duration = pun_get_audio_duration(input_path)
if duration <= max_seconds:
return input_path
except Exception:
pass
temp_out = f"{input_path}.trim.mp3"
ffmpeg_bin = shutil.which("ffmpeg") or "ffmpeg"
cmd = [ffmpeg_bin, "-y", "-i", input_path, "-ss", "0", "-t", str(max_seconds), "-c:a", "libmp3lame", "-b:a", "128k", temp_out]
try:
subprocess.run(cmd, check=True, capture_output=True, text=True)
os.replace(temp_out, input_path)
return input_path
except Exception:
if os.path.exists(temp_out):
os.remove(temp_out)
return input_path
def pad_audio_to_duration(input_path: str, target_seconds: float) -> str:
"""Ensure `input_path` audio is at least `target_seconds` long.
If shorter, append silence to reach target duration. Returns the path to the (possibly modified) file.
Operates in-place (creates a temporary file and replaces original).
"""
if not os.path.exists(input_path):
return input_path
try:
audio = AudioSegment.from_file(input_path)
current = len(audio) / 1000.0
except Exception:
return input_path
if current >= float(target_seconds):
return input_path
silence_ms = int((float(target_seconds) - current) * 1000)
padding = AudioSegment.silent(duration=silence_ms)
new_audio = audio + padding
temp_out = f"{input_path}.pad.mp3"
try:
new_audio.export(temp_out, format='mp3', bitrate='128k')
os.replace(temp_out, input_path)
return input_path
except Exception:
if os.path.exists(temp_out):
os.remove(temp_out)
return input_path
# ---------------- Media combine ----------------
def get_media_duration(file_path: str) -> float:
try:
cmd = ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", file_path]
result = subprocess.run(cmd, check=True, capture_output=True, text=True)
return float(result.stdout.strip())
except Exception:
return 0.0
def combine_audio_with_video(video_path: str, audio_path: str, output_path: str) -> bool:
try:
if not os.path.exists(video_path) or not os.path.exists(audio_path):
return False
video_duration = get_media_duration(video_path)
audio_duration = get_media_duration(audio_path)
ffmpeg_bin = shutil.which("ffmpeg") or "ffmpeg"
if audio_duration > video_duration:
pad = audio_duration - video_duration
command = [
ffmpeg_bin, "-y",
"-i", video_path,
"-i", audio_path,
"-filter_complex", f"[0:v]tpad=stop_mode=clone:stop_duration={pad}[v]",
"-map", "[v]",
"-map", "1:a:0",
"-c:v", "libx264",
"-preset", "medium",
"-crf", "23",
"-c:a", "aac",
"-shortest",
output_path
]
else:
command = [
ffmpeg_bin, "-y",
"-i", video_path,
"-i", audio_path,
"-map", "0:v:0",
"-map", "1:a:0",
"-c:v", "copy",
"-c:a", "aac",
"-shortest",
output_path
]
subprocess.run(command, check=True, capture_output=True, text=True)
return os.path.exists(output_path) and os.path.getsize(output_path) > 0
except subprocess.CalledProcessError as e:
print(f"FFmpeg failed: {e.stderr}")
return False
except Exception as e:
print(f"Unexpected combine error: {e}")
return False
# ---------------- Segment-level audio assembly (pun_) ----------------
def pun_generate_segment_audios(code_segments: list, temp_audio_dir: str,
client = None, txt_src_typ: Literal['file', 'string']='file',
target_language: str='english', tts_gender: str='female',
tts_voice_name: str='nova', toggle_hinglish: bool=False, delay_ms: int=1000, default_chars_per_second: int=25, extra_padding_s: float=0.5) -> tuple[str | None, float]:
os.makedirs(temp_audio_dir, exist_ok=True)
final_audio_clips_for_master = []
total_duration = 0.0
for i, segment in enumerate(code_segments):
if segment.explanation:
try:
generated_audio_path = pun_audio_fn_from_string(
text_input=pun_remove_special_characters(segment.explanation),
client=client,
target_language=target_language,
tts_gender=tts_gender,
tts_voice_name=tts_voice_name,
toggle_hinglish=toggle_hinglish,
file_name_prefix=f'segment_{i:02d}',
text_source_type=txt_src_typ
)
if generated_audio_path:
segment.audio_path = generated_audio_path
segment.audio_duration = pun_get_audio_duration(generated_audio_path)
else:
segment.audio_duration = 0.0
except Exception:
segment.audio_duration = 0.0
else:
segment.audio_duration = 0.0
typing_duration = len(segment.code_snippet) / default_chars_per_second if default_chars_per_second > 0 else 0.1
segment_visual_duration = max(segment.audio_duration, typing_duration) + extra_padding_s
segment_visual_duration_ms = int(segment_visual_duration * 1000)
segment_audio_block = AudioSegment.silent(duration=segment_visual_duration_ms)
if segment.audio_duration > 0 and os.path.exists(segment.audio_path):
try:
explanation_audio = AudioSegment.from_file(segment.audio_path)
segment_audio_block = segment_audio_block.overlay(explanation_audio, position=0)
except Exception:
pass
final_audio_clips_for_master.append(segment_audio_block)
total_duration += segment_visual_duration
if i < len(code_segments) - 1:
final_audio_clips_for_master.append(AudioSegment.silent(duration=delay_ms))
total_duration += delay_ms / 1000.0
master_audio_path = os.path.join(temp_audio_dir, 'master_explanation.mp3')
if final_audio_clips_for_master:
try:
combined_audio = sum(final_audio_clips_for_master)
combined_audio.export(master_audio_path, format='mp3')
return master_audio_path, total_duration
except Exception:
return None, 0.0
else:
return None, 0.0