feat(telegram): download channel comment media

Add optional comment-thread scanning for manual channel downloads and route mixed channel/comment message refs through the download queue.

Skip Telegram stickers when extracting downloadable media and add a regression verifier for sticker exclusion.
This commit is contained in:
hicocos committed 2026-07-03 17:41:22 +08:00
1 parent 050fedf185
commit b9297e4db3
8 files changed
+1060 -482

No files matched your search

+558 -312
View File
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,17 @@
import { extractFileInfo } from '../utils/telegramMedia.js';
function assert(condition: unknown, message: string) {
if (!condition) {
throw new Error(message);
}
}
const stickerMessage = {
id: 123,
media: { className: 'MessageMediaDocument' },
sticker: { className: 'Document' },
};
const result = extractFileInfo(stickerMessage as any);
assert(result === null, `expected Telegram sticker to be ignored, got ${JSON.stringify(result)}`);
console.log('telegramMedia sticker exclusion test passed');
+109 -16
View File
@@ -18,6 +18,7 @@ import {
subscribeTelegramChannel,
unsubscribeTelegramChannel,
updateTelegramSubscriptionFolder,
TELEGRAM_COMMENTS_MAX_PER_POST,
} from './telegramChannelJobs.js';
import { cleanupOrphanFiles, isAutoCleanupEnabled, startPeriodicCleanup } from './orphanCleanup.js';
import { MSG, buildStartPrompt, buildAuthSuccess, build2FASetupCaption, buildCleanupNotice } from '../utils/telegramMessages.js';
@@ -33,7 +34,7 @@ const SESSION_FILE = process.env.TELEGRAM_SESSION_FILE || './data/telegram_sessi
let client: TelegramClient | null = null;
type TelegramWizardKind = 'tg_sub_manage' | 'tg_download' | 'tg_date' | 'tg_tag';
type TelegramWizardStep = 'mode' | 'source' | 'path' | 'start_date' | 'end_date' | 'tag';
type TelegramWizardStep = 'mode' | 'source' | 'path' | 'comments' | 'start_date' | 'end_date' | 'tag';
interface TelegramWizardState {
kind: TelegramWizardKind;
@@ -42,6 +43,8 @@ interface TelegramWizardState {
startDate?: string;
tag?: string;
customFolder?: string;
includeComments?: boolean;
commentsMaxPerPost?: number;
subscriptionId?: string;
subscriptionTitle?: string;
subscriptionSource?: string;
@@ -63,6 +66,22 @@ function buildTelegramDownloadModeKeyboard(): Api.ReplyInlineMarkup {
});
}
function buildTelegramCommentsKeyboard(): Api.ReplyInlineMarkup {
return new Api.ReplyInlineMarkup({
rows: [
new Api.KeyboardButtonRow({
buttons: [
new Api.KeyboardButtonCallback({ text: '仅频道正文', data: Buffer.from('tgd_comments_off') }),
new Api.KeyboardButtonCallback({ text: '频道 + 评论区', data: Buffer.from('tgd_comments_on') }),
],
}),
new Api.KeyboardButtonRow({
buttons: [new Api.KeyboardButtonCallback({ text: '取消', data: Buffer.from('tgd_cancel') })],
}),
],
});
}
const telegramWizardStates = new Map<number, TelegramWizardState>();
interface RateBucket {
@@ -187,6 +206,8 @@ function buildTelegramWizardPrompt(state: TelegramWizardState): string {
'请发送频道用户名或链接:',
'例如:`@channel_username` 或 `https://t.me/channel_username`',
'',
'也可以直接发送:`@频道 comments` 或 `@频道 no-comments`。',
'',
'发送“取消”可退出。',
].join('\n');
}
@@ -207,6 +228,22 @@ function buildTelegramWizardPrompt(state: TelegramWizardState): string {
].join('\n');
}
if (state.step === 'comments') {
return [
title,
`📍 频道:${state.subscriptionSource || state.source}`,
state.customFolder ? `📁 保存目录:${state.customFolder}` : '📁 保存策略:默认自动分类',
'',
'是否同时扫描频道帖子下方的评论区文件?',
'',
`默认关闭;开启后每个频道帖子最多扫描 ${state.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST} 条评论。`,
'文字评论、普通链接和其它无文件消息会自动忽略。',
'',
'也可以发送:`开` / `关` / `yes` / `no`。',
'发送“取消”可退出。',
].join('\n');
}
if (state.step === 'tag') {
return [
title,
@@ -251,9 +288,12 @@ function isDateOnly(text: string): boolean {
async function replyWithJobResult(statusMessage: Api.Message, fallbackMessage: Api.Message, promise: Promise<any>, kind: 'date' | 'tag'): Promise<void> {
promise
.then(result => {
const commentLine = result.commentMediaFound || result.commentMessagesScanned
? `\n评论区: 扫描 ${result.commentMessagesScanned || 0} 条,发现 ${result.commentMediaFound || 0} 个文件`
: '';
const text = kind === 'tag'
? `✅ 标签下载任务完成\n标签: ${result.tag}\nID: ${String(result.jobId).slice(0, 8)}\n入队: ${result.found}\n跳过: ${result.skipped}\n失败: ${result.failed}`
: `✅ 日期范围任务完成\nID: ${String(result.jobId).slice(0, 8)}\n入队: ${result.found}\n跳过: ${result.skipped}\n失败: ${result.failed}`;
? `✅ 标签下载任务完成\n标签: ${result.tag}\nID: ${String(result.jobId).slice(0, 8)}\n入队: ${result.found}\n跳过: ${result.skipped}\n失败: ${result.failed}${commentLine}`
: `✅ 日期范围任务完成\nID: ${String(result.jobId).slice(0, 8)}\n入队: ${result.found}\n跳过: ${result.skipped}\n失败: ${result.failed}${commentLine}`;
statusMessage.edit({ text }).catch(() => fallbackMessage.reply({ message: text }).catch(() => undefined));
})
.catch(error => {
@@ -305,7 +345,18 @@ async function handleTelegramWizardMessage(message: Api.Message, senderId: numbe
}
if (state.step === 'source') {
state.source = input;
const sourceParts = input.split(/\s+/).filter(Boolean);
const commentFlag = sourceParts[sourceParts.length - 1]?.toLowerCase();
if (['comments', '--comments', 'include-comments', '评论', '评论区'].includes(commentFlag)) {
state.includeComments = true;
state.commentsMaxPerPost = TELEGRAM_COMMENTS_MAX_PER_POST;
sourceParts.pop();
} else if (['no-comments', '--no-comments', 'channel-only', '仅频道'].includes(commentFlag)) {
state.includeComments = false;
state.commentsMaxPerPost = TELEGRAM_COMMENTS_MAX_PER_POST;
sourceParts.pop();
}
state.source = sourceParts.join(' ') || input;
if (state.kind === 'tg_sub_manage') {
if (/^\d+$/.test(input)) {
const rows = await listTelegramSubscriptions(senderId);
@@ -387,11 +438,25 @@ async function handleTelegramWizardMessage(message: Api.Message, senderId: numbe
return true;
}
if (state.kind === 'tg_tag') {
state.step = 'tag';
} else {
state.step = 'start_date';
if (state.kind === 'tg_tag' || state.kind === 'tg_date') {
state.step = state.includeComments !== undefined ? (state.kind === 'tg_tag' ? 'tag' : 'start_date') : 'comments';
await message.reply({ message: buildTelegramWizardPrompt(state), buttons: state.step === 'comments' ? buildTelegramCommentsKeyboard() : undefined });
return true;
}
return true;
}
if (state.step === 'comments') {
const enabled = /^(开|开启|是|包含|评论|评论区|yes|y|on|true|1)$/i.test(input);
const disabled = /^(关|关闭|否|不包含|仅频道|no|n|off|false|0)$/i.test(input);
if (!enabled && !disabled) {
await message.reply({ message: '❌ 请发送 `开`/`关`,或点击按钮选择是否包含评论区文件。' });
return true;
}
state.includeComments = enabled;
state.commentsMaxPerPost = TELEGRAM_COMMENTS_MAX_PER_POST;
state.step = state.kind === 'tg_tag' ? 'tag' : 'start_date';
await message.reply({ message: buildTelegramWizardPrompt(state) });
return true;
}
@@ -401,7 +466,10 @@ async function handleTelegramWizardMessage(message: Api.Message, senderId: numbe
try {
const queuedMsg = await message.reply({ message: `⏳ 已开始后台扫描 ${state.source} 中带有 ${input.startsWith('#') ? input : `#${input}`} 的媒体消息...
完成后会自动更新结果,可用 /tasks 查看后台任务。` });
await replyWithJobResult(queuedMsg as Api.Message, message, enqueueTelegramTagDownload(client!, message, senderId, state.source!, input, state.customFolder), 'tag');
await replyWithJobResult(queuedMsg as Api.Message, message, enqueueTelegramTagDownload(client!, message, senderId, state.source!, input, state.customFolder, {
includeComments: Boolean(state.includeComments),
commentsMaxPerPost: state.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST,
}), 'tag');
} catch (error) {
await message.reply({ message: `❌ 标签下载失败: ${error instanceof Error ? error.message : String(error)}` });
}
@@ -428,7 +496,10 @@ async function handleTelegramWizardMessage(message: Api.Message, senderId: numbe
try {
const queuedMsg = await message.reply({ message: `⏳ 已开始后台扫描 ${state.source}:${state.startDate} → ${input}...
完成后会自动更新结果,可用 /tasks 查看后台任务。` });
await replyWithJobResult(queuedMsg as Api.Message, message, enqueueTelegramDateDownload(client!, message, senderId, state.source!, state.startDate!, input, state.customFolder), 'date');
await replyWithJobResult(queuedMsg as Api.Message, message, enqueueTelegramDateDownload(client!, message, senderId, state.source!, state.startDate!, input, state.customFolder, {
includeComments: Boolean(state.includeComments),
commentsMaxPerPost: state.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST,
}), 'date');
} catch (error) {
await message.reply({ message: `❌ 日期下载失败: ${error instanceof Error ? error.message : String(error)}` });
}
@@ -754,12 +825,34 @@ async function handleTelegramDownloadModeCallback(update: Api.UpdateBotCallbackQ
return;
}
const state = telegramWizardStates.get(userId) || { kind: 'tg_download' as TelegramWizardKind, step: 'mode' as TelegramWizardStep };
if (data === 'tgd_mode_date') state.kind = 'tg_date';
if (data === 'tgd_mode_tag') state.kind = 'tg_tag';
state.step = 'source';
telegramWizardStates.set(userId, state);
await client.editMessage(update.peer, { message: update.msgId, text: buildTelegramWizardPrompt(state) });
await client.invoke(new Api.messages.SetBotCallbackAnswer({ queryId: update.queryId, message: data === 'tgd_mode_date' ? '按日期下载' : '按标签下载' }));
if (data === 'tgd_mode_date') {
state.kind = 'tg_date';
state.step = 'source';
telegramWizardStates.set(userId, state);
await client.editMessage(update.peer, { message: update.msgId, text: buildTelegramWizardPrompt(state) });
await client.invoke(new Api.messages.SetBotCallbackAnswer({ queryId: update.queryId, message: '按日期下载' }));
return;
}
if (data === 'tgd_mode_tag') {
state.kind = 'tg_tag';
state.step = 'source';
telegramWizardStates.set(userId, state);
await client.editMessage(update.peer, { message: update.msgId, text: buildTelegramWizardPrompt(state) });
await client.invoke(new Api.messages.SetBotCallbackAnswer({ queryId: update.queryId, message: '按标签下载' }));
return;
}
if (data === 'tgd_comments_on' || data === 'tgd_comments_off') {
state.includeComments = data === 'tgd_comments_on';
state.commentsMaxPerPost = TELEGRAM_COMMENTS_MAX_PER_POST;
state.step = state.kind === 'tg_tag' ? 'tag' : 'start_date';
telegramWizardStates.set(userId, state);
await client.editMessage(update.peer, { message: update.msgId, text: buildTelegramWizardPrompt(state) });
await client.invoke(new Api.messages.SetBotCallbackAnswer({
queryId: update.queryId,
message: state.includeComments ? '将包含评论区文件' : '仅下载频道正文文件',
}));
return;
}
}
async function handleTelegramSubscriptionCallback(update: Api.UpdateBotCallbackQuery, data: string): Promise<void> {
+145 -13
View File
@@ -2,11 +2,13 @@ import { Api, TelegramClient } from 'telegram';
import { query } from '../db/index.js';
import { storageManager } from './storage.js';
import { getTelegramUserClient, isTelegramUserClientReady } from './telegramUserClient.js';
import { downloadTelegramChannelRange, getTelegramDownloadPreview } from './telegramUpload.js';
import { downloadTelegramChannelRange, getTelegramDownloadPreview, type TelegramDownloadMessageRef } from './telegramUpload.js';
import { getSetting } from '../utils/settings.js';
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);
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;
function parseTelegramSourceAllowlist(raw: string | undefined): string[] {
@@ -158,6 +160,117 @@ async function persistDownloadItems(jobId: string, source: string, messages: Api
}
}
interface TelegramCommentScanOptions {
includeComments?: boolean;
commentsMaxPerPost?: number;
}
interface TelegramDownloadScanResult {
messages: Api.Message[];
refs: TelegramDownloadMessageRef[];
commentMessagesScanned: number;
commentMediaFound: number;
}
async function getDiscussionMediaRefs(
userClient: TelegramClient,
source: string,
postMessages: Api.Message[],
options: TelegramCommentScanOptions & { tag?: string; startDate?: Date; endDate?: Date } = {},
): Promise<{ refs: TelegramDownloadMessageRef[]; scanned: number; mediaFound: number }> {
if (!options.includeComments || postMessages.length === 0) {
return { refs: [], scanned: 0, mediaFound: 0 };
}
const maxPerPost = Math.max(1, Math.floor(options.commentsMaxPerPost || TELEGRAM_COMMENTS_MAX_PER_POST));
const refs: TelegramDownloadMessageRef[] = [];
let scanned = 0;
let mediaFound = 0;
const seen = new Set<string>();
for (const post of postMessages) {
const declaredReplies = Number((post as any).replies?.replies || 0);
if (declaredReplies <= 0) continue;
let offsetId = 0;
let scannedForPost = 0;
while (scannedForPost < maxPerPost) {
const batch = await userClient.getMessages(source as any, {
limit: Math.min(100, maxPerPost - scannedForPost),
offsetId,
replyTo: post.id,
});
if (!batch.length) break;
for (const comment of batch) {
if (!comment) continue;
scanned += 1;
scannedForPost += 1;
offsetId = comment.id;
if (options.startDate || options.endDate) {
const commentDate = new Date((comment.date || 0) * 1000);
if (options.startDate && commentDate < options.startDate) continue;
if (options.endDate && commentDate > options.endDate) continue;
}
if (options.tag && !messageMatchesHashtag(comment, options.tag)) continue;
const fileInfo = extractFileInfo(comment);
if (!fileInfo) continue;
const sourceKey = `${comment.chatId?.toString() || source}:${comment.id}`;
if (seen.has(sourceKey)) continue;
seen.add(sourceKey);
mediaFound += 1;
refs.push({
id: comment.id,
source: comment.chatId || source,
origin: 'comment',
channelPostId: post.id,
fileInfo,
totalSize: getEstimatedFileSize(comment),
message: comment,
});
}
if (batch.length === 0 || scannedForPost >= maxPerPost) break;
}
}
return { refs, scanned, mediaFound };
}
function toChannelDownloadRef(source: string, message: Api.Message): TelegramDownloadMessageRef | null {
const fileInfo = extractFileInfo(message);
if (!fileInfo) return null;
return {
id: message.id,
source,
origin: 'channel',
fileInfo,
totalSize: getEstimatedFileSize(message),
};
}
async function buildDownloadScanResult(
userClient: TelegramClient,
source: string,
messages: Api.Message[],
options: TelegramCommentScanOptions & { tag?: string; startDate?: Date; endDate?: Date } = {},
): Promise<TelegramDownloadScanResult> {
const refs = messages
.map(message => toChannelDownloadRef(source, message))
.filter((ref): ref is TelegramDownloadMessageRef => Boolean(ref));
const commentScan = await getDiscussionMediaRefs(userClient, source, messages, options);
refs.push(...commentScan.refs);
return {
messages,
refs,
commentMessagesScanned: commentScan.scanned,
commentMediaFound: commentScan.mediaFound,
};
}
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;
@@ -263,7 +376,7 @@ 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) {
export async function enqueueTelegramDateDownload(botClient: TelegramClient, requestMessage: Api.Message, userId: number, sourceInput: string, startDateText: string, endDateText: string, folderOverride?: string | null, options: TelegramCommentScanOptions = {}) {
const userClient = requireUserClient();
const source = normalizeSource(sourceInput);
await assertTelegramSourceAllowed(source);
@@ -273,13 +386,23 @@ export async function enqueueTelegramDateDownload(botClient: TelegramClient, req
const baseMessages = await getMessagesByDateRange(userClient, source, startDate, endDate);
const messages = await expandMessagesWithMediaGroups(userClient, source, baseMessages);
const messageIds = messages.map(message => message.id);
const jobId = await createJob(userId, requestMessage.chatId?.toString(), 'date_range', source, { startDate: startDateText, endDate: endDateText, messageIds, folderOverride: folderOverride || null });
const scan = await buildDownloadScanResult(userClient, source, messages, { ...options, startDate, endDate });
const messageIds = scan.refs.map(ref => ref.id);
const jobId = await createJob(userId, requestMessage.chatId?.toString(), 'date_range', source, {
startDate: startDateText,
endDate: endDateText,
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: messages.length });
await updateJob(jobId, { status: 'running', started_at: new Date(), total_count: scan.refs.length });
try {
const result = await downloadTelegramChannelRange(botClient, requestMessage, source, 0, messages.length, 'older', messages.map(message => message.id), folderOverride);
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');
@@ -290,7 +413,7 @@ export async function enqueueTelegramDateDownload(botClient: TelegramClient, req
error: result.failed > 0 ? `${result.failed} 个文件下载失败` : null,
finished_at: new Date(),
});
return { jobId, ...result };
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;
@@ -321,7 +444,7 @@ async function getMessagesByHashtag(userClient: TelegramClient, source: string,
return result.sort((a, b) => a.id - b.id);
}
export async function enqueueTelegramTagDownload(botClient: TelegramClient, requestMessage: Api.Message, userId: number, sourceInput: string, tagInput: string, folderOverride?: string | null) {
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);
@@ -329,13 +452,22 @@ export async function enqueueTelegramTagDownload(botClient: TelegramClient, requ
const baseMessages = await getMessagesByHashtag(userClient, source, tag);
const messages = await expandMessagesWithMediaGroups(userClient, source, baseMessages);
const messageIds = messages.map(message => message.id);
const jobId = await createJob(userId, requestMessage.chatId?.toString(), 'tag_download', source, { tag, messageIds, folderOverride: folderOverride || null });
const scan = await buildDownloadScanResult(userClient, source, messages, { ...options, tag });
const messageIds = scan.refs.map(ref => ref.id);
const jobId = await createJob(userId, requestMessage.chatId?.toString(), 'tag_download', source, {
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: messages.length });
await updateJob(jobId, { status: 'running', started_at: new Date(), total_count: scan.refs.length });
try {
const result = await downloadTelegramChannelRange(botClient, requestMessage, source, 0, messages.length, 'older', messages.map(message => message.id), folderOverride);
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');
@@ -346,7 +478,7 @@ export async function enqueueTelegramTagDownload(botClient: TelegramClient, requ
error: result.failed > 0 ? `${result.failed} 个文件下载失败` : null,
finished_at: new Date(),
});
return { jobId, tag, ...result };
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;
+114 -141
View File
@@ -10,7 +10,8 @@ import { storageManager } from '../services/storage.js';
import { getTelegramUserClient, isTelegramUserClientReady } from './telegramUserClient.js';
import { getSetting } from '../utils/settings.js';
import { isAuthenticatedAsync } from './telegramState.js';
import { formatBytes, getTypeEmoji, getFileType, getMimeTypeFromFilename, sanitizeFilename } from '../utils/telegramUtils.js';
import { formatBytes, getTypeEmoji, getFileType, sanitizeFilename } from '../utils/telegramUtils.js';
import { extractFileInfo, getDownloadableMedia, getEstimatedFileSize, isTelegramPhotoMedia, type TelegramFileInfo } from '../utils/telegramMedia.js';
import {
MSG,
buildUploadSuccess,
@@ -65,10 +66,25 @@ function appendTelegramDebugLog(line: string) {
}
export interface TelegramDownloadMessageRef {
id: number;
source?: Api.TypeEntityLike | string;
origin?: 'channel' | 'comment';
channelPostId?: number;
fileInfo?: TelegramFileInfo;
totalSize?: number;
message?: Api.Message;
}
interface DownloadableMessageRef {
id: number;
fileInfo: { fileName: string; mimeType: string };
sourceKey: string;
sourceEntity: Api.TypeEntityLike | string;
origin: 'channel' | 'comment';
channelPostId?: number;
fileInfo: TelegramFileInfo;
totalSize: number;
message?: Api.Message;
}
function clampDownloadWorkers(value: unknown): number {
@@ -1247,119 +1263,6 @@ const mediaGroupQueues = new Map<string, MediaGroupQueue>();
// 多文件上传处理延迟(毫秒),等待所有文件消息到达
const MEDIA_GROUP_DELAY = 1500;
function getDownloadableMedia(message: Api.Message): any | null {
if (!message.media) return null;
const media: any = message.media;
if (message.document || message.photo || message.video || message.audio || message.voice || message.sticker) {
return message.media;
}
if (media.document || media.photo) {
return media.document || media.photo;
}
if (media.webpage?.document || media.webpage?.photo) {
return media.webpage.document || media.webpage.photo;
}
return null;
}
function isTelegramPhotoMedia(media: any): boolean {
const inner = media?.photo || media;
return media?.className === 'MessageMediaPhoto' || inner?.className === 'Photo' || Boolean(inner?.sizes);
}
// 获取文件预估大小
function getEstimatedFileSize(message: Api.Message): number {
const media = getDownloadableMedia(message);
if (isTelegramPhotoMedia(media)) {
return 0;
}
const document = (media as any)?.document || media;
if (document?.size) {
return Number(document.size) || 0;
}
return 0;
}
// 进度条函数已移至 telegramMessages.ts (generateProgressBar)
function getDocumentFilename(document: any, fallback: string): string {
const fileNameAttr = document.attributes?.find((a: any) => a.className === 'DocumentAttributeFilename') as any;
return fileNameAttr?.fileName || fallback;
}
// 提取文件信息
function extractFileInfo(message: Api.Message): { fileName: string; mimeType: string } | null {
const downloadableMedia = getDownloadableMedia(message);
if (!downloadableMedia) return null;
let fileName = 'unknown';
let mimeType = 'application/octet-stream';
try {
if (message.document) {
const doc = message.document as Api.Document;
const fileNameAttr = doc.attributes?.find((a: any) => a.className === 'DocumentAttributeFilename') as any;
fileName = fileNameAttr?.fileName || `file_${message.id}`;
mimeType = doc.mimeType || getMimeTypeFromFilename(fileName);
// 如果是音频/视频但没有文件名属性,尝试根据类型生成
if (fileName.startsWith('file_')) {
const videoAttr = doc.attributes?.find((a: any) => a.className === 'DocumentAttributeVideo');
const audioAttr = doc.attributes?.find((a: any) => a.className === 'DocumentAttributeAudio');
if (videoAttr) fileName = `video_${message.id}.mp4`;
else if (audioAttr) fileName = `audio_${message.id}.mp3`;
}
} else if (message.photo) {
const date = new Date();
const timestamp = date.toISOString().replace(/[-:T]/g, '').slice(0, 14);
fileName = `Img_${timestamp}.jpg`;
mimeType = 'image/jpeg';
} else if (message.video) {
const video = message.video as Api.Document;
const fileNameAttr = video.attributes?.find((a: any) => a.className === 'DocumentAttributeFilename') as any;
fileName = fileNameAttr?.fileName || `video_${message.id}.mp4`;
mimeType = video.mimeType || 'video/mp4';
} else if (message.audio) {
const audio = message.audio as Api.Document;
const fileNameAttr = audio.attributes?.find((a: any) => a.className === 'DocumentAttributeFilename') as any;
fileName = fileNameAttr?.fileName || `audio_${message.id}.mp3`;
mimeType = audio.mimeType || 'audio/mpeg';
} else if (message.voice) {
fileName = `voice_${message.id}.ogg`;
mimeType = 'audio/ogg';
} else if (message.sticker) {
fileName = `sticker_${message.id}.webp`;
mimeType = 'image/webp';
} else {
const media = message.media as any;
if (media.document && media.document instanceof Api.Document) {
const doc = media.document;
const fileNameAttr = doc.attributes?.find((a: any) => a.className === 'DocumentAttributeFilename') as any;
fileName = fileNameAttr?.fileName || `file_${message.id}`;
mimeType = doc.mimeType || getMimeTypeFromFilename(fileName);
} else {
const document = (downloadableMedia as any).document || (downloadableMedia as any);
const photo = (downloadableMedia as any).photo || (downloadableMedia as any);
if (document?.className === 'Document' || document?.attributes) {
fileName = getDocumentFilename(document, `file_${message.id}`);
mimeType = document.mimeType || getMimeTypeFromFilename(fileName);
} else if (photo?.className === 'Photo' || photo?.sizes) {
const date = new Date((message.date || Math.floor(Date.now() / 1000)) * 1000);
const timestamp = date.toISOString().replace(/[-:T]/g, '').slice(0, 14);
fileName = `Img_${timestamp}_${message.id}.jpg`;
mimeType = 'image/jpeg';
} else {
return null;
}
}
}
} catch (e) {
console.error('🤖 提取文件信息出错:', e);
return null;
}
return { fileName: sanitizeFilename(fileName), mimeType };
}
// 下载并保存文件
async function downloadAndSaveFile(
@@ -1884,6 +1787,26 @@ export async function getTelegramDownloadPreview(messages: Api.Message[]): Promi
};
}
function sourceKeyForDownloadRef(source: Api.TypeEntityLike | string): string {
if (typeof source === 'string') return source;
const anySource: any = source as any;
const id = anySource?.channelId || anySource?.chatId || anySource?.userId || anySource?.id;
if (id !== undefined && id !== null) {
return `${anySource?.className || 'peer'}:${id.toString()}`;
}
return JSON.stringify(source);
}
function normalizeTelegramDownloadRefs(
refs: TelegramDownloadMessageRef[] | undefined,
defaultSourceEntity: Api.TypeEntityLike | string,
): TelegramDownloadMessageRef[] | undefined {
if (!refs) return undefined;
return refs
.filter(ref => ref.id > 0)
.map(ref => ({ ...ref, source: ref.source || defaultSourceEntity }));
}
export async function downloadTelegramChannelRange(
botClient: TelegramClient,
requestMessage: Api.Message,
@@ -1893,6 +1816,7 @@ export async function downloadTelegramChannelRange(
direction: 'older' | 'newer' = 'older',
explicitIds?: number[],
folderOverride?: string | null,
explicitRefs?: TelegramDownloadMessageRef[],
): 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()) {
@@ -1900,7 +1824,11 @@ export async function downloadTelegramChannelRange(
}
const safeLimit = Math.max(1, Math.floor(limit || TG_BATCH_DEFAULT_LIMIT));
const ids = explicitIds?.filter(id => id > 0) || Array.from({ length: safeLimit }, (_, index) => (
const sourceEntity = source.startsWith('@') || /^-?\d+$/.test(source) || /^https?:\/\//i.test(source)
? source
: `@${source}`;
const normalizedExplicitRefs = normalizeTelegramDownloadRefs(explicitRefs, sourceEntity);
const ids = normalizedExplicitRefs?.map(ref => ref.id) || explicitIds?.filter(id => id > 0) || Array.from({ length: safeLimit }, (_, index) => (
direction === 'newer' ? startMessageId + index : startMessageId - index
)).filter(id => id > 0);
@@ -1908,10 +1836,6 @@ export async function downloadTelegramChannelRange(
throw new Error('起始消息 ID 无效');
}
const sourceEntity = source.startsWith('@') || /^-?\d+$/.test(source) || /^https?:\/\//i.test(source)
? source
: `@${source}`;
const chatId = requestMessage.chatId;
if (!chatId) {
throw new Error('无法识别当前 Bot 会话');
@@ -1929,25 +1853,54 @@ export async function downloadTelegramChannelRange(
const failedMessageIds: number[] = [];
const skippedMessageIds: number[] = [];
for (let offset = 0; offset < ids.length; offset += TG_LARGE_TASK_SEGMENT_SIZE) {
const scanIds = ids.slice(offset, offset + TG_LARGE_TASK_SEGMENT_SIZE);
const scanMessages = await userClient.getMessages(sourceEntity as any, { ids: scanIds });
const returnedIds = new Set<number>();
for (const sourceMessage of scanMessages) {
if (!sourceMessage) continue;
returnedIds.add(sourceMessage.id);
const fileInfo = extractFileInfo(sourceMessage);
if (normalizedExplicitRefs) {
for (const ref of normalizedExplicitRefs) {
const refSource = ref.source || sourceEntity;
const fileInfo = ref.fileInfo;
if (!fileInfo) {
skipped += 1;
skippedMessageIds.push(sourceMessage.id);
skippedMessageIds.push(ref.id);
continue;
}
downloadableRefs.push({ id: sourceMessage.id, fileInfo, totalSize: getEstimatedFileSize(sourceMessage) });
downloadableRefs.push({
id: ref.id,
sourceKey: sourceKeyForDownloadRef(refSource),
sourceEntity: refSource,
origin: ref.origin || 'channel',
channelPostId: ref.channelPostId,
fileInfo,
totalSize: ref.totalSize || 0,
message: ref.message,
});
}
for (const requestedId of scanIds) {
if (!returnedIds.has(requestedId)) {
skipped += 1;
skippedMessageIds.push(requestedId);
} else {
for (let offset = 0; offset < ids.length; offset += TG_LARGE_TASK_SEGMENT_SIZE) {
const scanIds = ids.slice(offset, offset + TG_LARGE_TASK_SEGMENT_SIZE);
const scanMessages = await userClient.getMessages(sourceEntity as any, { ids: scanIds });
const returnedIds = new Set<number>();
for (const sourceMessage of scanMessages) {
if (!sourceMessage) continue;
returnedIds.add(sourceMessage.id);
const fileInfo = extractFileInfo(sourceMessage);
if (!fileInfo) {
skipped += 1;
skippedMessageIds.push(sourceMessage.id);
continue;
}
downloadableRefs.push({
id: sourceMessage.id,
sourceKey: sourceKeyForDownloadRef(sourceEntity),
sourceEntity,
origin: 'channel',
fileInfo,
totalSize: getEstimatedFileSize(sourceMessage),
});
}
for (const requestedId of scanIds) {
if (!returnedIds.has(requestedId)) {
skipped += 1;
skippedMessageIds.push(requestedId);
}
}
}
}
@@ -1958,7 +1911,7 @@ export async function downloadTelegramChannelRange(
if (taskFolderOverride !== undefined) {
taskResolvedStorageFolder = taskFolderOverride;
} else if (firstRef) {
const firstMessage = (await userClient.getMessages(sourceEntity as any, { ids: [firstRef.id] }))[0] as Api.Message | undefined;
const firstMessage = (await userClient.getMessages(firstRef.sourceEntity as any, { ids: [firstRef.id] }))[0] as Api.Message | undefined;
if (firstMessage) {
const chatName = await getTelegramChatName(firstMessage);
const storageRules = await getStoragePathRules();
@@ -2014,16 +1967,36 @@ export async function downloadTelegramChannelRange(
const segment = downloadableRefs.slice(offset, offset + TG_LARGE_TASK_SEGMENT_SIZE);
const segmentBytes = segment.reduce((sum, item) => sum + (item.totalSize || 0), 0);
await waitForDiskWatermark(segmentBytes);
const segmentIds = segment.map(item => item.id);
const segmentMessages = await userClient.getMessages(sourceEntity as any, { ids: segmentIds });
const segmentMessageById = new Map<number, Api.Message>();
for (const segmentMessage of segmentMessages) {
if (segmentMessage) segmentMessageById.set(segmentMessage.id, segmentMessage as Api.Message);
const segmentMessagesBySource = new Map<string, Map<number, Api.Message>>();
const refsBySource = new Map<string, DownloadableMessageRef[]>();
for (const item of segment) {
const items = refsBySource.get(item.sourceKey) || [];
items.push(item);
refsBySource.set(item.sourceKey, items);
}
for (const [sourceKey, sourceItems] of refsBySource) {
const preloadedMessageById = new Map<number, Api.Message>();
const missingSourceItems: DownloadableMessageRef[] = [];
for (const sourceItem of sourceItems) {
if (sourceItem.message) {
preloadedMessageById.set(sourceItem.id, sourceItem.message);
} else {
missingSourceItems.push(sourceItem);
}
}
if (missingSourceItems.length > 0) {
const segmentIds = missingSourceItems.map(item => item.id);
const segmentMessages = await userClient.getMessages(sourceItems[0].sourceEntity as any, { ids: segmentIds });
for (const segmentMessage of segmentMessages) {
if (segmentMessage) preloadedMessageById.set(segmentMessage.id, segmentMessage as Api.Message);
}
}
segmentMessagesBySource.set(sourceKey, preloadedMessageById);
}
await Promise.all(segment.map(async (item) => {
const { fileName, mimeType } = item.fileInfo;
const message = segmentMessageById.get(item.id);
const message = segmentMessagesBySource.get(item.sourceKey)?.get(item.id);
if (!message) {
skipped += 1;
failed += 1;
+114
View File
@@ -0,0 +1,114 @@
import { Api } from 'telegram';
import { getMimeTypeFromFilename, sanitizeFilename } from './telegramUtils.js';
export interface TelegramFileInfo {
fileName: string;
mimeType: string;
}
export function getDownloadableMedia(message: Api.Message): any | null {
if (!message.media) return null;
const media: any = message.media;
if (message.sticker) return null;
if (message.document || message.photo || message.video || message.audio || message.voice) {
return message.media;
}
if (media.document || media.photo) {
return media.document || media.photo;
}
if (media.webpage?.document || media.webpage?.photo) {
return media.webpage.document || media.webpage.photo;
}
return null;
}
export function isTelegramPhotoMedia(media: any): boolean {
const inner = media?.photo || media;
return media?.className === 'MessageMediaPhoto' || inner?.className === 'Photo' || Boolean(inner?.sizes);
}
export function getEstimatedFileSize(message: Api.Message): number {
const media = getDownloadableMedia(message);
if (isTelegramPhotoMedia(media)) {
return 0;
}
const document = (media as any)?.document || media;
if (document?.size) {
return Number(document.size) || 0;
}
return 0;
}
function getDocumentFilename(document: any, fallback: string): string {
const fileNameAttr = document.attributes?.find((a: any) => a.className === 'DocumentAttributeFilename') as any;
return fileNameAttr?.fileName || fallback;
}
export function extractFileInfo(message: Api.Message): TelegramFileInfo | null {
const downloadableMedia = getDownloadableMedia(message);
if (!downloadableMedia) return null;
let fileName = 'unknown';
let mimeType = 'application/octet-stream';
try {
if (message.document) {
const doc = message.document as Api.Document;
const fileNameAttr = doc.attributes?.find((a: any) => a.className === 'DocumentAttributeFilename') as any;
fileName = fileNameAttr?.fileName || `file_${message.id}`;
mimeType = doc.mimeType || getMimeTypeFromFilename(fileName);
if (fileName.startsWith('file_')) {
const videoAttr = doc.attributes?.find((a: any) => a.className === 'DocumentAttributeVideo');
const audioAttr = doc.attributes?.find((a: any) => a.className === 'DocumentAttributeAudio');
if (videoAttr) fileName = `video_${message.id}.mp4`;
else if (audioAttr) fileName = `audio_${message.id}.mp3`;
}
} else if (message.photo) {
const date = new Date((message.date || Math.floor(Date.now() / 1000)) * 1000);
const timestamp = date.toISOString().replace(/[-:T]/g, '').slice(0, 14);
fileName = `Img_${timestamp}_${message.id}.jpg`;
mimeType = 'image/jpeg';
} else if (message.video) {
const video = message.video as Api.Document;
const fileNameAttr = video.attributes?.find((a: any) => a.className === 'DocumentAttributeFilename') as any;
fileName = fileNameAttr?.fileName || `video_${message.id}.mp4`;
mimeType = video.mimeType || 'video/mp4';
} else if (message.audio) {
const audio = message.audio as Api.Document;
const fileNameAttr = audio.attributes?.find((a: any) => a.className === 'DocumentAttributeFilename') as any;
fileName = fileNameAttr?.fileName || `audio_${message.id}.mp3`;
mimeType = audio.mimeType || 'audio/mpeg';
} else if (message.voice) {
fileName = `voice_${message.id}.ogg`;
mimeType = 'audio/ogg';
} else {
const media = message.media as any;
if (media.document && media.document instanceof Api.Document) {
const doc = media.document;
const fileNameAttr = doc.attributes?.find((a: any) => a.className === 'DocumentAttributeFilename') as any;
fileName = fileNameAttr?.fileName || `file_${message.id}`;
mimeType = doc.mimeType || getMimeTypeFromFilename(fileName);
} else {
const document = (downloadableMedia as any).document || (downloadableMedia as any);
const photo = (downloadableMedia as any).photo || (downloadableMedia as any);
if (document?.className === 'Document' || document?.attributes) {
fileName = getDocumentFilename(document, `file_${message.id}`);
mimeType = document.mimeType || getMimeTypeFromFilename(fileName);
} else if (photo?.className === 'Photo' || photo?.sizes) {
const date = new Date((message.date || Math.floor(Date.now() / 1000)) * 1000);
const timestamp = date.toISOString().replace(/[-:T]/g, '').slice(0, 14);
fileName = `Img_${timestamp}_${message.id}.jpg`;
mimeType = 'image/jpeg';
} else {
return null;
}
}
}
} catch (e) {
console.error('🤖 提取文件信息出错:', e);
return null;
}
return { fileName: sanitizeFilename(fileName), mimeType };
}
+2
View File
@@ -215,6 +215,8 @@ export function buildHelp(): string {
` /tg_download — 按日期/标签下载频道文件`,
` /tg_download date <频道> <开始日期> <结束日期> — 按日期下载`,
` /tg_download tag <频道> <#标签> — 按标签下载`,
` /tg_download 向导中可选择“频道 + 评论区”;开启后只下载评论区里的文件/图片/视频/音频,文字评论会忽略`,
` 评论区每个帖子默认最多扫描 ${process.env.TELEGRAM_COMMENTS_MAX_PER_POST || '200'} 条评论,可用 TELEGRAM_COMMENTS_MAX_PER_POST 调整`,
` /download_workers — 设置单文件分片并发`,
` /file_concurrency — 设置同时下载文件数`,
` /duplicate_mode — 设置重复文件处理`,
+1
View File
@@ -51,6 +51,7 @@ services:
- TELEGRAM_RATE_MAX=${TELEGRAM_RATE_MAX:-30}
- TELEGRAM_HEAVY_RATE_WINDOW_MS=${TELEGRAM_HEAVY_RATE_WINDOW_MS:-600000}
- TELEGRAM_HEAVY_RATE_MAX=${TELEGRAM_HEAVY_RATE_MAX:-5}
- TELEGRAM_COMMENTS_MAX_PER_POST=${TELEGRAM_COMMENTS_MAX_PER_POST:-200}
- TOTP_SECRET=${TOTP_SECRET:-}
volumes:
- file-storage:/data