src/tools/audio_utils.py

# 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"(<named_entities>(.*?)</named_entities>)|([^<]+)"
    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'(<named_entities>(.*?)</named_entities>)|([^<]+)'
    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