diff --git a/backend/dist/index.js b/backend/dist/index.js index d50dfc4..83eb87e 100644 --- a/backend/dist/index.js +++ b/backend/dist/index.js @@ -4597,7 +4597,7 @@ function normalizeTelegramDownloadRefs(refs, defaultSourceEntity) { if (!refs) return void 0; return refs.filter((ref) => ref.id > 0).map((ref) => ({ ...ref, source: ref.source || defaultSourceEntity })); } -async function downloadTelegramChannelRange(botClient, requestMessage, source, startMessageId, limit = 50, direction = "older", explicitIds, folderOverride, explicitRefs) { +async function downloadTelegramChannelRange(botClient, requestMessage, source, startMessageId, limit = 50, direction = "older", explicitIds, folderOverride, explicitRefs, onItemSettled) { const userClient2 = getTelegramUserClient(); if (!userClient2 || !isTelegramUserClientReady()) { throw new Error("Telegram \u7528\u6237\u8D26\u53F7\u4E0B\u8F7D\u5668\u672A\u5C31\u7EEA\uFF1A\u8BF7\u5148\u914D\u7F6E TELEGRAM_API_ID / TELEGRAM_API_HASH \u5E76\u751F\u6210 user session"); @@ -4630,6 +4630,7 @@ async function downloadTelegramChannelRange(botClient, requestMessage, source, s if (!fileInfo) { skipped += 1; skippedMessageIds.push(ref.id); + await onItemSettled?.(ref, "skipped"); continue; } downloadableRefs.push({ @@ -4768,6 +4769,7 @@ async function downloadTelegramChannelRange(botClient, requestMessage, source, s completed += 1; failedMessageIds.push(item.id); await refreshSegmentStatus(false, fileName); + await onItemSettled?.(item, "failed", "\u6D88\u606F\u4E0D\u5B58\u5728\u6216\u65E0\u6CD5\u91CD\u65B0\u8BFB\u53D6"); return; } const uploadItem = { @@ -4798,14 +4800,17 @@ async function downloadTelegramChannelRange(botClient, requestMessage, source, s if (uploadItem.status === "success") { successful += 1; successfulMessageIds.push(item.id); + await onItemSettled?.(item, "success"); } else { failed += 1; failedMessageIds.push(item.id); + await onItemSettled?.(item, "failed", uploadItem.error || "\u4E0B\u8F7D\u5931\u8D25"); } } catch (err) { console.error(`\u{1F916} \u9891\u9053\u5206\u6BB5\u4E0B\u8F7D\u4EFB\u52A1\u5F02\u5E38: ${fileName}`, err); failed += 1; failedMessageIds.push(item.id); + await onItemSettled?.(item, "failed", err instanceof Error ? err.message : String(err)); } finally { completed += 1; found += 1; @@ -5192,8 +5197,14 @@ async function handleFileUpload(client2, event) { init_db(); var SUBSCRIPTION_INTERVAL_MS = Math.max(6e4, parseInt(process.env.TELEGRAM_SUBSCRIPTION_INTERVAL_MS || "300000", 10) || 3e5); var SUBSCRIPTION_SCAN_LIMIT = Math.max(1, parseInt(process.env.TELEGRAM_SUBSCRIPTION_SCAN_LIMIT || "100", 10) || 100); +var TG_JOB_RECOVERY_DELAY_MS = Math.max(1e3, parseInt(process.env.TG_JOB_RECOVERY_DELAY_MS || "10000", 10) || 1e4); +var TG_JOB_SCAN_SEGMENT_SIZE = Math.max(20, parseInt(process.env.TG_JOB_SCAN_SEGMENT_SIZE || "100", 10) || 100); +var TG_JOB_DOWNLOAD_BATCH_SIZE = Math.max(1, parseInt(process.env.TG_JOB_DOWNLOAD_BATCH_SIZE || "20", 10) || 20); +var TG_JOB_MAX_ATTEMPTS = Math.max(1, parseInt(process.env.TG_JOB_MAX_ATTEMPTS || "3", 10) || 3); var TELEGRAM_COMMENTS_MAX_PER_POST = Math.max(1, parseInt(process.env.TELEGRAM_COMMENTS_MAX_PER_POST || "200", 10) || 200); var subscriptionTimer = null; +var recoveryStarted = false; +var recoveryRunning = false; function parseTelegramSourceAllowlist(raw) { return (raw || "").split(",").map((item) => item.trim()).filter(Boolean).map((item) => normalizeSource(item).toLowerCase()); } @@ -5264,27 +5275,6 @@ async function getLatestMessageId(userClient2, source) { const [latest] = await userClient2.getMessages(source, { limit: 1 }); return latest?.id || 0; } -async function getMessagesByDateRange(userClient2, source, startDate, endDate, maxScan = 5e3) { - const result = []; - let offsetId = 0; - while (result.length < maxScan) { - const batch = await userClient2.getMessages(source, { limit: Math.min(100, maxScan - result.length), offsetId }); - if (!batch.length) break; - let reachedOlder = false; - for (const message of batch) { - offsetId = message.id; - const messageDate = new Date((message.date || 0) * 1e3); - if (messageDate > endDate) continue; - if (messageDate < startDate) { - reachedOlder = true; - break; - } - if (messageHasMedia(message)) result.push(message); - } - if (reachedOlder) break; - } - return result.sort((a, b) => a.id - b.id); -} function messageGroupId(message) { const groupedId = message?.groupedId; return groupedId ? groupedId.toString() : void 0; @@ -5307,13 +5297,37 @@ async function expandMessagesWithMediaGroups(userClient2, source, messages) { } return Array.from(byId.values()).sort((a, b) => a.id - b.id); } -async function persistDownloadItems(jobId, source, messages) { - for (const message of messages) { +function sourcePeerKey(value, fallback) { + if (value === void 0 || value === null) return fallback; + return String(value); +} +async function persistDownloadRefs(jobId, source, refs, folderOverride) { + for (const ref of refs) { await query( - `INSERT INTO telegram_download_items (job_id, source, message_id, grouped_id, status) - VALUES ($1, $2, $3, $4, 'pending') - ON CONFLICT (job_id, message_id) DO NOTHING`, - [jobId, source, message.id, messageGroupId(message) || null] + `INSERT INTO telegram_download_items ( + job_id, source, source_peer, origin, message_id, grouped_id, channel_post_id, + file_name, mime_type, total_size, folder_override, status, error, last_error, locked_at, completed_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, 'pending', NULL, NULL, NULL, NULL) + ON CONFLICT (job_id, source_peer, message_id) + DO UPDATE SET + file_name = COALESCE(EXCLUDED.file_name, telegram_download_items.file_name), + mime_type = COALESCE(EXCLUDED.mime_type, telegram_download_items.mime_type), + total_size = COALESCE(EXCLUDED.total_size, telegram_download_items.total_size), + folder_override = EXCLUDED.folder_override, + updated_at = NOW()`, + [ + jobId, + source, + sourcePeerKey(ref.source, source), + ref.origin || "channel", + ref.id, + null, + ref.channelPostId || null, + ref.fileInfo?.fileName || null, + ref.fileInfo?.mimeType || null, + ref.totalSize || 0, + folderOverride || null + ] ); } } @@ -5355,7 +5369,7 @@ async function getDiscussionMediaRefs(userClient2, source, postMessages, options if (seen.has(sourceKey)) continue; seen.add(sourceKey); mediaFound += 1; - refs.push({ + const ref = { id: comment.id, source: comment.chatId || source, origin: "comment", @@ -5363,7 +5377,9 @@ async function getDiscussionMediaRefs(userClient2, source, postMessages, options fileInfo, totalSize: getEstimatedFileSize(comment), message: comment - }); + }; + refs.push(ref); + await options.onRefDiscovered?.(ref); } if (batch.length === 0 || scannedForPost >= maxPerPost) break; } @@ -5383,6 +5399,9 @@ function toChannelDownloadRef(source, message) { } async function buildDownloadScanResult(userClient2, source, messages, options = {}) { const refs = messages.map((message) => toChannelDownloadRef(source, message)).filter((ref) => Boolean(ref)); + for (const ref of refs) { + await options.onRefDiscovered?.(ref); + } const commentScan = await getDiscussionMediaRefs(userClient2, source, messages, options); refs.push(...commentScan.refs); return { @@ -5393,16 +5412,37 @@ async function buildDownloadScanResult(userClient2, source, messages, options = commentMediaFound: commentScan.mediaFound }; } -async function updateDownloadItemsStatus(jobId, messageIds, status, error) { - const ids = Array.from(new Set((messageIds || []).filter((id) => id > 0))); - if (ids.length === 0) return; +async function markDownloadRefsDownloading(jobId, refs) { + for (const ref of refs) { + const sourcePeer = sourcePeerKey(ref.source, ref.origin === "comment" ? "comment" : "channel"); + await query( + `UPDATE telegram_download_items + SET status = 'downloading', locked_at = NOW(), updated_at = NOW() + WHERE job_id = $1 AND source_peer = $2 AND message_id = $3 + AND status IN ('pending', 'failed')`, + [jobId, sourcePeer, ref.id] + ); + } +} +async function markDownloadRefStatus(jobId, ref, status, error) { + const sourcePeer = sourcePeerKey(ref.source, ref.origin === "comment" ? "comment" : "channel"); await query( `UPDATE telegram_download_items - SET status = $2, error = $3, updated_at = NOW() - WHERE job_id = $1 AND message_id = ANY($4::int[])`, - [jobId, status, error || null, ids] + SET status = $3, + error = $4, + last_error = $4, + attempts = CASE WHEN $3 = 'failed' THEN attempts + 1 ELSE attempts END, + completed_at = CASE WHEN $3 IN ('success', 'skipped') THEN NOW() ELSE completed_at END, + locked_at = NULL, + updated_at = NOW() + WHERE job_id = $1 AND source_peer = $2 AND message_id = $5`, + [jobId, sourcePeer, status, error || null, ref.id] ); } +async function persistDownloadMessages(jobId, source, messages, folderOverride) { + const refs = messages.map((message) => toChannelDownloadRef(source, message)).filter((ref) => Boolean(ref)); + await persistDownloadRefs(jobId, source, refs, folderOverride); +} function parseDateOnly(value, endOfDay = false) { const match = value.match(/^(\d{4})-(\d{2})-(\d{2})$/); if (!match) throw new Error("\u65E5\u671F\u683C\u5F0F\u5FC5\u987B\u662F YYYY-MM-DD"); @@ -5422,13 +5462,17 @@ function parseDateOnly(value, endOfDay = false) { } async function createJob(userId, chatId, kind, source, params) { const result = await query( - `INSERT INTO telegram_background_jobs (user_id, chat_id, kind, source, params) - VALUES ($1, $2, $3, $4, $5) + `INSERT INTO telegram_background_jobs (user_id, chat_id, kind, source, params, status, scan_status, download_status, scan_cursor) + VALUES ($1, $2, $3, $4, $5, 'queued', 'pending', 'pending', '{}'::jsonb) RETURNING id`, [userId, chatId || null, kind, source, JSON.stringify(params)] ); return result.rows[0].id; } +async function getJob(jobId) { + const result = await query(`SELECT * FROM telegram_background_jobs WHERE id = $1`, [jobId]); + return result.rows[0] || null; +} async function updateJob(jobId, updates) { const entries = Object.entries(updates); if (entries.length === 0) return; @@ -5486,132 +5530,272 @@ async function unsubscribeTelegramChannel(userId, selector) { ); return result.rows[0] || null; } -async function enqueueTelegramDateDownload(botClient, requestMessage, userId, sourceInput, startDateText, endDateText, folderOverride, options = {}) { +async function getJobItemStats(jobId) { + const result = await query( + `SELECT status, COUNT(*)::int AS count + FROM telegram_download_items + WHERE job_id = $1 + GROUP BY status`, + [jobId] + ); + const stats = { pending: 0, downloading: 0, success: 0, failed: 0, skipped: 0 }; + for (const row of result.rows) stats[row.status] = Number(row.count || 0); + return stats; +} +async function getJobProgress(jobId) { + const job = await getJob(jobId); + if (!job) return null; + const params = job.params || {}; + const cursor = job.scan_cursor || params.scan || {}; + const stats = await getJobItemStats(jobId); + return { + jobId, + source: job.source, + mode: job.kind === "tag_download" ? "tag" : "date", + status: job.status, + scanStatus: job.scan_status || "pending", + downloadStatus: job.download_status || "pending", + channelMessagesScanned: Number(cursor.channelMessagesScanned || 0), + channelMediaFound: Number(cursor.channelMediaFound || 0), + commentMessagesScanned: Number(cursor.commentMessagesScanned || 0), + commentMediaFound: Number(cursor.commentMediaFound || 0), + totalMediaFound: Number(job.total_count || 0), + completed: Number(stats.success || 0), + pending: Number(stats.pending || 0), + downloading: Number(stats.downloading || 0), + failed: Number(stats.failed || 0), + skipped: Number(stats.skipped || 0), + cooldownUntil: job.cooldown_until ? new Date(job.cooldown_until).toISOString() : null + }; +} +async function notifyProgress(jobId, options) { + const progress = await getJobProgress(jobId); + if (progress) await options.onProgress?.(progress); +} +function isFloodWait(error) { + const anyErr = error; + const text = `${anyErr?.message || ""} ${anyErr?.errorMessage || ""}`; + const seconds = Number(anyErr?.seconds || anyErr?.value || text.match(/FLOOD_WAIT_?(\d+)/i)?.[1] || 0); + if (seconds > 0 || /FLOOD|Too many requests/i.test(text)) return { seconds: Math.max(30, seconds || 60) }; + return null; +} +async function ensureJobCanRun(jobId) { + const job = await getJob(jobId); + if (!job) return "cancelled"; + if (job.cancelled_at || job.status === "cancelled") return "cancelled"; + if (job.paused_at || job.status === "paused") return "paused"; + if (job.cooldown_until && new Date(job.cooldown_until).getTime() > Date.now()) return "cooldown"; + return "run"; +} +async function waitUntilRunnable(jobId, options) { + while (true) { + const state = await ensureJobCanRun(jobId); + if (state === "run") return true; + if (state === "cancelled") return false; + await notifyProgress(jobId, options); + await new Promise((resolve) => setTimeout(resolve, state === "cooldown" ? 5e3 : 2e3)); + } +} +async function claimPendingDownloadRefs(jobId, limit = TG_JOB_DOWNLOAD_BATCH_SIZE) { + const result = await query( + `UPDATE telegram_download_items i + SET status = 'downloading', locked_at = NOW(), updated_at = NOW() + WHERE i.id IN ( + SELECT id FROM telegram_download_items + WHERE job_id = $1 AND status = 'pending' AND attempts < $2 + ORDER BY created_at ASC + LIMIT $3 + ) + RETURNING source, source_peer, origin, message_id, channel_post_id, file_name, mime_type, total_size, folder_override`, + [jobId, TG_JOB_MAX_ATTEMPTS, limit] + ); + return result.rows.filter((row) => row.file_name && row.mime_type).map((row) => ({ + id: Number(row.message_id), + source: row.source_peer || row.source, + origin: row.origin === "comment" ? "comment" : "channel", + channelPostId: row.channel_post_id || void 0, + fileInfo: { fileName: row.file_name, mimeType: row.mime_type }, + totalSize: Number(row.total_size || 0) + })); +} +async function downloadClaimedRefs(botClient, requestMessage, jobId, source, refs, folderOverride, options) { + if (refs.length === 0) return { found: 0, skipped: 0, failed: 0, successful: 0 }; + await updateJob(jobId, { status: "running", download_status: "active", error: null }); + try { + const result = await downloadTelegramChannelRange(botClient, requestMessage, source, 0, refs.length, "older", refs.map((ref) => ref.id), folderOverride, refs, async (ref, status, error) => { + await markDownloadRefStatus(jobId, ref, status, error); + await notifyProgress(jobId, options); + }); + return result; + } catch (error) { + const flood = isFloodWait(error); + if (flood) { + const cooldownUntil = new Date(Date.now() + flood.seconds * 1e3); + await updateJob(jobId, { status: "running", cooldown_until: cooldownUntil, error: `Telegram FloodWait\uFF0C\u51B7\u5374\u5230 ${cooldownUntil.toISOString()}` }); + for (const ref of refs) await markDownloadRefStatus(jobId, ref, "failed", `FloodWait ${flood.seconds}s`); + return { found: 0, skipped: 0, failed: refs.length, successful: 0 }; + } + for (const ref of refs) await markDownloadRefStatus(jobId, ref, "failed", error instanceof Error ? error.message : String(error)); + throw error; + } +} +async function downloadPendingForJob(botClient, requestMessage, jobId, source, folderOverride, options, drain = false) { + let aggregate = { found: 0, skipped: 0, failed: 0, successful: 0 }; + while (await waitUntilRunnable(jobId, options)) { + const refs = await claimPendingDownloadRefs(jobId); + if (refs.length === 0) break; + const result = await downloadClaimedRefs(botClient, requestMessage, jobId, source, refs, folderOverride, options); + aggregate = { + found: aggregate.found + (result.found || 0), + skipped: aggregate.skipped + (result.skipped || 0), + failed: aggregate.failed + (result.failed || 0), + successful: aggregate.successful + (result.successful || 0) + }; + if (!drain) break; + } + return aggregate; +} +async function finalizeTelegramJob(jobId, options) { + const stats = await getJobItemStats(jobId); + const pending = Number(stats.pending || 0) + Number(stats.downloading || 0); + const failed = Number(stats.failed || 0); + const status = pending > 0 ? "running" : failed > 0 ? "completed_with_errors" : "completed"; + await updateJob(jobId, { + status, + download_status: pending > 0 ? "active" : "done", + enqueued_count: Number(stats.success || 0), + skipped_count: Number(stats.skipped || 0), + error: failed > 0 ? `${failed} \u4E2A\u6587\u4EF6\u4E0B\u8F7D\u5931\u8D25` : null, + finished_at: pending > 0 ? null : /* @__PURE__ */ new Date() + }); + await notifyProgress(jobId, options); +} +async function scanChannelSegment(userClient2, jobId, source, params, cursor, options) { + const mode = params.mode; + const offsetId = Number(cursor.offsetId || 0); + const batch = await userClient2.getMessages(source, { + limit: TG_JOB_SCAN_SEGMENT_SIZE, + offsetId, + ...mode === "tag" ? { search: params.tag } : {} + }); + if (!batch.length) return { messages: [], done: true, nextOffsetId: offsetId }; + let done = false; + let nextOffsetId = offsetId; + const matched = []; + for (const message of batch) { + nextOffsetId = message.id; + if (mode === "date") { + const messageDate = new Date((message.date || 0) * 1e3); + const startDate = new Date(params.startDateIso); + const endDate = new Date(params.endDateIso); + if (messageDate > endDate) continue; + if (messageDate < startDate) { + done = true; + break; + } + if (messageHasMedia(message)) matched.push(message); + } else if (messageHasMedia(message) && messageMatchesHashtag(message, params.tag)) { + matched.push(message); + } + } + const expanded = await expandMessagesWithMediaGroups(userClient2, source, matched); + return { messages: expanded, done: done || batch.length < TG_JOB_SCAN_SEGMENT_SIZE, nextOffsetId }; +} +async function runSegmentedTelegramJob(botClient, requestMessage, jobId, source, folderOverride, options) { const userClient2 = requireUserClient(); + const job = await getJob(jobId); + const params = job?.params || {}; + let cursor = job?.scan_cursor || {}; + const discoveredRefKeys = /* @__PURE__ */ new Set(); + let totals = { found: 0, skipped: 0, failed: 0, successful: 0 }; + await updateJob(jobId, { status: "running", scan_status: "scanning", download_status: "active", started_at: job?.started_at || /* @__PURE__ */ new Date(), error: null }); + while (await waitUntilRunnable(jobId, options)) { + const current = await getJob(jobId); + cursor = current?.scan_cursor || cursor || {}; + if (current?.scan_status === "done") break; + try { + const segment = await scanChannelSegment(userClient2, jobId, source, params, cursor, options); + const onRefDiscovered = async (ref) => { + const key = `${sourcePeerKey(ref.source, source)}:${ref.id}`; + if (discoveredRefKeys.has(key)) return; + discoveredRefKeys.add(key); + await persistDownloadRefs(jobId, source, [ref], folderOverride); + }; + const scan = await buildDownloadScanResult(userClient2, source, segment.messages, { + ...options, + tag: params.tag, + startDate: params.startDateIso ? new Date(params.startDateIso) : void 0, + endDate: params.endDateIso ? new Date(params.endDateIso) : void 0, + onRefDiscovered + }); + cursor = { + ...cursor, + phase: segment.done ? "done" : "channel", + offsetId: segment.nextOffsetId, + channelMessagesScanned: Number(cursor.channelMessagesScanned || 0) + segment.messages.length, + channelMediaFound: Number(cursor.channelMediaFound || 0) + scan.channelMediaFound, + commentMessagesScanned: Number(cursor.commentMessagesScanned || 0) + scan.commentMessagesScanned, + commentMediaFound: Number(cursor.commentMediaFound || 0) + scan.commentMediaFound + }; + const stats = await getJobItemStats(jobId); + await updateJob(jobId, { scan_cursor: JSON.stringify(cursor), total_count: Number(stats.pending || 0) + Number(stats.downloading || 0) + Number(stats.success || 0) + Number(stats.failed || 0) + Number(stats.skipped || 0), scan_status: segment.done ? "done" : "scanning" }); + await notifyProgress(jobId, options); + const partial = await downloadPendingForJob(botClient, requestMessage, jobId, source, folderOverride, options, false); + totals = { found: totals.found + partial.found, skipped: totals.skipped + partial.skipped, failed: totals.failed + partial.failed, successful: totals.successful + partial.successful }; + if (segment.done) break; + } catch (error) { + const flood = isFloodWait(error); + if (!flood) throw error; + const cooldownUntil = new Date(Date.now() + flood.seconds * 1e3); + await updateJob(jobId, { cooldown_until: cooldownUntil, error: `Telegram FloodWait\uFF0C\u51B7\u5374\u5230 ${cooldownUntil.toISOString()}` }); + } + } + const runnable = await waitUntilRunnable(jobId, options); + if (!runnable) { + await updateJob(jobId, { status: "cancelled", scan_status: "cancelled", download_status: "cancelled", finished_at: /* @__PURE__ */ new Date() }); + return { jobId, ...totals, requested: 0, commentMessagesScanned: Number(cursor.commentMessagesScanned || 0), commentMediaFound: Number(cursor.commentMediaFound || 0) }; + } + await updateJob(jobId, { scan_status: "done" }); + const drained = await downloadPendingForJob(botClient, requestMessage, jobId, source, folderOverride, options, true); + totals = { found: totals.found + drained.found, skipped: totals.skipped + drained.skipped, failed: totals.failed + drained.failed, successful: totals.successful + drained.successful }; + await finalizeTelegramJob(jobId, options); + return { jobId, ...totals, requested: totals.found + totals.skipped, commentMessagesScanned: Number(cursor.commentMessagesScanned || 0), commentMediaFound: Number(cursor.commentMediaFound || 0) }; +} +async function enqueueTelegramDateDownload(botClient, requestMessage, userId, sourceInput, startDateText, endDateText, folderOverride, options = {}) { const source = normalizeSource(sourceInput); await assertTelegramSourceAllowed(source); const startDate = parseDateOnly(startDateText); const endDate = parseDateOnly(endDateText, true); if (startDate > endDate) throw new Error("\u5F00\u59CB\u65E5\u671F\u4E0D\u80FD\u665A\u4E8E\u7ED3\u675F\u65E5\u671F"); - const baseMessages = await getMessagesByDateRange(userClient2, source, startDate, endDate); - const messages = await expandMessagesWithMediaGroups(userClient2, source, baseMessages); - const scan = await buildDownloadScanResult(userClient2, source, messages, { ...options, startDate, endDate }); - const commentLimit = options.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST; - await options.onScanComplete?.({ - source, - mode: "date", - channelMessagesScanned: messages.length, - channelMediaFound: scan.channelMediaFound, - commentMessagesScanned: scan.commentMessagesScanned, - commentMediaFound: scan.commentMediaFound, - totalMediaFound: scan.refs.length, - commentsEnabled: Boolean(options.includeComments), - commentsMaxPerPost: commentLimit - }); - const messageIds = scan.refs.map((ref) => ref.id); const jobId = await createJob(userId, requestMessage.chatId?.toString(), "date_range", source, { + mode: "date", startDate: startDateText, endDate: endDateText, - messageIds, + startDateIso: startDate.toISOString(), + endDateIso: endDate.toISOString(), folderOverride: folderOverride || null, includeComments: Boolean(options.includeComments), - commentsMaxPerPost: options.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST, - commentMessagesScanned: scan.commentMessagesScanned, - commentMediaFound: scan.commentMediaFound + commentsMaxPerPost: options.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST }); - await persistDownloadItems(jobId, source, messages); - await updateJob(jobId, { status: "running", started_at: /* @__PURE__ */ new Date(), total_count: scan.refs.length }); - try { - const result = await downloadTelegramChannelRange(botClient, requestMessage, source, 0, scan.refs.length, "older", messageIds, folderOverride, scan.refs); - await updateDownloadItemsStatus(jobId, result.successfulMessageIds, "success"); - await updateDownloadItemsStatus(jobId, result.failedMessageIds, "failed", "\u4E0B\u8F7D\u5931\u8D25"); - await updateDownloadItemsStatus(jobId, result.skippedMessageIds, "skipped"); - await updateJob(jobId, { - status: result.failed > 0 ? "completed_with_errors" : "completed", - enqueued_count: result.found, - skipped_count: result.skipped, - error: result.failed > 0 ? `${result.failed} \u4E2A\u6587\u4EF6\u4E0B\u8F7D\u5931\u8D25` : null, - finished_at: /* @__PURE__ */ new Date() - }); - return { jobId, commentMessagesScanned: scan.commentMessagesScanned, commentMediaFound: scan.commentMediaFound, ...result }; - } catch (error) { - await updateJob(jobId, { status: "failed", error: error instanceof Error ? error.message : String(error), finished_at: /* @__PURE__ */ new Date() }); - throw error; - } -} -async function getMessagesByHashtag(userClient2, source, tag, maxScan = 1e4) { - const normalizedTag = normalizeHashtag(tag); - const result = []; - let offsetId = 0; - while (result.length < maxScan) { - const batch = await userClient2.getMessages(source, { - limit: Math.min(100, maxScan - result.length), - offsetId, - search: normalizedTag - }); - if (!batch.length) break; - for (const message of batch) { - offsetId = message.id; - if (messageHasMedia(message) && messageMatchesHashtag(message, normalizedTag)) { - result.push(message); - } - } - } - return result.sort((a, b) => a.id - b.id); + return runSegmentedTelegramJob(botClient, requestMessage, jobId, source, folderOverride, options); } async function enqueueTelegramTagDownload(botClient, requestMessage, userId, sourceInput, tagInput, folderOverride, options = {}) { - const userClient2 = requireUserClient(); const source = normalizeSource(sourceInput); await assertTelegramSourceAllowed(source); const tag = normalizeHashtag(tagInput); - const baseMessages = await getMessagesByHashtag(userClient2, source, tag); - const messages = await expandMessagesWithMediaGroups(userClient2, source, baseMessages); - const scan = await buildDownloadScanResult(userClient2, source, messages, { ...options, tag }); - const commentLimit = options.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST; - await options.onScanComplete?.({ - source, - mode: "tag", - channelMessagesScanned: messages.length, - channelMediaFound: scan.channelMediaFound, - commentMessagesScanned: scan.commentMessagesScanned, - commentMediaFound: scan.commentMediaFound, - totalMediaFound: scan.refs.length, - commentsEnabled: Boolean(options.includeComments), - commentsMaxPerPost: commentLimit - }); - const messageIds = scan.refs.map((ref) => ref.id); const jobId = await createJob(userId, requestMessage.chatId?.toString(), "tag_download", source, { + mode: "tag", tag, - messageIds, folderOverride: folderOverride || null, includeComments: Boolean(options.includeComments), - commentsMaxPerPost: options.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST, - commentMessagesScanned: scan.commentMessagesScanned, - commentMediaFound: scan.commentMediaFound + commentsMaxPerPost: options.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST }); - await persistDownloadItems(jobId, source, messages); - await updateJob(jobId, { status: "running", started_at: /* @__PURE__ */ new Date(), total_count: scan.refs.length }); - try { - const result = await downloadTelegramChannelRange(botClient, requestMessage, source, 0, scan.refs.length, "older", messageIds, folderOverride, scan.refs); - await updateDownloadItemsStatus(jobId, result.successfulMessageIds, "success"); - await updateDownloadItemsStatus(jobId, result.failedMessageIds, "failed", "\u4E0B\u8F7D\u5931\u8D25"); - await updateDownloadItemsStatus(jobId, result.skippedMessageIds, "skipped"); - await updateJob(jobId, { - status: result.failed > 0 ? "completed_with_errors" : "completed", - enqueued_count: result.found, - skipped_count: result.skipped, - error: result.failed > 0 ? `${result.failed} \u4E2A\u6587\u4EF6\u4E0B\u8F7D\u5931\u8D25` : null, - finished_at: /* @__PURE__ */ new Date() - }); - return { jobId, tag, commentMessagesScanned: scan.commentMessagesScanned, commentMediaFound: scan.commentMediaFound, ...result }; - } catch (error) { - await updateJob(jobId, { status: "failed", error: error instanceof Error ? error.message : String(error), finished_at: /* @__PURE__ */ new Date() }); - throw error; - } + const result = await runSegmentedTelegramJob(botClient, requestMessage, jobId, source, folderOverride, options); + return { ...result, tag }; } async function listTelegramBackgroundJobs(userId, limit = 10) { const result = await query( - `SELECT id, kind, source, status, total_count, enqueued_count, skipped_count, duplicate_count, error, created_at, updated_at + `SELECT id, kind, source, status, scan_status, download_status, scan_cursor, cooldown_until, paused_at, cancelled_at, total_count, enqueued_count, skipped_count, duplicate_count, error, created_at, updated_at FROM telegram_background_jobs WHERE user_id = $1 ORDER BY created_at DESC @@ -5620,6 +5804,56 @@ async function listTelegramBackgroundJobs(userId, limit = 10) { ); return result.rows; } +async function pauseTelegramBackgroundJob(userId, selector) { + const result = await query( + `UPDATE telegram_background_jobs + SET status = 'paused', paused_at = NOW(), updated_at = NOW() + WHERE user_id = $1 AND id::text LIKE $2 AND status NOT IN ('completed', 'cancelled') + RETURNING id, source, status`, + [userId, `${selector}%`] + ); + return result.rows[0] || null; +} +async function resumeTelegramBackgroundJob(userId, selector) { + const result = await query( + `UPDATE telegram_background_jobs + SET status = 'running', paused_at = NULL, updated_at = NOW() + WHERE user_id = $1 AND id::text LIKE $2 AND status = 'paused' + RETURNING id, source, status`, + [userId, `${selector}%`] + ); + return result.rows[0] || null; +} +async function cancelTelegramBackgroundJob(userId, selector) { + const result = await query( + `UPDATE telegram_background_jobs + SET status = 'cancelled', scan_status = 'cancelled', download_status = 'cancelled', cancelled_at = NOW(), finished_at = NOW(), updated_at = NOW() + WHERE user_id = $1 AND id::text LIKE $2 AND status NOT IN ('completed', 'cancelled') + RETURNING id, source, status`, + [userId, `${selector}%`] + ); + if (result.rows[0]) { + await query(`UPDATE telegram_download_items SET status = 'skipped', locked_at = NULL, updated_at = NOW() WHERE job_id = $1 AND status IN ('pending', 'downloading')`, [result.rows[0].id]); + } + return result.rows[0] || null; +} +async function retryTelegramBackgroundJob(userId, selector) { + const result = await query( + `SELECT id FROM telegram_background_jobs WHERE user_id = $1 AND id::text LIKE $2 LIMIT 1`, + [userId, `${selector}%`] + ); + const row = result.rows[0]; + if (!row) return null; + const retry = await query( + `UPDATE telegram_download_items + SET status = 'pending', locked_at = NULL, last_error = NULL, error = NULL, updated_at = NOW() + WHERE job_id = $1 AND status = 'failed' + RETURNING id`, + [row.id] + ); + await updateJob(row.id, { status: "running", download_status: "active", error: null, finished_at: null }); + return { id: row.id, retried: retry.rowCount || 0 }; +} async function runSubscriptionScan(botClient) { const userClient2 = getTelegramUserClient(); if (!userClient2 || !isTelegramUserClientReady()) return; @@ -5639,14 +5873,13 @@ async function runSubscriptionScan(botClient) { const ids = Array.from({ length: count }, (_, index) => lastMessageId + index + 1); const jobId = await createJob(Number(row.user_id), row.chat_id?.toString(), "subscription_sync", row.source, { fromId: lastMessageId + 1, toId: latestMessageId }); const candidateMessages = await expandMessagesWithMediaGroups(userClient2, row.source, (await userClient2.getMessages(row.source, { ids })).filter(Boolean)); - await persistDownloadItems(jobId, row.source, candidateMessages); + await persistDownloadMessages(jobId, row.source, candidateMessages, row.folder_override || null); await updateJob(jobId, { status: "running", started_at: /* @__PURE__ */ new Date(), total_count: ids.length }); const targetChat = row.chat_id || row.user_id; const requestMessage = { chatId: targetChat, id: latestMessageId }; - const downloadResult = await downloadTelegramChannelRange(botClient, requestMessage, row.source, 0, ids.length, "newer", ids, row.folder_override || null); - await updateDownloadItemsStatus(jobId, downloadResult.successfulMessageIds, "success"); - await updateDownloadItemsStatus(jobId, downloadResult.failedMessageIds, "failed", "\u4E0B\u8F7D\u5931\u8D25"); - await updateDownloadItemsStatus(jobId, downloadResult.skippedMessageIds, "skipped"); + const subscriptionRefs = candidateMessages.map((message) => toChannelDownloadRef(row.source, message)).filter((ref) => Boolean(ref)); + await markDownloadRefsDownloading(jobId, subscriptionRefs); + const downloadResult = await downloadTelegramChannelRange(botClient, requestMessage, row.source, 0, ids.length, "newer", ids, row.folder_override || null, subscriptionRefs, (ref, status, error) => markDownloadRefStatus(jobId, ref, status, error)); await updateJob(jobId, { status: downloadResult.failed > 0 ? "completed_with_errors" : "completed", enqueued_count: downloadResult.found, @@ -5665,6 +5898,95 @@ async function runSubscriptionScan(botClient) { } } } +async function recoverTelegramJob(botClient, job) { + const itemResult = await query( + `SELECT id, source, source_peer, origin, message_id, channel_post_id, file_name, mime_type, total_size, folder_override + FROM telegram_download_items + WHERE job_id = $1 AND status = 'pending' + ORDER BY created_at ASC`, + [job.id] + ); + if (itemResult.rows.length === 0) return; + const refs = itemResult.rows.filter((row) => row.file_name && row.mime_type).map((row) => ({ + id: Number(row.message_id), + source: row.source_peer || row.source, + origin: row.origin === "comment" ? "comment" : "channel", + channelPostId: row.channel_post_id || void 0, + fileInfo: { fileName: row.file_name, mimeType: row.mime_type }, + totalSize: Number(row.total_size || 0) + })); + if (refs.length === 0) return; + const targetChat = job.chat_id || job.user_id; + const requestMessage = { chatId: targetChat, id: 0 }; + console.log(`\u267B\uFE0F \u6062\u590D Telegram \u4E0B\u8F7D\u4EFB\u52A1 ${String(job.id).slice(0, 8)}\uFF0C\u5F85\u5904\u7406 ${refs.length} \u4E2A\u6587\u4EF6`); + await updateJob(job.id, { status: "running", started_at: job.started_at || /* @__PURE__ */ new Date(), error: null }); + try { + const result = await downloadTelegramChannelRange( + botClient, + requestMessage, + job.source, + 0, + refs.length, + "older", + refs.map((ref) => ref.id), + itemResult.rows[0]?.folder_override || null, + refs, + (ref, status, error) => markDownloadRefStatus(job.id, ref, status, error) + ); + await updateJob(job.id, { + status: result.failed > 0 ? "completed_with_errors" : "completed", + enqueued_count: result.found, + skipped_count: result.skipped, + error: result.failed > 0 ? `${result.failed} \u4E2A\u6587\u4EF6\u4E0B\u8F7D\u5931\u8D25` : null, + finished_at: /* @__PURE__ */ new Date() + }); + await botClient.sendMessage(targetChat, { + message: `\u267B\uFE0F \u5DF2\u6062\u590D\u5E76\u5B8C\u6210\u4EFB\u52A1 ${String(job.id).slice(0, 8)}\uFF1A\u6210\u529F ${result.successful}\uFF0C\u8DF3\u8FC7 ${result.skipped}\uFF0C\u5931\u8D25 ${result.failed}` + }).catch(() => void 0); + } catch (error) { + await updateJob(job.id, { status: "failed", error: error instanceof Error ? error.message : String(error), finished_at: /* @__PURE__ */ new Date() }); + throw error; + } +} +async function recoverInterruptedTelegramJobs(botClient) { + if (recoveryRunning) return; + recoveryRunning = true; + try { + await query( + `UPDATE telegram_download_items + SET status = 'pending', locked_at = NULL, updated_at = NOW() + WHERE status = 'downloading' + AND (locked_at IS NULL OR locked_at < NOW() - INTERVAL '30 minutes')` + ); + const jobs = await query( + `SELECT DISTINCT j.* + FROM telegram_background_jobs j + JOIN telegram_download_items i ON i.job_id = j.id + WHERE j.kind IN ('date_range', 'tag_download') + AND j.status IN ('pending', 'running', 'failed', 'completed_with_errors') + AND i.status = 'pending' + ORDER BY j.created_at ASC + LIMIT 5` + ); + for (const job of jobs.rows) { + if (job.scan_status !== "done" && (job.kind === "date_range" || job.kind === "tag_download") && job.params?.mode) { + const targetChat = job.chat_id || job.user_id; + const requestMessage = { chatId: targetChat, id: 0 }; + await runSegmentedTelegramJob(botClient, requestMessage, job.id, job.source, job.params?.folderOverride || null, {}).catch((error) => console.error("\u267B\uFE0F Telegram \u5206\u6BB5\u4EFB\u52A1\u6062\u590D\u5931\u8D25:", error)); + } else { + await recoverTelegramJob(botClient, job).catch((error) => console.error("\u267B\uFE0F Telegram \u4EFB\u52A1\u6062\u590D\u5931\u8D25:", error)); + } + } + } finally { + recoveryRunning = false; + } +} +function startTelegramJobRecoveryWorker(botClient) { + if (recoveryStarted) return; + recoveryStarted = true; + setTimeout(() => recoverInterruptedTelegramJobs(botClient).catch((error) => console.error("\u267B\uFE0F Telegram \u4EFB\u52A1\u6062\u590D\u626B\u63CF\u5931\u8D25:", error)), TG_JOB_RECOVERY_DELAY_MS); + setInterval(() => recoverInterruptedTelegramJobs(botClient).catch((error) => console.error("\u267B\uFE0F Telegram \u4EFB\u52A1\u6062\u590D\u626B\u63CF\u5931\u8D25:", error)), SUBSCRIPTION_INTERVAL_MS); +} function startTelegramSubscriptionWorker(botClient) { if (subscriptionTimer) return; subscriptionTimer = setInterval(() => { @@ -6384,6 +6706,15 @@ async function handleStopTasks(message) { } async function handlePauseTasks(message, args = []) { const taskId = args[0]; + const senderId = message.senderId?.toJSNumber(); + if (taskId && senderId) { + const job = await pauseTelegramBackgroundJob(senderId, taskId); + if (job) { + await message.reply({ message: `\u23F8\uFE0F \u5DF2\u6682\u505C\u9891\u9053\u4EFB\u52A1 ${String(job.id).slice(0, 8)} +\u6765\u6E90\uFF1A${job.source}` }); + return; + } + } const result = pauseDownloadTasks(taskId); await message.reply({ message: `\u23F8\uFE0F \u5DF2\u6682\u505C\u5168\u5C40\u4E0B\u8F7D\u961F\u5217${taskId ? ` \u4EFB\u52A1\u5361: ${taskId}` : ""} @@ -6395,6 +6726,15 @@ async function handlePauseTasks(message, args = []) { } async function handleResumeTasks(message, args = []) { const taskId = args[0]; + const senderId = message.senderId?.toJSNumber(); + if (taskId && senderId) { + const job = await resumeTelegramBackgroundJob(senderId, taskId); + if (job) { + await message.reply({ message: `\u25B6\uFE0F \u5DF2\u7EE7\u7EED\u9891\u9053\u4EFB\u52A1 ${String(job.id).slice(0, 8)} +\u6765\u6E90\uFF1A${job.source}` }); + return; + } + } const result = resumeDownloadTasks(taskId); await message.reply({ message: `\u25B6\uFE0F \u5DF2\u7EE7\u7EED\u5168\u5C40\u4E0B\u8F7D\u961F\u5217${taskId ? ` \u4EFB\u52A1\u5361: ${taskId}` : ""} @@ -6404,6 +6744,15 @@ async function handleResumeTasks(message, args = []) { } async function handleCancelTask(message, args) { const selector = args.join(" ").trim() || "all"; + const senderId = message.senderId?.toJSNumber(); + if (selector !== "all" && senderId) { + const job = await cancelTelegramBackgroundJob(senderId, selector); + if (job) { + await message.reply({ message: `\u{1F6D1} \u5DF2\u53D6\u6D88\u9891\u9053\u4EFB\u52A1 ${String(job.id).slice(0, 8)} +\u6765\u6E90\uFF1A${job.source}` }); + return; + } + } const result = cancelDownloadTask(selector); await message.reply({ message: result.total > 0 ? `\u{1F6D1} \u5DF2\u53D6\u6D88\u5339\u914D\u4EFB\u52A1 @@ -6412,6 +6761,16 @@ async function handleCancelTask(message, args) { \u7B49\u5F85\u4E2D: ${result.pending}` : `\u{1F4EE} \u6CA1\u6709\u627E\u5230\u5339\u914D\u7684\u4EFB\u52A1\uFF0C\u672A\u6E05\u7A7A\u5168\u5C40\u961F\u5217\uFF1A${selector}` }); } async function handleRetryFailedTasks(message, args) { + const senderId = message.senderId?.toJSNumber(); + const jobSelector = args.find((arg) => /^[0-9a-f-]{4,}$/i.test(arg)); + if (senderId && jobSelector) { + const jobRetry = await retryTelegramBackgroundJob(senderId, jobSelector); + if (jobRetry) { + await message.reply({ message: jobRetry.retried > 0 ? `\u{1F504} \u5DF2\u91CD\u65B0\u52A0\u5165\u9891\u9053\u4EFB\u52A1\u5931\u8D25\u9879 ${jobRetry.retried} \u4E2A +\u4EFB\u52A1: ${String(jobRetry.id).slice(0, 8)}` : "\u{1F4EE} \u8BE5\u9891\u9053\u4EFB\u52A1\u6CA1\u6709\u53EF\u91CD\u8BD5\u5931\u8D25\u9879" }); + return; + } + } const taskId = args.find((arg) => /^t[a-z0-9]+/i.test(arg)); const numericArg = args.find((arg) => /^\d+$/.test(arg)); const limit = Math.max(1, Math.min(50, parseInt(numericArg || "10", 10) || 10)); @@ -7180,6 +7539,25 @@ function buildTelegramWizardPrompt(state) { function isDateOnly(text) { return /^\d{4}-\d{2}-\d{2}$/.test(text.trim()); } +async function updateJobProgressMessage(statusMessage, summary) { + const totalDone = summary.completed + summary.failed + summary.skipped; + const lines = [ + summary.status === "paused" ? `\u23F8\uFE0F **\u9891\u9053\u4E0B\u8F7D\u5DF2\u6682\u505C**` : summary.status === "cancelled" ? `\u{1F6D1} **\u9891\u9053\u4E0B\u8F7D\u5DF2\u53D6\u6D88**` : totalDone >= summary.totalMediaFound && summary.scanStatus === "done" ? `\u2705 **\u9891\u9053\u4EFB\u52A1\u5B8C\u6210**` : `\u{1F50E} **\u9891\u9053\u4EFB\u52A1\u8FD0\u884C\u4E2D**`, + `\u{1F194} job: ${summary.jobId.slice(0, 8)}`, + `\u{1F4CD} \u9891\u9053\uFF1A${summary.source}`, + ``, + `\u{1F50E} \u626B\u63CF\uFF1A${summary.scanStatus}`, + `\u{1F4C4} \u9891\u9053\u6B63\u6587\uFF1A\u5DF2\u626B ${summary.channelMessagesScanned} \u6761\uFF0C\u53D1\u73B0 ${summary.channelMediaFound} \u4E2A\u6587\u4EF6`, + `\u{1F4AC} \u8BC4\u8BBA\u533A\uFF1A\u5DF2\u626B ${summary.commentMessagesScanned} \u6761\uFF0C\u53D1\u73B0 ${summary.commentMediaFound} \u4E2A\u6587\u4EF6`, + ``, + `\u2B07\uFE0F \u4E0B\u8F7D\uFF1A${summary.downloadStatus}`, + `\u2705 \u6210\u529F ${summary.completed}\u3000\u23F3 \u5F85\u4E0B\u8F7D ${summary.pending}\u3000\u{1F504} \u4E0B\u8F7D\u4E2D ${summary.downloading}\u3000\u274C \u5931\u8D25 ${summary.failed}\u3000\u23ED \u8DF3\u8FC7 ${summary.skipped}`, + summary.cooldownUntil ? `\u23F3 FloodWait \u51B7\u5374\u5230\uFF1A${summary.cooldownUntil}` : "", + ``, + `\u63A7\u5236\uFF1A/task_pause ${summary.jobId.slice(0, 8)} \xB7 /task_resume ${summary.jobId.slice(0, 8)} \xB7 /task_cancel ${summary.jobId.slice(0, 8)}` + ].filter(Boolean); + await statusMessage.edit({ text: lines.join("\n") }).catch(() => void 0); +} async function updateScanStatusMessage(statusMessage, summary) { const lines = [ `\u{1F50E} **\u626B\u63CF\u5B8C\u6210\uFF0C\u5F00\u59CB\u4E0B\u8F7D**`, @@ -7372,7 +7750,8 @@ ${buildPathPreviewLine(state.customFolder)}` : "\u{1F4C1} \u672C\u8BA2\u9605\u4F await replyWithJobResult(queuedMsg, message, enqueueTelegramTagDownload(client, message, senderId, state.source, input, state.customFolder, { includeComments: Boolean(state.includeComments), commentsMaxPerPost: state.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST, - onScanComplete: (summary) => updateScanStatusMessage(queuedMsg, summary) + onScanComplete: (summary) => updateScanStatusMessage(queuedMsg, summary), + onProgress: (summary) => updateJobProgressMessage(queuedMsg, summary) }), "tag"); } catch (error) { await message.reply({ message: `\u274C \u6807\u7B7E\u4E0B\u8F7D\u5931\u8D25: ${error instanceof Error ? error.message : String(error)}` }); @@ -7400,7 +7779,8 @@ ${buildPathPreviewLine(state.customFolder)}` : "\u{1F4C1} \u672C\u8BA2\u9605\u4F await replyWithJobResult(queuedMsg, message, enqueueTelegramDateDownload(client, message, senderId, state.source, state.startDate, input, state.customFolder, { includeComments: Boolean(state.includeComments), commentsMaxPerPost: state.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST, - onScanComplete: (summary) => updateScanStatusMessage(queuedMsg, summary) + onScanComplete: (summary) => updateScanStatusMessage(queuedMsg, summary), + onProgress: (summary) => updateJobProgressMessage(queuedMsg, summary) }), "date"); } catch (error) { await message.reply({ message: `\u274C \u65E5\u671F\u4E0B\u8F7D\u5931\u8D25: ${error instanceof Error ? error.message : String(error)}` }); @@ -7903,6 +8283,7 @@ async function initTelegramBot() { } startPeriodicCleanup(); startTelegramSubscriptionWorker(client); + startTelegramJobRecoveryWorker(client); client.addEventHandler(async (event) => { if (!client) return; try { diff --git a/backend/src/db/schema.sql b/backend/src/db/schema.sql index 7c4e4b1..ed3b88a 100644 --- a/backend/src/db/schema.sql +++ b/backend/src/db/schema.sql @@ -116,6 +116,12 @@ CREATE TABLE IF NOT EXISTS telegram_background_jobs ( kind VARCHAR(50) NOT NULL, source TEXT NOT NULL, status VARCHAR(20) NOT NULL DEFAULT 'pending', + scan_status TEXT DEFAULT 'pending', + download_status TEXT DEFAULT 'pending', + scan_cursor JSONB DEFAULT '{}'::jsonb, + cooldown_until TIMESTAMPTZ, + paused_at TIMESTAMPTZ, + cancelled_at TIMESTAMPTZ, params JSONB DEFAULT '{}'::jsonb, total_count INT DEFAULT 0, enqueued_count INT DEFAULT 0, @@ -130,6 +136,14 @@ CREATE TABLE IF NOT EXISTS telegram_background_jobs ( CREATE INDEX IF NOT EXISTS idx_tg_background_jobs_user_created ON telegram_background_jobs(user_id, created_at DESC); CREATE INDEX IF NOT EXISTS idx_tg_background_jobs_status ON telegram_background_jobs(status); +ALTER TABLE telegram_background_jobs ADD COLUMN IF NOT EXISTS scan_status TEXT DEFAULT 'pending'; +ALTER TABLE telegram_background_jobs ADD COLUMN IF NOT EXISTS download_status TEXT DEFAULT 'pending'; +ALTER TABLE telegram_background_jobs ADD COLUMN IF NOT EXISTS scan_cursor JSONB DEFAULT '{}'::jsonb; +ALTER TABLE telegram_background_jobs ADD COLUMN IF NOT EXISTS cooldown_until TIMESTAMPTZ; +ALTER TABLE telegram_background_jobs ADD COLUMN IF NOT EXISTS paused_at TIMESTAMPTZ; +ALTER TABLE telegram_background_jobs ADD COLUMN IF NOT EXISTS cancelled_at TIMESTAMPTZ; +CREATE INDEX IF NOT EXISTS idx_tg_background_jobs_pipeline ON telegram_background_jobs(status, scan_status, download_status); +CREATE INDEX IF NOT EXISTS idx_tg_background_jobs_cooldown ON telegram_background_jobs(cooldown_until); CREATE OR REPLACE TRIGGER telegram_background_jobs_updated_at BEFORE UPDATE ON telegram_background_jobs @@ -141,16 +155,44 @@ CREATE TABLE IF NOT EXISTS telegram_download_items ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), job_id UUID REFERENCES telegram_background_jobs(id) ON DELETE CASCADE, source TEXT NOT NULL, + source_peer TEXT, + origin TEXT DEFAULT 'channel', message_id INT NOT NULL, grouped_id TEXT, + channel_post_id INT, + file_name TEXT, + mime_type TEXT, + total_size BIGINT DEFAULT 0, + folder_override TEXT, status VARCHAR(20) NOT NULL DEFAULT 'pending', + attempts INT DEFAULT 0, error TEXT, + last_error TEXT, + locked_at TIMESTAMPTZ, + completed_at TIMESTAMPTZ, created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW(), UNIQUE(job_id, message_id) ); +ALTER TABLE telegram_download_items ADD COLUMN IF NOT EXISTS source_peer TEXT; +ALTER TABLE telegram_download_items ADD COLUMN IF NOT EXISTS origin TEXT DEFAULT 'channel'; +ALTER TABLE telegram_download_items ADD COLUMN IF NOT EXISTS channel_post_id INT; +ALTER TABLE telegram_download_items ADD COLUMN IF NOT EXISTS file_name TEXT; +ALTER TABLE telegram_download_items ADD COLUMN IF NOT EXISTS mime_type TEXT; +ALTER TABLE telegram_download_items ADD COLUMN IF NOT EXISTS total_size BIGINT DEFAULT 0; +ALTER TABLE telegram_download_items ADD COLUMN IF NOT EXISTS folder_override TEXT; +ALTER TABLE telegram_download_items ADD COLUMN IF NOT EXISTS attempts INT DEFAULT 0; +ALTER TABLE telegram_download_items ADD COLUMN IF NOT EXISTS last_error TEXT; +ALTER TABLE telegram_download_items ADD COLUMN IF NOT EXISTS locked_at TIMESTAMPTZ; +ALTER TABLE telegram_download_items ADD COLUMN IF NOT EXISTS completed_at TIMESTAMPTZ; +UPDATE telegram_download_items SET source_peer = COALESCE(source_peer, source) WHERE source_peer IS NULL; +ALTER TABLE telegram_download_items DROP CONSTRAINT IF EXISTS telegram_download_items_job_id_message_id_key; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_tg_download_items_job_peer_msg + ON telegram_download_items(job_id, source_peer, message_id); CREATE INDEX IF NOT EXISTS idx_tg_download_items_job_status ON telegram_download_items(job_id, status); +CREATE INDEX IF NOT EXISTS idx_tg_download_items_recover ON telegram_download_items(status, locked_at); CREATE OR REPLACE TRIGGER telegram_download_items_updated_at BEFORE UPDATE ON telegram_download_items diff --git a/backend/src/services/telegramBot.ts b/backend/src/services/telegramBot.ts index 0941db0..6690679 100644 --- a/backend/src/services/telegramBot.ts +++ b/backend/src/services/telegramBot.ts @@ -14,6 +14,8 @@ import { enqueueTelegramDateDownload, enqueueTelegramTagDownload, listTelegramSubscriptions, + type TelegramJobProgressSummary, + startTelegramJobRecoveryWorker, startTelegramSubscriptionWorker, subscribeTelegramChannel, unsubscribeTelegramChannel, @@ -297,6 +299,26 @@ interface TelegramDownloadScanSummary { commentsMaxPerPost: number; } +async function updateJobProgressMessage(statusMessage: Api.Message, summary: TelegramJobProgressSummary): Promise { + const totalDone = summary.completed + summary.failed + summary.skipped; + const lines = [ + summary.status === 'paused' ? `⏸️ **频道下载已暂停**` : summary.status === 'cancelled' ? `🛑 **频道下载已取消**` : totalDone >= summary.totalMediaFound && summary.scanStatus === 'done' ? `✅ **频道任务完成**` : `🔎 **频道任务运行中**`, + `🆔 job: ${summary.jobId.slice(0, 8)}`, + `📍 频道：${summary.source}`, + ``, + `🔎 扫描：${summary.scanStatus}`, + `📄 频道正文：已扫 ${summary.channelMessagesScanned} 条，发现 ${summary.channelMediaFound} 个文件`, + `💬 评论区：已扫 ${summary.commentMessagesScanned} 条，发现 ${summary.commentMediaFound} 个文件`, + ``, + `⬇️ 下载：${summary.downloadStatus}`, + `✅ 成功 ${summary.completed}　⏳ 待下载 ${summary.pending}　🔄 下载中 ${summary.downloading}　❌ 失败 ${summary.failed}　⏭ 跳过 ${summary.skipped}`, + summary.cooldownUntil ? `⏳ FloodWait 冷却到：${summary.cooldownUntil}` : '', + ``, + `控制：/task_pause ${summary.jobId.slice(0, 8)} · /task_resume ${summary.jobId.slice(0, 8)} · /task_cancel ${summary.jobId.slice(0, 8)}`, + ].filter(Boolean); + await statusMessage.edit({ text: lines.join('\n') }).catch(() => undefined); +} + async function updateScanStatusMessage(statusMessage: Api.Message, summary: TelegramDownloadScanSummary): Promise { const lines = [ `🔎 **扫描完成，开始下载**`, @@ -498,6 +520,7 @@ async function handleTelegramWizardMessage(message: Api.Message, senderId: numbe includeComments: Boolean(state.includeComments), commentsMaxPerPost: state.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST, onScanComplete: summary => updateScanStatusMessage(queuedMsg as Api.Message, summary), + onProgress: summary => updateJobProgressMessage(queuedMsg as Api.Message, summary), }), 'tag'); } catch (error) { await message.reply({ message: `❌ 标签下载失败: ${error instanceof Error ? error.message : String(error)}` }); @@ -529,6 +552,7 @@ async function handleTelegramWizardMessage(message: Api.Message, senderId: numbe includeComments: Boolean(state.includeComments), commentsMaxPerPost: state.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST, onScanComplete: summary => updateScanStatusMessage(queuedMsg as Api.Message, summary), + onProgress: summary => updateJobProgressMessage(queuedMsg as Api.Message, summary), }), 'date'); } catch (error) { await message.reply({ message: `❌ 日期下载失败: ${error instanceof Error ? error.message : String(error)}` }); @@ -1087,6 +1111,7 @@ export async function initTelegramBot(): Promise { // 启动定期清理（每小时） startPeriodicCleanup(); startTelegramSubscriptionWorker(client); + startTelegramJobRecoveryWorker(client); // Handle Messages client.addEventHandler(async (event: NewMessageEvent) => { diff --git a/backend/src/services/telegramChannelJobs.ts b/backend/src/services/telegramChannelJobs.ts index 7e33cdc..9dc320a 100644 --- a/backend/src/services/telegramChannelJobs.ts +++ b/backend/src/services/telegramChannelJobs.ts @@ -8,8 +8,14 @@ import { extractFileInfo, getEstimatedFileSize } from '../utils/telegramMedia.js const SUBSCRIPTION_INTERVAL_MS = Math.max(60_000, parseInt(process.env.TELEGRAM_SUBSCRIPTION_INTERVAL_MS || '300000', 10) || 300_000); const SUBSCRIPTION_SCAN_LIMIT = Math.max(1, parseInt(process.env.TELEGRAM_SUBSCRIPTION_SCAN_LIMIT || '100', 10) || 100); +const TG_JOB_RECOVERY_DELAY_MS = Math.max(1000, parseInt(process.env.TG_JOB_RECOVERY_DELAY_MS || '10000', 10) || 10_000); +const TG_JOB_SCAN_SEGMENT_SIZE = Math.max(20, parseInt(process.env.TG_JOB_SCAN_SEGMENT_SIZE || '100', 10) || 100); +const TG_JOB_DOWNLOAD_BATCH_SIZE = Math.max(1, parseInt(process.env.TG_JOB_DOWNLOAD_BATCH_SIZE || '20', 10) || 20); +const TG_JOB_MAX_ATTEMPTS = Math.max(1, parseInt(process.env.TG_JOB_MAX_ATTEMPTS || '3', 10) || 3); export const TELEGRAM_COMMENTS_MAX_PER_POST = Math.max(1, parseInt(process.env.TELEGRAM_COMMENTS_MAX_PER_POST || '200', 10) || 200); let subscriptionTimer: NodeJS.Timeout | null = null; +let recoveryStarted = false; +let recoveryRunning = false; function parseTelegramSourceAllowlist(raw: string | undefined): string[] { return (raw || '') @@ -149,13 +155,38 @@ async function expandMessagesWithMediaGroups(userClient: TelegramClient, source: return Array.from(byId.values()).sort((a, b) => a.id - b.id); } -async function persistDownloadItems(jobId: string, source: string, messages: Api.Message[]) { - for (const message of messages) { +function sourcePeerKey(value: unknown, fallback: string): string { + if (value === undefined || value === null) return fallback; + return String(value); +} + +async function persistDownloadRefs(jobId: string, source: string, refs: TelegramDownloadMessageRef[], folderOverride?: string | null) { + for (const ref of refs) { await query( - `INSERT INTO telegram_download_items (job_id, source, message_id, grouped_id, status) - VALUES ($1, $2, $3, $4, 'pending') - ON CONFLICT (job_id, message_id) DO NOTHING`, - [jobId, source, message.id, messageGroupId(message) || null] + `INSERT INTO telegram_download_items ( + job_id, source, source_peer, origin, message_id, grouped_id, channel_post_id, + file_name, mime_type, total_size, folder_override, status, error, last_error, locked_at, completed_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, 'pending', NULL, NULL, NULL, NULL) + ON CONFLICT (job_id, source_peer, message_id) + DO UPDATE SET + file_name = COALESCE(EXCLUDED.file_name, telegram_download_items.file_name), + mime_type = COALESCE(EXCLUDED.mime_type, telegram_download_items.mime_type), + total_size = COALESCE(EXCLUDED.total_size, telegram_download_items.total_size), + folder_override = EXCLUDED.folder_override, + updated_at = NOW()`, + [ + jobId, + source, + sourcePeerKey(ref.source, source), + ref.origin || 'channel', + ref.id, + null, + ref.channelPostId || null, + ref.fileInfo?.fileName || null, + ref.fileInfo?.mimeType || null, + ref.totalSize || 0, + folderOverride || null, + ] ); } } @@ -176,6 +207,29 @@ interface TelegramCommentScanOptions { includeComments?: boolean; commentsMaxPerPost?: number; onScanComplete?: (summary: TelegramDownloadScanSummary) => Promise | void; + onProgress?: (summary: TelegramJobProgressSummary) => Promise | void; + onRefDiscovered?: (ref: TelegramDownloadMessageRef) => Promise | void; +} + +export interface TelegramJobProgressSummary { + jobId: string; + source: string; + mode: 'date' | 'tag'; + status: string; + scanStatus: string; + downloadStatus: string; + channelMessagesScanned: number; + channelMediaFound: number; + commentMessagesScanned: number; + commentMediaFound: number; + totalMediaFound: number; + completed: number; + pending: number; + downloading: number; + failed: number; + skipped: number; + currentFileName?: string; + cooldownUntil?: string | null; } interface TelegramDownloadScanResult { @@ -236,7 +290,7 @@ async function getDiscussionMediaRefs( if (seen.has(sourceKey)) continue; seen.add(sourceKey); mediaFound += 1; - refs.push({ + const ref: TelegramDownloadMessageRef = { id: comment.id, source: comment.chatId || source, origin: 'comment', @@ -244,7 +298,9 @@ async function getDiscussionMediaRefs( fileInfo, totalSize: getEstimatedFileSize(comment), message: comment, - }); + }; + refs.push(ref); + await options.onRefDiscovered?.(ref); } if (batch.length === 0 || scannedForPost >= maxPerPost) break; @@ -275,6 +331,9 @@ async function buildDownloadScanResult( const refs = messages .map(message => toChannelDownloadRef(source, message)) .filter((ref): ref is TelegramDownloadMessageRef => Boolean(ref)); + for (const ref of refs) { + await options.onRefDiscovered?.(ref); + } const commentScan = await getDiscussionMediaRefs(userClient, source, messages, options); refs.push(...commentScan.refs); return { @@ -286,12 +345,50 @@ async function buildDownloadScanResult( }; } +async function markDownloadRefsDownloading(jobId: string, refs: TelegramDownloadMessageRef[]) { + for (const ref of refs) { + const sourcePeer = sourcePeerKey(ref.source, ref.origin === 'comment' ? 'comment' : 'channel'); + await query( + `UPDATE telegram_download_items + SET status = 'downloading', locked_at = NOW(), updated_at = NOW() + WHERE job_id = $1 AND source_peer = $2 AND message_id = $3 + AND status IN ('pending', 'failed')`, + [jobId, sourcePeer, ref.id] + ); + } +} + +async function markDownloadRefStatus(jobId: string, ref: TelegramDownloadMessageRef, status: 'success' | 'failed' | 'skipped', error?: string) { + const sourcePeer = sourcePeerKey(ref.source, ref.origin === 'comment' ? 'comment' : 'channel'); + await query( + `UPDATE telegram_download_items + SET status = $3, + error = $4, + last_error = $4, + attempts = CASE WHEN $3 = 'failed' THEN attempts + 1 ELSE attempts END, + completed_at = CASE WHEN $3 IN ('success', 'skipped') THEN NOW() ELSE completed_at END, + locked_at = NULL, + updated_at = NOW() + WHERE job_id = $1 AND source_peer = $2 AND message_id = $5`, + [jobId, sourcePeer, status, error || null, ref.id] + ); +} + +async function persistDownloadMessages(jobId: string, source: string, messages: Api.Message[], folderOverride?: string | null) { + const refs = messages + .map(message => toChannelDownloadRef(source, message)) + .filter((ref): ref is TelegramDownloadMessageRef => Boolean(ref)); + await persistDownloadRefs(jobId, source, refs, folderOverride); +} + async function updateDownloadItemsStatus(jobId: string, messageIds: number[] | undefined, status: 'success' | 'failed' | 'skipped', error?: string) { const ids = Array.from(new Set((messageIds || []).filter(id => id > 0))); if (ids.length === 0) return; await query( `UPDATE telegram_download_items - SET status = $2, error = $3, updated_at = NOW() + SET status = $2, error = $3, last_error = $3, updated_at = NOW(), + completed_at = CASE WHEN $2 IN ('success', 'skipped') THEN NOW() ELSE completed_at END, + locked_at = NULL WHERE job_id = $1 AND message_id = ANY($4::int[])`, [jobId, status, error || null, ids] ); @@ -318,14 +415,19 @@ export function parseDateOnly(value: string, endOfDay = false): Date { async function createJob(userId: number, chatId: string | undefined, kind: string, source: string, params: Record) { const result = await query( - `INSERT INTO telegram_background_jobs (user_id, chat_id, kind, source, params) - VALUES ($1, $2, $3, $4, $5) + `INSERT INTO telegram_background_jobs (user_id, chat_id, kind, source, params, status, scan_status, download_status, scan_cursor) + VALUES ($1, $2, $3, $4, $5, 'queued', 'pending', 'pending', '{}'::jsonb) RETURNING id`, [userId, chatId || null, kind, source, JSON.stringify(params)] ); return result.rows[0].id as string; } +async function getJob(jobId: string) { + const result = await query(`SELECT * FROM telegram_background_jobs WHERE id = $1`, [jobId]); + return result.rows[0] || null; +} + async function updateJob(jobId: string, updates: Record) { const entries = Object.entries(updates); if (entries.length === 0) return; @@ -391,60 +493,269 @@ export async function unsubscribeTelegramChannel(userId: number, selector: strin return result.rows[0] || null; } -export async function enqueueTelegramDateDownload(botClient: TelegramClient, requestMessage: Api.Message, userId: number, sourceInput: string, startDateText: string, endDateText: string, folderOverride?: string | null, options: TelegramCommentScanOptions = {}) { + +async function getJobItemStats(jobId: string) { + const result = await query( + `SELECT status, COUNT(*)::int AS count + FROM telegram_download_items + WHERE job_id = $1 + GROUP BY status`, + [jobId] + ); + const stats: Record = { pending: 0, downloading: 0, success: 0, failed: 0, skipped: 0 }; + for (const row of result.rows) stats[row.status] = Number(row.count || 0); + return stats; +} + +async function getJobProgress(jobId: string): Promise { + const job = await getJob(jobId); + if (!job) return null; + const params = job.params || {}; + const cursor = job.scan_cursor || params.scan || {}; + const stats = await getJobItemStats(jobId); + return { + jobId, + source: job.source, + mode: job.kind === 'tag_download' ? 'tag' : 'date', + status: job.status, + scanStatus: job.scan_status || 'pending', + downloadStatus: job.download_status || 'pending', + channelMessagesScanned: Number(cursor.channelMessagesScanned || 0), + channelMediaFound: Number(cursor.channelMediaFound || 0), + commentMessagesScanned: Number(cursor.commentMessagesScanned || 0), + commentMediaFound: Number(cursor.commentMediaFound || 0), + totalMediaFound: Number(job.total_count || 0), + completed: Number(stats.success || 0), + pending: Number(stats.pending || 0), + downloading: Number(stats.downloading || 0), + failed: Number(stats.failed || 0), + skipped: Number(stats.skipped || 0), + cooldownUntil: job.cooldown_until ? new Date(job.cooldown_until).toISOString() : null, + }; +} + +async function notifyProgress(jobId: string, options: TelegramCommentScanOptions) { + const progress = await getJobProgress(jobId); + if (progress) await options.onProgress?.(progress); +} + +function isFloodWait(error: unknown): { seconds: number } | null { + const anyErr = error as any; + const text = `${anyErr?.message || ''} ${anyErr?.errorMessage || ''}`; + const seconds = Number(anyErr?.seconds || anyErr?.value || text.match(/FLOOD_WAIT_?(\d+)/i)?.[1] || 0); + if (seconds > 0 || /FLOOD|Too many requests/i.test(text)) return { seconds: Math.max(30, seconds || 60) }; + return null; +} + +async function ensureJobCanRun(jobId: string): Promise<'run' | 'paused' | 'cancelled' | 'cooldown'> { + const job = await getJob(jobId); + if (!job) return 'cancelled'; + if (job.cancelled_at || job.status === 'cancelled') return 'cancelled'; + if (job.paused_at || job.status === 'paused') return 'paused'; + if (job.cooldown_until && new Date(job.cooldown_until).getTime() > Date.now()) return 'cooldown'; + return 'run'; +} + +async function waitUntilRunnable(jobId: string, options: TelegramCommentScanOptions): Promise { + while (true) { + const state = await ensureJobCanRun(jobId); + if (state === 'run') return true; + if (state === 'cancelled') return false; + await notifyProgress(jobId, options); + await new Promise(resolve => setTimeout(resolve, state === 'cooldown' ? 5000 : 2000)); + } +} + +async function claimPendingDownloadRefs(jobId: string, limit = TG_JOB_DOWNLOAD_BATCH_SIZE): Promise { + const result = await query( + `UPDATE telegram_download_items i + SET status = 'downloading', locked_at = NOW(), updated_at = NOW() + WHERE i.id IN ( + SELECT id FROM telegram_download_items + WHERE job_id = $1 AND status = 'pending' AND attempts < $2 + ORDER BY created_at ASC + LIMIT $3 + ) + RETURNING source, source_peer, origin, message_id, channel_post_id, file_name, mime_type, total_size, folder_override`, + [jobId, TG_JOB_MAX_ATTEMPTS, limit] + ); + return result.rows + .filter(row => row.file_name && row.mime_type) + .map(row => ({ + id: Number(row.message_id), + source: row.source_peer || row.source, + origin: row.origin === 'comment' ? 'comment' : 'channel', + channelPostId: row.channel_post_id || undefined, + fileInfo: { fileName: row.file_name, mimeType: row.mime_type }, + totalSize: Number(row.total_size || 0), + })); +} + +async function downloadClaimedRefs(botClient: TelegramClient, requestMessage: Api.Message, jobId: string, source: string, refs: TelegramDownloadMessageRef[], folderOverride: string | null | undefined, options: TelegramCommentScanOptions) { + if (refs.length === 0) return { found: 0, skipped: 0, failed: 0, successful: 0 }; + await updateJob(jobId, { status: 'running', download_status: 'active', error: null }); + try { + const result = await downloadTelegramChannelRange(botClient, requestMessage, source, 0, refs.length, 'older', refs.map(ref => ref.id), folderOverride, refs, async (ref, status, error) => { + await markDownloadRefStatus(jobId, ref, status, error); + await notifyProgress(jobId, options); + }); + return result; + } catch (error) { + const flood = isFloodWait(error); + if (flood) { + const cooldownUntil = new Date(Date.now() + flood.seconds * 1000); + await updateJob(jobId, { status: 'running', cooldown_until: cooldownUntil, error: `Telegram FloodWait，冷却到 ${cooldownUntil.toISOString()}` }); + for (const ref of refs) await markDownloadRefStatus(jobId, ref, 'failed', `FloodWait ${flood.seconds}s`); + return { found: 0, skipped: 0, failed: refs.length, successful: 0 }; + } + for (const ref of refs) await markDownloadRefStatus(jobId, ref, 'failed', error instanceof Error ? error.message : String(error)); + throw error; + } +} + +async function downloadPendingForJob(botClient: TelegramClient, requestMessage: Api.Message, jobId: string, source: string, folderOverride: string | null | undefined, options: TelegramCommentScanOptions, drain = false) { + let aggregate = { found: 0, skipped: 0, failed: 0, successful: 0 }; + while (await waitUntilRunnable(jobId, options)) { + const refs = await claimPendingDownloadRefs(jobId); + if (refs.length === 0) break; + const result = await downloadClaimedRefs(botClient, requestMessage, jobId, source, refs, folderOverride, options); + aggregate = { + found: aggregate.found + (result.found || 0), + skipped: aggregate.skipped + (result.skipped || 0), + failed: aggregate.failed + (result.failed || 0), + successful: aggregate.successful + (result.successful || 0), + }; + if (!drain) break; + } + return aggregate; +} + +async function finalizeTelegramJob(jobId: string, options: TelegramCommentScanOptions) { + const stats = await getJobItemStats(jobId); + const pending = Number(stats.pending || 0) + Number(stats.downloading || 0); + const failed = Number(stats.failed || 0); + const status = pending > 0 ? 'running' : failed > 0 ? 'completed_with_errors' : 'completed'; + await updateJob(jobId, { + status, + download_status: pending > 0 ? 'active' : 'done', + enqueued_count: Number(stats.success || 0), + skipped_count: Number(stats.skipped || 0), + error: failed > 0 ? `${failed} 个文件下载失败` : null, + finished_at: pending > 0 ? null : new Date(), + }); + await notifyProgress(jobId, options); +} + +async function scanChannelSegment(userClient: TelegramClient, jobId: string, source: string, params: any, cursor: any, options: TelegramCommentScanOptions): Promise<{ messages: Api.Message[]; done: boolean; nextOffsetId: number }> { + const mode = params.mode as 'date' | 'tag'; + const offsetId = Number(cursor.offsetId || 0); + const batch = await userClient.getMessages(source as any, { + limit: TG_JOB_SCAN_SEGMENT_SIZE, + offsetId, + ...(mode === 'tag' ? { search: params.tag } : {}), + }); + if (!batch.length) return { messages: [], done: true, nextOffsetId: offsetId }; + let done = false; + let nextOffsetId = offsetId; + const matched: Api.Message[] = []; + for (const message of batch) { + nextOffsetId = message.id; + if (mode === 'date') { + const messageDate = new Date((message.date || 0) * 1000); + const startDate = new Date(params.startDateIso); + const endDate = new Date(params.endDateIso); + if (messageDate > endDate) continue; + if (messageDate < startDate) { done = true; break; } + if (messageHasMedia(message)) matched.push(message); + } else if (messageHasMedia(message) && messageMatchesHashtag(message, params.tag)) { + matched.push(message); + } + } + const expanded = await expandMessagesWithMediaGroups(userClient, source, matched); + return { messages: expanded, done: done || batch.length < TG_JOB_SCAN_SEGMENT_SIZE, nextOffsetId }; +} + +async function runSegmentedTelegramJob(botClient: TelegramClient, requestMessage: Api.Message, jobId: string, source: string, folderOverride: string | null | undefined, options: TelegramCommentScanOptions) { const userClient = requireUserClient(); + const job = await getJob(jobId); + const params = job?.params || {}; + let cursor = job?.scan_cursor || {}; + const discoveredRefKeys = new Set(); + let totals = { found: 0, skipped: 0, failed: 0, successful: 0 }; + await updateJob(jobId, { status: 'running', scan_status: 'scanning', download_status: 'active', started_at: job?.started_at || new Date(), error: null }); + + while (await waitUntilRunnable(jobId, options)) { + const current = await getJob(jobId); + cursor = current?.scan_cursor || cursor || {}; + if (current?.scan_status === 'done') break; + try { + const segment = await scanChannelSegment(userClient, jobId, source, params, cursor, options); + const onRefDiscovered = async (ref: TelegramDownloadMessageRef) => { + const key = `${sourcePeerKey(ref.source, source)}:${ref.id}`; + if (discoveredRefKeys.has(key)) return; + discoveredRefKeys.add(key); + await persistDownloadRefs(jobId, source, [ref], folderOverride); + }; + const scan = await buildDownloadScanResult(userClient, source, segment.messages, { + ...options, + tag: params.tag, + startDate: params.startDateIso ? new Date(params.startDateIso) : undefined, + endDate: params.endDateIso ? new Date(params.endDateIso) : undefined, + onRefDiscovered, + }); + cursor = { + ...cursor, + phase: segment.done ? 'done' : 'channel', + offsetId: segment.nextOffsetId, + channelMessagesScanned: Number(cursor.channelMessagesScanned || 0) + segment.messages.length, + channelMediaFound: Number(cursor.channelMediaFound || 0) + scan.channelMediaFound, + commentMessagesScanned: Number(cursor.commentMessagesScanned || 0) + scan.commentMessagesScanned, + commentMediaFound: Number(cursor.commentMediaFound || 0) + scan.commentMediaFound, + }; + const stats = await getJobItemStats(jobId); + await updateJob(jobId, { scan_cursor: JSON.stringify(cursor), total_count: Number(stats.pending || 0) + Number(stats.downloading || 0) + Number(stats.success || 0) + Number(stats.failed || 0) + Number(stats.skipped || 0), scan_status: segment.done ? 'done' : 'scanning' }); + await notifyProgress(jobId, options); + const partial = await downloadPendingForJob(botClient, requestMessage, jobId, source, folderOverride, options, false); + totals = { found: totals.found + partial.found, skipped: totals.skipped + partial.skipped, failed: totals.failed + partial.failed, successful: totals.successful + partial.successful }; + if (segment.done) break; + } catch (error) { + const flood = isFloodWait(error); + if (!flood) throw error; + const cooldownUntil = new Date(Date.now() + flood.seconds * 1000); + await updateJob(jobId, { cooldown_until: cooldownUntil, error: `Telegram FloodWait，冷却到 ${cooldownUntil.toISOString()}` }); + } + } + + const runnable = await waitUntilRunnable(jobId, options); + if (!runnable) { + await updateJob(jobId, { status: 'cancelled', scan_status: 'cancelled', download_status: 'cancelled', finished_at: new Date() }); + return { jobId, ...totals, requested: 0, commentMessagesScanned: Number(cursor.commentMessagesScanned || 0), commentMediaFound: Number(cursor.commentMediaFound || 0) }; + } + await updateJob(jobId, { scan_status: 'done' }); + const drained = await downloadPendingForJob(botClient, requestMessage, jobId, source, folderOverride, options, true); + totals = { found: totals.found + drained.found, skipped: totals.skipped + drained.skipped, failed: totals.failed + drained.failed, successful: totals.successful + drained.successful }; + await finalizeTelegramJob(jobId, options); + return { jobId, ...totals, requested: totals.found + totals.skipped, commentMessagesScanned: Number(cursor.commentMessagesScanned || 0), commentMediaFound: Number(cursor.commentMediaFound || 0) }; +} + +export async function enqueueTelegramDateDownload(botClient: TelegramClient, requestMessage: Api.Message, userId: number, sourceInput: string, startDateText: string, endDateText: string, folderOverride?: string | null, options: TelegramCommentScanOptions = {}) { const source = normalizeSource(sourceInput); await assertTelegramSourceAllowed(source); const startDate = parseDateOnly(startDateText); const endDate = parseDateOnly(endDateText, true); if (startDate > endDate) throw new Error('开始日期不能晚于结束日期'); - const baseMessages = await getMessagesByDateRange(userClient, source, startDate, endDate); - const messages = await expandMessagesWithMediaGroups(userClient, source, baseMessages); - const scan = await buildDownloadScanResult(userClient, source, messages, { ...options, startDate, endDate }); - const commentLimit = options.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST; - await options.onScanComplete?.({ - source, - mode: 'date', - channelMessagesScanned: messages.length, - channelMediaFound: scan.channelMediaFound, - commentMessagesScanned: scan.commentMessagesScanned, - commentMediaFound: scan.commentMediaFound, - totalMediaFound: scan.refs.length, - commentsEnabled: Boolean(options.includeComments), - commentsMaxPerPost: commentLimit, - }); - const messageIds = scan.refs.map(ref => ref.id); const jobId = await createJob(userId, requestMessage.chatId?.toString(), 'date_range', source, { + mode: 'date', startDate: startDateText, endDate: endDateText, - messageIds, + startDateIso: startDate.toISOString(), + endDateIso: endDate.toISOString(), folderOverride: folderOverride || null, includeComments: Boolean(options.includeComments), commentsMaxPerPost: options.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST, - commentMessagesScanned: scan.commentMessagesScanned, - commentMediaFound: scan.commentMediaFound, }); - await persistDownloadItems(jobId, source, messages); - await updateJob(jobId, { status: 'running', started_at: new Date(), total_count: scan.refs.length }); - - try { - const result = await downloadTelegramChannelRange(botClient, requestMessage, source, 0, scan.refs.length, 'older', messageIds, folderOverride, scan.refs); - await updateDownloadItemsStatus(jobId, result.successfulMessageIds, 'success'); - await updateDownloadItemsStatus(jobId, result.failedMessageIds, 'failed', '下载失败'); - await updateDownloadItemsStatus(jobId, result.skippedMessageIds, 'skipped'); - await updateJob(jobId, { - status: result.failed > 0 ? 'completed_with_errors' : 'completed', - enqueued_count: result.found, - skipped_count: result.skipped, - error: result.failed > 0 ? `${result.failed} 个文件下载失败` : null, - finished_at: new Date(), - }); - return { jobId, commentMessagesScanned: scan.commentMessagesScanned, commentMediaFound: scan.commentMediaFound, ...result }; - } catch (error) { - await updateJob(jobId, { status: 'failed', error: error instanceof Error ? error.message : String(error), finished_at: new Date() }); - throw error; - } + return runSegmentedTelegramJob(botClient, requestMessage, jobId, source, folderOverride, options); } async function getMessagesByHashtag(userClient: TelegramClient, source: string, tag: string, maxScan = 10000): Promise { @@ -472,61 +783,24 @@ async function getMessagesByHashtag(userClient: TelegramClient, source: string, } export async function enqueueTelegramTagDownload(botClient: TelegramClient, requestMessage: Api.Message, userId: number, sourceInput: string, tagInput: string, folderOverride?: string | null, options: TelegramCommentScanOptions = {}) { - const userClient = requireUserClient(); const source = normalizeSource(sourceInput); await assertTelegramSourceAllowed(source); const tag = normalizeHashtag(tagInput); - const baseMessages = await getMessagesByHashtag(userClient, source, tag); - const messages = await expandMessagesWithMediaGroups(userClient, source, baseMessages); - const scan = await buildDownloadScanResult(userClient, source, messages, { ...options, tag }); - const commentLimit = options.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST; - await options.onScanComplete?.({ - source, - mode: 'tag', - channelMessagesScanned: messages.length, - channelMediaFound: scan.channelMediaFound, - commentMessagesScanned: scan.commentMessagesScanned, - commentMediaFound: scan.commentMediaFound, - totalMediaFound: scan.refs.length, - commentsEnabled: Boolean(options.includeComments), - commentsMaxPerPost: commentLimit, - }); - const messageIds = scan.refs.map(ref => ref.id); const jobId = await createJob(userId, requestMessage.chatId?.toString(), 'tag_download', source, { + mode: 'tag', tag, - messageIds, folderOverride: folderOverride || null, includeComments: Boolean(options.includeComments), commentsMaxPerPost: options.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST, - commentMessagesScanned: scan.commentMessagesScanned, - commentMediaFound: scan.commentMediaFound, }); - await persistDownloadItems(jobId, source, messages); - await updateJob(jobId, { status: 'running', started_at: new Date(), total_count: scan.refs.length }); - - try { - const result = await downloadTelegramChannelRange(botClient, requestMessage, source, 0, scan.refs.length, 'older', messageIds, folderOverride, scan.refs); - await updateDownloadItemsStatus(jobId, result.successfulMessageIds, 'success'); - await updateDownloadItemsStatus(jobId, result.failedMessageIds, 'failed', '下载失败'); - await updateDownloadItemsStatus(jobId, result.skippedMessageIds, 'skipped'); - await updateJob(jobId, { - status: result.failed > 0 ? 'completed_with_errors' : 'completed', - enqueued_count: result.found, - skipped_count: result.skipped, - error: result.failed > 0 ? `${result.failed} 个文件下载失败` : null, - finished_at: new Date(), - }); - return { jobId, tag, commentMessagesScanned: scan.commentMessagesScanned, commentMediaFound: scan.commentMediaFound, ...result }; - } catch (error) { - await updateJob(jobId, { status: 'failed', error: error instanceof Error ? error.message : String(error), finished_at: new Date() }); - throw error; - } + const result = await runSegmentedTelegramJob(botClient, requestMessage, jobId, source, folderOverride, options); + return { ...result, tag }; } export async function listTelegramBackgroundJobs(userId: number, limit = 10) { const result = await query( - `SELECT id, kind, source, status, total_count, enqueued_count, skipped_count, duplicate_count, error, created_at, updated_at + `SELECT id, kind, source, status, scan_status, download_status, scan_cursor, cooldown_until, paused_at, cancelled_at, total_count, enqueued_count, skipped_count, duplicate_count, error, created_at, updated_at FROM telegram_background_jobs WHERE user_id = $1 ORDER BY created_at DESC @@ -536,6 +810,61 @@ export async function listTelegramBackgroundJobs(userId: number, limit = 10) { return result.rows; } + +export async function pauseTelegramBackgroundJob(userId: number, selector: string) { + const result = await query( + `UPDATE telegram_background_jobs + SET status = 'paused', paused_at = NOW(), updated_at = NOW() + WHERE user_id = $1 AND id::text LIKE $2 AND status NOT IN ('completed', 'cancelled') + RETURNING id, source, status`, + [userId, `${selector}%`] + ); + return result.rows[0] || null; +} + +export async function resumeTelegramBackgroundJob(userId: number, selector: string) { + const result = await query( + `UPDATE telegram_background_jobs + SET status = 'running', paused_at = NULL, updated_at = NOW() + WHERE user_id = $1 AND id::text LIKE $2 AND status = 'paused' + RETURNING id, source, status`, + [userId, `${selector}%`] + ); + return result.rows[0] || null; +} + +export async function cancelTelegramBackgroundJob(userId: number, selector: string) { + const result = await query( + `UPDATE telegram_background_jobs + SET status = 'cancelled', scan_status = 'cancelled', download_status = 'cancelled', cancelled_at = NOW(), finished_at = NOW(), updated_at = NOW() + WHERE user_id = $1 AND id::text LIKE $2 AND status NOT IN ('completed', 'cancelled') + RETURNING id, source, status`, + [userId, `${selector}%`] + ); + if (result.rows[0]) { + await query(`UPDATE telegram_download_items SET status = 'skipped', locked_at = NULL, updated_at = NOW() WHERE job_id = $1 AND status IN ('pending', 'downloading')`, [result.rows[0].id]); + } + return result.rows[0] || null; +} + +export async function retryTelegramBackgroundJob(userId: number, selector: string) { + const result = await query( + `SELECT id FROM telegram_background_jobs WHERE user_id = $1 AND id::text LIKE $2 LIMIT 1`, + [userId, `${selector}%`] + ); + const row = result.rows[0]; + if (!row) return null; + const retry = await query( + `UPDATE telegram_download_items + SET status = 'pending', locked_at = NULL, last_error = NULL, error = NULL, updated_at = NOW() + WHERE job_id = $1 AND status = 'failed' + RETURNING id`, + [row.id] + ); + await updateJob(row.id, { status: 'running', download_status: 'active', error: null, finished_at: null }); + return { id: row.id, retried: retry.rowCount || 0 }; +} + async function runSubscriptionScan(botClient: TelegramClient) { const userClient = getTelegramUserClient(); if (!userClient || !isTelegramUserClientReady()) return; @@ -558,15 +887,16 @@ async function runSubscriptionScan(botClient: TelegramClient) { const ids = Array.from({ length: count }, (_, index) => lastMessageId + index + 1); const jobId = await createJob(Number(row.user_id), row.chat_id?.toString(), 'subscription_sync', row.source, { fromId: lastMessageId + 1, toId: latestMessageId }); const candidateMessages = await expandMessagesWithMediaGroups(userClient, row.source, (await userClient.getMessages(row.source as any, { ids })).filter(Boolean) as Api.Message[]); - await persistDownloadItems(jobId, row.source, candidateMessages); + await persistDownloadMessages(jobId, row.source, candidateMessages, row.folder_override || null); await updateJob(jobId, { status: 'running', started_at: new Date(), total_count: ids.length }); const targetChat = row.chat_id || row.user_id; const requestMessage = ({ chatId: targetChat, id: latestMessageId } as unknown) as Api.Message; - const downloadResult = await downloadTelegramChannelRange(botClient, requestMessage, row.source, 0, ids.length, 'newer', ids, row.folder_override || null); - await updateDownloadItemsStatus(jobId, downloadResult.successfulMessageIds, 'success'); - await updateDownloadItemsStatus(jobId, downloadResult.failedMessageIds, 'failed', '下载失败'); - await updateDownloadItemsStatus(jobId, downloadResult.skippedMessageIds, 'skipped'); + const subscriptionRefs = candidateMessages + .map(message => toChannelDownloadRef(row.source, message)) + .filter((ref): ref is TelegramDownloadMessageRef => Boolean(ref)); + await markDownloadRefsDownloading(jobId, subscriptionRefs); + const downloadResult = await downloadTelegramChannelRange(botClient, requestMessage, row.source, 0, ids.length, 'newer', ids, row.folder_override || null, subscriptionRefs, (ref, status, error) => markDownloadRefStatus(jobId, ref, status, error)); await updateJob(jobId, { status: downloadResult.failed > 0 ? 'completed_with_errors' : 'completed', enqueued_count: downloadResult.found, @@ -586,6 +916,104 @@ async function runSubscriptionScan(botClient: TelegramClient) { } } +async function recoverTelegramJob(botClient: TelegramClient, job: any): Promise { + const itemResult = await query( + `SELECT id, source, source_peer, origin, message_id, channel_post_id, file_name, mime_type, total_size, folder_override + FROM telegram_download_items + WHERE job_id = $1 AND status = 'pending' + ORDER BY created_at ASC`, + [job.id] + ); + if (itemResult.rows.length === 0) return; + + const refs: TelegramDownloadMessageRef[] = itemResult.rows + .filter(row => row.file_name && row.mime_type) + .map(row => ({ + id: Number(row.message_id), + source: row.source_peer || row.source, + origin: row.origin === 'comment' ? 'comment' : 'channel', + channelPostId: row.channel_post_id || undefined, + fileInfo: { fileName: row.file_name, mimeType: row.mime_type }, + totalSize: Number(row.total_size || 0), + })); + + if (refs.length === 0) return; + + const targetChat = job.chat_id || job.user_id; + const requestMessage = ({ chatId: targetChat, id: 0 } as unknown) as Api.Message; + console.log(`♻️ 恢复 Telegram 下载任务 ${String(job.id).slice(0, 8)}，待处理 ${refs.length} 个文件`); + await updateJob(job.id, { status: 'running', started_at: job.started_at || new Date(), error: null }); + + try { + const result = await downloadTelegramChannelRange( + botClient, + requestMessage, + job.source, + 0, + refs.length, + 'older', + refs.map(ref => ref.id), + itemResult.rows[0]?.folder_override || null, + refs, + (ref, status, error) => markDownloadRefStatus(job.id, ref, status, error), + ); + await updateJob(job.id, { + status: result.failed > 0 ? 'completed_with_errors' : 'completed', + enqueued_count: result.found, + skipped_count: result.skipped, + error: result.failed > 0 ? `${result.failed} 个文件下载失败` : null, + finished_at: new Date(), + }); + await botClient.sendMessage(targetChat, { + message: `♻️ 已恢复并完成任务 ${String(job.id).slice(0, 8)}：成功 ${result.successful}，跳过 ${result.skipped}，失败 ${result.failed}`, + }).catch(() => undefined); + } catch (error) { + await updateJob(job.id, { status: 'failed', error: error instanceof Error ? error.message : String(error), finished_at: new Date() }); + throw error; + } +} + +export async function recoverInterruptedTelegramJobs(botClient: TelegramClient): Promise { + if (recoveryRunning) return; + recoveryRunning = true; + try { + await query( + `UPDATE telegram_download_items + SET status = 'pending', locked_at = NULL, updated_at = NOW() + WHERE status = 'downloading' + AND (locked_at IS NULL OR locked_at < NOW() - INTERVAL '30 minutes')` + ); + const jobs = await query( + `SELECT DISTINCT j.* + FROM telegram_background_jobs j + JOIN telegram_download_items i ON i.job_id = j.id + WHERE j.kind IN ('date_range', 'tag_download') + AND j.status IN ('pending', 'running', 'failed', 'completed_with_errors') + AND i.status = 'pending' + ORDER BY j.created_at ASC + LIMIT 5` + ); + for (const job of jobs.rows) { + if (job.scan_status !== 'done' && (job.kind === 'date_range' || job.kind === 'tag_download') && job.params?.mode) { + const targetChat = job.chat_id || job.user_id; + const requestMessage = ({ chatId: targetChat, id: 0 } as unknown) as Api.Message; + await runSegmentedTelegramJob(botClient, requestMessage, job.id, job.source, job.params?.folderOverride || null, {}).catch(error => console.error('♻️ Telegram 分段任务恢复失败:', error)); + } else { + await recoverTelegramJob(botClient, job).catch(error => console.error('♻️ Telegram 任务恢复失败:', error)); + } + } + } finally { + recoveryRunning = false; + } +} + +export function startTelegramJobRecoveryWorker(botClient: TelegramClient) { + if (recoveryStarted) return; + recoveryStarted = true; + setTimeout(() => recoverInterruptedTelegramJobs(botClient).catch(error => console.error('♻️ Telegram 任务恢复扫描失败:', error)), TG_JOB_RECOVERY_DELAY_MS); + setInterval(() => recoverInterruptedTelegramJobs(botClient).catch(error => console.error('♻️ Telegram 任务恢复扫描失败:', error)), SUBSCRIPTION_INTERVAL_MS); +} + export function startTelegramSubscriptionWorker(botClient: TelegramClient) { if (subscriptionTimer) return; subscriptionTimer = setInterval(() => { diff --git a/backend/src/services/telegramCommands.ts b/backend/src/services/telegramCommands.ts index e08845a..fe1a659 100644 --- a/backend/src/services/telegramCommands.ts +++ b/backend/src/services/telegramCommands.ts @@ -17,7 +17,7 @@ import { import { authenticatedUsers, passwordInputState, isAuthenticatedAsync } from './telegramState.js'; import { forceStopDownloadTasks, getDownloadQueueStats, getTaskStatus, pauseDownloadTasks, resumeDownloadTasks, cancelDownloadTask, retryFailedDownloadTasks, getFileDownloadConcurrency, setFileDownloadConcurrency } from './telegramUpload.js'; import { storageManager } from './storage.js'; -import { listTelegramBackgroundJobs } from './telegramChannelJobs.js'; +import { cancelTelegramBackgroundJob, listTelegramBackgroundJobs, pauseTelegramBackgroundJob, resumeTelegramBackgroundJob, retryTelegramBackgroundJob } from './telegramChannelJobs.js'; import { getSetting, setSetting } from '../utils/settings.js'; import { DuplicateMode, getDuplicateMode } from '../utils/duplicatePolicy.js'; import { startPeriodicCleanup, stopPeriodicCleanup } from './orphanCleanup.js'; @@ -627,23 +627,56 @@ export async function handleStopTasks(message: Api.Message): Promise { export async function handlePauseTasks(message: Api.Message, args: string[] = []): Promise { const taskId = args[0]; + const senderId = message.senderId?.toJSNumber(); + if (taskId && senderId) { + const job = await pauseTelegramBackgroundJob(senderId, taskId); + if (job) { + await message.reply({ message: `⏸️ 已暂停频道任务 ${String(job.id).slice(0, 8)}\n来源：${job.source}` }); + return; + } + } const result = pauseDownloadTasks(taskId); await message.reply({ message: `⏸️ 已暂停全局下载队列${taskId ? `\n任务卡: ${taskId}` : ''}\n\n进行中: ${result.active}\n等待中: ${result.pending}\n\n当前正在下载的文件会继续完成，新的等待任务暂不开始。` }); } export async function handleResumeTasks(message: Api.Message, args: string[] = []): Promise { const taskId = args[0]; + const senderId = message.senderId?.toJSNumber(); + if (taskId && senderId) { + const job = await resumeTelegramBackgroundJob(senderId, taskId); + if (job) { + await message.reply({ message: `▶️ 已继续频道任务 ${String(job.id).slice(0, 8)}\n来源：${job.source}` }); + return; + } + } const result = resumeDownloadTasks(taskId); await message.reply({ message: `▶️ 已继续全局下载队列${taskId ? `\n任务卡: ${taskId}` : ''}\n\n进行中: ${result.active}\n等待中: ${result.pending}` }); } export async function handleCancelTask(message: Api.Message, args: string[]): Promise { const selector = args.join(' ').trim() || 'all'; + const senderId = message.senderId?.toJSNumber(); + if (selector !== 'all' && senderId) { + const job = await cancelTelegramBackgroundJob(senderId, selector); + if (job) { + await message.reply({ message: `🛑 已取消频道任务 ${String(job.id).slice(0, 8)}\n来源：${job.source}` }); + return; + } + } const result = cancelDownloadTask(selector); await message.reply({ message: result.total > 0 ? `🛑 已取消匹配任务\n\n匹配: ${selector}\n处理中: ${result.active}\n等待中: ${result.pending}` : `📮 没有找到匹配的任务，未清空全局队列：${selector}` }); } export async function handleRetryFailedTasks(message: Api.Message, args: string[]): Promise { + const senderId = message.senderId?.toJSNumber(); + const jobSelector = args.find(arg => /^[0-9a-f-]{4,}$/i.test(arg)); + if (senderId && jobSelector) { + const jobRetry = await retryTelegramBackgroundJob(senderId, jobSelector); + if (jobRetry) { + await message.reply({ message: jobRetry.retried > 0 ? `🔄 已重新加入频道任务失败项 ${jobRetry.retried} 个\n任务: ${String(jobRetry.id).slice(0, 8)}` : '📮 该频道任务没有可重试失败项' }); + return; + } + } const taskId = args.find(arg => /^t[a-z0-9]+/i.test(arg)); const numericArg = args.find(arg => /^\d+$/.test(arg)); const limit = Math.max(1, Math.min(50, parseInt(numericArg || '10', 10) || 10)); diff --git a/backend/src/services/telegramUpload.ts b/backend/src/services/telegramUpload.ts index f217000..f22daf3 100644 --- a/backend/src/services/telegramUpload.ts +++ b/backend/src/services/telegramUpload.ts @@ -1817,6 +1817,7 @@ export async function downloadTelegramChannelRange( explicitIds?: number[], folderOverride?: string | null, explicitRefs?: TelegramDownloadMessageRef[], + onItemSettled?: (item: TelegramDownloadMessageRef, status: 'success' | 'failed' | 'skipped', error?: string) => Promise | void, ): Promise<{ requested: number; found: number; skipped: number; failed: number; successful: number; successfulMessageIds: number[]; failedMessageIds: number[]; skippedMessageIds: number[]; firstId: number; lastId: number }> { const userClient = getTelegramUserClient(); if (!userClient || !isTelegramUserClientReady()) { @@ -1860,6 +1861,7 @@ export async function downloadTelegramChannelRange( if (!fileInfo) { skipped += 1; skippedMessageIds.push(ref.id); + await onItemSettled?.(ref, 'skipped'); continue; } downloadableRefs.push({ @@ -2003,6 +2005,7 @@ export async function downloadTelegramChannelRange( completed += 1; failedMessageIds.push(item.id); await refreshSegmentStatus(false, fileName); + await onItemSettled?.(item, 'failed', '消息不存在或无法重新读取'); return; } const uploadItem: FileUploadItem = { @@ -2033,14 +2036,17 @@ export async function downloadTelegramChannelRange( if (uploadItem.status === 'success') { successful += 1; successfulMessageIds.push(item.id); + await onItemSettled?.(item, 'success'); } else { failed += 1; failedMessageIds.push(item.id); + await onItemSettled?.(item, 'failed', uploadItem.error || '下载失败'); } } catch (err) { console.error(`🤖 频道分段下载任务异常: ${fileName}`, err); failed += 1; failedMessageIds.push(item.id); + await onItemSettled?.(item, 'failed', err instanceof Error ? err.message : String(err)); } finally { completed += 1; found += 1;