feat: Telegram Bridge 图片接收能力

新增 telegram-media.ts 模块,处理 Telegram 图片下载与 base64 转换:
- selectOptimalPhoto: 选最优尺寸(长边 ≥ 1568px 的最小版本)
- downloadPhoto/downloadDocumentImage: getFile API → 下载 → base64,3 次重试 + 指数退避
- document 类型图片超 20MB 预检拒绝,提示用户按照片发送
- kill switch: bridge_telegram_image_enabled 设置项

扩展 telegram-adapter.ts poll loop:
- TelegramUpdate 类型增加 caption/photo/document/media_group_id
- 单图消息直接下载入队
- 相册消息 500ms 防抖合并后批量下载,构建单条 InboundMessage
- offset 安全:相册使用 max(updateIds) 保证不丢消息

透传 attachments 至 Claude vision pipeline:
- types.ts: InboundMessage 新增 attachments 可选字段
- conversation-engine.ts: processMessage 新增 files 参数,传入 streamClaude
- bridge-manager.ts: 允许纯图片消息通过,透传 attachments

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
7418andClaude Opus 4.6 committed 2026-03-02 12:37:13 +08:00
1 parent c9f1b62809
commit 570d26de3a
5 files changed
+597 -31

No files matched your search

+294 -24
View File
@@ -12,8 +12,17 @@ import type {
OutboundMessage,
SendResult,
} from '../types';
import type { FileAttachment } from '@/types';
import { BaseChannelAdapter, registerAdapterFactory } from '../channel-adapter';
import { callTelegramApi, escapeHtml } from './telegram-utils';
import {
isImageEnabled,
downloadPhoto,
downloadDocumentImage,
isSupportedImageMime,
inferMimeType,
} from './telegram-media';
import type { TelegramPhotoSize, TelegramDocument } from './telegram-media';
import { getChannelOffset, setChannelOffset, insertAuditLog } from '../../db';
import { getSetting } from '../../db';
@@ -34,6 +43,10 @@ interface TelegramUpdate {
chat: { id: number; first_name?: string; title?: string; username?: string };
from?: { id: number; first_name: string; username?: string };
text?: string;
caption?: string;
photo?: TelegramPhotoSize[];
document?: TelegramDocument;
media_group_id?: string;
date: number;
};
callback_query?: {
@@ -52,6 +65,19 @@ interface QueueEntry {
resolve: () => void;
}
/** Media group debounce buffer entry for album messages. */
interface MediaGroupBufferEntry {
updates: TelegramUpdate[];
updateIds: number[];
timer: ReturnType<typeof setTimeout>;
chatId: string;
userId: string;
displayName: string;
}
/** Debounce window for media group messages (ms). */
const MEDIA_GROUP_DEBOUNCE_MS = 500;
export class TelegramAdapter extends BaseChannelAdapter {
readonly channelType: ChannelType = 'telegram';
@@ -60,6 +86,7 @@ export class TelegramAdapter extends BaseChannelAdapter {
private queue: InboundMessage[] = [];
private waiters: Array<(msg: InboundMessage | null) => void> = [];
private typingIntervals = new Map<string, ReturnType<typeof setInterval>>();
private mediaGroupBuffers = new Map<string, MediaGroupBufferEntry>();
/** Committed offset — the highest update_id that has been safely enqueued or skipped. */
private committedOffset = 0;
@@ -120,6 +147,12 @@ export class TelegramAdapter extends BaseChannelAdapter {
}
this.typingIntervals.clear();
// Clear media group debounce timers
for (const [, entry] of this.mediaGroupBuffers) {
clearTimeout(entry.timer);
}
this.mediaGroupBuffers.clear();
console.log('[telegram-adapter] Stopped');
}
@@ -461,10 +494,11 @@ export class TelegramAdapter extends BaseChannelAdapter {
// Answer callback to dismiss the loading state
this.answerCallback(cb.id).catch(() => {});
} else if (update.message?.text) {
} else if (update.message) {
const m = update.message;
const chatId = String(m.chat.id);
const userId = m.from ? String(m.from.id) : chatId;
const displayName = m.from?.username || m.from?.first_name || chatId;
if (!this.isAuthorized(userId, chatId)) {
console.warn('[telegram-adapter] Unauthorized message from userId:', userId, 'chatId:', chatId);
@@ -472,32 +506,51 @@ export class TelegramAdapter extends BaseChannelAdapter {
continue;
}
const msg: InboundMessage = {
messageId: String(m.message_id),
address: {
channelType: 'telegram',
chatId,
userId,
displayName: m.from?.username || m.from?.first_name || chatId,
},
text: m.text!,
timestamp: m.date * 1000,
raw: update,
updateId: update.update_id,
};
const hasPhoto = m.photo && m.photo.length > 0;
const hasDocImage = m.document && this.isDocumentImage(m.document);
const hasMedia = hasPhoto || hasDocImage;
// Audit log
try {
insertAuditLog({
channelType: 'telegram',
chatId,
direction: 'inbound',
if (hasMedia && isImageEnabled()) {
if (m.media_group_id) {
// Album message — buffer for debounce, advance fetchOffset immediately
this.bufferMediaGroup(m.media_group_id, update, chatId, userId, displayName);
// Don't markUpdateProcessed yet — offset will be committed on flush
} else {
// Single image message — process immediately
await this.processSingleImageMessage(update, chatId, userId, displayName);
}
} else if (m.text) {
// Pure text message — original logic
const msg: InboundMessage = {
messageId: String(m.message_id),
summary: m.text!.slice(0, 200),
});
} catch { /* best effort */ }
address: {
channelType: 'telegram',
chatId,
userId,
displayName,
},
text: m.text,
timestamp: m.date * 1000,
raw: update,
updateId: update.update_id,
};
this.enqueue(msg);
// Audit log
try {
insertAuditLog({
channelType: 'telegram',
chatId,
direction: 'inbound',
messageId: String(m.message_id),
summary: m.text.slice(0, 200),
});
} catch { /* best effort */ }
this.enqueue(msg);
} else {
// Unhandled message type (sticker, voice, etc.) — skip
this.markUpdateProcessed(update.update_id);
}
} else {
// Unhandled update type — still safe to advance past it
this.markUpdateProcessed(update.update_id);
@@ -517,6 +570,223 @@ export class TelegramAdapter extends BaseChannelAdapter {
}
}
}
/**
* Check if a Telegram document is a supported image type.
*/
private isDocumentImage(doc: TelegramDocument): boolean {
if (doc.mime_type && isSupportedImageMime(doc.mime_type)) return true;
if (doc.file_name) {
const mime = inferMimeType(doc.file_name);
if (mime && isSupportedImageMime(mime)) return true;
}
return false;
}
/**
* Process a single image message (no media_group_id).
* Downloads the image and enqueues a message with attachments.
*/
private async processSingleImageMessage(
update: TelegramUpdate,
chatId: string,
userId: string,
displayName: string,
): Promise<void> {
const m = update.message!;
const token = this.botToken;
if (!token) {
this.markUpdateProcessed(update.update_id);
return;
}
const attachments: FileAttachment[] = [];
let rejectedReason: string | undefined;
if (m.photo && m.photo.length > 0) {
const att = await downloadPhoto(token, m.photo, String(m.message_id));
if (att) attachments.push(att);
} else if (m.document) {
const result = await downloadDocumentImage(token, m.document, String(m.message_id));
if (result.attachment) {
attachments.push(result.attachment);
} else if (result.rejected) {
rejectedReason = result.rejected;
}
}
const text = m.caption || m.text || '';
const hasContent = attachments.length > 0 || text.trim();
if (!hasContent) {
// Nothing usable — if rejected, we'll still mark processed
if (rejectedReason && rejectedReason !== 'unsupported_type') {
// Enqueue a rejection notification as text
const msg: InboundMessage = {
messageId: String(m.message_id),
address: { channelType: 'telegram', chatId, userId, displayName },
text: `[Image rejected] ${rejectedReason}`,
timestamp: m.date * 1000,
raw: update,
updateId: update.update_id,
};
this.enqueue(msg);
} else {
this.markUpdateProcessed(update.update_id);
}
return;
}
const summary = attachments.length > 0
? `[${attachments.length} image(s)] ${text.slice(0, 150)}`
: text.slice(0, 200);
// Audit log
try {
insertAuditLog({
channelType: 'telegram',
chatId,
direction: 'inbound',
messageId: String(m.message_id),
summary,
});
} catch { /* best effort */ }
const msg: InboundMessage = {
messageId: String(m.message_id),
address: { channelType: 'telegram', chatId, userId, displayName },
text: text,
timestamp: m.date * 1000,
raw: update,
updateId: update.update_id,
attachments: attachments.length > 0 ? attachments : undefined,
};
this.enqueue(msg);
}
/**
* Buffer a media group update for debounced processing.
* Resets the 500ms timer on each new update in the same group.
*/
private bufferMediaGroup(
mediaGroupId: string,
update: TelegramUpdate,
chatId: string,
userId: string,
displayName: string,
): void {
const existing = this.mediaGroupBuffers.get(mediaGroupId);
if (existing) {
// Add to existing buffer, reset timer
clearTimeout(existing.timer);
existing.updates.push(update);
existing.updateIds.push(update.update_id);
existing.timer = setTimeout(() => this.flushMediaGroup(mediaGroupId), MEDIA_GROUP_DEBOUNCE_MS);
} else {
// New buffer
const timer = setTimeout(() => this.flushMediaGroup(mediaGroupId), MEDIA_GROUP_DEBOUNCE_MS);
this.mediaGroupBuffers.set(mediaGroupId, {
updates: [update],
updateIds: [update.update_id],
timer,
chatId,
userId,
displayName,
});
}
}
/**
* Flush a media group buffer — download all images and enqueue a single message.
*/
private async flushMediaGroup(mediaGroupId: string): Promise<void> {
const entry = this.mediaGroupBuffers.get(mediaGroupId);
if (!entry) return;
this.mediaGroupBuffers.delete(mediaGroupId);
const token = this.botToken;
if (!token) {
// Can't download — mark all as processed
for (const uid of entry.updateIds) {
this.markUpdateProcessed(uid);
}
this.persistCommittedOffset();
return;
}
const attachments: FileAttachment[] = [];
let caption = '';
let firstMessageId = '';
let firstDate = 0;
// Download all images in the group
for (const update of entry.updates) {
const m = update.message!;
if (!firstMessageId) {
firstMessageId = String(m.message_id);
firstDate = m.date;
}
// Use caption from whichever update has it (Telegram only sends caption on one)
if (m.caption && !caption) {
caption = m.caption;
}
if (m.photo && m.photo.length > 0) {
const att = await downloadPhoto(token, m.photo, String(m.message_id));
if (att) attachments.push(att);
} else if (m.document && this.isDocumentImage(m.document)) {
const result = await downloadDocumentImage(token, m.document, String(m.message_id));
if (result.attachment) attachments.push(result.attachment);
}
}
const text = caption;
const hasContent = attachments.length > 0 || text.trim();
if (!hasContent) {
// All downloads failed and no caption — mark all processed
for (const uid of entry.updateIds) {
this.markUpdateProcessed(uid);
}
this.persistCommittedOffset();
return;
}
const summary = attachments.length > 0
? `[Album: ${attachments.length} image(s)] ${text.slice(0, 150)}`
: text.slice(0, 200);
try {
insertAuditLog({
channelType: 'telegram',
chatId: entry.chatId,
direction: 'inbound',
messageId: firstMessageId,
summary,
});
} catch { /* best effort */ }
// Use the max updateId so acknowledgeUpdate advances offset past all buffered updates
const maxUpdateId = Math.max(...entry.updateIds);
const msg: InboundMessage = {
messageId: firstMessageId,
address: {
channelType: 'telegram',
chatId: entry.chatId,
userId: entry.userId,
displayName: entry.displayName,
},
text,
timestamp: firstDate * 1000,
updateId: maxUpdateId,
attachments: attachments.length > 0 ? attachments : undefined,
};
this.enqueue(msg);
}
}
// Self-register so bridge-manager can create TelegramAdapter via the registry.
+285
View File
@@ -0,0 +1,285 @@
/**
* Telegram Media — download and process images from Telegram messages.
*
* Handles photo[] size selection, file download via Bot API, base64 conversion,
* and document-type image validation. Produces FileAttachment objects that plug
* directly into the existing streamClaude vision pipeline.
*/
import type { FileAttachment } from '@/types';
import { getSetting } from '../../db';
const TELEGRAM_API = 'https://api.telegram.org';
/** Claude vision optimal long-edge size (px). */
const OPTIMAL_LONG_EDGE = 1568;
/** Default max image size in bytes (20 MB). */
const DEFAULT_MAX_IMAGE_SIZE = 20 * 1024 * 1024;
/** Max retry attempts for download. */
const MAX_RETRIES = 3;
/** Supported image MIME types for Claude vision. */
const SUPPORTED_IMAGE_MIMES = new Set([
'image/jpeg',
'image/png',
'image/gif',
'image/webp',
]);
// ── Telegram Photo Types ─────────────────────────────────────
export interface TelegramPhotoSize {
file_id: string;
file_unique_id: string;
width: number;
height: number;
file_size?: number;
}
export interface TelegramDocument {
file_id: string;
file_unique_id: string;
file_name?: string;
mime_type?: string;
file_size?: number;
}
// ── Public API ───────────────────────────────────────────────
/**
* Check whether the Telegram image feature is enabled.
*/
export function isImageEnabled(): boolean {
const setting = getSetting('bridge_telegram_image_enabled');
// Default to true if not explicitly set to 'false'
return setting !== 'false';
}
/**
* Get the configured max image size in bytes.
*/
function getMaxImageSize(): number {
const setting = getSetting('bridge_telegram_max_image_size');
if (setting) {
const parsed = parseInt(setting, 10);
if (!isNaN(parsed) && parsed > 0) return parsed;
}
return DEFAULT_MAX_IMAGE_SIZE;
}
/**
* Check if a MIME type is a supported image format.
*/
export function isSupportedImageMime(mime: string): boolean {
return SUPPORTED_IMAGE_MIMES.has(mime.toLowerCase());
}
/**
* Infer MIME type from a file path/name extension.
* Returns undefined if the extension is not a recognized image type.
*/
export function inferMimeType(filePath: string): string | undefined {
const ext = filePath.split('.').pop()?.toLowerCase();
switch (ext) {
case 'jpg':
case 'jpeg':
return 'image/jpeg';
case 'png':
return 'image/png';
case 'gif':
return 'image/gif';
case 'webp':
return 'image/webp';
default:
return undefined;
}
}
/**
* Select the optimal photo size from Telegram's photo[] array.
*
* Strategy: sort by long edge ascending, pick the smallest version whose
* long edge >= OPTIMAL_LONG_EDGE (1568px, Claude vision optimal). If none
* are large enough, take the largest available.
*/
export function selectOptimalPhoto(photos: TelegramPhotoSize[]): TelegramPhotoSize {
if (photos.length === 1) return photos[0];
// Sort by long edge ascending
const sorted = [...photos].sort((a, b) => {
const aLong = Math.max(a.width, a.height);
const bLong = Math.max(b.width, b.height);
return aLong - bLong;
});
// Find smallest version with long edge >= optimal
for (const photo of sorted) {
const longEdge = Math.max(photo.width, photo.height);
if (longEdge >= OPTIMAL_LONG_EDGE) {
return photo;
}
}
// None large enough — take the largest
return sorted[sorted.length - 1];
}
/**
* Download a photo from Telegram's photo[] array and return a FileAttachment.
*
* Selects the optimal size, calls getFile API, downloads the binary,
* and converts to base64.
*
* Returns null if download fails after retries.
*/
export async function downloadPhoto(
botToken: string,
photos: TelegramPhotoSize[],
messageId: string,
): Promise<FileAttachment | null> {
const selected = selectOptimalPhoto(photos);
return downloadFileById(botToken, selected.file_id, messageId, selected.file_size);
}
/**
* Download a document-type image from Telegram.
*
* Pre-checks file_size against the max limit before initiating download.
* Returns null with a reason string if the document is too large or not a supported image.
*/
export async function downloadDocumentImage(
botToken: string,
doc: TelegramDocument,
messageId: string,
): Promise<{ attachment: FileAttachment | null; rejected?: string }> {
// Check MIME type
const mime = doc.mime_type || inferMimeType(doc.file_name || '');
if (!mime || !isSupportedImageMime(mime)) {
return { attachment: null, rejected: 'unsupported_type' };
}
// Pre-check file size before downloading
const maxSize = getMaxImageSize();
if (doc.file_size && doc.file_size > maxSize) {
const sizeMB = (doc.file_size / (1024 * 1024)).toFixed(1);
const limitMB = (maxSize / (1024 * 1024)).toFixed(0);
return {
attachment: null,
rejected: `Image file too large (${sizeMB} MB). Max allowed: ${limitMB} MB. Please send as a photo instead of a file.`,
};
}
const attachment = await downloadFileById(botToken, doc.file_id, messageId, doc.file_size);
return { attachment };
}
// ── Internal ─────────────────────────────────────────────────
/**
* Download a file by its Telegram file_id.
* Calls getFile → download URL → binary → base64 FileAttachment.
* Retries up to MAX_RETRIES with exponential backoff.
*/
async function downloadFileById(
botToken: string,
fileId: string,
messageId: string,
expectedSize?: number,
): Promise<FileAttachment | null> {
const maxSize = getMaxImageSize();
for (let attempt = 1; attempt <= MAX_RETRIES; attempt++) {
try {
// Step 1: Get file path from Telegram
const getFileUrl = `${TELEGRAM_API}/bot${botToken}/getFile`;
const getFileRes = await fetch(getFileUrl, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ file_id: fileId }),
signal: AbortSignal.timeout(15_000),
});
const getFileData = await getFileRes.json();
if (!getFileData.ok || !getFileData.result?.file_path) {
console.warn(`[telegram-media] getFile failed for ${fileId}:`, getFileData.description);
if (attempt < MAX_RETRIES) {
await sleep(1000 * Math.pow(2, attempt - 1));
continue;
}
return null;
}
const filePath: string = getFileData.result.file_path;
const fileSize: number | undefined = getFileData.result.file_size;
// Pre-check size from API response
if (fileSize && fileSize > maxSize) {
console.warn(`[telegram-media] File too large: ${fileSize} bytes (max ${maxSize})`);
return null;
}
// Step 2: Download the file
const downloadUrl = `${TELEGRAM_API}/file/bot${botToken}/${filePath}`;
const downloadRes = await fetch(downloadUrl, {
signal: AbortSignal.timeout(60_000),
});
if (!downloadRes.ok) {
console.warn(`[telegram-media] Download failed: HTTP ${downloadRes.status}`);
if (attempt < MAX_RETRIES) {
await sleep(1000 * Math.pow(2, attempt - 1));
continue;
}
return null;
}
// Check Content-Length header
const contentLength = downloadRes.headers.get('content-length');
if (contentLength && parseInt(contentLength, 10) > maxSize) {
console.warn(`[telegram-media] Content-Length exceeds max: ${contentLength}`);
return null;
}
// Step 3: Read buffer and validate actual size
const buffer = Buffer.from(await downloadRes.arrayBuffer());
if (buffer.length > maxSize) {
console.warn(`[telegram-media] Downloaded buffer too large: ${buffer.length} bytes`);
return null;
}
// Step 4: Determine MIME type
const mime = inferMimeType(filePath) || 'image/jpeg';
// Step 5: Convert to base64 and build FileAttachment
const base64 = buffer.toString('base64');
const fileName = filePath.split('/').pop() || `image_${messageId}`;
const attachment: FileAttachment = {
id: `tg-${messageId}-${fileId.slice(0, 8)}`,
name: fileName,
type: mime,
size: buffer.length,
data: base64,
};
return attachment;
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
console.warn(`[telegram-media] Download attempt ${attempt}/${MAX_RETRIES} failed:`, errMsg);
if (attempt < MAX_RETRIES) {
await sleep(1000 * Math.pow(2, attempt - 1));
continue;
}
return null;
}
}
return null;
}
function sleep(ms: number): Promise<void> {
return new Promise(resolve => setTimeout(resolve, ms));
}
+8 -4
View File
@@ -328,7 +328,8 @@ async function handleMessage(
}
const rawText = msg.text.trim();
if (!rawText) { ack(); return; }
const hasAttachments = msg.attachments && msg.attachments.length > 0;
if (!rawText && !hasAttachments) { ack(); return; }
// Check for IM commands (before sanitization — commands are validated individually)
if (rawText.startsWith('/')) {
@@ -350,7 +351,7 @@ async function handleMessage(
});
}
if (!text) { ack(); return; }
if (!text && !hasAttachments) { ack(); return; }
// Regular message — route to conversation engine
const binding = router.resolve(msg.address);
@@ -366,7 +367,10 @@ async function handleMessage(
try {
// Pass permission callback so requests are forwarded to IM immediately
// during streaming (the stream blocks until permission is resolved).
const result = await engine.processMessage(binding, text, async (perm) => {
// Use text or empty string for image-only messages (prompt is still required by streamClaude)
const promptText = text || (hasAttachments ? 'Describe this image.' : '');
const result = await engine.processMessage(binding, promptText, async (perm) => {
await broker.forwardPermissionRequest(
adapter,
msg.address,
@@ -376,7 +380,7 @@ async function handleMessage(
binding.codepilotSessionId,
perm.suggestions,
);
}, taskAbort.signal);
}, taskAbort.signal, hasAttachments ? msg.attachments : undefined);
// Send response text
if (result.responseText) {
+8 -3
View File
@@ -7,7 +7,7 @@
*/
import type { ChannelBinding } from './types';
import type { SSEEvent, TokenUsage, MessageContentBlock } from '@/types';
import type { SSEEvent, TokenUsage, MessageContentBlock, FileAttachment } from '@/types';
import { streamClaude } from '../claude-client';
import {
addMessage,
@@ -61,6 +61,7 @@ export async function processMessage(
text: string,
onPermissionRequest?: OnPermissionRequest,
abortSignal?: AbortSignal,
files?: FileAttachment[],
): Promise<ConversationResult> {
const sessionId = binding.codepilotSessionId;
@@ -86,8 +87,11 @@ export async function processMessage(
}, 60_000);
try {
// Save user message
addMessage(sessionId, 'user', text);
// Save user message (with image indicator if attachments present)
const userMsgText = files && files.length > 0
? `[${files.length} image(s) attached] ${text}`
: text;
addMessage(sessionId, 'user', userMsgText);
// Resolve provider
const session = getSession(sessionId);
@@ -139,6 +143,7 @@ export async function processMessage(
permissionMode,
provider: resolvedProvider,
conversationHistory: historyMsgs,
files,
onRuntimeStatusChange: (status: string) => {
try { setSessionRuntimeStatus(sessionId, status); } catch { /* best effort */ }
},
+2
View File
@@ -49,6 +49,8 @@ export interface InboundMessage {
raw?: unknown;
/** Adapter-specific update ID for deferred offset acknowledgement */
updateId?: number;
/** File attachments (images, documents) from the IM channel */
attachments?: import('@/types').FileAttachment[];
}
/** Outbound message to send to an IM channel */