app.py

import io
import json
import os
import re
import subprocess
import sys
import tempfile
import traceback
from functools import lru_cache
from pathlib import Path
from typing import Literal

import numpy as np
import soundfile as sf
import torch
from fastapi import FastAPI, HTTPException
from fastapi.responses import Response
from huggingface_hub import hf_hub_download
from pydantic import BaseModel, Field
from transformers import AutoModelForCausalLM, AutoTokenizer


APP_VERSION = '2.0.0'
DEFAULT_SAMPLE_RATE = 22050

LLM_MODEL_ID = os.getenv('PODCAST_LLM_MODEL', 'Qwen/Qwen2.5-0.5B-Instruct')
LLM_FALLBACK_MODEL_ID = os.getenv('PODCAST_LLM_FALLBACK_MODEL', 'Vikhrmodels/Vikhr-Qwen-2.5-0.5b-Instruct')
LLM_DEVICE_MODE = os.getenv('PODCAST_LLM_DEVICE', 'auto').lower()
LLM_MAX_FACTS = int(os.getenv('PODCAST_LLM_MAX_FACTS', '4'))
LLM_MAX_INPUT_CHARS = int(os.getenv('PODCAST_LLM_MAX_INPUT_CHARS', '7000'))
LLM_MAX_NEW_TOKENS = int(os.getenv('PODCAST_LLM_MAX_NEW_TOKENS', '420'))
PODCAST_LLM_ENABLED = os.getenv('PODCAST_LLM_ENABLED', 'false').lower() in {'1', 'true', 'yes', 'on'}

TTS_ENGINE = os.getenv('TTS_ENGINE', 'piper').lower()
TTS_MAX_REQUEST_CHARS = int(os.getenv('TTS_MAX_REQUEST_CHARS', os.getenv('TTS_MAX_TEXT_CHARS', '1800')))
TTS_CHUNK_MAX_CHARS = int(os.getenv('TTS_CHUNK_MAX_CHARS', '420'))
TTS_RUSSIAN_NORMALIZATION_ENABLED = os.getenv('TTS_RUSSIAN_NORMALIZATION_ENABLED', 'true').lower() in {'1', 'true', 'yes', 'on'}
PIPER_VOICE_REPO = os.getenv('PIPER_VOICE_REPO', 'rhasspy/piper-voices')
PIPER_DEFAULT_VOICE = os.getenv('PIPER_DEFAULT_VOICE', 'ru_RU-irina-medium')
PIPER_FALLBACK_VOICE = 'ru_RU-dmitri-medium'
PIPER_VOICES = {
    'ru_RU-irina-medium': 'ru/ru_RU/irina/medium',
    'ru_RU-dmitri-medium': 'ru/ru_RU/dmitri/medium',
}
if PIPER_DEFAULT_VOICE not in PIPER_VOICES:
    PIPER_DEFAULT_VOICE = PIPER_FALLBACK_VOICE

app = FastAPI(title='Bike Podcast Space', version=APP_VERSION)

LLM_LOADED_MODEL_ID: str | None = None
LAST_SCRIPT_ERROR = ''
LAST_SCRIPT_ERROR_TYPE = ''
LAST_SCRIPT_MODEL = ''
LAST_SCRIPT_PROVIDER = ''


class RouteSummary(BaseModel):
    distance_m: float | None = None
    duration_sec: float | None = None
    average_speed_kmh: float | None = None


class PodcastFact(BaseModel):
    id: str = ''
    index: int = 0
    name: str = ''
    type: str = ''
    subtype: str = ''
    address: str | None = None
    description: str | None = None
    wikipedia_description: str | None = None
    wikipedia_extract: str | None = None
    research_summary: str | None = None
    research_source: str | None = None
    website: str | None = None
    source_url: str | None = None


class PodcastScriptRequest(BaseModel):
    route: RouteSummary = Field(default_factory=RouteSummary)
    facts: list[PodcastFact] = Field(default_factory=list)
    language: str = 'ru'
    style: str = 'friendly'


