use std::time::Duration; use crate::constants; use crate::database::{self, DbPool}; use crate::events::{build_file_event, BroadcastEventBus}; use crate::telegram::service::TelegramService; use crate::telegram::types::*; pub async fn run_bot_polling( bot_token: String, channel_name: String, db_pool: DbPool, event_bus: BroadcastEventBus, base_url: String, http_client: reqwest::Client, mut shutdown_rx: tokio::sync::oneshot::Receiver<()>, ) { // Reuse the shared `AppState::http_client` instead of constructing a new // one on every bot restart. Apart from avoiding per-restart connection // pool churn, this also ensures Telegram requests issued from the bot // polling loop honour the same timeouts / TLS config as every other // Telegram call in the process. let client = http_client; let tg_service = TelegramService::new( bot_token.clone(), channel_name.clone(), client.clone(), ); let mut offset: i64 = 0; // Drop pending updates (equivalent to drop_pending_updates=True in python-telegram-bot) match get_updates(&client, &bot_token, -1, 0).await { Ok(updates) => { if let Some(last) = updates.last() { offset = last.update_id + 1; tracing::info!("跳过 {} 个待处理更新", updates.len()); } } Err(e) => { tracing::warn!("清除待处理更新失败: {}", e); } } tracing::info!("Bot 轮询已启动"); loop { tokio::select! { _ = &mut shutdown_rx => { tracing::info!("Bot 轮询收到关闭信号"); break; } result = get_updates(&client, &bot_token, offset, constants::BOT_POLL_TIMEOUT_SECS as i64) => { match result { Ok(updates) => { for update in updates { offset = update.update_id + 1; process_update( &update, &tg_service, &channel_name, &db_pool, &event_bus, &base_url, ).await; } } Err(e) => { tracing::error!("getUpdates 失败: {}", e); tokio::time::sleep(Duration::from_secs(5)).await; } } } } } } async fn get_updates( client: &reqwest::Client, bot_token: &str, offset: i64, timeout: i64, ) -> Result, String> { let url = format!("https://api.telegram.org/bot{}/getUpdates", bot_token); let resp = client .post(&url) .json(&serde_json::json!({ "offset": offset, "timeout": timeout, "allowed_updates": ["message", "channel_post", "edited_message", "edited_channel_post"] })) .timeout(Duration::from_secs(timeout as u64 + 10)) .send() .await .map_err(|e| format!("Request error: {}", e))?; let data: TelegramResponse> = resp .json() .await .map_err(|e| format!("Parse error: {}", e))?; if data.ok { Ok(data.result.unwrap_or_default()) } else { Err(format!( "getUpdates error: {}", data.description.unwrap_or_default() )) } } async fn process_update( update: &Update, tg_service: &TelegramService, channel_name: &str, db_pool: &DbPool, event_bus: &BroadcastEventBus, base_url: &str, ) { // Handle new file (message or channel_post) let message = update.message.as_ref().or(update.channel_post.as_ref()); if let Some(msg) = message { if msg.document.is_some() || msg.photo.is_some() { handle_new_file(msg, channel_name, db_pool, event_bus).await; } // Handle "get" reply if let Some(text) = &msg.text { if text.trim().to_lowercase() == "get" && msg.reply_to_message.is_some() { handle_get_reply(msg, tg_service, base_url).await; } } } // Handle edited/deleted messages let edited = update .edited_message .as_ref() .or(update.edited_channel_post.as_ref()); if let Some(msg) = edited { if msg.text.is_none() && msg.document.is_none() && msg.photo.is_none() { handle_deleted_message(msg, db_pool, event_bus).await; } } } async fn handle_new_file( message: &Message, channel_name: &str, db_pool: &DbPool, event_bus: &BroadcastEventBus, ) { // Check source let chat = &message.chat; let is_allowed = if channel_name.starts_with('@') { chat.username .as_deref() .map_or(false, |u| u == channel_name.trim_start_matches('@')) } else { chat.id.to_string() == channel_name }; if !is_allowed { return; } // Extract file info let (file_id, file_name, file_size) = if let Some(doc) = &message.document { ( doc.file_id.clone(), doc.file_name.clone().unwrap_or_else(|| format!("file_{}", message.message_id)), doc.file_size.unwrap_or(0), ) } else if let Some(photos) = &message.photo { if let Some(photo) = photos.last() { ( photo.file_id.clone(), format!("photo_{}.jpg", message.message_id), photo.file_size.unwrap_or(0), ) } else { return; } } else { return; }; // Skip large files and manifests. Anything larger than the per-chunk // Telegram limit (`TELEGRAM_CHUNK_SIZE`) has to have been uploaded via // the multi-part / manifest flow, which the bot-side ingestion path // does not know how to reconstruct. if file_size as usize >= constants::TELEGRAM_CHUNK_SIZE || file_name.ends_with(".manifest") { return; } let composite_id = format!("{}:{}", message.message_id, file_id); match database::add_file_metadata(db_pool, &file_name, &composite_id, file_size) { Ok(short_id) => { let upload_date = message .date .map(|ts| { chrono::DateTime::from_timestamp(ts, 0) .map(|dt| dt.to_rfc3339()) .unwrap_or_default() }) .unwrap_or_default(); let event = build_file_event( "add", &composite_id, Some(&file_name), Some(file_size), Some(&upload_date), Some(&short_id), ); event_bus.publish(serde_json::to_string(&event).unwrap_or_default()); } Err(e) => { tracing::error!("添加文件元数据失败: {}", e); } } } async fn handle_get_reply( message: &Message, tg_service: &TelegramService, base_url: &str, ) { let replied = match &message.reply_to_message { Some(r) => r, None => return, }; let (file_id, file_name) = if let Some(doc) = &replied.document { ( doc.file_id.clone(), doc.file_name.clone().unwrap_or_else(|| format!("file_{}", replied.message_id)), ) } else if let Some(photos) = &replied.photo { if let Some(photo) = photos.last() { ( photo.file_id.clone(), format!("photo_{}.jpg", replied.message_id), ) } else { let _ = send_message( &tg_service.client, &tg_service.bot_token, message.chat.id, "请回复到一个文件/图片消息", ) .await; return; } } else { let _ = send_message( &tg_service.client, &tg_service.bot_token, message.chat.id, "请回复到一个文件/图片消息", ) .await; return; }; let composite_id = format!("{}:{}", replied.message_id, file_id); let mut final_filename = file_name.clone(); // Check if manifest if file_name.ends_with(".manifest") { match tg_service .try_get_manifest_original_filename(&file_id) .await { Ok(name) => final_filename = name, Err(e) => { let _ = send_message( &tg_service.client, &tg_service.bot_token, message.chat.id, &format!("错误:解析清单文件失败:{}", e), ) .await; return; } } } let encoded = percent_encoding::utf8_percent_encode( &final_filename, percent_encoding::NON_ALPHANUMERIC, ); let file_path = format!("/d/{}/{}", composite_id, encoded); let text = if !base_url.is_empty() { let base = base_url.trim_end_matches('/'); format!( "这是 '{}' 的下载链接:\n{}{}", final_filename, base, file_path ) } else { format!( "这是 '{}' 的下载路径 (请自行拼接域名):\n`{}`", final_filename, file_path ) }; let _ = send_message( &tg_service.client, &tg_service.bot_token, message.chat.id, &text, ) .await; } async fn handle_deleted_message( message: &Message, db_pool: &DbPool, event_bus: &BroadcastEventBus, ) { let message_id = message.message_id; match database::delete_file_by_message_id(db_pool, message_id) { Ok(Some(file_id)) => { let event = build_file_event("delete", &file_id, None, None, None, None); event_bus.publish(serde_json::to_string(&event).unwrap_or_default()); } _ => {} } } async fn send_message( client: &reqwest::Client, bot_token: &str, chat_id: i64, text: &str, ) -> Result<(), String> { let url = format!("https://api.telegram.org/bot{}/sendMessage", bot_token); client .post(&url) .json(&serde_json::json!({ "chat_id": chat_id, "text": text })) .send() .await .map_err(|e| e.to_string())?; Ok(()) }