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 доведён, протестирован, закоммичен.
236 lines
No EOL
12 KiB
Python
236 lines
No EOL
12 KiB
Python
"""Simbiot Harness — транскрипция голосовых сообщений Telegram.
|
||
|
||
Пайплайн: file_id (в MessageReceived) → getFile → download → ASR →
|
||
MessageTranscribed (транскрипт в проекцию inbox) → минутный уровень
|
||
видит голосовое как обычный текст и отвечает по смыслу `tg-reply`.
|
||
|
||
ASR — два яруса:
|
||
1) whisper-asr-webservice (docker 127.0.0.1:8084, openai_whisper base):
|
||
быстрый, но короткие русские фразы иногда глуховатые;
|
||
2) fallback: faster-whisper large-v3-turbo (CPU int8, модель уже в
|
||
HF-кэше сервера, ~5-15с на короткое голосовое) — точнее на русском.
|
||
Живёт в системном python3 (в venv харнеса пакета нет) — вызываем
|
||
через subprocess-обёртку `_fw_asr_worker.py`.
|
||
|
||
Вызывает инбокс-поллер сразу после MessageReceived (транскрипт в сторе
|
||
ещё до первого минутного тика — уровню не нужно качать файлы, его
|
||
таймаут мал). Аудио кэшируется по file_unique_id: повторная расшифровка
|
||
не качает из Telegram и не перегоняет через ASR.
|
||
"""
|
||
import base64
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
from . import es
|
||
|
||
# ─── конфигурация ───────────────────────────────────────────────────────────
|
||
_HARNESS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
_ENV_PATH = os.path.join(_HARNESS_DIR, ".env")
|
||
_FW_WORKER = os.path.join(_HARNESS_DIR, "harness", "_fw_asr_worker.py")
|
||
_FW_PYTHON = "/usr/bin/python3" # faster-whisper живёт в системном python
|
||
|
||
ASR_URL = os.environ.get("SIMBIOT_ASR_URL", "http://127.0.0.1:8084")
|
||
# что расшифровываем: voice — главное; audio/video_note — та же речь
|
||
TRANSCRIBABLE_KINDS = {"voice", "audio", "video_note"}
|
||
# длина, сек: длиннее не тянем (поллер не должен вязнуть в 10-мин озвучках)
|
||
MAX_DURATION = 300
|
||
# аудио-кэш: file_unique_id → ogg-файл
|
||
VOICE_DIR = os.environ.get("SIMBIOT_VOICE_DIR", "/var/lib/simbiot-harness/voice")
|
||
|
||
TG_TOKEN = None
|
||
|
||
|
||
def _tg_init():
|
||
"""Токен бота из .env (кэш в глобале)."""
|
||
global TG_TOKEN
|
||
if TG_TOKEN is None:
|
||
if os.path.exists(_ENV_PATH):
|
||
with open(_ENV_PATH) as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line.startswith("SIMBIOT_TG_TOKEN="):
|
||
TG_TOKEN = line.split("=", 1)[1].strip()
|
||
return TG_TOKEN
|
||
|
||
|
||
# ───────────────────────────── TG download ──────────────────────────────────
|
||
|
||
def _download_voice(file_id: str, file_unique_id: str = ""):
|
||
"""getFile → download. Аудио кэшируется на диске по file_unique_id.
|
||
|
||
Возвращает (bytes, cache_path) или (None, None) при неудаче.
|
||
Старые голосовые (файл истёк у Telegram) дают 400 — это не ошибка
|
||
поллера, просто расшифровка невозможна.
|
||
"""
|
||
if not _tg_init():
|
||
return None, None
|
||
cache = os.path.join(VOICE_DIR, (file_unique_id or file_id) + ".ogg")
|
||
if os.path.exists(cache): # кэш: повтор не качает из Telegram
|
||
with open(cache, "rb") as f:
|
||
return f.read(), cache
|
||
try:
|
||
url = f"https://api.telegram.org/bot{TG_TOKEN}/getFile"
|
||
data = urllib.parse.urlencode({"file_id": file_id}).encode()
|
||
with urllib.request.urlopen(
|
||
urllib.request.Request(url, data=data), timeout=30) as r:
|
||
resp = json.loads(r.read().decode())
|
||
if not resp.get("ok"):
|
||
return None, None
|
||
path = resp["result"].get("file_path")
|
||
if not path:
|
||
return None, None
|
||
durl = f"https://api.telegram.org/file/bot{TG_TOKEN}/{path}"
|
||
with urllib.request.urlopen(durl, timeout=60) as r:
|
||
audio = r.read()
|
||
except urllib.error.HTTPError as e:
|
||
# 400 = file_id истёк (TG хранит файлы ограниченно) — тихо отдаём отказ
|
||
print(f"[voice] getFile HTTP {e.code}: файл недоступен", flush=True)
|
||
return None, None
|
||
except Exception as e:
|
||
print(f"[voice] download error: {e}", flush=True)
|
||
return None, None
|
||
try: # кэш — best effort
|
||
os.makedirs(VOICE_DIR, exist_ok=True)
|
||
with open(cache, "wb") as f:
|
||
f.write(audio)
|
||
except Exception:
|
||
pass
|
||
return audio, cache
|
||
|
||
|
||
# ─────────────────────────────── ASR ярусы ──────────────────────────────────
|
||
|
||
def transcribe_bytes(audio: bytes, filename: str = "audio.ogg",
|
||
long: bool = False):
|
||
"""Байты аудио → текст. '' = тишина/пусто, None = не смогли.
|
||
|
||
Ярус 1 — локальный ASR-контейнер whisper base (быстро, ~1с; короткие
|
||
русские фразы распознаёт с ошибками). Ярус 2 — faster-whisper
|
||
large-v3-turbo (точнее на русском, ~7-15с на короткое голосовое).
|
||
Второй ярус включается: при пустом результате первого или при
|
||
long=True (короткие фразы и длинные записи — где base слабее всего).
|
||
"""
|
||
text = _asr_webservice(audio, filename)
|
||
if text:
|
||
return text, "whisper-asr-base"
|
||
# короткие фразы (≤30с) — зона, где base ошибается чаще всего;
|
||
# длинные записи base тянет хорошо, но large-v3-turbo точнее
|
||
fw = _asr_faster_whisper(audio, filename)
|
||
if fw: # непустой результат второго яруса
|
||
return fw, "faster-whisper-large-v3-turbo"
|
||
return text, None # '' = тишина, None = оба яруса недоступны
|
||
|
||
|
||
def _asr_webservice(audio: bytes, filename: str):
|
||
"""POST /asr в whisper-asr-webservice. None при недоступности сервиса."""
|
||
import uuid
|
||
boundary = "----simbiot" + uuid.uuid4().hex
|
||
body = (
|
||
f'--{boundary}\r\nContent-Disposition: form-data; name="audio_file"; '
|
||
f'filename="{filename}"\r\nContent-Type: audio/ogg\r\n\r\n'
|
||
).encode() + audio + f"\r\n--{boundary}--\r\n".encode()
|
||
req = urllib.request.Request(
|
||
f"{ASR_URL}/asr?output=json", data=body,
|
||
headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=120) as r:
|
||
data = json.loads(r.read().decode())
|
||
return (data.get("text") or "").strip()
|
||
except Exception as e:
|
||
print(f"[voice] asr-webservice error: {e}", flush=True)
|
||
return None
|
||
|
||
|
||
def _asr_faster_whisper(audio: bytes, filename: str):
|
||
"""faster-whisper large-v3-turbo через системный python3 (subprocess).
|
||
|
||
В venv харнеса пакета нет (и не тащим: ctranslate2 тянет за собой
|
||
тяжёлые бинарники) — модель живёт в системном python, зовём воркером.
|
||
None при отказе, '' при пустой расшифровке.
|
||
"""
|
||
try:
|
||
req = json.dumps({
|
||
"audio_b64": base64.b64encode(audio).decode(),
|
||
"filename": filename,
|
||
}, ensure_ascii=False)
|
||
proc = subprocess.run(
|
||
[_FW_PYTHON, _FW_WORKER], input=req.encode(),
|
||
capture_output=True, timeout=600)
|
||
out = proc.stdout.decode() if proc.stdout else ""
|
||
try:
|
||
data = json.loads(out.strip().splitlines()[-1]) if out.strip() else {}
|
||
except (ValueError, IndexError):
|
||
data = {"error": (proc.stderr or b"").decode()[:300]}
|
||
if data.get("error") or proc.returncode != 0:
|
||
print(f"[voice] faster-whisper error: "
|
||
f"{data.get('error') or proc.returncode}", flush=True)
|
||
return None
|
||
return (data.get("text") or "").strip()
|
||
except Exception as e:
|
||
print(f"[voice] faster-whisper subprocess error: {e}", flush=True)
|
||
return None
|
||
|
||
|
||
# ─────────────────────────── оркестрация ────────────────────────────────────
|
||
|
||
def transcribe_message(msg_id: str, file_id: str, file_unique_id: str = "",
|
||
duration=None, kind: str = "voice"):
|
||
"""Полный путь одного входящего: download + ASR + MessageTranscribed.
|
||
|
||
msg_id — 'm-NNN' (aggregate_id). Возвращает (текст, engine):
|
||
текст '' = тишина/пусто, None = расшифровать не смогли;
|
||
идемпотентно: повторный вызов по уже расшифрованному — транскрипт из
|
||
read model без хождений в Telegram/ASR.
|
||
"""
|
||
model = es.run_projections()
|
||
m = (model.get("inbox") or {}).get(msg_id)
|
||
if m is None:
|
||
return None, None # входящее не найдено (нет MessageReceived)
|
||
if m.get("transcribed"):
|
||
return m.get("transcript") or "", m.get("engine")
|
||
|
||
reason = None
|
||
if kind not in TRANSCRIBABLE_KINDS:
|
||
reason = "not_transcribable"
|
||
elif duration and duration > MAX_DURATION:
|
||
reason = "too_long"
|
||
if reason:
|
||
es.append_event("inbox", msg_id, "MessageTranscribed", {
|
||
"transcript": "", "engine": None, "reason": reason,
|
||
"duration": duration,
|
||
})
|
||
return "", None
|
||
|
||
audio, _path = _download_voice(file_id, file_unique_id)
|
||
if audio is None:
|
||
es.append_event("inbox", msg_id, "MessageTranscribed", {
|
||
"transcript": "", "engine": None, "reason": "download_failed",
|
||
"duration": duration,
|
||
})
|
||
return None, None
|
||
# короткие русские фразы base-модель коверкает («Туше, что ты тебя
|
||
# слышишь» вместо «Ты теперь слышишь») — для voice ≤30с сразу идём
|
||
# в точный large-v3-turbo, long-записи и audio — как получится
|
||
short_voice = kind == "voice" and (duration or 0) <= 30
|
||
if short_voice:
|
||
text, engine = _asr_faster_whisper(audio, "audio.ogg"), "faster-whisper-large-v3-turbo"
|
||
if text: # точный путь сработал
|
||
es.append_event("inbox", msg_id, "MessageTranscribed", {
|
||
"transcript": text[:4000], "engine": engine,
|
||
"bytes": len(audio), "duration": duration,
|
||
})
|
||
return text, engine
|
||
text, engine = transcribe_bytes(audio, "audio.ogg")
|
||
es.append_event("inbox", msg_id, "MessageTranscribed", {
|
||
"transcript": (text or "")[:4000],
|
||
"engine": engine if text else None,
|
||
"reason": None if text is not None else "asr_failed",
|
||
"bytes": len(audio),
|
||
"duration": duration,
|
||
})
|
||
return text, engine |