class PodcastScriptSegment(BaseModel):
    role: Literal['host', 'guide']
    title: str = ''
    text: str = Field(min_length=1)
    source_ids: list[str] = Field(default_factory=list)


class PodcastScriptResponse(BaseModel):
    segments: list[PodcastScriptSegment]
    provider: str
    model: str


class TtsRequest(BaseModel):
    text: str = Field(min_length=1)
    role: str = 'narrator'
    voice: str = ''
    speed: float = Field(default=1.0, ge=0.7, le=1.3)
    format: str = 'wav'
    language: str = 'ru'


def select_device() -> str:
    if LLM_DEVICE_MODE == 'cpu':
        return 'cpu'

    if LLM_DEVICE_MODE == 'cuda':
        return 'cuda'

    return 'cuda' if torch.cuda.is_available() else 'cpu'


def clean_text(value: str | None, limit: int = 900) -> str:
    if not value:
        return ''

    text = re.sub(r'[\x00-\x1f]+', ' ', str(value))
    text = re.sub(r'\s+', ' ', text).strip()
    if len(text) <= limit:
        return text

    shortened = text[:limit]
    last_sentence = max(shortened.rfind('. '), shortened.rfind('! '), shortened.rfind('? '))
    if last_sentence > 180:
        return shortened[:last_sentence + 1].strip()

    return shortened.rsplit(' ', 1)[0].strip()


def clean_full_text(value: str | None) -> str:
    if not value:
        return ''

    text = re.sub(r'[\x00-\x1f]+', ' ', str(value))
    return re.sub(r'\s+', ' ', text).strip()


def russian_number_unit(raw_number: str, forms: tuple[str, str, str]) -> str:
    spoken_number = raw_number.replace('.', ',')
    if ',' in spoken_number:
        return f'{spoken_number} {forms[1]}'

    value = int(spoken_number)
    remainder_100 = value % 100
    remainder_10 = value % 10
    if 11 <= remainder_100 <= 14:
        form = forms[2]
    elif remainder_10 == 1:
        form = forms[0]
    elif 2 <= remainder_10 <= 4:
        form = forms[1]
    else:
        form = forms[2]
    return f'{spoken_number} {form}'


def normalize_spoken_russian(text: str) -> str:
    if not TTS_RUSSIAN_NORMALIZATION_ENABLED:
        return text

    normalized = text
    address_rules = [
        (r'\bв\s+г\.\s*(?=[А-ЯЁ])', 'в городе '),
        (r'\bиз\s+г\.\s*(?=[А-ЯЁ])', 'из города '),
        (r'\bпо\s+г\.\s*(?=[А-ЯЁ])', 'по городу '),
        (r'\bг\.\s*(?=[А-ЯЁ])', 'город '),
        (r'\bна\s+ул\.\s*', 'на улице '),
        (r'\bпо\s+ул\.\s*', 'по улице '),
        (r'\bс\s+ул\.\s*', 'с улицы '),
        (r'\bул\.\s*', 'улица '),
        (r'\bна\s+(?:пр-т|просп\.)\s*', 'на проспекте '),
        (r'\bпо\s+(?:пр-т|просп\.)\s*', 'по проспекту '),
        (r'\b(?:пр-т|просп\.)\s*', 'проспект '),
        (r'\bв\s+пер\.\s*', 'в переулке '),
        (r'\bпо\s+пер\.\s*', 'по переулку '),
        (r'\bпер\.\s*', 'переулок '),
        (r'\bна\s+пл\.\s*', 'на площади '),
        (r'\bпл\.\s*', 'площадь '),
        (r'\bд\.\s*(?=\d)', 'дом '),
        (r'\bкорп\.\s*(?=\d)', 'корпус '),
        (r'\bстр\.\s*(?=\d)', 'строение '),
    ]
    normalized = re.sub(r'\bв\s+(\d{3,4})\s*г\.\s*', r'в \1 году ', normalized)
    normalized = re.sub(r'\b(\d{3,4})\s*г\.\s*', r'\1 год ', normalized)
    for pattern, replacement in address_rules:
        normalized = re.sub(pattern, replacement, normalized, flags=re.IGNORECASE)

    unit_rules = [
        (r'(\d+(?:[.,]\d+)?)\s*км\s*/\s*ч\b', ('километр в час', 'километра в час', 'километров в час')),
        (r'(\d+(?:[.,]\d+)?)\s*км\b', ('километр', 'километра', 'километров')),
        (r'(\d+(?:[.,]\d+)?)\s*мин\b', ('минута', 'минуты', 'минут')),
        (r'(\d+(?:[.,]\d+)?)\s*ч\b', ('час', 'часа', 'часов')),
        (r'(\d+(?:[.,]\d+)?)\s*м\b', ('метр', 'метра', 'метров')),
    ]
    for pattern, forms in unit_rules:
        normalized = re.sub(pattern, lambda match: russian_number_unit(match.group(1), forms), normalized)
    return re.sub(r'\s+', ' ', normalized).strip()


