app.py

import glob
import os
import subprocess
import tempfile

# ВАЖНО: для Space на тарифе ZeroGPU HF теперь при старте контейнера
# сканирует код на наличие хотя бы одной функции с @spaces.GPU — если её
# нет, падает "Runtime error: No @spaces.GPU function detected during
# startup" (раньше хватало просто импорта spaces, сейчас уже нет).
# Реальная модель у нас работает на CPU, GPU не нужен, поэтому заводим
# функцию-пустышку только ради прохождения этой проверки — она нигде не
# вызывается, так что GPU-квота ZeroGPU не тратится.
import spaces


@spaces.GPU
def _zerogpu_startup_probe():
    """Никогда не вызывается — нужна только чтобы Space прошёл валидацию
    ZeroGPU на старте (см. комментарий выше)."""
    return None

import gigaam
import gradio as gr
from faster_whisper import WhisperModel
from faster_whisper.audio import decode_audio

# Модель грузим ОДИН раз при старте контейнера, не на каждый запрос.
# v3_e2e_ctc — версия с пунктуацией и нормализацией текста (не голый поток
# слов без знаков препинания) и CTC-декодер (быстрее RNNT, для транскрибации
# голосовых в чате этого достаточно).
# Никакого @spaces.GPU нигде не используем — Space создан на тарифе ZeroGPU
# (сейчас у бесплатных аккаунтов HF нет варианта CPU Basic при создании),
# но раз GPU нигде не запрашивается, приложение работает на обычном CPU
# этого же контейнера и не трогает дневную квоту GPU-секунд. GigaAM
# компактная (220-240М параметров), GPU ей и не нужен.
MODEL = gigaam.load_model("v3_e2e_ctc", device="cpu")

# GigaAM обучена только на русской речи — на других языках она не "не
# понимает" в смысле пустого результата, а галлюцинирует русский текст по
# созвучию (особенно опасно для карточки "о чём песня" — см. index.js:
# ошибочная "русская" расшифровка иностранного текста песни ломает весь
# последующий ответ модели). Поэтому перед прогоном через GigaAM отдельно
# определяем язык лёгкой моделью faster-whisper (tiny, только
# language-id — полную транскрибацию ею не делаем, для этого дальше в
# пайплайне бота уже есть Groq Whisper). tiny достаточно для language-id:
# точность определения языка у tiny почти не хуже, чем у крупных моделей,
# в отличие от точности самой транскрибации.
LANG_ID_MODEL = WhisperModel("tiny", device="cpu", compute_type="int8")

# Ниже этого порога уверенности определённому языку не доверяем и всё
# равно отдаём на GigaAM — так безопаснее для коротких/тихих голосовых,
# где language-id менее надёжен, чем ошибочно отбраковывать русскую речь.
LANG_ID_MIN_CONFIDENCE = 0.5


def _detect_language(wav_path: str) -> tuple[str, float]:
    """Определяет язык по первому 30-секундному сегменту WAV (больше и не
    нужно — language-id у Whisper всегда смотрит только на первый сегмент,
    длинное аудио тут не даёт точности, только замедляет)."""
    audio = decode_audio(wav_path, sampling_rate=16000)
    language, probability, _ = LANG_ID_MODEL.detect_language(audio)
    return language, probability


# Нативный model.transcribe() ограничен 25 секундами аудио за вызов.
# Официальный обход (.transcribe_longform) тянет за собой pyannote.audio
# и токен Hugging Face — ради простой последовательной нарезки заводить
# это не стоит, поэтому режем длинные голосовые сами через ffmpeg.
CHUNK_SECONDS = 20


def _to_wav(src_path: str) -> str:
    """Конвертирует присланный файл (у Telegram голосовые — .ogg/opus, но
    подойдёт любой формат, который понимает ffmpeg — mp3, flac и т.д.) в
    16kHz mono WAV. Модели нужен предсказуемый формат на входе."""
    dst_path = tempfile.mktemp(suffix=".wav")
    subprocess.run(
        ["ffmpeg", "-y", "-i", src_path, "-ar", "16000", "-ac", "1", dst_path],
        check=True,
        capture_output=True,
    )
    return dst_path


def _split_into_chunks(wav_path: str) -> list[str]:
    """Режет WAV на куски по CHUNK_SECONDS через ffmpeg segment muxer.
    Простая последовательная нарезка без перехлёста — изредка может
    срезать слово ровно на границе куска, но для голосовых в чате это
    приемлемый компромисс против того, чтобы тянуть pyannote+HF-токен
    только ради длинных сообщений."""
    out_dir = tempfile.mkdtemp()
    pattern = os.path.join(out_dir, "chunk_%03d.wav")
    subprocess.run(
        [
            "ffmpeg", "-y", "-i", wav_path,
            "-f", "segment", "-segment_time", str(CHUNK_SECONDS),
            "-c", "copy", pattern,
        ],
        check=True,
        capture_output=True,
    )
    return sorted(glob.glob(os.path.join(out_dir, "chunk_*.wav")))


def transcribe(audio_path: str, force_giga: bool = False) -> str:
    """force_giga=True пропускает language-id и всегда гонит через GigaAM —
    для ручного выбора движка на стороне бота ("Жень текст giga"). В
    остальных случаях (auto) сначала проверяем язык и на не-русском сразу
    возвращаем "" — вызывающий код в боте трактует пустой ответ как "GigaAM
    не справился" и молча уходит на Groq Whisper (см. transcribeViaGigaAM в
    index.js), поэтому отдельный сигнал "чужой язык" наружу прокидывать не
    нужно, пустой строки достаточно."""
    if not audio_path:
        return ""
    wav_path = _to_wav(audio_path)
    chunks: list[str] = []
    try:
        if not force_giga:
            language, probability = _detect_language(wav_path)
            if language != "ru" or probability < LANG_ID_MIN_CONFIDENCE:
                return ""
        chunks = _split_into_chunks(wav_path)
        parts = [str(MODEL.transcribe(chunk)).strip() for chunk in chunks]
        return " ".join(p for p in parts if p).strip()
    finally:
        os.remove(wav_path)
        for c in chunks:
            try:
                os.remove(c)
            except OSError:
                pass
        if chunks:
            try:
                os.rmdir(os.path.dirname(chunks[0]))
            except OSError:
                pass


# api_name фиксирует имя эндпоинта — так со стороны бота адрес вызова
# предсказуем (/call/transcribe) и не завязан на автогенерируемое имя.
demo = gr.Interface(
    fn=transcribe,
    inputs=[
        gr.Audio(type="filepath", label="Аудио"),
        gr.Checkbox(value=False, label="Принудительно GigaAM (без проверки языка)"),
    ],
    outputs=gr.Textbox(label="Транскрипция"),
    title="GigaAM-v3 — распознавание речи (RU)",
    api_name="transcribe",
)

if __name__ == "__main__":
    demo.launch()