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"}