def normalize_tts_text(text: str) -> str:
    normalized = clean_full_text(text)
    if not normalized:
        raise HTTPException(status_code=400, detail='Text is required.')

    if len(normalized) > TTS_MAX_REQUEST_CHARS:
        raise HTTPException(status_code=400, detail='TTS text exceeds maximum request length.')

    return normalize_spoken_russian(normalized)


def split_tts_chunks(text: str, max_chars: int = TTS_CHUNK_MAX_CHARS) -> list[str]:
    if max_chars < 20:
        raise ValueError('TTS chunk size must be at least 20 characters.')

    chunks: list[str] = []
    current = ''
    phrases = re.findall(r'.+?(?:[.!?;:]+(?=\s|$)|$)', text)
    for phrase in phrases:
        phrase = phrase.strip()
        if not phrase:
            continue
        if len(phrase) > max_chars:
            words = phrase.split()
            pieces: list[str] = []
            part = ''
            for word in words:
                if len(word) > max_chars:
                    if part:
                        pieces.append(part)
                        part = ''
                    # An unusually long token is preserved rather than read as broken fragments.
                    pieces.append(word)
                elif not part or len(part) + len(word) + 1 <= max_chars:
                    part = f'{part} {word}'.strip()
                else:
                    pieces.append(part)
                    part = word
            if part:
                pieces.append(part)
        else:
            pieces = [phrase]

        for piece in pieces:
            if not current:
                current = piece
            elif len(current) + len(piece) + 1 <= max_chars:
                current = f'{current} {piece}'
            else:
                chunks.append(current)
                current = piece

    if current:
        chunks.append(current)
    return chunks or [text]


def resolve_piper_voice(voice: str, role: str) -> str:
    requested = clean_text(voice, 80)
    if requested in PIPER_VOICES:
        return requested

    # Older PHP configs sent the semantic role as voice; keep that contract safe.
    if not requested or requested in {'host', 'guide', 'narrator'}:
        return PIPER_DEFAULT_VOICE if PIPER_DEFAULT_VOICE in PIPER_VOICES else PIPER_FALLBACK_VOICE

    raise HTTPException(status_code=400, detail='Unsupported Piper voice.')


def facts_payload(facts: list[PodcastFact]) -> list[dict]:
    payload = []
    for fact in facts[:max(1, LLM_MAX_FACTS)]:
        name = clean_text(fact.name, 140)
        if not name:
            continue

        payload.append({
            'id': clean_text(fact.id, 80) or f'fact-{len(payload) + 1}',
            'index': fact.index,
            'name': name,
            'type': clean_text(fact.type, 80),
            'subtype': clean_text(fact.subtype, 80),
            'address': clean_text(fact.address, 160),
            'description': clean_text(fact.description, 500),
            'wikipedia_description': clean_text(fact.wikipedia_description, 220),
            'wikipedia_extract': clean_text(fact.wikipedia_extract, 900),
            'research_summary': clean_text(fact.research_summary, 700),
            'research_source': clean_text(fact.research_source, 180),
            'website': clean_text(fact.website, 180),
            'source_url': clean_text(fact.source_url, 180),
        })

    return payload


