feat(api,userbot,frontend): fetch only new messages in backfill by default
This commit is contained in:
@@ -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)
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -358,11 +358,12 @@ export function listJobs(status?: JobStatus): Promise<JobView[]> {
|
||||
|
||||
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 },
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -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="Догрузить новые сообщения"
|
||||
>
|
||||
<Icon name="cloud-download" />
|
||||
</Button>
|
||||
|
||||
@@ -48,8 +48,8 @@
|
||||
}
|
||||
busy = true;
|
||||
try {
|
||||
await enqueueBackfill(chatId, true);
|
||||
toasts.success("Бэкфилл запущен");
|
||||
await enqueueBackfill(chatId, true, true);
|
||||
toasts.success("Полный бэкфилл запущен");
|
||||
} catch {
|
||||
toasts.error("Не удалось запустить бэкфилл");
|
||||
} finally {
|
||||
|
||||
@@ -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)}
|
||||
<div class="job">
|
||||
<div class="job-head">
|
||||
<span class="kind">{kindLabel(job.kind)}</span>
|
||||
<span class="kind">{kindLabel(job)}</span>
|
||||
{#if canCancel(job)}
|
||||
<button
|
||||
type="button"
|
||||
|
||||
@@ -56,8 +56,8 @@
|
||||
}
|
||||
starting = true;
|
||||
try {
|
||||
await enqueueBackfill(selected, media);
|
||||
toasts.success("Бэкфилл запущен");
|
||||
await enqueueBackfill(selected, media, true);
|
||||
toasts.success("Полный бэкфилл запущен");
|
||||
version += 1;
|
||||
} catch {
|
||||
toasts.error("Не удалось запустить бэкфилл");
|
||||
@@ -135,6 +135,10 @@
|
||||
|
||||
<section>
|
||||
<div class="section-title">Бэкфилл</div>
|
||||
<p class="hint">
|
||||
Полный бэкфилл перечитывает всю историю чата с самого начала. Кнопка в
|
||||
шапке чата догружает только сообщения новее последнего сохранённого.
|
||||
</p>
|
||||
|
||||
<button
|
||||
type="button"
|
||||
@@ -192,7 +196,7 @@
|
||||
onclick={start}
|
||||
>
|
||||
<Icon name="cloud-download" />
|
||||
<span>Запустить бэкфилл</span>
|
||||
<span>Запустить полный бэкфилл</span>
|
||||
</Button>
|
||||
</div>
|
||||
</section>
|
||||
|
||||
Reference in New Issue
Block a user