diff --git a/backend/dist/index.js b/backend/dist/index.js index bc5d1e8..6c9f859 100644 --- a/backend/dist/index.js +++ b/backend/dist/index.js @@ -2291,46 +2291,35 @@ function buildStorageReport(data) { ` \u{1F504} \u5904\u7406\u4E2D ${data.queueActive}\u3000\u23F3 \u7B49\u5F85\u4E2D ${data.queuePending}` ].join("\n"); } -function buildTasksReport(active, pending, history) { +function buildTasksReport(active, pending, _history = []) { const lines = [ - `\u{1F4CB} **\u4EFB\u52A1\u961F\u5217\u72B6\u6001**`, - `\u{1F504} ${active.length} \u8FDB\u884C\u4E2D\u3000\u23F3 ${pending.length} \u7B49\u5F85\u4E2D`, + `\u{1F4CB} **\u5B9E\u65F6\u4E0B\u8F7D\u961F\u5217**`, + `\u{1F504} ${active.length} \u6B63\u5728\u4E0B\u8F7D\u3000\u23F3 ${pending.length} \u7B49\u5F85\u5F00\u59CB`, LINE ]; if (active.length > 0) { lines.push(""); - lines.push(`**\u{1F504} \u6B63\u5728\u5904\u7406**`); + lines.push(`**\u{1F504} \u6B63\u5728\u4E0B\u8F7D**`); active.forEach((task) => { lines.push(` \u25B8 ${task.fileName}`); if (task.totalSize && task.downloadedSize) { const bar = generateProgressBar(task.downloadedSize, task.totalSize, 10); lines.push(` ${bar} (${formatBytes(task.downloadedSize)}/${formatBytes(task.totalSize)})`); } else { - lines.push(` \u23F3 \u4E0B\u8F7D\u4E2D...`); + lines.push(` \u4F20\u8F93\u4E2D\uFF0C\u8BF7\u7A0D\u5019...`); } }); } if (pending.length > 0) { lines.push(""); - lines.push(`**\u23F3 \u7B49\u5F85\u961F\u5217** (\u524D 5 \u4E2A)`); + lines.push(`**\u23F3 \u7B49\u5F85\u5F00\u59CB** (\u524D 5 \u4E2A)`); pending.slice(0, 5).forEach((task, i) => { lines.push(` ${i + 1}. ${task.fileName}`); }); if (pending.length > 5) { - lines.push(` ... \u8FD8\u6709 ${pending.length - 5} \u4E2A\u4EFB\u52A1`); + lines.push(` ... \u8FD8\u6709 ${pending.length - 5} \u4E2A\u7B49\u5F85\u4EFB\u52A1`); } } - if (history.length > 0) { - lines.push(""); - lines.push(`**\u{1F552} \u6700\u8FD1\u5B8C\u6210** (\u524D 5 \u4E2A)`); - history.slice(0, 5).forEach((task) => { - const icon = task.status === "success" ? "\u2705" : task.status === "cancelled" ? "\u{1F6D1}" : "\u274C"; - lines.push(` ${icon} ${task.fileName}`); - if ((task.status === "failed" || task.status === "cancelled") && task.error) { - lines.push(` \u539F\u56E0: ${task.error}`); - } - }); - } return lines.join("\n"); } function buildUploadSuccess(fileName, size, fileType, providerName, folder) { @@ -5539,11 +5528,11 @@ 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 = $3, + SET status = $3::varchar, 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, + attempts = CASE WHEN $3::text = 'failed' THEN attempts + 1 ELSE attempts END, + completed_at = CASE WHEN $3::text 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`, @@ -5590,6 +5579,54 @@ async function updateJob(jobId, updates) { const setSql = entries.map(([key], index) => `${key} = $${index + 2}`).join(", "); await query(`UPDATE telegram_background_jobs SET ${setSql}, updated_at = NOW() WHERE id = $1`, [jobId, ...entries.map(([, value]) => value)]); } +async function hydratePendingDownloadRefs(userClient2, jobId) { + const result = await query( + `SELECT id, source_peer, message_id + FROM telegram_download_items + WHERE job_id = $1 + AND status = 'pending' + AND (file_name IS NULL OR mime_type IS NULL) + ORDER BY created_at ASC + LIMIT 100`, + [jobId] + ); + let hydrated = 0; + for (const row of result.rows) { + try { + const messages = await userClient2.getMessages(row.source_peer, { ids: [Number(row.message_id)] }); + const message = messages?.[0]; + if (!message) { + await query( + `UPDATE telegram_download_items + SET status = 'failed', error = $2, last_error = $2, attempts = attempts + 1, updated_at = NOW() + WHERE id = $1`, + [row.id, "\u6D88\u606F\u4E0D\u5B58\u5728\uFF0C\u65E0\u6CD5\u8865\u5168\u6587\u4EF6\u5143\u6570\u636E"] + ); + continue; + } + const fileInfo = extractFileInfo(message); + if (!fileInfo) { + await query( + `UPDATE telegram_download_items + SET status = 'skipped', error = $2, last_error = $2, completed_at = NOW(), updated_at = NOW() + WHERE id = $1`, + [row.id, "\u6D88\u606F\u4E0D\u5305\u542B\u53EF\u4E0B\u8F7D\u5A92\u4F53\uFF0C\u65E0\u6CD5\u8865\u5168\u6587\u4EF6\u5143\u6570\u636E"] + ); + continue; + } + await query( + `UPDATE telegram_download_items + SET file_name = $2, mime_type = $3, total_size = $4, updated_at = NOW() + WHERE id = $1`, + [row.id, fileInfo.fileName, fileInfo.mimeType, getEstimatedFileSize(message)] + ); + hydrated += 1; + } catch (error) { + console.warn("\u267B\uFE0F \u8865\u5168 Telegram \u4E0B\u8F7D\u6761\u76EE\u5143\u6570\u636E\u5931\u8D25:", error); + } + } + return hydrated; +} async function subscribeTelegramChannel(userId, chatId, sourceInput, folderOverride) { const userClient2 = requireUserClient(); const source = normalizeSource(sourceInput); @@ -5713,7 +5750,11 @@ async function claimPendingDownloadRefs(jobId, limit = TG_JOB_DOWNLOAD_BATCH_SIZ 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 + WHERE job_id = $1 + AND status = 'pending' + AND attempts < $2 + AND file_name IS NOT NULL + AND mime_type IS NOT NULL ORDER BY created_at ASC LIMIT $3 ) @@ -5752,7 +5793,9 @@ async function downloadClaimedRefs(botClient, requestMessage, jobId, source, ref } async function downloadPendingForJob(botClient, requestMessage, jobId, source, folderOverride, options, drain = false) { let aggregate = { found: 0, skipped: 0, failed: 0, successful: 0 }; + const userClient2 = getTelegramUserClient(); while (await waitUntilRunnable(jobId, options)) { + if (userClient2) await hydratePendingDownloadRefs(userClient2, jobId); const refs = await claimPendingDownloadRefs(jobId); if (refs.length === 0) break; const result = await downloadClaimedRefs(botClient, requestMessage, jobId, source, refs, folderOverride, options); @@ -5904,12 +5947,65 @@ async function enqueueTelegramTagDownload(botClient, requestMessage, userId, sou const result = await runSegmentedTelegramJob(botClient, requestMessage, jobId, source, folderOverride, options); return { ...result, tag }; } -async function listTelegramBackgroundJobs(userId, limit = 10) { +async function listTelegramActiveTaskQueues(userId, limit = 10) { const result = await query( - `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 + `WITH item_stats AS ( + SELECT + job_id, + COUNT(*)::int AS item_count, + COUNT(*) FILTER (WHERE status = 'pending')::int AS pending_count, + COUNT(*) FILTER (WHERE status = 'downloading')::int AS downloading_count, + COUNT(*) FILTER (WHERE status = 'success')::int AS success_count, + COUNT(*) FILTER (WHERE status = 'failed')::int AS failed_count, + COUNT(*) FILTER (WHERE status = 'skipped')::int AS skipped_count_items, + COUNT(*) FILTER (WHERE status = 'pending' AND (file_name IS NULL OR mime_type IS NULL))::int AS missing_metadata_count, + MAX(updated_at) FILTER (WHERE status IN ('pending', 'downloading')) AS queue_updated_at + FROM telegram_download_items + GROUP BY job_id + ) + SELECT + j.id, j.kind, j.source, j.status, j.scan_status, j.download_status, + j.scan_cursor, j.cooldown_until, j.paused_at, j.cancelled_at, + j.total_count, j.enqueued_count, j.skipped_count, j.duplicate_count, + j.error, j.started_at, j.finished_at, j.created_at, j.updated_at, + COALESCE(s.item_count, 0)::int AS item_count, + COALESCE(s.pending_count, 0)::int AS pending_count, + COALESCE(s.downloading_count, 0)::int AS downloading_count, + COALESCE(s.success_count, 0)::int AS success_count, + COALESCE(s.failed_count, 0)::int AS failed_count, + COALESCE(s.skipped_count_items, 0)::int AS skipped_count_items, + COALESCE(s.missing_metadata_count, 0)::int AS missing_metadata_count, + s.queue_updated_at, + ( + j.status = 'running' + AND ( + COALESCE(s.downloading_count, 0) > 0 + OR j.scan_status = 'scanning' + OR (j.cooldown_until IS NOT NULL AND j.cooldown_until > NOW()) + ) + ) AS is_actively_running + FROM telegram_background_jobs j + LEFT JOIN item_stats s ON s.job_id = j.id + WHERE j.user_id = $1 + AND j.cancelled_at IS NULL + AND j.finished_at IS NULL + AND ( + ( + j.status = 'running' + AND ( + COALESCE(s.downloading_count, 0) > 0 + OR j.scan_status = 'scanning' + OR (j.cooldown_until IS NOT NULL AND j.cooldown_until > NOW()) + ) + ) + OR ( + j.status = 'paused' + AND (COALESCE(s.pending_count, 0) > 0 OR COALESCE(s.downloading_count, 0) > 0 OR j.scan_status = 'scanning') + ) + ) + ORDER BY + CASE WHEN j.status = 'paused' THEN 1 ELSE 0 END, + COALESCE(s.queue_updated_at, j.updated_at) DESC LIMIT $2`, [userId, limit] ); @@ -5919,7 +6015,7 @@ 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') + WHERE user_id = $1 AND id::text LIKE $2 AND finished_at IS NULL AND status NOT IN ('completed', 'completed_with_errors', 'cancelled') RETURNING id, source, status`, [userId, `${selector}%`] ); @@ -5928,8 +6024,8 @@ async function pauseTelegramBackgroundJob(userId, selector) { 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' + SET status = 'running', paused_at = NULL, finished_at = NULL, error = NULL, download_status = 'active', updated_at = NOW() + WHERE user_id = $1 AND id::text LIKE $2 AND cancelled_at IS NULL AND status = 'paused' RETURNING id, source, status`, [userId, `${selector}%`] ); @@ -5939,7 +6035,7 @@ 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') + WHERE user_id = $1 AND id::text LIKE $2 AND finished_at IS NULL AND status NOT IN ('completed', 'completed_with_errors', 'cancelled') RETURNING id, source, status`, [userId, `${selector}%`] ); @@ -5948,6 +6044,27 @@ async function cancelTelegramBackgroundJob(userId, selector) { } return result.rows[0] || null; } +async function cancelAllTelegramBackgroundJobs(userId) { + 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 status NOT IN ('completed', 'completed_with_errors', 'cancelled') + RETURNING id, source, status`, + [userId] + ); + const ids = result.rows.map((row) => row.id); + if (ids.length > 0) { + await query( + `UPDATE telegram_download_items + SET status = 'skipped', locked_at = NULL, updated_at = NOW() + WHERE job_id = ANY($1::uuid[]) + AND status IN ('pending', 'downloading')`, + [ids] + ); + } + return result.rows; +} 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`, @@ -6018,6 +6135,16 @@ async function recoverTelegramJob(botClient, job) { [job.id] ); if (itemResult.rows.length === 0) return; + const missingMetadata = itemResult.rows.filter((row) => !row.file_name || !row.mime_type).length; + if (missingMetadata > 0) { + const userClient2 = getTelegramUserClient(); + if (userClient2) { + await hydratePendingDownloadRefs(userClient2, job.id); + return recoverTelegramJob(botClient, job); + } + await updateJob(job.id, { status: "failed", error: `${missingMetadata} \u4E2A\u5F85\u4E0B\u8F7D\u6761\u76EE\u7F3A\u5C11\u6587\u4EF6\u5143\u6570\u636E\uFF0C\u65E0\u6CD5\u6062\u590D`, finished_at: /* @__PURE__ */ new Date() }); + 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, @@ -6074,6 +6201,8 @@ async function recoverInterruptedTelegramJobs(botClient) { 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.finished_at IS NULL + AND j.cancelled_at IS NULL AND j.status IN ('pending', 'running', 'failed', 'completed_with_errors') AND i.status = 'pending' ORDER BY j.created_at ASC @@ -6907,32 +7036,113 @@ async function handleDeleteConfirmCallback(client2, update, data) { await client2.invoke(new Api5.messages.SetBotCallbackAnswer({ queryId: update.queryId, message: `\u5220\u9664\u5931\u8D25: ${error.message}`, alert: true })); } } +function formatTaskAge(value) { + if (!value) return null; + const ms = new Date(value).getTime(); + if (!Number.isFinite(ms) || ms <= 0) return null; + const diffSeconds = Math.max(0, Math.floor((Date.now() - ms) / 1e3)); + if (diffSeconds < 60) return "\u521A\u521A"; + const diffMinutes = Math.floor(diffSeconds / 60); + if (diffMinutes < 60) return `${diffMinutes} \u5206\u949F\u524D`; + const diffHours = Math.floor(diffMinutes / 60); + if (diffHours < 24) return `${diffHours} \u5C0F\u65F6\u524D`; + return `${Math.floor(diffHours / 24)} \u5929\u524D`; +} +function buildChannelTaskQueueReport(jobs) { + const labelStatus = (value) => ({ + pending: "\u7B49\u5F85", + queued: "\u6392\u961F", + scanning: "\u626B\u63CF\u4E2D", + active: "\u4F20\u8F93\u4E2D", + done: "\u5B8C\u6210", + running: "\u8FD0\u884C\u4E2D", + paused: "\u5DF2\u6682\u505C", + failed: "\u5931\u8D25", + cancelled: "\u5DF2\u53D6\u6D88", + completed: "\u5B8C\u6210", + completed_with_errors: "\u90E8\u5206\u5B8C\u6210" + })[value || ""] || (value || "\u672A\u77E5"); + const lines = ["\u{1F4E1} **\u9891\u9053\u4EFB\u52A1\u961F\u5217**"]; + jobs.forEach((job, index) => { + const total = Number(job.total_count || job.item_count || 0); + const pending = Number(job.pending_count || 0); + const downloading = Number(job.downloading_count || 0); + const success = Number(job.success_count || 0); + const failed = Number(job.failed_count || 0); + const skipped = Number(job.skipped_count_items || job.skipped_count || 0); + const done = success + failed + skipped; + const missing = Number(job.missing_metadata_count || 0); + const activeNow = Boolean(job.is_actively_running); + const paused = job.status === "paused"; + const cooldownUntilMs = job.cooldown_until ? new Date(job.cooldown_until).getTime() : 0; + const inCooldown = cooldownUntilMs > Date.now(); + const statusText = paused ? "\u5DF2\u6682\u505C" : inCooldown ? "\u51B7\u5374\u7B49\u5F85\u4E2D" : activeNow ? "\u6B63\u5728\u8FD0\u884C" : "\u7B49\u5F85\u63A5\u624B"; + const icon = paused ? "\u23F8\uFE0F" : inCooldown ? "\u{1F9CA}" : activeNow ? "\u{1F7E2}" : "\u23F3"; + const updatedText = formatTaskAge(job.updated_at); + const id = String(job.id).slice(0, 8); + lines.push([ + "", + `${index + 1}. ${icon} **${statusText}** \xB7 ${job.kind}`, + ` \u6765\u6E90\uFF1A${job.source}`, + ` \u9636\u6BB5\uFF1A\u626B\u63CF ${labelStatus(job.scan_status)} \xB7 \u4E0B\u8F7D ${labelStatus(job.download_status)}`, + ` \u961F\u5217\uFF1A\u4E0B\u8F7D\u4E2D ${downloading} \xB7 \u5F85\u5904\u7406 ${pending} \xB7 \u5DF2\u5B8C\u6210 ${done}/${total}`, + failed > 0 ? ` \u5F02\u5E38\uFF1A\u5931\u8D25 ${failed} \xB7 \u8DF3\u8FC7 ${skipped}` : skipped > 0 ? ` \u8DF3\u8FC7\uFF1A${skipped}` : "", + missing > 0 ? ` \u63D0\u793A\uFF1A${missing} \u4E2A\u5F85\u5904\u7406\u6761\u76EE\u6B63\u5728\u8865\u5168\u6587\u4EF6\u4FE1\u606F` : "", + inCooldown ? ` \u51B7\u5374\u5230\uFF1A${new Date(job.cooldown_until).toLocaleString("zh-CN", { hour12: false })}` : "", + updatedText ? ` \u6700\u8FD1\u6D3B\u52A8\uFF1A${updatedText}` : "", + ` ID\uFF1A${id}` + ].filter(Boolean).join("\n")); + }); + return lines.join("\n"); +} +function buildTasksKeyboard(jobs) { + if (jobs.length === 0) return void 0; + const rows = []; + for (const job of jobs.slice(0, 8)) { + const id = String(job.id).slice(0, 8); + const paused = job.status === "paused"; + rows.push(new Api5.KeyboardButtonRow({ + buttons: [ + new Api5.KeyboardButtonCallback({ + text: `${paused ? "\u25B6\uFE0F \u7EE7\u7EED" : "\u23F8 \u6682\u505C"} ${id}`, + data: Buffer.from(`ctq_${paused ? "resume" : "pause"}_${id}`) + }), + new Api5.KeyboardButtonCallback({ + text: `\u{1F6D1} \u53D6\u6D88 ${id}`, + data: Buffer.from(`ctq_cancel_${id}`) + }) + ] + })); + } + if (jobs.length > 1) { + rows.push(new Api5.KeyboardButtonRow({ + buttons: [new Api5.KeyboardButtonCallback({ text: "\u{1F6D1} \u53D6\u6D88\u5168\u90E8\u9891\u9053\u4EFB\u52A1", data: Buffer.from("ctq_cancel_all") })] + })); + } + return new Api5.ReplyInlineMarkup({ rows }); +} async function handleTasks(message) { try { const status = getTaskStatus(); const activeCount = status.active.length; const pendingCount = status.pending.length; - const historyCount = status.history.length; const senderId = message.senderId?.toJSNumber(); - const jobs = senderId ? await listTelegramBackgroundJobs(senderId, 5) : []; - if (activeCount === 0 && pendingCount === 0 && historyCount === 0 && jobs.length === 0) { + const jobs = senderId ? await listTelegramActiveTaskQueues(senderId, 10) : []; + if (activeCount === 0 && pendingCount === 0 && jobs.length === 0) { await message.reply({ message: MSG.EMPTY_TASKS }); return; } - const sections = [buildTasksReport(status.active, status.pending, status.history)]; - if (jobs.length > 0) { - sections.push([ - "\u{1F4E1} **\u9891\u9053\u540E\u53F0\u4EFB\u52A1**", - ...jobs.map((job, index) => [ - `${index + 1}. ${job.kind} \xB7 ${job.status}`, - ` \u6765\u6E90\uFF1A${job.source}`, - ` \u8FDB\u5EA6\uFF1A${job.enqueued_count || 0}/${job.total_count || 0}\u3000\u8DF3\u8FC7\uFF1A${job.skipped_count || 0}`, - job.error ? ` \u26A0\uFE0F ${job.error}` : "", - ` ID: ${String(job.id).slice(0, 8)}` - ].filter(Boolean).join("\n")) - ].join("\n")); + const sections = []; + if (activeCount > 0 || pendingCount > 0) { + sections.push(buildTasksReport(status.active, status.pending)); } - await message.reply({ message: sections.filter(Boolean).join("\n\n") }); + if (jobs.length > 0) { + sections.push(buildChannelTaskQueueReport(jobs)); + } + await message.reply({ + message: sections.filter(Boolean).join("\n\n"), + buttons: buildTasksKeyboard(jobs) + }); } catch (error) { console.error("\u{1F916} \u83B7\u53D6\u4EFB\u52A1\u5217\u8868\u5931\u8D25:", error); await message.reply({ message: MSG.ERR_TASKS }); @@ -6999,12 +7209,24 @@ 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)} + if (senderId) { + if (selector === "all") { + const jobs = await cancelAllTelegramBackgroundJobs(senderId); + const result2 = cancelDownloadTask(selector); + if (jobs.length > 0 || result2.total > 0) { + await message.reply({ message: `\u{1F6D1} \u5DF2\u53D6\u6D88\u4EFB\u52A1 + +\u9891\u9053\u4EFB\u52A1: ${jobs.length} +\u666E\u901A\u4E0B\u8F7D: \u5904\u7406\u4E2D ${result2.active} / \u7B49\u5F85 ${result2.pending}` }); + return; + } + } else { + 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; + return; + } } } const result = cancelDownloadTask(selector); @@ -7014,6 +7236,46 @@ async function handleCancelTask(message, args) { \u5904\u7406\u4E2D: ${result.active} \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 handleChannelTaskQueueCallback(client2, update, data) { + const userId = update.userId.toJSNumber(); + if (!await isAuthenticatedAsync(userId)) { + await client2.invoke(new Api5.messages.SetBotCallbackAnswer({ queryId: update.queryId, message: MSG.AUTH_REQUIRED, alert: true })); + return; + } + try { + const match = data.match(/^ctq_(pause|resume|cancel)_([0-9a-f]{4,}|all)$/i); + if (!match) return; + const [, action, selector] = match; + let toast = ""; + if (selector === "all" && action === "cancel") { + const jobs2 = await cancelAllTelegramBackgroundJobs(userId); + toast = jobs2.length > 0 ? `\u5DF2\u53D6\u6D88 ${jobs2.length} \u4E2A\u9891\u9053\u4EFB\u52A1` : "\u6CA1\u6709\u53EF\u53D6\u6D88\u7684\u9891\u9053\u4EFB\u52A1"; + } else if (action === "pause") { + const job = await pauseTelegramBackgroundJob(userId, selector); + toast = job ? `\u5DF2\u6682\u505C ${String(job.id).slice(0, 8)}` : "\u4EFB\u52A1\u4E0D\u5B58\u5728\u6216\u5DF2\u7ED3\u675F"; + } else if (action === "resume") { + const job = await resumeTelegramBackgroundJob(userId, selector); + toast = job ? `\u5DF2\u7EE7\u7EED ${String(job.id).slice(0, 8)}` : "\u4EFB\u52A1\u4E0D\u5B58\u5728\u6216\u4E0D\u5728\u6682\u505C\u72B6\u6001"; + } else { + const job = await cancelTelegramBackgroundJob(userId, selector); + toast = job ? `\u5DF2\u53D6\u6D88 ${String(job.id).slice(0, 8)}` : "\u4EFB\u52A1\u4E0D\u5B58\u5728\u6216\u5DF2\u7ED3\u675F"; + } + const jobs = await listTelegramActiveTaskQueues(userId, 10); + const status = getTaskStatus(); + const sections = []; + if (status.active.length > 0 || status.pending.length > 0) sections.push(buildTasksReport(status.active, status.pending)); + if (jobs.length > 0) sections.push(buildChannelTaskQueueReport(jobs)); + await client2.editMessage(update.peer, { + message: Number(update.msgId), + text: sections.length > 0 ? sections.join("\\n\\n") : MSG.EMPTY_TASKS, + buttons: buildTasksKeyboard(jobs) || new Api5.ReplyInlineMarkup({ rows: [] }) + }); + await client2.invoke(new Api5.messages.SetBotCallbackAnswer({ queryId: update.queryId, message: toast })); + } catch (error) { + console.error("\u{1F916} \u9891\u9053\u4EFB\u52A1\u6309\u94AE\u64CD\u4F5C\u5931\u8D25:", error); + await client2.invoke(new Api5.messages.SetBotCallbackAnswer({ queryId: update.queryId, message: `\u64CD\u4F5C\u5931\u8D25: ${error.message}`, alert: true })); + } +} async function handleRetryFailedTasks(message, args) { const senderId = message.senderId?.toJSNumber(); const jobSelector = args.find((arg) => /^[0-9a-f-]{4,}$/i.test(arg)); @@ -9053,6 +9315,10 @@ ${buildPathPreviewLine(appliedPath.folder)} await handleTelegramSubscriptionCallback(callbackUpdate, data); return; } + if (data.startsWith("ctq_")) { + await handleChannelTaskQueueCallback(activeClient, callbackUpdate, data); + return; + } if (data.startsWith("tq_")) { await handleTaskQueueCallback(callbackUpdate, data); return; diff --git a/backend/scripts/verify_telegram_task_queue_status.py b/backend/scripts/verify_telegram_task_queue_status.py new file mode 100644 index 0000000..6b1555e --- /dev/null +++ b/backend/scripts/verify_telegram_task_queue_status.py @@ -0,0 +1,46 @@ +from pathlib import Path + +root = Path(__file__).resolve().parents[1] +jobs = (root / 'src/services/telegramChannelJobs.ts').read_text() +commands = (root / 'src/services/telegramCommands.ts').read_text() +messages = (root / 'src/utils/telegramMessages.ts').read_text() + +checks = [ + ('listTelegramActiveTaskQueues', jobs), + ('finished_at IS NULL', jobs), + ('cancelled_at IS NULL', jobs), + ("j.status = 'running'", jobs), + ("j.status = 'paused'", jobs), + ("pending_count", jobs), + ("downloading_count", jobs), + ("is_actively_running", jobs), + ("completed_with_errors", jobs), + ('buildChannelTaskQueueReport', commands), + ('buildTasksKeyboard', commands), + ('handleChannelTaskQueueCallback', commands), + ('ctq_cancel_all', commands), + ('频道任务队列', commands), + ('正在运行', commands), + ('已暂停', commands), + ('等待接手', commands), + ('buildTasksReport(status.active, status.pending)', commands), + ('historyCount', commands, False), + ('最近完成', messages, False), + ('实时下载队列', messages), + ('等待开始', messages), +] + +missing = [] +for check in checks: + if len(check) == 2: + needle, haystack = check + should_exist = True + else: + needle, haystack, should_exist = check + present = needle in haystack + if present != should_exist: + missing.append(("missing" if should_exist else "unexpected") + f": {needle}") + +if missing: + raise SystemExit('Telegram /tasks queue rendering verification failed: ' + ', '.join(missing)) +print('Telegram /tasks queue rendering verifies active/paused-only output and hides history') diff --git a/backend/src/services/telegramBot.ts b/backend/src/services/telegramBot.ts index 301bc3d..c59f6b5 100644 --- a/backend/src/services/telegramBot.ts +++ b/backend/src/services/telegramBot.ts @@ -7,7 +7,7 @@ import path from 'path'; import { storageManager } from '../services/storage.js'; import { authenticatedUsers, passwordInputState, isAuthenticatedAsync, loadAuthenticatedUsers, persistAuthenticatedUser, userStates, TelegramUserState } from './telegramState.js'; import { is2FAEnabled, generateOTPAuthUrl, verifyTOTP, activate2FA } from '../utils/security.js'; -import { handleStart, handleHelp, handleStorage, handleStorageSwitch, handleStorageSwitchCallback, handleDelete, handleDeleteConfirmCallback, handleTasks, handleStopTasks, handlePauseTasks, handleResumeTasks, handleCancelTask, handleRetryFailedTasks, handleDownloadWorkers, handleDownloadWorkersCallback, handleFileConcurrency, handleFileConcurrencyCallback, handleStorageCleanupCallback, handlePathRules, handlePathOnce, handlePathSession, handlePathClear, handlePathRulesCallback, handleDuplicateMode, handleDuplicateModeCallback, handleCleanupSettings, handleCleanupSettingsCallback } from './telegramCommands.js'; +import { handleStart, handleHelp, handleStorage, handleStorageSwitch, handleStorageSwitchCallback, handleDelete, handleDeleteConfirmCallback, handleTasks, handleStopTasks, handlePauseTasks, handleResumeTasks, handleCancelTask, handleChannelTaskQueueCallback, handleRetryFailedTasks, handleDownloadWorkers, handleDownloadWorkersCallback, handleFileConcurrency, handleFileConcurrencyCallback, handleStorageCleanupCallback, handlePathRules, handlePathOnce, handlePathSession, handlePathClear, handlePathRulesCallback, handleDuplicateMode, handleDuplicateModeCallback, handleCleanupSettings, handleCleanupSettingsCallback } from './telegramCommands.js'; import { handleFileUpload, handleCleanupCallback, pauseDownloadTasks, resumeDownloadTasks, resolveTaskChatIdForControl, refreshSilentProgress, cancelSilentTask, canControlTask, loadFileDownloadConcurrencySetting } from './telegramUpload.js'; import { handleYtDlpCommand } from './ytDlpDownload.js'; import { @@ -1713,6 +1713,12 @@ export async function initTelegramBot(): Promise { return; } + // 处理 /tasks 频道任务队列按钮 + if (data.startsWith('ctq_')) { + await handleChannelTaskQueueCallback(activeClient, callbackUpdate, data); + return; + } + // 处理任务队列控制回调 if (data.startsWith('tq_')) { await handleTaskQueueCallback(callbackUpdate, data); diff --git a/backend/src/services/telegramChannelJobs.ts b/backend/src/services/telegramChannelJobs.ts index 9dc320a..6c0b672 100644 --- a/backend/src/services/telegramChannelJobs.ts +++ b/backend/src/services/telegramChannelJobs.ts @@ -362,11 +362,11 @@ async function markDownloadRefStatus(jobId: string, ref: TelegramDownloadMessage const sourcePeer = sourcePeerKey(ref.source, ref.origin === 'comment' ? 'comment' : 'channel'); await query( `UPDATE telegram_download_items - SET status = $3, + SET status = $3::varchar, 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, + attempts = CASE WHEN $3::text = 'failed' THEN attempts + 1 ELSE attempts END, + completed_at = CASE WHEN $3::text 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`, @@ -386,8 +386,8 @@ async function updateDownloadItemsStatus(jobId: string, messageIds: number[] | u if (ids.length === 0) return; await query( `UPDATE telegram_download_items - 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, + SET status = $2::varchar, error = $3, last_error = $3, updated_at = NOW(), + completed_at = CASE WHEN $2::text 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] @@ -435,6 +435,55 @@ async function updateJob(jobId: string, updates: Record) { await query(`UPDATE telegram_background_jobs SET ${setSql}, updated_at = NOW() WHERE id = $1`, [jobId, ...entries.map(([, value]) => value)]); } +async function hydratePendingDownloadRefs(userClient: TelegramClient, jobId: string): Promise { + const result = await query( + `SELECT id, source_peer, message_id + FROM telegram_download_items + WHERE job_id = $1 + AND status = 'pending' + AND (file_name IS NULL OR mime_type IS NULL) + ORDER BY created_at ASC + LIMIT 100`, + [jobId] + ); + let hydrated = 0; + for (const row of result.rows) { + try { + const messages = await userClient.getMessages(row.source_peer as any, { ids: [Number(row.message_id)] }); + const message = messages?.[0] as Api.Message | undefined; + if (!message) { + await query( + `UPDATE telegram_download_items + SET status = 'failed', error = $2, last_error = $2, attempts = attempts + 1, updated_at = NOW() + WHERE id = $1`, + [row.id, '消息不存在,无法补全文件元数据'] + ); + continue; + } + const fileInfo = extractFileInfo(message); + if (!fileInfo) { + await query( + `UPDATE telegram_download_items + SET status = 'skipped', error = $2, last_error = $2, completed_at = NOW(), updated_at = NOW() + WHERE id = $1`, + [row.id, '消息不包含可下载媒体,无法补全文件元数据'] + ); + continue; + } + await query( + `UPDATE telegram_download_items + SET file_name = $2, mime_type = $3, total_size = $4, updated_at = NOW() + WHERE id = $1`, + [row.id, fileInfo.fileName, fileInfo.mimeType, getEstimatedFileSize(message)] + ); + hydrated += 1; + } catch (error) { + console.warn('♻️ 补全 Telegram 下载条目元数据失败:', error); + } + } + return hydrated; +} + export async function subscribeTelegramChannel(userId: number, chatId: string | undefined, sourceInput: string, folderOverride?: string | null) { const userClient = requireUserClient(); const source = normalizeSource(sourceInput); @@ -572,7 +621,11 @@ async function claimPendingDownloadRefs(jobId: string, limit = TG_JOB_DOWNLOAD_B 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 + WHERE job_id = $1 + AND status = 'pending' + AND attempts < $2 + AND file_name IS NOT NULL + AND mime_type IS NOT NULL ORDER BY created_at ASC LIMIT $3 ) @@ -615,7 +668,9 @@ async function downloadClaimedRefs(botClient: TelegramClient, requestMessage: Ap 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 }; + const userClient = getTelegramUserClient(); while (await waitUntilRunnable(jobId, options)) { + if (userClient) await hydratePendingDownloadRefs(userClient, jobId); const refs = await claimPendingDownloadRefs(jobId); if (refs.length === 0) break; const result = await downloadClaimedRefs(botClient, requestMessage, jobId, source, refs, folderOverride, options); @@ -798,24 +853,79 @@ export async function enqueueTelegramTagDownload(botClient: TelegramClient, requ return { ...result, tag }; } -export async function listTelegramBackgroundJobs(userId: number, limit = 10) { +export async function listTelegramActiveTaskQueues(userId: number, limit = 10) { const result = await query( - `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 + `WITH item_stats AS ( + SELECT + job_id, + COUNT(*)::int AS item_count, + COUNT(*) FILTER (WHERE status = 'pending')::int AS pending_count, + COUNT(*) FILTER (WHERE status = 'downloading')::int AS downloading_count, + COUNT(*) FILTER (WHERE status = 'success')::int AS success_count, + COUNT(*) FILTER (WHERE status = 'failed')::int AS failed_count, + COUNT(*) FILTER (WHERE status = 'skipped')::int AS skipped_count_items, + COUNT(*) FILTER (WHERE status = 'pending' AND (file_name IS NULL OR mime_type IS NULL))::int AS missing_metadata_count, + MAX(updated_at) FILTER (WHERE status IN ('pending', 'downloading')) AS queue_updated_at + FROM telegram_download_items + GROUP BY job_id + ) + SELECT + j.id, j.kind, j.source, j.status, j.scan_status, j.download_status, + j.scan_cursor, j.cooldown_until, j.paused_at, j.cancelled_at, + j.total_count, j.enqueued_count, j.skipped_count, j.duplicate_count, + j.error, j.started_at, j.finished_at, j.created_at, j.updated_at, + COALESCE(s.item_count, 0)::int AS item_count, + COALESCE(s.pending_count, 0)::int AS pending_count, + COALESCE(s.downloading_count, 0)::int AS downloading_count, + COALESCE(s.success_count, 0)::int AS success_count, + COALESCE(s.failed_count, 0)::int AS failed_count, + COALESCE(s.skipped_count_items, 0)::int AS skipped_count_items, + COALESCE(s.missing_metadata_count, 0)::int AS missing_metadata_count, + s.queue_updated_at, + ( + j.status = 'running' + AND ( + COALESCE(s.downloading_count, 0) > 0 + OR j.scan_status = 'scanning' + OR (j.cooldown_until IS NOT NULL AND j.cooldown_until > NOW()) + ) + ) AS is_actively_running + FROM telegram_background_jobs j + LEFT JOIN item_stats s ON s.job_id = j.id + WHERE j.user_id = $1 + AND j.cancelled_at IS NULL + AND j.finished_at IS NULL + AND ( + ( + j.status = 'running' + AND ( + COALESCE(s.downloading_count, 0) > 0 + OR j.scan_status = 'scanning' + OR (j.cooldown_until IS NOT NULL AND j.cooldown_until > NOW()) + ) + ) + OR ( + j.status = 'paused' + AND (COALESCE(s.pending_count, 0) > 0 OR COALESCE(s.downloading_count, 0) > 0 OR j.scan_status = 'scanning') + ) + ) + ORDER BY + CASE WHEN j.status = 'paused' THEN 1 ELSE 0 END, + COALESCE(s.queue_updated_at, j.updated_at) DESC LIMIT $2`, [userId, limit] ); return result.rows; } +export const listTelegramBackgroundJobs = listTelegramActiveTaskQueues; + 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') + WHERE user_id = $1 AND id::text LIKE $2 AND finished_at IS NULL AND status NOT IN ('completed', 'completed_with_errors', 'cancelled') RETURNING id, source, status`, [userId, `${selector}%`] ); @@ -825,8 +935,8 @@ export async function pauseTelegramBackgroundJob(userId: number, selector: strin 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' + SET status = 'running', paused_at = NULL, finished_at = NULL, error = NULL, download_status = 'active', updated_at = NOW() + WHERE user_id = $1 AND id::text LIKE $2 AND cancelled_at IS NULL AND status = 'paused' RETURNING id, source, status`, [userId, `${selector}%`] ); @@ -837,7 +947,7 @@ export async function cancelTelegramBackgroundJob(userId: number, selector: stri 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') + WHERE user_id = $1 AND id::text LIKE $2 AND finished_at IS NULL AND status NOT IN ('completed', 'completed_with_errors', 'cancelled') RETURNING id, source, status`, [userId, `${selector}%`] ); @@ -847,6 +957,28 @@ export async function cancelTelegramBackgroundJob(userId: number, selector: stri return result.rows[0] || null; } +export async function cancelAllTelegramBackgroundJobs(userId: number) { + 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 status NOT IN ('completed', 'completed_with_errors', 'cancelled') + RETURNING id, source, status`, + [userId] + ); + const ids = result.rows.map(row => row.id); + if (ids.length > 0) { + await query( + `UPDATE telegram_download_items + SET status = 'skipped', locked_at = NULL, updated_at = NOW() + WHERE job_id = ANY($1::uuid[]) + AND status IN ('pending', 'downloading')`, + [ids] + ); + } + return result.rows; +} + 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`, @@ -926,6 +1058,17 @@ async function recoverTelegramJob(botClient: TelegramClient, job: any): Promise< ); if (itemResult.rows.length === 0) return; + const missingMetadata = itemResult.rows.filter(row => !row.file_name || !row.mime_type).length; + if (missingMetadata > 0) { + const userClient = getTelegramUserClient(); + if (userClient) { + await hydratePendingDownloadRefs(userClient, job.id); + return recoverTelegramJob(botClient, job); + } + await updateJob(job.id, { status: 'failed', error: `${missingMetadata} 个待下载条目缺少文件元数据,无法恢复`, finished_at: new Date() }); + return; + } + const refs: TelegramDownloadMessageRef[] = itemResult.rows .filter(row => row.file_name && row.mime_type) .map(row => ({ @@ -988,6 +1131,8 @@ export async function recoverInterruptedTelegramJobs(botClient: TelegramClient): 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.finished_at IS NULL + AND j.cancelled_at IS NULL AND j.status IN ('pending', 'running', 'failed', 'completed_with_errors') AND i.status = 'pending' ORDER BY j.created_at ASC diff --git a/backend/src/services/telegramCommands.ts b/backend/src/services/telegramCommands.ts index 7065b70..8e8b1a4 100644 --- a/backend/src/services/telegramCommands.ts +++ b/backend/src/services/telegramCommands.ts @@ -18,7 +18,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 { cancelTelegramBackgroundJob, listTelegramBackgroundJobs, pauseTelegramBackgroundJob, resumeTelegramBackgroundJob, retryTelegramBackgroundJob } from './telegramChannelJobs.js'; +import { cancelAllTelegramBackgroundJobs, cancelTelegramBackgroundJob, listTelegramActiveTaskQueues, 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'; @@ -744,34 +744,119 @@ export async function handleDeleteConfirmCallback(client: TelegramClient, update } } +function formatTaskAge(value: unknown): string | null { + if (!value) return null; + const ms = new Date(value as string).getTime(); + if (!Number.isFinite(ms) || ms <= 0) return null; + const diffSeconds = Math.max(0, Math.floor((Date.now() - ms) / 1000)); + if (diffSeconds < 60) return '刚刚'; + const diffMinutes = Math.floor(diffSeconds / 60); + if (diffMinutes < 60) return `${diffMinutes} 分钟前`; + const diffHours = Math.floor(diffMinutes / 60); + if (diffHours < 24) return `${diffHours} 小时前`; + return `${Math.floor(diffHours / 24)} 天前`; +} + +function buildChannelTaskQueueReport(jobs: any[]): string { + const labelStatus = (value: string | undefined) => ({ + pending: '等待', + queued: '排队', + scanning: '扫描中', + active: '传输中', + done: '完成', + running: '运行中', + paused: '已暂停', + failed: '失败', + cancelled: '已取消', + completed: '完成', + completed_with_errors: '部分完成', + } as Record)[value || ''] || (value || '未知'); + + const lines = ['📡 **频道任务队列**']; + jobs.forEach((job: any, index: number) => { + const total = Number(job.total_count || job.item_count || 0); + const pending = Number(job.pending_count || 0); + const downloading = Number(job.downloading_count || 0); + const success = Number(job.success_count || 0); + const failed = Number(job.failed_count || 0); + const skipped = Number(job.skipped_count_items || job.skipped_count || 0); + const done = success + failed + skipped; + const missing = Number(job.missing_metadata_count || 0); + const activeNow = Boolean(job.is_actively_running); + const paused = job.status === 'paused'; + const cooldownUntilMs = job.cooldown_until ? new Date(job.cooldown_until).getTime() : 0; + const inCooldown = cooldownUntilMs > Date.now(); + const statusText = paused ? '已暂停' : inCooldown ? '冷却等待中' : activeNow ? '正在运行' : '等待接手'; + const icon = paused ? '⏸️' : inCooldown ? '🧊' : activeNow ? '🟢' : '⏳'; + const updatedText = formatTaskAge(job.updated_at); + const id = String(job.id).slice(0, 8); + lines.push([ + '', + `${index + 1}. ${icon} **${statusText}** · ${job.kind}`, + ` 来源:${job.source}`, + ` 阶段:扫描 ${labelStatus(job.scan_status)} · 下载 ${labelStatus(job.download_status)}`, + ` 队列:下载中 ${downloading} · 待处理 ${pending} · 已完成 ${done}/${total}`, + failed > 0 ? ` 异常:失败 ${failed} · 跳过 ${skipped}` : (skipped > 0 ? ` 跳过:${skipped}` : ''), + missing > 0 ? ` 提示:${missing} 个待处理条目正在补全文件信息` : '', + inCooldown ? ` 冷却到:${new Date(job.cooldown_until).toLocaleString('zh-CN', { hour12: false })}` : '', + updatedText ? ` 最近活动:${updatedText}` : '', + ` ID:${id}`, + ].filter(Boolean).join('\n')); + }); + return lines.join('\n'); +} + +function buildTasksKeyboard(jobs: any[]): Api.ReplyInlineMarkup | undefined { + if (jobs.length === 0) return undefined; + const rows: Api.KeyboardButtonRow[] = []; + for (const job of jobs.slice(0, 8)) { + const id = String(job.id).slice(0, 8); + const paused = job.status === 'paused'; + rows.push(new Api.KeyboardButtonRow({ + buttons: [ + new Api.KeyboardButtonCallback({ + text: `${paused ? '▶️ 继续' : '⏸ 暂停'} ${id}`, + data: Buffer.from(`ctq_${paused ? 'resume' : 'pause'}_${id}`), + }), + new Api.KeyboardButtonCallback({ + text: `🛑 取消 ${id}`, + data: Buffer.from(`ctq_cancel_${id}`), + }), + ], + })); + } + if (jobs.length > 1) { + rows.push(new Api.KeyboardButtonRow({ + buttons: [new Api.KeyboardButtonCallback({ text: '🛑 取消全部频道任务', data: Buffer.from('ctq_cancel_all') })], + })); + } + return new Api.ReplyInlineMarkup({ rows }); +} + export async function handleTasks(message: Api.Message): Promise { try { const status = getTaskStatus(); const activeCount = status.active.length; const pendingCount = status.pending.length; - const historyCount = status.history.length; const senderId = message.senderId?.toJSNumber(); - const jobs = senderId ? await listTelegramBackgroundJobs(senderId, 5) : []; + const jobs = senderId ? await listTelegramActiveTaskQueues(senderId, 10) : []; - if (activeCount === 0 && pendingCount === 0 && historyCount === 0 && jobs.length === 0) { + if (activeCount === 0 && pendingCount === 0 && jobs.length === 0) { await message.reply({ message: MSG.EMPTY_TASKS }); return; } - const sections = [buildTasksReport(status.active, status.pending, status.history)]; - if (jobs.length > 0) { - sections.push([ - '📡 **频道后台任务**', - ...jobs.map((job: any, index: number) => [ - `${index + 1}. ${job.kind} · ${job.status}`, - ` 来源:${job.source}`, - ` 进度:${job.enqueued_count || 0}/${job.total_count || 0} 跳过:${job.skipped_count || 0}`, - job.error ? ` ⚠️ ${job.error}` : '', - ` ID: ${String(job.id).slice(0, 8)}`, - ].filter(Boolean).join('\n')), - ].join('\n')); + const sections: string[] = []; + if (activeCount > 0 || pendingCount > 0) { + sections.push(buildTasksReport(status.active, status.pending)); } - await message.reply({ message: sections.filter(Boolean).join('\n\n') }); + if (jobs.length > 0) { + sections.push(buildChannelTaskQueueReport(jobs)); + } + await message.reply({ + message: sections.filter(Boolean).join('\n\n'), + buttons: buildTasksKeyboard(jobs), + }); } catch (error) { console.error('🤖 获取任务列表失败:', error); @@ -827,17 +912,68 @@ export async function handleResumeTasks(message: Api.Message, args: string[] = [ 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; + if (senderId) { + if (selector === 'all') { + const jobs = await cancelAllTelegramBackgroundJobs(senderId); + const result = cancelDownloadTask(selector); + if (jobs.length > 0 || result.total > 0) { + await message.reply({ message: `🛑 已取消任务\n\n频道任务: ${jobs.length}\n普通下载: 处理中 ${result.active} / 等待 ${result.pending}` }); + return; + } + } else { + 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 handleChannelTaskQueueCallback(client: TelegramClient, update: Api.UpdateBotCallbackQuery, data: string): Promise { + const userId = update.userId.toJSNumber(); + if (!(await isAuthenticatedAsync(userId))) { + await client.invoke(new Api.messages.SetBotCallbackAnswer({ queryId: update.queryId, message: MSG.AUTH_REQUIRED, alert: true })); + return; + } + try { + const match = data.match(/^ctq_(pause|resume|cancel)_([0-9a-f]{4,}|all)$/i); + if (!match) return; + const [, action, selector] = match; + let toast = ''; + if (selector === 'all' && action === 'cancel') { + const jobs = await cancelAllTelegramBackgroundJobs(userId); + toast = jobs.length > 0 ? `已取消 ${jobs.length} 个频道任务` : '没有可取消的频道任务'; + } else if (action === 'pause') { + const job = await pauseTelegramBackgroundJob(userId, selector); + toast = job ? `已暂停 ${String(job.id).slice(0, 8)}` : '任务不存在或已结束'; + } else if (action === 'resume') { + const job = await resumeTelegramBackgroundJob(userId, selector); + toast = job ? `已继续 ${String(job.id).slice(0, 8)}` : '任务不存在或不在暂停状态'; + } else { + const job = await cancelTelegramBackgroundJob(userId, selector); + toast = job ? `已取消 ${String(job.id).slice(0, 8)}` : '任务不存在或已结束'; + } + + const jobs = await listTelegramActiveTaskQueues(userId, 10); + const status = getTaskStatus(); + const sections: string[] = []; + if (status.active.length > 0 || status.pending.length > 0) sections.push(buildTasksReport(status.active, status.pending)); + if (jobs.length > 0) sections.push(buildChannelTaskQueueReport(jobs)); + await client.editMessage(update.peer, { + message: Number(update.msgId), + text: sections.length > 0 ? sections.join('\\n\\n') : MSG.EMPTY_TASKS, + buttons: buildTasksKeyboard(jobs) || new Api.ReplyInlineMarkup({ rows: [] }), + }); + await client.invoke(new Api.messages.SetBotCallbackAnswer({ queryId: update.queryId, message: toast })); + } catch (error) { + console.error('🤖 频道任务按钮操作失败:', error); + await client.invoke(new Api.messages.SetBotCallbackAnswer({ queryId: update.queryId, message: `操作失败: ${(error as Error).message}`, alert: true })); + } +} + 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)); diff --git a/backend/src/utils/telegramMessages.ts b/backend/src/utils/telegramMessages.ts index 24a9cca..8ee8a4c 100644 --- a/backend/src/utils/telegramMessages.ts +++ b/backend/src/utils/telegramMessages.ts @@ -348,51 +348,39 @@ interface TaskItem { export function buildTasksReport( active: TaskItem[], pending: TaskItem[], - history: TaskItem[] + _history: TaskItem[] = [] ): string { const lines: string[] = [ - `📋 **任务队列状态**`, - `🔄 ${active.length} 进行中 ⏳ ${pending.length} 等待中`, + `📋 **实时下载队列**`, + `🔄 ${active.length} 正在下载 ⏳ ${pending.length} 等待开始`, LINE, ]; if (active.length > 0) { lines.push(''); - lines.push(`**🔄 正在处理**`); + lines.push(`**🔄 正在下载**`); active.forEach(task => { lines.push(` ▸ ${task.fileName}`); if (task.totalSize && task.downloadedSize) { const bar = generateProgressBar(task.downloadedSize, task.totalSize, 10); lines.push(` ${bar} (${formatBytes(task.downloadedSize)}/${formatBytes(task.totalSize)})`); } else { - lines.push(` ⏳ 下载中...`); + lines.push(` 传输中,请稍候...`); } }); } if (pending.length > 0) { lines.push(''); - lines.push(`**⏳ 等待队列** (前 5 个)`); + lines.push(`**⏳ 等待开始** (前 5 个)`); pending.slice(0, 5).forEach((task, i) => { lines.push(` ${i + 1}. ${task.fileName}`); }); if (pending.length > 5) { - lines.push(` ... 还有 ${pending.length - 5} 个任务`); + lines.push(` ... 还有 ${pending.length - 5} 个等待任务`); } } - if (history.length > 0) { - lines.push(''); - lines.push(`**🕒 最近完成** (前 5 个)`); - history.slice(0, 5).forEach(task => { - const icon = task.status === 'success' ? '✅' : task.status === 'cancelled' ? '🛑' : '❌'; - lines.push(` ${icon} ${task.fileName}`); - if ((task.status === 'failed' || task.status === 'cancelled') && task.error) { - lines.push(` 原因: ${task.error}`); - } - }); - } - return lines.join('\n'); }