def build_prompt(request: PodcastScriptRequest) -> str:
    facts = facts_payload(request.facts)
    if not facts:
        raise HTTPException(status_code=400, detail='At least one fact with a name is required.')

    source_json = json.dumps({
        'route': request.route.model_dump(),
        'facts': facts,
    }, ensure_ascii=False)
    if len(source_json) > LLM_MAX_INPUT_CHARS:
        source_json = source_json[:LLM_MAX_INPUT_CHARS]

    return (
        'Ты сценарист короткого аудиоподкаста для велосипедного маршрута.\n'
        'Пиши только на русском языке.\n'
        'Используй только факты из блока SOURCE. Не добавляй даты, имена, легенды, оценки и историю, которых нет в SOURCE.\n'
        'Если данных мало, говори честно и нейтрально: что это точка маршрута, что можно заметить по названию, типу или адресу.\n'
        'Верни только валидный JSON без markdown и пояснений.\n'
        'Формат ответа: {"segments":[{"role":"host","title":"...","text":"...","source_ids":["..."]},{"role":"guide","title":"...","text":"...","source_ids":["..."]}]}.\n'
        'Сделай 2 реплики на каждую точку: короткий вопрос/подводка ведущего и короткий ответ гида.\n'
        'Каждая реплика: 1-2 предложения, до 230 символов.\n'
        f'SOURCE:\n{source_json}\n'
        'JSON:'
    )


def extract_json_object(text: str) -> dict:
    stripped = text.strip()
    if stripped.startswith('```'):
        stripped = re.sub(r'^```(?:json)?\s*', '', stripped)
        stripped = re.sub(r'\s*```$', '', stripped)

    try:
        return json.loads(stripped)
    except json.JSONDecodeError:
        start = stripped.find('{')
        end = stripped.rfind('}')
        if start < 0 or end <= start:
            raise
        return json.loads(stripped[start:end + 1])


def source_text_for_validation(facts: list[dict]) -> str:
    values = []
    for fact in facts:
        for key in [
            'name',
            'type',
            'subtype',
            'address',
            'description',
            'wikipedia_description',
            'wikipedia_extract',
            'research_summary',
            'research_source',
            'website',
            'source_url',
        ]:
            value = clean_text(fact.get(key), 1200)
            if value:
                values.append(value)

    return ' '.join(values).lower()


def source_allowed_capitalized_words(facts: list[dict]) -> set[str]:
    source = source_text_for_validation(facts)
    return {
        word.lower()
        for word in re.findall(r'\b[А-ЯЁA-Z][А-ЯЁA-Zа-яёa-zA-Z-]{2,}\b', source)
    }


def generated_unknown_capitalized_words(text: str, allowed_words: set[str]) -> list[str]:
    common_words = {
        'это',
        'здесь',
        'точка',
        'маршрут',
        'ведущий',
        'гид',
        'если',
        'можно',
        'название',
        'тип',
        'адрес',
        'путь',
        'по',
        'на',
        'для',
        'вот',
        'коротко',
        'сегодня',
        'рядом',
        'велосипеде',
        'велосипедный',
    }
    unknown = []
    for match in re.finditer(r'\b[А-ЯЁA-Z][А-ЯЁA-Zа-яёa-zA-Z-]{2,}\b', text):
        word = match.group(0).lower()
        if word in allowed_words or word in common_words:
            continue

        unknown.append(match.group(0))

    return unknown


