From d7eb17ba94210e56c8ed51fbb57357c864670245 Mon Sep 17 00:00:00 2001 From: h Date: Wed, 28 Oct 2026 02:06:11 +0000 Subject: [PATCH] feat(transcribe,texts): voice notes and round videos to the bot come with a transcript, beavergram finds them by sha256 and telegram premium transcribes --- .env.example | 3 +- README.md | 5 ++ beaver_agent/texts.py | 1 + beaver_agent/transcribe.py | 154 +++++++++++++++++++++++++++++++ config.py | 5 ++ tests/test_transcribe.py | 179 +++++++++++++++++++++++++++++++++++++ 6 files changed, 346 insertions(+), 1 deletion(-) create mode 100644 beaver_agent/transcribe.py create mode 100644 tests/test_transcribe.py diff --git a/.env.example b/.env.example index 6858c1b..9ad22ea 100644 --- a/.env.example +++ b/.env.example @@ -80,7 +80,8 @@ FIREFLY_PAT= # тот же ical напрямую - урл фида она берёт из параметра `icsUrl` этих же # записей; записи без `icsUrl` остаются рукой, но в справку не попадают. CALENDAR_MCPS= -# MCP beavergram (telegram-history) +# MCP beavergram (telegram-history): https://…/mcp?token=…; с TELEGRAM_USER_ID +# тот же токен расшифровывает голосовые и кружки боту через Premium BEAVERGRAM_MCP= # MCP msos (таскер рабочих проектов): полный URL с токеном, https://…/mcp?token=… MSOS_MCP= diff --git a/README.md b/README.md index 76ba5d6..3117fb1 100644 --- a/README.md +++ b/README.md @@ -165,6 +165,11 @@ pieces live in `beaver_agent/`, one concern per module: files for deep chats, `/api`, `/admin`, `/anthropic`, `/mcp` - `texts.py` - everything the gateway says to the model or to the user, in Russian (the gateway's own defaults are English) +- `transcribe.py` - voice notes and round videos sent to the bot get a + transcript under the attachment note: the file's sha256 finds the same + message in beavergram (the owner's account sees it too), and Telegram + Premium transcribes it there; needs `BEAVERGRAM_MCP` with its `token` and + `TELEGRAM_USER_ID`, music and audio files are left alone - `memory/` - the recall block under a message, the reply log, handouts and seeds, the vault watcher, the curator briefing - `jobs/` - crons and webhooks, one file per job diff --git a/beaver_agent/texts.py b/beaver_agent/texts.py index ebb4efd..e3eb690 100644 --- a/beaver_agent/texts.py +++ b/beaver_agent/texts.py @@ -159,6 +159,7 @@ TELEGRAM = TelegramTexts( "[вложение: {kind} {path}{size}; хранится {days} дн., " "перенеси в vault, если нужно]" ), + attachment_transcript="{note}\nрасшифровка: {text}", merge_nothing="сливать нечего: это не открытая ветка", merging="🔀 сливаю в мастер", merged="✅ слито в мастер", diff --git a/beaver_agent/transcribe.py b/beaver_agent/transcribe.py new file mode 100644 index 0000000..e06d3aa --- /dev/null +++ b/beaver_agent/transcribe.py @@ -0,0 +1,154 @@ +"""Голосовые и кружки Бобра расшифровывает Telegram Premium, а достаёт beavergram. + +Бот получает файл из лички с Бобром; userbot beavergram видит то же +сообщение с аккаунта Бобра и кладёт медиа по sha256 содержимого. По хешу +находится сообщение, по сообщению - расшифровка: готовая или свежая через +`/api/media/transcribe`. Треки и аудиофайлы не расшифровываются - только +`voice` и `video_note`; трек агент при нужде расшифрует MCP-тулзой сам. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import logging +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any +from urllib.parse import parse_qs, urlsplit + +import aiohttp + +if TYPE_CHECKING: + from collections.abc import Mapping + from pathlib import Path + + from beaver_gateway.transcribe import Audio + +__all__ = ["BeavergramTranscriber"] + +_log = logging.getLogger(__name__) + +HEARD = frozenset({"audio/ogg", "video/mp4"}) +"""Типы, под которыми Telegram-окно отдаёт голосовое и кружок.""" +SPOKEN = ("voice", "video_note") + + +@dataclass(slots=True) +class BeavergramTranscriber: + origin: str + token: str = field(repr=False) + tg_user_id: int + """Аккаунт Бобра в beavergram ищется по его Telegram id.""" + find_for: float = 15.0 + """Сколько ждать, пока userbot поймает и скачает то же сообщение.""" + find_every: float = 1.0 + wait: int = 60 + """Сколько beavergram ждёт Premium-расшифровку, прежде чем ответить pending.""" + request_timeout: float = 10.0 + _account: int | None = field(default=None, init=False, repr=False) + + @classmethod + def from_env(cls, env: Mapping[str, str]) -> BeavergramTranscriber | None: + """Из `BEAVERGRAM_MCP` (`https://host/mcp?token=…`) и `TELEGRAM_USER_ID`.""" + url = urlsplit(env.get("BEAVERGRAM_MCP", "")) + token = parse_qs(url.query).get("token", [""])[0] + user = env.get("TELEGRAM_USER_ID", "") + if not (url.scheme and url.netloc and token and user.isdigit()): + return None + return cls( + origin=f"{url.scheme}://{url.netloc}", token=token, tg_user_id=int(user) + ) + + async def __call__(self, audio: Audio) -> str | None: + if audio.media_type not in HEARD: + return None + try: + digest = await asyncio.to_thread(_sha256, audio.path) + async with aiohttp.ClientSession( + base_url=self.origin, + headers={"Authorization": f"Bearer {self.token}"}, + timeout=aiohttp.ClientTimeout(total=self.request_timeout), + ) as http: + return await self._transcribe(http, digest) + except (aiohttp.ClientError, TimeoutError, OSError, ValueError) as exc: + _log.warning( + "beavergram: %s не расшифровано: %s: %s", + audio.path.name, + type(exc).__name__, + exc, + ) + return None + + async def _transcribe(self, http: aiohttp.ClientSession, digest: str) -> str | None: + account = await self._account_id(http) + if account is None: + _log.warning("beavergram: нет аккаунта с tg_user_id %s", self.tg_user_id) + return None + place = await self._find(http, digest, account) + if place is None: + _log.info("beavergram: голосового с sha256 %s нет", digest[:12]) + return None + if place.get("extracted_text"): + return place["extracted_text"] + result = await _json( + http.post( + "/api/media/transcribe", + json={ + "account_id": account, + "chat_id": place["chat_id"], + "message_id": place["message_id"], + "wait": self.wait, + }, + timeout=aiohttp.ClientTimeout(total=self.wait + self.request_timeout), + ) + ) + if result.get("status") == "done": + return result.get("text") + _log.info( + "beavergram: расшифровка %s/%s - %s %s", + place["chat_id"], + place["message_id"], + result.get("status"), + result.get("error") or "", + ) + return None + + async def _account_id(self, http: aiohttp.ClientSession) -> int | None: + if self._account is None: + accounts = await _json(http.get("/api/accounts")) + self._account = next( + ( + a["account_id"] + for a in accounts + if a.get("tg_user_id") == self.tg_user_id + ), + None, + ) + return self._account + + async def _find( + self, http: aiohttp.ClientSession, digest: str, account: int + ) -> dict[str, Any] | None: + loop = asyncio.get_running_loop() + deadline = loop.time() + self.find_for + while True: + places = await _json( + http.get( + "/api/media/lookup", + params={"sha256": digest, "account_id": account}, + ) + ) + if places or loop.time() >= deadline: + return next((p for p in places if p.get("kind") in SPOKEN), None) + await asyncio.sleep(self.find_every) + + +async def _json(request: Any) -> Any: + async with request as response: + response.raise_for_status() + return await response.json() + + +def _sha256(path: Path) -> str: + with path.open("rb") as file: + return hashlib.file_digest(file, "sha256").hexdigest() diff --git a/config.py b/config.py index fc1fed7..f70725b 100644 --- a/config.py +++ b/config.py @@ -5,6 +5,7 @@ - hands/ - MCP-руки и кому какая выдана; agents.py - роли и их экземпляры; - frontends.py - окна; voice/ - голосовые точки по квартире и их ветка; - texts.py - всё, что gateway говорит, по-русски; +- transcribe.py - расшифровка голосовых и кружков через beavergram; - memory/ - справка, реплики, хендауты, вотчер, куратор; jobs/ - кроны и вебхуки. """ @@ -21,8 +22,11 @@ from beaver_agent.frontends import frontends from beaver_agent.hands import mcps from beaver_agent.jobs import jobs from beaver_agent.memory.watch import recall, replies, watch +from beaver_agent.transcribe import BeavergramTranscriber from beaver_agent.vault import DIGESTS, INDEX, TZ +beavergram = BeavergramTranscriber.from_env(os.environ) + gateway = Gateway( agents=agents, mcps=mcps, @@ -33,6 +37,7 @@ gateway = Gateway( watch=watch, recall=recall.block, user_sink=replies.write, + transcribers=[beavergram] if beavergram else [], budget=Budget(threshold=0.7), distiller=Distiller( agent="beaver-distiller", diff --git a/tests/test_transcribe.py b/tests/test_transcribe.py new file mode 100644 index 0000000..374fef7 --- /dev/null +++ b/tests/test_transcribe.py @@ -0,0 +1,179 @@ +import hashlib +import tempfile +from pathlib import Path + +import pytest +from aiohttp import web +from aiohttp.test_utils import TestServer +from beaver_gateway.transcribe import Audio + +from beaver_agent.transcribe import BeavergramTranscriber + +OWNER = 777 +VOICE = b"OggS voice" +DIGEST = hashlib.sha256(VOICE).hexdigest() + + +class Beavergram: + """Поддельный beavergram: аккаунты, поиск по sha256 и расшифровка.""" + + def __init__(self, *, empty_lookups=0, kind="voice", text=None, fail=None): + self.empty_lookups = empty_lookups + self.kind = kind + self.text = text + self.fail = fail + self.lookups = [] + self.transcribes = [] + self.auth = set() + + def app(self): + app = web.Application(middlewares=[self.gate]) + app.router.add_get("/api/accounts", self.accounts) + app.router.add_get("/api/media/lookup", self.lookup) + app.router.add_post("/api/media/transcribe", self.transcribe) + return app + + @web.middleware + async def gate(self, request, handler): + self.auth.add(request.headers.get("Authorization")) + if self.fail: + return web.json_response({"detail": "nope"}, status=self.fail) + return await handler(request) + + async def accounts(self, request): + return web.json_response( + [ + {"account_id": 1, "tg_user_id": 5, "is_bot": True}, + {"account_id": 2, "tg_user_id": OWNER, "is_bot": False}, + ] + ) + + async def lookup(self, request): + self.lookups.append(dict(request.query)) + if len(self.lookups) <= self.empty_lookups: + return web.json_response([]) + return web.json_response( + [ + { + "account_id": 2, + "chat_id": 42, + "message_id": 9, + "kind": self.kind, + "extracted_text": self.text, + } + ] + ) + + async def transcribe(self, request): + self.transcribes.append(await request.json()) + return web.json_response({"status": "done", "text": "привет из премиума"}) + + +@pytest.fixture +def voice(): + path = Path(tempfile.mkdtemp(prefix="beaver-bg-")) / "1-vu1.ogg" + path.write_bytes(VOICE) + return path + + +async def serve(fake): + server = TestServer(fake.app()) + await server.start_server() + return server + + +def transcriber(server): + return BeavergramTranscriber( + origin=str(server.make_url("")).rstrip("/"), + token="t0k", + tg_user_id=OWNER, + find_for=2.0, + find_every=0.01, + ) + + +async def test_finds_the_voice_after_a_retry_and_asks_for_a_transcript(voice): + fake = Beavergram(empty_lookups=2) + server = await serve(fake) + try: + stt = transcriber(server) + assert await stt(Audio(voice, "audio/ogg", 3)) == "привет из премиума" + finally: + await server.close() + assert len(fake.lookups) == 3 + assert fake.lookups[-1] == {"sha256": DIGEST, "account_id": "2"} + assert fake.transcribes == [ + {"account_id": 2, "chat_id": 42, "message_id": 9, "wait": 60} + ] + assert fake.auth == {"Bearer t0k"} + + +async def test_an_existing_transcript_comes_back_without_a_transcribe(voice): + fake = Beavergram(kind="video_note", text="уже расшифровано") + server = await serve(fake) + try: + assert ( + await transcriber(server)(Audio(voice, "video/mp4")) == "уже расшифровано" + ) + finally: + await server.close() + assert fake.transcribes == [] + + +async def test_music_and_other_files_are_not_touched(voice): + fake = Beavergram() + server = await serve(fake) + try: + stt = transcriber(server) + for media_type in ("audio/mpeg", "audio/mp4", None): + assert await stt(Audio(voice, media_type)) is None + assert fake.lookups == [] + fake.kind = "audio" + assert await stt(Audio(voice, "audio/ogg")) is None + finally: + await server.close() + assert len(fake.lookups) == 1 and fake.transcribes == [] + + +async def test_nothing_found_in_time_is_none(voice): + fake = Beavergram(empty_lookups=1000) + server = await serve(fake) + try: + stt = transcriber(server) + stt.find_for = 0.05 + assert await stt(Audio(voice, "audio/ogg")) is None + finally: + await server.close() + assert len(fake.lookups) >= 2 and fake.transcribes == [] + + +@pytest.mark.parametrize("status", [401, 500]) +async def test_http_errors_are_none_not_raised(voice, status): + fake = Beavergram(fail=status) + server = await serve(fake) + try: + assert await transcriber(server)(Audio(voice, "audio/ogg")) is None + finally: + await server.close() + assert fake.auth == {"Bearer t0k"} + + +def test_from_env_needs_the_token_and_the_owner(): + url = "https://tg.example.com/mcp?token=" + "a" * 64 + stt = BeavergramTranscriber.from_env( + {"BEAVERGRAM_MCP": url, "TELEGRAM_USER_ID": "777"} + ) + assert stt is not None + assert (stt.origin, stt.token, stt.tg_user_id) == ( + "https://tg.example.com", + "a" * 64, + 777, + ) + assert BeavergramTranscriber.from_env({"TELEGRAM_USER_ID": "777"}) is None + assert ( + BeavergramTranscriber.from_env( + {"BEAVERGRAM_MCP": "https://tg.example.com/mcp", "TELEGRAM_USER_ID": "777"} + ) + is None + ) + assert BeavergramTranscriber.from_env({"BEAVERGRAM_MCP": url}) is None