From ee63f8b783631f96ef4d8b0a7b6d542b05846ac6 Mon Sep 17 00:00:00 2001 From: h Date: Thu, 6 Aug 2026 12:30:37 +0200 Subject: [PATCH] feat(api,userbot,frontend): fetch only new messages in backfill by default --- backend/src/api/routers/backfill.py | 3 +- backend/src/api/routers/discover.py | 5 +- .../src/userbot/modules/capture/repository.py | 10 ++++ .../userbot/modules/jobs/handlers/backfill.py | 50 +++++++++++++++---- frontend/src/lib/api/endpoints.ts | 5 +- frontend/src/lib/components/ChatHeader.svelte | 4 +- frontend/src/lib/components/TrackChat.svelte | 4 +- .../src/lib/components/jobs/JobList.svelte | 10 ++-- .../src/lib/components/jobs/JobsPanel.svelte | 10 ++-- 9 files changed, 76 insertions(+), 25 deletions(-) diff --git a/backend/src/api/routers/backfill.py b/backend/src/api/routers/backfill.py index 0e39439..87304ac 100644 --- a/backend/src/api/routers/backfill.py +++ b/backend/src/api/routers/backfill.py @@ -16,6 +16,7 @@ class BackfillRequest(BaseModel): account_id: int chat_id: int media: bool = False + full: bool = False class FetchMediaRequest(BaseModel): @@ -70,7 +71,7 @@ async def enqueue_backfill( pool, body.account_id, "backfill", - {"chat_id": body.chat_id, "media": body.media}, + {"chat_id": body.chat_id, "media": body.media, "full": body.full}, ) return EnqueueResponse(job_id=job_id) diff --git a/backend/src/api/routers/discover.py b/backend/src/api/routers/discover.py index 4fcb524..72e3ba4 100644 --- a/backend/src/api/routers/discover.py +++ b/backend/src/api/routers/discover.py @@ -103,7 +103,10 @@ async def track_chat( await enqueue(pool, body.account_id, "enrich_chat", {"chat_id": chat_id}) if body.backfill: await enqueue( - pool, body.account_id, "backfill", {"chat_id": chat_id, "media": True} + pool, + body.account_id, + "backfill", + {"chat_id": chat_id, "media": True, "full": True}, ) return await discover.get_item(pool, body.account_id, chat_id) diff --git a/backend/src/userbot/modules/capture/repository.py b/backend/src/userbot/modules/capture/repository.py index c813ea4..4023805 100644 --- a/backend/src/userbot/modules/capture/repository.py +++ b/backend/src/userbot/modules/capture/repository.py @@ -118,6 +118,16 @@ async def upsert_message( # noqa: PLR0913 ) +async def max_message_id( + pool: asyncpg.Pool, account_id: int, chat_id: int +) -> int | None: + return await pool.fetchval( + "SELECT max(message_id) FROM messages WHERE account_id = $1 AND chat_id = $2", + account_id, + chat_id, + ) + + async def mark_deleted_box( pool: asyncpg.Pool, account_id: int, message_ids: list[int] ) -> None: diff --git a/backend/src/userbot/modules/jobs/handlers/backfill.py b/backend/src/userbot/modules/jobs/handlers/backfill.py index 7d82aa9..84a442c 100644 --- a/backend/src/userbot/modules/jobs/handlers/backfill.py +++ b/backend/src/userbot/modules/jobs/handlers/backfill.py @@ -1,7 +1,11 @@ +from pyrogram import Client from pyrogram.errors import PeerIdInvalid +from pyrogram.types import Message from userbot.modules.capture import capture_message +from userbot.modules.capture import repository as capture_repo from userbot.modules.capture.chat_meta import meta_from_chat +from userbot.modules.capture.context import CaptureContext from userbot.modules.jobs.context import JobContext from userbot.modules.jobs.registry import register from userbot.modules.stt import repository as stt_repo @@ -12,6 +16,33 @@ from utils.policy.models import CaptureToggles SAVE_EVERY = 100 +async def resolve_min_id(ctx: JobContext, chat_id: int) -> int: + cursor = ctx.job.cursor or {} + if "min_id" in cursor: + return int(cursor["min_id"]) + if ctx.job.params.get("full"): + return 0 + newest = await capture_repo.max_message_id(ctx.pool, ctx.account_id, chat_id) + return newest + 1 if newest else 0 + + +async def maybe_transcribe( + client: Client, + capture: CaptureContext, + chat_id: int, + message: Message, + self_id: int | None, +) -> None: + if not (should_transcribe_on_backfill(message, self_id) and message.chat): + return + meta = meta_from_chat(message.chat, capture.contacts.ids) + already = await stt_repo.is_transcribed( + capture.pool, capture.account_id, chat_id, message.id + ) + if capture.resolve(meta).stt and not already: + await safe_transcribe(client, capture, chat_id, message.id) + + @register("backfill") async def backfill(ctx: JobContext) -> None: client = ctx.client @@ -26,24 +57,21 @@ async def backfill(ctx: JobContext) -> None: media=bool(ctx.job.params.get("media")), self_destruct_media=False, ) - max_id = (ctx.job.cursor or {}).get("max_id", 0) + max_id = int((ctx.job.cursor or {}).get("max_id", 0)) + min_id = await resolve_min_id(ctx, chat_id) + await ctx.save_cursor({"max_id": max_id, "min_id": min_id}) processed = ctx.job.progress.get("processed", 0) - kwargs = {"max_id": max_id} if max_id else {} self_id = client.me.id if client.me else None try: - async for message in client.get_chat_history(chat_id, **kwargs): + async for message in client.get_chat_history( + chat_id, max_id=max_id, min_id=min_id + ): await capture_message(client, message, capture, toggles) - if should_transcribe_on_backfill(message, self_id) and message.chat: - meta = meta_from_chat(message.chat, capture.contacts.ids) - already = await stt_repo.is_transcribed( - capture.pool, capture.account_id, chat_id, message.id - ) - if capture.resolve(meta).stt and not already: - await safe_transcribe(client, capture, chat_id, message.id) + await maybe_transcribe(client, capture, chat_id, message, self_id) processed += 1 if processed % SAVE_EVERY == 0: next_max = message.id - 1 - await ctx.save_cursor({"max_id": next_max}) + await ctx.save_cursor({"max_id": next_max, "min_id": min_id}) await ctx.report_progress({"processed": processed, "max_id": next_max}) if await ctx.is_canceled(): return diff --git a/frontend/src/lib/api/endpoints.ts b/frontend/src/lib/api/endpoints.ts index 77cf490..8580586 100644 --- a/frontend/src/lib/api/endpoints.ts +++ b/frontend/src/lib/api/endpoints.ts @@ -358,11 +358,12 @@ export function listJobs(status?: JobStatus): Promise { export function enqueueBackfill( chatId: number, - media: boolean + media: boolean, + full = false ): Promise<{ job_id: number }> { return request<{ job_id: number }>("/backfill", { method: "POST", - body: { account_id: accounts.selectedId, chat_id: chatId, media }, + body: { account_id: accounts.selectedId, chat_id: chatId, media, full }, }); } diff --git a/frontend/src/lib/components/ChatHeader.svelte b/frontend/src/lib/components/ChatHeader.svelte index 666525c..947ecec 100644 --- a/frontend/src/lib/components/ChatHeader.svelte +++ b/frontend/src/lib/components/ChatHeader.svelte @@ -41,7 +41,7 @@ backfilling = true; try { await enqueueBackfill(chatId, true); - toasts.success("Бэкфилл запущен"); + toasts.success("Догружаем новые сообщения"); } catch { toasts.error("Не удалось запустить бэкфилл"); } finally { @@ -211,7 +211,7 @@ smaller loading={backfilling} onclick={backfill} - aria-label="Скачать историю" + aria-label="Догрузить новые сообщения" > diff --git a/frontend/src/lib/components/TrackChat.svelte b/frontend/src/lib/components/TrackChat.svelte index 47faa06..08f9f51 100644 --- a/frontend/src/lib/components/TrackChat.svelte +++ b/frontend/src/lib/components/TrackChat.svelte @@ -48,8 +48,8 @@ } busy = true; try { - await enqueueBackfill(chatId, true); - toasts.success("Бэкфилл запущен"); + await enqueueBackfill(chatId, true, true); + toasts.success("Полный бэкфилл запущен"); } catch { toasts.error("Не удалось запустить бэкфилл"); } finally { diff --git a/frontend/src/lib/components/jobs/JobList.svelte b/frontend/src/lib/components/jobs/JobList.svelte index ae8b426..3f60e1b 100644 --- a/frontend/src/lib/components/jobs/JobList.svelte +++ b/frontend/src/lib/components/jobs/JobList.svelte @@ -79,8 +79,12 @@ schedule(); } - function kindLabel(kind: string): string { - return KIND_LABELS[kind] ?? kind; + function kindLabel(job: JobView): string { + const label = KIND_LABELS[job.kind] ?? job.kind; + if (job.kind !== "backfill") { + return label; + } + return job.params.full ? `${label} (полный)` : `${label} (новые)`; } function processed(job: JobView): number | null { @@ -117,7 +121,7 @@ {#each jobs as job (job.id)}
- {kindLabel(job.kind)} + {kindLabel(job)} {#if canCancel(job)}