def build_template_script_response(request: PodcastScriptRequest, reason: str = '') -> PodcastScriptResponse:
    facts = facts_payload(request.facts)
    if not facts:
        raise HTTPException(status_code=400, detail='At least one fact with a name is required.')

    route_distance = request.route.distance_m
    route_text = ''
    if route_distance and route_distance > 0:
        route_text = f' Маршрут получился примерно {round(route_distance / 1000, 1)} км.'

    segments = []
    for fact in facts[:max(1, LLM_MAX_FACTS)]:
        source_id = fact['id']
        name = fact['name']
        address = fact.get('address') or ''
        point_type = fact.get('subtype') or fact.get('type') or ''
        description = (
            fact.get('wikipedia_description')
            or fact.get('research_summary')
            or fact.get('description')
            or fact.get('wikipedia_extract')
            or ''
        )

        context_parts = []
        if point_type:
            context_parts.append(f'тип: {point_type}')
        if address:
            context_parts.append(f'адрес: {address}')

        context = ', '.join(context_parts)
        host_text = f'Точка {fact.get("index") or len(segments) + 1}: {name}. Что здесь стоит заметить по данным маршрута?'
        if context:
            guide_text = f'{name}: {context}.'
        else:
            guide_text = f'{name} отмечена как точка этого маршрута.'

        if description:
            guide_text = f'{guide_text} {clean_text(description, 180)}'

        guide_text = f'{guide_text}{route_text}'.strip()
        segments.append(PodcastScriptSegment(
            role='host',
            title=name,
            text=clean_text(host_text, 240),
            source_ids=[source_id],
        ))
        segments.append(PodcastScriptSegment(
            role='guide',
            title=name,
            text=clean_text(guide_text, 260),
            source_ids=[source_id],
        ))

    if reason:
        print(f'[podcast-script] using template fallback: {reason}', file=sys.stderr)

    return PodcastScriptResponse(segments=segments, provider='template-fallback', model='route-facts-template')


def configured_llm_model_ids() -> list[str]:
    model_ids = []
    for model_id in [LLM_MODEL_ID, LLM_FALLBACK_MODEL_ID]:
        if model_id and model_id not in model_ids:
            model_ids.append(model_id)

    return model_ids


@lru_cache(maxsize=2)
def get_llm(model_id: str):
    global LLM_LOADED_MODEL_ID

    device = select_device()
    tokenizer = AutoTokenizer.from_pretrained(model_id)
    model = AutoModelForCausalLM.from_pretrained(
        model_id,
        torch_dtype=torch.float16 if device == 'cuda' else torch.float32,
    )
    model.to(device)
    model.eval()
    LLM_LOADED_MODEL_ID = model_id
    return tokenizer, model, model_id


def generate_script_text(prompt: str, model_id: str) -> tuple[str, str]:
    tokenizer, model, model_id = get_llm(model_id)
    device = select_device()

    rendered_prompt = prompt
    if getattr(tokenizer, 'chat_template', None):
        try:
            messages = [
                {'role': 'system', 'content': 'Ты строго следуешь фактам и возвращаешь только JSON.'},
                {'role': 'user', 'content': prompt},
            ]
            rendered_prompt = tokenizer.apply_chat_template(
                messages,
                add_generation_prompt=True,
                tokenize=False,
            )
        except Exception as exc:
            print(
                f'[podcast-script] {model_id} chat template failed, using plain prompt: {exc.__class__.__name__}: {exc}',
                file=sys.stderr,
            )
            rendered_prompt = prompt

    inputs = tokenizer(
        rendered_prompt,
        return_tensors='pt',
        truncation=True,
        max_length=2048,
    ).to(device)
    eos_token_id = tokenizer.eos_token_id
    pad_token_id = tokenizer.pad_token_id if tokenizer.pad_token_id is not None else eos_token_id

    with torch.inference_mode():
        output_ids = model.generate(
            **inputs,
            max_new_tokens=LLM_MAX_NEW_TOKENS,
            do_sample=False,
            repetition_penalty=1.08,
            pad_token_id=pad_token_id,
            eos_token_id=eos_token_id,
        )

    prompt_len = inputs['input_ids'].shape[-1]
    generated_ids = output_ids[0][prompt_len:]
    return tokenizer.decode(generated_ids, skip_special_tokens=True), model_id


