From 2dd4e589c3301b2c57b88c38968e1fcbb196fd3f Mon Sep 17 00:00:00 2001 From: h Date: Wed, 28 Oct 2026 01:35:56 +0000 Subject: [PATCH] feat(backend,frontend,mcp): any voice or round video can be transcribed on request, media is found by its bytes --- .../b7d2e9f4a1c6_media_storage_key_index.py | 27 ++++ backend/src/api/mcp/server.py | 23 +++ backend/src/api/routers/backfill.py | 16 +- backend/src/api/routers/media.py | 13 +- backend/src/api/transcribe.py | 64 ++++++++ backend/src/userbot/modules/stt/gate.py | 4 +- backend/src/userbot/modules/stt/service.py | 51 ++++-- backend/src/utils/read/media.py | 15 +- backend/src/utils/read/models.py | 8 + frontend/src/lib/api/endpoints.ts | 14 +- frontend/src/lib/api/media.ts | 68 +++++--- .../src/lib/components/MessageMedia.svelte | 28 +++- .../src/lib/components/media/VideoNote.svelte | 147 ++++++++++++------ .../lib/components/media/VoiceMessage.svelte | 122 ++++++++------- .../lib/components/profile/SharedMedia.svelte | 1 + frontend/src/lib/i18n/messages/media.ts | 16 ++ frontend/src/lib/media/transcript.svelte.ts | 65 ++++++++ 17 files changed, 523 insertions(+), 159 deletions(-) create mode 100644 backend/migrations/versions/b7d2e9f4a1c6_media_storage_key_index.py create mode 100644 backend/src/api/transcribe.py create mode 100644 frontend/src/lib/media/transcript.svelte.ts diff --git a/backend/migrations/versions/b7d2e9f4a1c6_media_storage_key_index.py b/backend/migrations/versions/b7d2e9f4a1c6_media_storage_key_index.py new file mode 100644 index 0000000..17e7091 --- /dev/null +++ b/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") diff --git a/backend/src/api/mcp/server.py b/backend/src/api/mcp/server.py index 5ab13c6..ed42120 100644 --- a/backend/src/api/mcp/server.py +++ b/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 `), 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 diff --git a/backend/src/api/routers/backfill.py b/backend/src/api/routers/backfill.py index d5b11fc..f5aecdc 100644 --- a/backend/src/api/routers/backfill.py +++ b/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) diff --git a/backend/src/api/routers/media.py b/backend/src/api/routers/media.py index af230c9..8222d60 100644 --- a/backend/src/api/routers/media.py +++ b/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], diff --git a/backend/src/api/transcribe.py b/backend/src/api/transcribe.py new file mode 100644 index 0000000..3fffb94 --- /dev/null +++ b/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) diff --git a/backend/src/userbot/modules/stt/gate.py b/backend/src/userbot/modules/stt/gate.py index 3759432..68c3759 100644 --- a/backend/src/userbot/modules/stt/gate.py +++ b/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}[/]") diff --git a/backend/src/userbot/modules/stt/service.py b/backend/src/userbot/modules/stt/service.py index a382b11..2922f08 100644 --- a/backend/src/userbot/modules/stt/service.py +++ b/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 + ) diff --git a/backend/src/utils/read/media.py b/backend/src/utils/read/media.py index 1995497..eabe060 100644 --- a/backend/src/utils/read/media.py +++ b/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]: diff --git a/backend/src/utils/read/models.py b/backend/src/utils/read/models.py index 45a1858..9090bcd 100644 --- a/backend/src/utils/read/models.py +++ b/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 diff --git a/frontend/src/lib/api/endpoints.ts b/frontend/src/lib/api/endpoints.ts index 5bc42e2..cbd7c98 100644 --- a/frontend/src/lib/api/endpoints.ts +++ b/frontend/src/lib/api/endpoints.ts @@ -575,15 +575,23 @@ export function fetchMedia( }); } +export interface TranscribeResult { + error: string | null; + status: "done" | "failed" | "pending"; + text: string | null; +} + export function transcribeMedia( chatId: number, - messageId: number -): Promise<{ job_id: number }> { - return request<{ job_id: number }>("/media/transcribe", { + messageId: number, + wait: number +): Promise { + return request("/media/transcribe", { body: { account_id: accounts.selectedId, chat_id: chatId, message_id: messageId, + wait, }, method: "POST", }); diff --git a/frontend/src/lib/api/media.ts b/frontend/src/lib/api/media.ts index 1199060..2c2f7af 100644 --- a/frontend/src/lib/api/media.ts +++ b/frontend/src/lib/api/media.ts @@ -3,8 +3,8 @@ import { getMessageMedia, transcribeMedia } from "#lib/api/endpoints.js"; import type { MediaRef } from "#lib/api/types.js"; import { accounts } from "#lib/stores/accounts.svelte.js"; -const TRANSCRIBE_TRIES = 10; -const TRANSCRIBE_DELAY = 2000; +const TRANSCRIBE_WAIT = 25; +const TRANSCRIBE_TRIES = 3; export type InlineMedia = | { @@ -15,7 +15,12 @@ export type InlineMedia = url: string; transcript: string | null; } - | { state: "not-downloaded"; mediaId: number; kind: string } + | { + state: "not-downloaded"; + mediaId: number; + kind: string; + transcript: string | null; + } | { state: "missing" }; export interface ViewerItem { @@ -173,7 +178,12 @@ async function resolve( return { state: "missing" }; } if (!meta.downloaded) { - return { kind: meta.kind, mediaId: meta.id, state: "not-downloaded" }; + return { + kind: meta.kind, + mediaId: meta.id, + state: "not-downloaded", + transcript: meta.extracted_text, + }; } const blob = await requestMedia(meta.id); if (blob.state === "ready") { @@ -187,7 +197,12 @@ async function resolve( }; } if (blob.state === "not-downloaded") { - return { kind: meta.kind, mediaId: meta.id, state: "not-downloaded" }; + return { + kind: meta.kind, + mediaId: meta.id, + state: "not-downloaded", + transcript: meta.extracted_text, + }; } return { state: "missing" }; } @@ -200,7 +215,12 @@ async function resolveById(media: MediaRef): Promise { return { state: "missing" }; } if (!media.downloaded) { - return { kind: media.kind, mediaId: media.id, state: "not-downloaded" }; + return { + kind: media.kind, + mediaId: media.id, + state: "not-downloaded", + transcript: null, + }; } const blob = await requestMedia(media.id); if (blob.state === "ready") { @@ -214,7 +234,12 @@ async function resolveById(media: MediaRef): Promise { }; } if (blob.state === "not-downloaded") { - return { kind: media.kind, mediaId: media.id, state: "not-downloaded" }; + return { + kind: media.kind, + mediaId: media.id, + state: "not-downloaded", + transcript: null, + }; } return { state: "missing" }; } @@ -257,32 +282,29 @@ function patchTranscript( return; } const cached = ready.get(cacheKey(account, chatId, messageId)); - if (cached?.state === "ready") { + if (cached && cached.state !== "missing") { cached.transcript = text; } } +export type TranscriptOutcome = { text: string } | { error: string }; + export async function requestTranscription( chatId: number, messageId: number -): Promise { - await transcribeMedia(chatId, messageId); +): Promise { for (let i = 0; i < TRANSCRIBE_TRIES; i += 1) { - // biome-ignore lint/performance/noAwaitInLoops: polling waits between tries on purpose - await new Promise((done) => { - setTimeout(done, TRANSCRIBE_DELAY); - }); - try { - const meta = await getMessageMedia(chatId, messageId); - if (meta.extracted_text) { - patchTranscript(chatId, messageId, meta.extracted_text); - return meta.extracted_text; - } - } catch { - return null; + // biome-ignore lint/performance/noAwaitInLoops: each try waits on the server + const result = await transcribeMedia(chatId, messageId, TRANSCRIBE_WAIT); + if (result.status === "done" && result.text) { + patchTranscript(chatId, messageId, result.text); + return { text: result.text }; + } + if (result.status === "failed") { + return { error: result.error ?? "failed" }; } } - return null; + return { error: "pending" }; } export function loadInlineMedia( diff --git a/frontend/src/lib/components/MessageMedia.svelte b/frontend/src/lib/components/MessageMedia.svelte index 9c68474..4798212 100644 --- a/frontend/src/lib/components/MessageMedia.svelte +++ b/frontend/src/lib/components/MessageMedia.svelte @@ -167,7 +167,12 @@ url={ready.url} /> {:else if ready && kind === "video_note"} - + {:else if ready && kind === "audio"} {:else if ready && isImage} @@ -210,6 +215,27 @@ {label} + {:else if media?.state === "not-downloaded" && kind === "voice"} + + {:else if media?.state === "not-downloaded" && kind === "video_note"} + {:else if media?.state === "not-downloaded" && vk !== "other"}