feat(backend,frontend,mcp): any voice or round video can be transcribed on request, media is found by its bytes

backend/migrations/versions/b7d2e9f4a1c6_media_storage_key_index.py

@@ -0,0 +1,27 @@
+"""media storage_key index for content lookup
+
+Revision ID: b7d2e9f4a1c6
+Revises: f1c4a8e2d6b3
+Create Date: 2026-10-28 01:00:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+
+revision: str = "b7d2e9f4a1c6"
+down_revision: str | None = "f1c4a8e2d6b3"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "CREATE INDEX ix_media_storage_key ON media (storage_key) "
+        "WHERE storage_key IS NOT NULL"
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS ix_media_storage_key")

backend/src/api/mcp/server.py

@@ -19,6 +19,7 @@ from api.mcp.download import (
     share_media,
 )
 from api.mcp.format import build_transcript, load_notes, load_photos, resolve_names
+from api.transcribe import request_transcript
 from dependencies.container import container
 from utils.files import resolve_mime
 from utils.read import (
@@ -87,6 +88,9 @@ Typical flows:
   returns a public link valid 24h that needs no auth (`curl -O <link>`), plus
   the content itself for small images and text files. `get_media` returns
   metadata only.
+- Voice: transcripts of played voices and round videos are already in the
+  history ("STT Content"). For "not transcribed" ones call
+  `transcribe_message`.
 - Monitoring: `set_watch` creates a local rule (kinds `keyword`,
   `new_message`, `message_deleted`, `message_edited`, `peer_online`; params in
   its description), `list_watches` lists rules, and `list_alerts` reads what
@@ -369,6 +373,25 @@ async def download_media(
     ]
 
 
+@mcp.tool
+async def transcribe_message(
+    account_id: int, chat_id: int, message_id: int, wait_seconds: int = 60
+) -> dict[str, Any]:
+    """Transcribe a voice message or round video through Telegram, any of them.
+
+    Returns the archived transcript at once when there is one; otherwise asks
+    Telegram (Premium transcription of the archived account) and waits up to
+    `wait_seconds`. `status` is `done` with `text`, `pending` (call again
+    later) or `failed` with `error`: `too_long`, `no_premium`, `not_voice`,
+    `empty`, `bot_account` or `failed`. Telegram may mark an unplayed voice as
+    listened for its sender.
+    """
+    result = await request_transcript(
+        await _pool(), account_id, chat_id, message_id, max(wait_seconds, 0)
+    )
+    return result.model_dump(mode="json", exclude_none=True)
+
+
 @mcp.tool
 async def get_callbacks(
     account_id: int, chat_id: int, message_id: int

backend/src/api/routers/backfill.py

@@ -7,6 +7,7 @@ from dishka.integrations.fastapi import DishkaRoute, FromDishka
 from fastapi import APIRouter, HTTPException, Query
 from pydantic import BaseModel
 
+from api.transcribe import TranscribeResult, request_transcript
 from utils.jobs import enqueue
 from utils.read.models import DEFAULT_LIMIT, Page
 
@@ -35,6 +36,7 @@ class TranscribeRequest(BaseModel):
     account_id: int
     chat_id: int
     message_id: int
+    wait: float = 0
 
 
 class SyncDialogsRequest(BaseModel):
@@ -105,17 +107,13 @@ async def enqueue_fetch_media(
     return EnqueueResponse(job_id=job_id)
 
 
-@router.post("/media/transcribe", status_code=201)
-async def enqueue_transcribe(
+@router.post("/media/transcribe")
+async def transcribe(
     pool: FromDishka[asyncpg.Pool], body: TranscribeRequest
-) -> EnqueueResponse:
-    job_id = await enqueue(
-        pool,
-        body.account_id,
-        "transcribe",
-        {"chat_id": body.chat_id, "message_id": body.message_id},
+) -> TranscribeResult:
+    return await request_transcript(
+        pool, body.account_id, body.chat_id, body.message_id, body.wait
     )
-    return EnqueueResponse(job_id=job_id)
 
 
 @router.post("/dialogs/sync", status_code=201)

backend/src/api/routers/media.py

@@ -8,17 +8,19 @@ from fastapi.responses import FileResponse
 from utils.cache import DAY_HEADERS, IMMUTABLE_HEADERS
 from utils.files import content_disposition, media_file_name, resolve_mime
 from utils.read.media import (
+    find_by_content,
     get_media,
     get_media_version,
     get_media_versions,
     get_message_media,
 )
-from utils.read.models import MediaVersionView, MediaView
+from utils.read.models import MediaPlaceView, MediaVersionView, MediaView
 from utils.storage import ContentAddressedStorage
 
 router = APIRouter(prefix="/api/media", tags=["media"], route_class=DishkaRoute)
 
 Download = Annotated[bool, Query()]
+Sha256 = Annotated[str, Query(pattern="^[0-9a-f]{64}$")]
 
 
 def _attachment(headers: dict[str, str], file_name: str) -> dict[str, str]:
@@ -81,6 +83,15 @@ async def message_media(
     return media
 
 
+@router.get("/lookup")
+async def lookup_media(
+    pool: FromDishka[asyncpg.Pool],
+    sha256: Sha256,
+    account_id: Annotated[int | None, Query()] = None,
+) -> list[MediaPlaceView]:
+    return await find_by_content(pool, sha256, account_id)
+
+
 @router.get("/{media_id}")
 async def serve_media(
     pool: FromDishka[asyncpg.Pool],

backend/src/api/transcribe.py

@@ -0,0 +1,64 @@
+import asyncio
+from typing import Literal
+
+import asyncpg
+from pydantic import BaseModel
+
+from utils.jobs import enqueue
+from utils.read import accounts, media
+
+MAX_WAIT_SECONDS = 60.0
+_POLL_SECONDS = 1.0
+_TRANSCRIBABLE = ("voice", "video_note")
+
+TranscribeStatus = Literal["done", "pending", "failed"]
+
+
+class TranscribeResult(BaseModel):
+    status: TranscribeStatus
+    text: str | None = None
+    error: str | None = None
+    job_id: int | None = None
+
+
+async def _text(
+    pool: asyncpg.Pool, account_id: int, chat_id: int, message_id: int
+) -> str | None:
+    item = await media.get_message_media(pool, account_id, chat_id, message_id)
+    return item.extracted_text if item else None
+
+
+async def _job_error(pool: asyncpg.Pool, job_id: int) -> str | None:
+    row = await pool.fetchrow("SELECT status, error FROM jobs WHERE id = $1", job_id)
+    if row is None or row["status"] == "canceled":
+        return "failed"
+    if row["status"] == "failed":
+        return row["error"] or "failed"
+    return None
+
+
+async def request_transcript(
+    pool: asyncpg.Pool, account_id: int, chat_id: int, message_id: int, wait: float = 0
+) -> TranscribeResult:
+    item = await media.get_message_media(pool, account_id, chat_id, message_id)
+    if item is None or item.kind not in _TRANSCRIBABLE:
+        return TranscribeResult(status="failed", error="not_voice")
+    if item.extracted_text:
+        return TranscribeResult(status="done", text=item.extracted_text)
+    account = await accounts.get_account(pool, account_id)
+    if account is None or account.is_bot:
+        return TranscribeResult(status="failed", error="bot_account")
+    job_id = await enqueue(
+        pool, account_id, "transcribe", {"chat_id": chat_id, "message_id": message_id}
+    )
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + min(wait, MAX_WAIT_SECONDS)
+    while loop.time() < deadline:
+        await asyncio.sleep(_POLL_SECONDS)
+        text = await _text(pool, account_id, chat_id, message_id)
+        if text:
+            return TranscribeResult(status="done", text=text, job_id=job_id)
+        error = await _job_error(pool, job_id)
+        if error is not None:
+            return TranscribeResult(status="failed", error=error, job_id=job_id)
+    return TranscribeResult(status="pending", job_id=job_id)

backend/src/userbot/modules/stt/gate.py

@@ -2,7 +2,7 @@ from pyrogram import Client
 from pyrogram.errors import FloodPremiumWait, FloodWait, RPCError
 
 from userbot.modules.capture.context import CaptureContext
-from userbot.modules.stt.service import transcribe_message
+from userbot.modules.stt.service import TranscribeError, transcribe_message
 from utils.logging import logger
 
 
@@ -20,5 +20,5 @@ async def safe_transcribe(
             "transcribe",
             {"chat_id": chat_id, "message_id": message_id},
         )
-    except RPCError as exc:
+    except (TranscribeError, RPCError) as exc:
         logger.warning(f"[yellow]STT failed for {chat_id}/{message_id}: {exc}[/]")

backend/src/userbot/modules/stt/service.py

@@ -1,4 +1,5 @@
 from pyrogram import Client, raw
+from pyrogram.errors import FloodPremiumWait, FloodWait, RPCError
 from pyrogram.types import Message
 
 from userbot.modules.capture.context import CaptureContext
@@ -6,6 +7,16 @@ from userbot.modules.media import self_destruct_ttl
 from userbot.modules.stt import repository
 from utils.logging import logger
 
+REFUSALS = {
+    "MSG_VOICE_TOO_LONG": "too_long",
+    "MSG_VOICE_MISSING": "not_voice",
+    "PREMIUM_ACCOUNT_REQUIRED": "no_premium",
+}
+
+
+class TranscribeError(Exception):
+    pass
+
 
 def is_transcribable(message: Message) -> bool:
     if self_destruct_ttl(message) is not None:
@@ -26,27 +37,37 @@ def should_transcribe_on_backfill(message: Message, self_id: int | None) -> bool
     return not message.unread_media
 
 
+async def _ask(
+    client: Client, chat_id: int, message_id: int
+) -> raw.types.messages.TranscribedAudio:
+    try:
+        peer = await client.resolve_peer(chat_id)
+        if peer is None:
+            msg = "failed"
+            raise TranscribeError(msg)
+        return await client.invoke(
+            raw.functions.messages.TranscribeAudio(peer=peer, msg_id=message_id)
+        )
+    except (FloodWait, FloodPremiumWait):
+        raise
+    except RPCError as exc:
+        logger.warning(f"[yellow]STT refused {chat_id}/{message_id}: {exc}[/]")
+        raise TranscribeError(REFUSALS.get(exc.ID or "", "failed")) from exc
+
+
 async def transcribe_message(
     client: Client, ctx: CaptureContext, chat_id: int, message_id: int
 ) -> None:
-    peer = await client.resolve_peer(chat_id)
-    if peer is None:
-        return
-    result = await client.invoke(
-        raw.functions.messages.TranscribeAudio(peer=peer, msg_id=message_id)
-    )
+    result = await _ask(client, chat_id, message_id)
     if result.pending:
         logger.info(
             f"[yellow]STT pending {chat_id}/{message_id} "
             f"(trial_remains={result.trial_remains_num})[/]"
         )
         return
-    if result.text:
-        await repository.set_extracted_text(
-            ctx.pool, ctx.account_id, chat_id, message_id, result.text
-        )
-    else:
-        logger.info(
-            f"[yellow]STT empty {chat_id}/{message_id} "
-            f"(trial_remains={result.trial_remains_num})[/]"
-        )
+    if not result.text:
+        msg = "empty"
+        raise TranscribeError(msg)
+    await repository.set_extracted_text(
+        ctx.pool, ctx.account_id, chat_id, message_id, result.text
+    )

backend/src/utils/read/media.py

@@ -3,7 +3,7 @@ import json
 import asyncpg
 
 from utils.files import media_file_name
-from utils.read.models import MediaVersionView, MediaView
+from utils.read.models import MediaPlaceView, MediaVersionView, MediaView
 
 MEDIA_COLS = (
     "m.id, m.account_id, m.chat_id, m.message_id, m.kind, m.storage_key, "
@@ -23,6 +23,12 @@ LEFT JOIN LATERAL (
 
 _VERSION_COLS = "id, kind, storage_key, file_size, mime, observed_at"
 
+_PLACES = """
+SELECT account_id, chat_id, message_id, kind, extracted_text FROM media
+WHERE storage_key = $1 AND ($2::bigint IS NULL OR account_id = $2)
+ORDER BY created_at DESC LIMIT 100
+"""
+
 _WEB_PAGE_MEDIA_KINDS = ("photo", "video", "animation", "document", "audio")
 
 
@@ -100,6 +106,13 @@ async def get_message_media(
     return await _web_page_media_stub(pool, account_id, chat_id, message_id)
 
 
+async def find_by_content(
+    pool: asyncpg.Pool, storage_key: str, account_id: int | None = None
+) -> list[MediaPlaceView]:
+    rows = await pool.fetch(_PLACES, storage_key, account_id)
+    return [MediaPlaceView(**dict(row)) for row in rows]
+
+
 async def get_media_versions(
     pool: asyncpg.Pool, account_id: int, chat_id: int, message_id: int
 ) -> list[MediaVersionView]:

backend/src/utils/read/models.py

@@ -275,6 +275,14 @@ class MediaView(BaseModel):
     file_name: str | None = None
 
 
+class MediaPlaceView(BaseModel):
+    account_id: int
+    chat_id: int
+    message_id: int
+    kind: str
+    extracted_text: str | None
+
+
 class MediaVersionView(BaseModel):
     id: int
     kind: str