def describe_error(error: Exception) -> tuple[str, str]:
    error_type = error.__class__.__name__
    message = str(error).strip()
    if isinstance(error, HTTPException):
        message = str(error.detail).strip()

    if not message:
        formatted = traceback.format_exception(error.__class__, error, error.__traceback__)
        tail = ''.join(formatted[-4:]).strip()
        message = tail or '<empty message>'

    return error_type, message[:700]


def remember_script_error(model_id: str, error: Exception) -> tuple[str, str]:
    global LAST_SCRIPT_ERROR
    global LAST_SCRIPT_ERROR_TYPE
    global LAST_SCRIPT_MODEL

    error_type, message = describe_error(error)
    LAST_SCRIPT_MODEL = model_id
    LAST_SCRIPT_ERROR_TYPE = error_type
    LAST_SCRIPT_ERROR = message
    print(f'[podcast-script] {model_id} failed with {error_type}: {message}', file=sys.stderr)
    return error_type, message


def clear_script_error(model_id: str, provider: str) -> None:
    global LAST_SCRIPT_ERROR
    global LAST_SCRIPT_ERROR_TYPE
    global LAST_SCRIPT_MODEL
    global LAST_SCRIPT_PROVIDER

    LAST_SCRIPT_MODEL = model_id
    LAST_SCRIPT_PROVIDER = provider
    LAST_SCRIPT_ERROR_TYPE = ''
    LAST_SCRIPT_ERROR = ''


def mark_script_provider(model_id: str, provider: str) -> None:
    global LAST_SCRIPT_MODEL
    global LAST_SCRIPT_PROVIDER

    LAST_SCRIPT_MODEL = model_id
    LAST_SCRIPT_PROVIDER = provider


def generate_script_response(request: PodcastScriptRequest, prompt: str, source_ids: set[str]) -> PodcastScriptResponse:
    errors = []
    facts = facts_payload(request.facts)
    allowed_words = source_allowed_capitalized_words(facts)

    for model_id in configured_llm_model_ids():
        try:
            generated, used_model_id = generate_script_text(prompt, model_id)
            raw = extract_json_object(generated)
            response = validate_script(raw, source_ids, used_model_id, allowed_words)
            clear_script_error(used_model_id, 'local-transformers')
            return response
        except Exception as exc:
            error_type, message = remember_script_error(model_id, exc)
            errors.append(f'{model_id}: {error_type}: {message}')
            continue

    detail = '; '.join(errors) if errors else 'No podcast LLM models configured.'
    response = build_template_script_response(request, detail)
    mark_script_provider(response.model, response.provider)
    return response


def validate_script(raw: dict, source_ids: set[str], model_id: str, allowed_words: set[str]) -> PodcastScriptResponse:
    segments = raw.get('segments')
    if not isinstance(segments, list) or not segments:
        raise HTTPException(status_code=502, detail='LLM returned no podcast segments.')

    validated = []
    for item in segments[:10]:
        if not isinstance(item, dict):
            continue

        role = item.get('role')
        if role not in ('host', 'guide'):
            continue

        text = clean_text(item.get('text'), 260)
        if not text:
            continue

        title = clean_text(item.get('title'), 90)
        unknown_words = generated_unknown_capitalized_words(f'{title} {text}', allowed_words)
        if unknown_words:
            raise HTTPException(
                status_code=502,
                detail=f'LLM introduced facts outside source: {", ".join(unknown_words[:8])}.',
            )

        item_source_ids = [
            clean_text(source_id, 80)
            for source_id in item.get('source_ids', [])
            if clean_text(source_id, 80) in source_ids
        ]
        if not item_source_ids:
            raise HTTPException(status_code=502, detail='LLM returned segment without valid source_ids.')

        validated.append(PodcastScriptSegment(
            role=role,
            title=title,
            text=text,
            source_ids=item_source_ids,
        ))

    if not validated:
        raise HTTPException(status_code=502, detail='LLM returned invalid podcast segments.')

    return PodcastScriptResponse(segments=validated, provider='local-transformers', model=model_id)


