diff --git a/cli.py b/cli.py index 7eec1ce..5eecfee 100644 --- a/cli.py +++ b/cli.py @@ -35,6 +35,10 @@ def main(): p_tgr.add_argument("msg_id", help="id входящего (m-NNN)") p_tgr.add_argument("text") + p_tgt = sub.add_parser("tg-transcribe", help="расшифровать голосовое (m-NNN) вручную") + p_tgt.add_argument("msg_id", help="id входящего (m-NNN)") + p_tgt.add_argument("--force", action="store_true", help="перерасшифровать заново") + p_lad = sub.add_parser("ladder", help="срез лестницы: уровни, занятость, задачи, эскалации") p_not = sub.add_parser("tg-notify", help="отправить сообщение в чат харнеса (уведомление)") p_not.add_argument("text") @@ -95,6 +99,36 @@ def main(): args.text = "📊 " + tg_inbox.ladder_snapshot() mid = tg_inbox.send_reply(args.msg_id, args.text) print(f"OK: ответ на {mid} отправлен") + elif args.cmd == "tg-transcribe": + es.init_db() + from harness import voice + from harness import tg_inbox + model = es.run_projections() + m = (model.get("inbox") or {}).get(args.msg_id) + if m is None: + print(f"ERROR: входящее {args.msg_id} не найдено") + sys.exit(1) + if args.force: + # сбрасываем прошлую расшифровку в read model (событие поверх) + # — transcribe_message идемпотентен по флагу transcribed + es.append_event("inbox", args.msg_id, "MessageTranscribed", { + "transcript": "", "engine": None, "reason": None, "reset": True, + }) + print("(force: прошлый транскрипт сброшен)") + if m.get("transcribed") and not args.force: + print(f"Уже расшифровано ({m.get('engine')}): {m.get('transcript')}") + else: + text, engine = voice.transcribe_message( + args.msg_id, file_id=m.get("file_id") or "", + file_unique_id=m.get("file_unique_id") or "", + duration=m.get("duration"), kind=m.get("kind") or "voice") + if text: + print(f"OK ({engine}): {text}") + else: + reason = m.get("transcribe_reason") + print(f"ПУСТО (engine={engine})" + + (f", причина: {reason}" if reason else "") + + " — транскрипт записан в стор как пустой") elif args.cmd == "ladder": es.init_db() from harness import tg_inbox diff --git a/harness/_fw_asr_worker.py b/harness/_fw_asr_worker.py new file mode 100755 index 0000000..6398a3b --- /dev/null +++ b/harness/_fw_asr_worker.py @@ -0,0 +1,36 @@ +#!/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() diff --git a/harness/es.py b/harness/es.py index 1cd5cb5..f951152 100644 --- a/harness/es.py +++ b/harness/es.py @@ -236,6 +236,43 @@ def _apply_event(model: dict, ev: dict): if m: m["replied"] = True m["reply_text"] = data.get("text", "") + elif etype == "MessageTranscribed": + # голосовое → текст: минутный уровень видит его как обычное + # входящее (kind=voice, text=транскрипт) и отвечает по смыслу. + # reset=True — сброс прошлой расшифровки (cli tg-transcribe --force). + if data.get("reset"): + m = model.setdefault("inbox", {}).get(aid) + if m: + m["transcribed"] = False + m.pop("transcript", None) + m.pop("transcribe_reason", None) + # текст возвращаем к исходнику (пустой у голосовых), + # чтобы не отвечали на устаревший транскрипт + m["text"] = (m.get("original_text") or "")[:4000] + return + m = model.setdefault("inbox", {}).get(aid) + if m: + if not m.get("original_text_recorded"): + m["original_text"] = m.get("text") or "" + m["original_text_recorded"] = True + m["transcribed"] = True + m["transcript"] = data.get("transcript") or "" + m["engine"] = data.get("engine") + if data.get("reason"): + m["transcribe_reason"] = data.get("reason") + else: + m.pop("transcribe_reason", None) + if data.get("transcript"): + m["text"] = data["transcript"][:4000] # расшифровка = текст сообщения + elif not m.get("original_text"): + # пустая расшифровка затирает старый транскрипт из text + m["text"] = "" + else: + model["inbox"][aid] = { + "transcribed": True, "transcript": data.get("transcript") or "", + "engine": data.get("engine"), "replied": False, "ignored": False, + "ts": ev["timestamp"], + } elif etype == "MessageIgnored": m = model.setdefault("inbox", {}).get(aid) if m: diff --git a/harness/scheduler.py b/harness/scheduler.py index bb521a9..3c193e9 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -119,15 +119,27 @@ def build_context(level_cfg, model: dict, task=None): parts.append("ЭСКАЛАЦИИ (важное с нижних уровней; отработай и закрой: " "./venv/bin/python cli.py escalation-ack ):\n" + "\n".join(lines)) - # инбокс: входящие TG-сообщения, ждущие ответа (в приоритете) + # инбокс: входящие TG-сообщения, ждущие ответа (в приоритете). + # Голосовое с пустой расшифровкой (тишина/отказ) тоже показываем — + # минутный уровень должен знать, что голосовое пришло, но текста нет, + # и ответить «не расслышал», а не делать вид, что сообщения не было. inbox = [m for m in (model.get("inbox") or {}).values() - if not m.get("replied") and not m.get("ignored") and m.get("text")] + if not m.get("replied") and not m.get("ignored") + and (m.get("text") or m.get("kind") in ("voice", "audio", "video_note"))] if inbox: lines = [] - for m in inbox[:10]: + for m in sorted(inbox, key=lambda x: x.get("ts") or "")[-10:]: m_id = "m-" + str(m.get("update_id")) who = m.get("username") or m.get("first_name") or m.get("user_id") - lines.append(f"- [{m_id}] от {who}: {m['text'][:300]}") + tag = "" + if m.get("kind") == "voice": + if m.get("transcribed") and m.get("text"): + tag = " [голосовое, расшифровка]" + else: + tag = (" [голосовое, расшифровка не удалась — попроси прислать текстом]" + if m.get("transcribed") else + " [голосовое, ещё не расшифровано]") + lines.append(f"- [{m_id}] от {who}{tag}: {m.get('text','')[:300]}") parts.append("ВХОДЯЩИЕ СООБЩЕНИЯ (ответь через: ./venv/bin/python cli.py tg-reply \"текст\"):\n" + "\n".join(lines)) # доска задач diff --git a/harness/tg_inbox.py b/harness/tg_inbox.py index cd93c1e..cb381b0 100644 --- a/harness/tg_inbox.py +++ b/harness/tg_inbox.py @@ -91,6 +91,9 @@ def receive_message(update: dict): "text": (msg.get("text") or msg.get("caption") or "")[:4000], "kind": kind, "file_id": file_id, + # file_unique_id: ключ аудио-кэша и идемпотентности расшифровки + # (file_id ротируется, unique — постоянный для одного файла) + "file_unique_id": file.get("file_unique_id") or (photo[-1].get("file_unique_id") if photo else None), "duration": duration, "date": msg.get("date"), } @@ -149,6 +152,25 @@ def poll_loop(): mid = receive_message(u) print(f"[tg] {mid} от {(msg.get('from') or {}).get('username') or user}: " f"{(msg.get('text') or '')[:60]}", flush=True) + # голосовые/аудио/видео-заметки: расшифровка сразу при приёме, + # чтобы минутный уровень уже видел текст (не качает файлы сам — + # таймаут тика мал). Сбой транскрипции не валит поллер. + fobj = (msg.get("voice") or msg.get("audio") or msg.get("video_note")) + if fobj: + try: + from . import voice + kind = ("voice" if msg.get("voice") else + "audio" if msg.get("audio") else "video_note") + text, engine = voice.transcribe_message( + mid, + file_id=fobj.get("file_id") or "", + file_unique_id=fobj.get("file_unique_id") or "", + duration=fobj.get("duration"), + kind=kind, + ) + print(f"[tg] {mid} транскрипция ({engine}): {(text or '')[:80]}", flush=True) + except Exception as e: + print(f"[tg] {mid} ошибка транскрипции: {e}", flush=True) except Exception as e: print(f"[tg] ошибка поллинга: {e}", flush=True) time.sleep(5) diff --git a/harness/voice.py b/harness/voice.py new file mode 100644 index 0000000..20ec290 --- /dev/null +++ b/harness/voice.py @@ -0,0 +1,236 @@ +"""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 \ No newline at end of file