343 lines
No EOL
17 KiB
Python
343 lines
No EOL
17 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__)))
|
||
from . import paths as _paths # t-261004113351-2858c1
|
||
_ENV_PATH = _paths.env_path()
|
||
_FW_WORKER = os.path.join(_HARNESS_DIR, "harness", "_fw_asr_worker.py")
|
||
_FW_PYTHON = os.environ.get("SIMBIOT_FW_PYTHON") or os.path.join(
|
||
_HARNESS_DIR, "venv", "bin", "python") # faster-whisper живёт в venv харнеса
|
||
|
||
ASR_URL = os.environ.get("SIMBIOT_ASR_URL", "http://127.0.0.1:8084")
|
||
|
||
VOICE_DIR = _paths.voice_dir() # SIMBIOT_HOME → <home>/voice; legacy без хома
|
||
# что расшифровываем: voice — главное; audio/video_note/video — та же речь
|
||
# (t-261004152714: видео-вложения Alex получали not_transcribable — звук
|
||
# из контейнера вообще не извлекался; теперь тянем дорожку ffmpeg'ом)
|
||
TRANSCRIBABLE_KINDS = {"voice", "audio", "video_note", "video"}
|
||
# длина, сек: длиннее не тянем (поллер не должен вязнуть в 10-мин озвучках)
|
||
MAX_DURATION = 300
|
||
# аудио-кэш: file_unique_id → ogg-файл (путь задаёт paths.voice_dir выше)
|
||
|
||
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
|
||
|
||
|
||
# ─────────── тяжёлые воркеры ВНЕ cgroup поллера (t-261004160531-780666) ─────
|
||
# faster-whisper large-v3-turbo (int8) живёт в ~2-3G RSS, ffmpeg тоже кусается:
|
||
# заспавненные поллером как обычные дети они падали в ЕГО cgroup — systemd
|
||
# Consumption показывал 3.7G peak у simbiot-harness-tg. Теперь тяжёлое уходит
|
||
# в собственный transient-scope (systemd-run --scope) со своим MemoryMax;
|
||
# поллер остаётся лёгким (<100M), его юнит держит MemoryMax=1G.
|
||
# Нет systemd (локалка/тесты) — прозрачный fallback на прямой запуск.
|
||
_ASR_SCOPE_MEM = os.environ.get("SIMBIOT_ASR_SCOPE_MEM", "3G")
|
||
_HAS_SYSTEMD = os.path.isdir("/run/systemd/system") and os.path.exists("/usr/bin/systemd-run")
|
||
|
||
|
||
def _heavy_argv(argv):
|
||
"""argv тяжёлого воркера → запуск в отдельном systemd-scope с MemoryMax."""
|
||
if not _HAS_SYSTEMD:
|
||
return list(argv)
|
||
import uuid
|
||
unit = f"simbiot-asr-{uuid.uuid4().hex[:10]}"
|
||
return ["systemd-run", "--scope", "--quiet", "--collect",
|
||
f"--unit={unit}", f"-p", f"MemoryMax={_ASR_SCOPE_MEM}",
|
||
"--"] + list(argv)
|
||
|
||
|
||
def _heavy_run(argv, input_bytes=None, timeout=600):
|
||
"""Тяжёлый subprocess: scope-изоляция памяти + killpg на таймауте.
|
||
|
||
systemd-run --scope выполняет команду синхронно, поэтому таймаут
|
||
работает как раньше; start_new_session + killpg (паттерн
|
||
isolated_runner, e-261003004801) не оставляют осиротевший воркер,
|
||
если тик умер раньше него.
|
||
"""
|
||
import signal
|
||
proc = subprocess.Popen(
|
||
_heavy_argv(argv), stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE, start_new_session=True)
|
||
try:
|
||
out, err = proc.communicate(input=input_bytes, timeout=timeout)
|
||
return proc.returncode, out or b"", err or b""
|
||
except subprocess.TimeoutExpired:
|
||
try:
|
||
os.killpg(proc.pid, signal.SIGKILL)
|
||
except (ProcessLookupError, PermissionError, OSError):
|
||
pass
|
||
try: # дождаться смерти группы, чтобы зомби не сидели в scope
|
||
proc.communicate(timeout=10)
|
||
except Exception:
|
||
pass
|
||
return None, b"", b""
|
||
|
||
|
||
# ────────────── извлечение аудио из видеоконтейнера ─────────────────────────
|
||
|
||
|
||
def extract_audio_track(container: bytes):
|
||
"""Видеоконтейнер (mp4 и т.п.) → ogg/opus аудиодорожка через ffmpeg.
|
||
|
||
Возвращает bytes дорожки или None (нет дорожки / ffmpeg отказал).
|
||
ffmpeg есть на хосте (6.1.1) — тянуть в venv нечего. t-261004152714
|
||
"""
|
||
import tempfile
|
||
src = dst = None
|
||
try:
|
||
with tempfile.NamedTemporaryFile(prefix="simbiot-v_", delete=False) as tf:
|
||
tf.write(container)
|
||
src = tf.name
|
||
# -vn: только звук; моно 16к — формат, который любят оба ASR-яруса.
|
||
# ffmpeg тяжёлый → в свой scope (память поллера не растёт)
|
||
rc, out, err = _heavy_run(
|
||
["ffmpeg", "-v", "error", "-i", src, "-vn", "-ac", "1",
|
||
"-ar", "16000", "-f", "ogg", "-c:a", "libopus", "-"],
|
||
timeout=180)
|
||
if rc != 0 or not out:
|
||
err = err.decode()[:200]
|
||
print(f"[voice] ffmpeg: дорожка не извлечена ({err})", flush=True)
|
||
return None
|
||
return out
|
||
except Exception as e:
|
||
print(f"[voice] ffmpeg error: {e}", flush=True)
|
||
return None
|
||
finally:
|
||
for p in (src, dst):
|
||
if p and os.path.exists(p):
|
||
try:
|
||
os.unlink(p)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
# ─────────────────────────────── 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)
|
||
# faster-whisper large-v3-turbo ~2-3G RSS: только в свой scope,
|
||
# иначе память ложится в cgroup поллера (3.7G peak, t-261004160531)
|
||
rc, out, err = _heavy_run(
|
||
[_FW_PYTHON, _FW_WORKER], input_bytes=req.encode(), timeout=600)
|
||
if rc is None:
|
||
print("[voice] faster-whisper: таймаут (группа убита)", flush=True)
|
||
return None
|
||
out = out.decode() if out else ""
|
||
try:
|
||
data = json.loads(out.strip().splitlines()[-1]) if out.strip() else {}
|
||
except (ValueError, IndexError):
|
||
data = {"error": err.decode()[:300]}
|
||
if data.get("error") or rc != 0:
|
||
print(f"[voice] faster-whisper error: "
|
||
f"{data.get('error') or rc}", 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
|
||
|
||
src_audio, _path = _download_voice(file_id, file_unique_id)
|
||
if src_audio is None:
|
||
es.append_event("inbox", msg_id, "MessageTranscribed", {
|
||
"transcript": "", "engine": None, "reason": "download_failed",
|
||
"duration": duration,
|
||
})
|
||
return None, None
|
||
# видео-контейнер: речь лежит в дорожке — вынимаем её ffmpeg'ом
|
||
if kind == "video":
|
||
audio = extract_audio_track(src_audio)
|
||
if audio is None:
|
||
es.append_event("inbox", msg_id, "MessageTranscribed", {
|
||
"transcript": "", "engine": None, "reason": "audio_track_failed",
|
||
"duration": duration,
|
||
})
|
||
return None, None
|
||
else:
|
||
audio = src_audio
|
||
# короткие русские фразы base-модель коверкает («Туше, что ты тебя
|
||
# слышишь» вместо «Ты теперь слышишь») — для voice/video_note/video ≤60с
|
||
# сразу идём в точный large-v3-turbo, long-записи — как получится
|
||
short_speech = kind in ("voice", "video_note", "video") and (duration or 0) <= 60
|
||
if short_speech:
|
||
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 |