@lru_cache(maxsize=2)
def piper_paths(voice_name: str) -> tuple[str, str]:
    voice_base = PIPER_VOICES[voice_name]
    model_file = f'{voice_base}/{voice_name}.onnx'
    config_file = f'{voice_base}/{voice_name}.onnx.json'
    model_path = hf_hub_download(PIPER_VOICE_REPO, model_file)
    config_path = hf_hub_download(PIPER_VOICE_REPO, config_file)
    return model_path, config_path


def synthesize_piper_chunk(text: str, speed: float, voice_name: str) -> tuple[np.ndarray, int]:
    model_path, config_path = piper_paths(voice_name)
    with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as wav_file:
        wav_path = wav_file.name

    length_scale = max(0.75, min(1.45, 1 / max(0.7, min(speed, 1.3))))
    common_args = [
        '--model',
        model_path,
        '--config',
        config_path,
        '--output_file',
        wav_path,
        '--length_scale',
        str(length_scale),
    ]
    commands = [
        [sys.executable, '-m', 'piper', *common_args],
        ['piper', *common_args],
    ]

    try:
        last_error = ''
        for command in commands:
            try:
                subprocess.run(
                    command,
                    input=text,
                    text=True,
                    capture_output=True,
                    check=True,
                    timeout=45,
                )
                break
            except FileNotFoundError as exc:
                last_error = str(exc)
                continue
            except subprocess.CalledProcessError as exc:
                last_error = (exc.stderr or exc.stdout or str(exc)).strip()
                if 'No module named piper' in last_error:
                    continue
                raise
        else:
            raise HTTPException(status_code=502, detail=f'Piper TTS failed: {last_error[:300]}')

        audio, sample_rate = sf.read(wav_path, dtype='float32')
        return np.asarray(audio), int(sample_rate or DEFAULT_SAMPLE_RATE)
    except subprocess.TimeoutExpired as exc:
        raise HTTPException(status_code=504, detail='Piper TTS timed out.') from exc
    except subprocess.CalledProcessError as exc:
        stderr = (exc.stderr or '').strip()
        raise HTTPException(status_code=502, detail=f'Piper TTS failed: {stderr[:300]}') from exc
    finally:
        try:
            Path(wav_path).unlink(missing_ok=True)
        except Exception:
            pass


def synthesize_with_piper(text: str, speed: float, voice_name: str) -> tuple[np.ndarray, int]:
    chunks = split_tts_chunks(text, TTS_CHUNK_MAX_CHARS)
    rendered_chunks: list[np.ndarray] = []
    sample_rate = DEFAULT_SAMPLE_RATE

    for index, chunk in enumerate(chunks):
        chunk_audio, chunk_sample_rate = synthesize_piper_chunk(chunk, speed, voice_name)
        if index == 0:
            sample_rate = chunk_sample_rate
        elif chunk_sample_rate != sample_rate:
            raise HTTPException(status_code=502, detail='Piper returned mismatched audio sample rates.')

        rendered_chunks.append(np.asarray(chunk_audio, dtype='float32'))
        if index < len(chunks) - 1:
            rendered_chunks.append(np.zeros(max(1, int(sample_rate * 0.06)), dtype='float32'))

    return np.concatenate(rendered_chunks), sample_rate


