e580b15653
- voice.py: download по file_id (getFile→file, кэш ogg по file_unique_id), два ASR-яруса: whisper-asr-webservice base (быстро ~1с) → fallback faster-whisper large-v3-turbo (CPU int8, ~11с, точный на русском; короткие voice ≤30с идут сразу в точный ярус — base коверкает фразы) - _fw_asr_worker.py: faster-whisper в системном python3 (в venv нет), subprocess JSON-обёртка (stdin JSON → stdout JSON) - tg_inbox.py: file_unique_id в MessageReceived + расшифровка сразу при приёме (поллер пишет MessageTranscribed до первого мин-тика; сбой транскрипции не валит поллер) + video_note как транскрибируемый kind - es.py: проекция MessageTranscribed (transcript→text, original_text, reset для --force), cli tg-transcribe <id> [--force] — ручная расшифровка - scheduler.py: голосовые в контексте мин-уровня даже с пустым текстом (неотвеченное не исчезает): пометки «расшифровано/не расшифровано/ расшифровка не удалась», сортировка по ts, срез 10 Е2Е проверено: m-516173163 — кэш-хит → turbo → «Ты теперь слышишь мои голосовые, да?» в стор; base на том же файле давал «Туше, что ты тебя слышишь» — выбор ярусов подтверждён фактом. Воркер past-hour умер по 2880с-таймауту до коммита; WIP доведён, протестирован, закоммичен.
36 lines
1.3 KiB
Python
Executable file
36 lines
1.3 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""Обёртка: faster-whisper в системном python3 (в venv харнеса его нет).
|
||
|
||
stdin: JSON {"audio_b64": "...", "filename": "..."}
|
||
stdout: JSON {"text": "..."} или {"error": "..."}
|
||
"""
|
||
import base64
|
||
import json
|
||
import sys
|
||
import os
|
||
|
||
def main():
|
||
req = json.load(sys.stdin)
|
||
try:
|
||
import tempfile
|
||
from faster_whisper import WhisperModel
|
||
model = WhisperModel("large-v3-turbo", device="cpu",
|
||
compute_type="int8", cpu_threads=8)
|
||
ext = os.path.splitext(req.get("filename") or "audio.ogg")[1] or ".ogg"
|
||
with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as tmp:
|
||
tmp.write(base64.b64decode(req["audio_b64"]))
|
||
tmp_path = tmp.name
|
||
try:
|
||
segments, info = model.transcribe(
|
||
tmp_path, language="ru", beam_size=5, vad_filter=True)
|
||
text = " ".join(s.text.strip() for s in segments
|
||
if s.text and s.text.strip()).strip()
|
||
print(json.dumps({"text": text, "language": info.language}, ensure_ascii=False))
|
||
finally:
|
||
os.unlink(tmp_path)
|
||
except Exception as e:
|
||
print(json.dumps({"error": str(e)[:500]}, ensure_ascii=False))
|
||
sys.exit(1)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|