def write_audio(audio: np.ndarray, sample_rate: int, requested_format: str) -> tuple[bytes, str]:
    buffer = io.BytesIO()
    requested_format = requested_format.lower()

    if requested_format == 'ogg':
        target_format = 'OGG'
        content_type = 'audio/ogg'
    else:
        target_format = 'WAV'
        content_type = 'audio/wav'

    try:
        sf.write(buffer, audio, sample_rate, format=target_format)
    except Exception:
        buffer = io.BytesIO()
        sf.write(buffer, audio, sample_rate, format='WAV')
        content_type = 'audio/wav'

    return buffer.getvalue(), content_type


@app.get('/health')
def health():
    return {
        'status': 'ok',
        'version': APP_VERSION,
        'llm_model': LLM_MODEL_ID,
        'llm_fallback_model': LLM_FALLBACK_MODEL_ID,
        'llm_device': select_device(),
        'podcast_llm_enabled': PODCAST_LLM_ENABLED,
        'tts_engine': TTS_ENGINE,
        'tts_voice': PIPER_DEFAULT_VOICE if TTS_ENGINE == 'piper' else 'disabled',
        'tts_russian_normalization_enabled': TTS_RUSSIAN_NORMALIZATION_ENABLED,
    }


@app.get('/diagnostics')
def diagnostics():
    return {
        'status': 'ok',
        'version': APP_VERSION,
        'torch_cuda_available': torch.cuda.is_available(),
        'llm_model': LLM_MODEL_ID,
        'llm_fallback_model': LLM_FALLBACK_MODEL_ID,
        'llm_loaded_model': LLM_LOADED_MODEL_ID,
        'llm_device': select_device(),
        'podcast_llm_enabled': PODCAST_LLM_ENABLED,
        'llm_max_facts': LLM_MAX_FACTS,
        'llm_max_new_tokens': LLM_MAX_NEW_TOKENS,
        'last_script_model': LAST_SCRIPT_MODEL,
        'last_script_provider': LAST_SCRIPT_PROVIDER,
        'last_script_error': LAST_SCRIPT_ERROR,
        'last_script_error_type': LAST_SCRIPT_ERROR_TYPE,
        'tts_engine': TTS_ENGINE,
        'tts_max_request_chars': TTS_MAX_REQUEST_CHARS,
        'tts_chunk_max_chars': TTS_CHUNK_MAX_CHARS,
        'tts_russian_normalization_enabled': TTS_RUSSIAN_NORMALIZATION_ENABLED,
        'piper_voice_repo': PIPER_VOICE_REPO,
        'piper_default_voice': PIPER_DEFAULT_VOICE,
        'piper_allowed_voices': list(PIPER_VOICES.keys()),
    }


@app.post('/podcast-script', response_model=PodcastScriptResponse)
def podcast_script(request: PodcastScriptRequest):
    if not PODCAST_LLM_ENABLED:
        response = build_template_script_response(
            request,
            'LLM disabled for responsive CPU Basic runtime.',
        )
        mark_script_provider(response.model, response.provider)
        return response

    prompt = build_prompt(request)
    source_ids = {item['id'] for item in facts_payload(request.facts)}

    return generate_script_response(request, prompt, source_ids)


@app.post('/tts')
def tts(request: TtsRequest):
    text = normalize_tts_text(request.text)
    if TTS_ENGINE != 'piper':
        raise HTTPException(status_code=503, detail='TTS engine is disabled.')

    voice_name = resolve_piper_voice(request.voice, request.role)
    audio, sample_rate = synthesize_with_piper(text, request.speed, voice_name)
    bytes_payload, content_type = write_audio(audio, sample_rate, request.format)
    return Response(
        content=bytes_payload,
        media_type=content_type,
        headers={
            'Cache-Control': 'no-store',
            'X-TTS-Provider': 'piper',
            'X-TTS-Voice': voice_name,
        },
    )


@app.get('/tts-test')
def tts_test():
    request = TtsRequest(
        text='Привет. Это тест аудиогида ВелоСёрч.',
        role='guide',
        format='wav',
        language='ru',
    )
    return tts(request)