tgstate-rust / src /routes /api_files.rs
bhgi's picture
Deploy tgstate-rust
71217a7
Raw
History Blame Contribute Delete
20.8 kB
use std::sync::Arc;
use axum::body::Body;
use axum::extract::{Path, Query, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::{delete, get, post};
use axum::{Json, Router};
use futures::StreamExt;
use serde::Deserialize;
use crate::config;
use crate::database;
use crate::error::http_error;
use crate::state::AppState;
use crate::telegram::service::TelegramService;
#[derive(Deserialize)]
pub struct DownloadQuery {
download: Option<String>,
}
#[derive(Deserialize)]
pub struct BatchDeleteRequest {
file_ids: Vec<String>,
}
fn get_telegram_service(state: &AppState) -> Result<TelegramService, impl IntoResponse> {
let app_settings = config::get_app_settings(&state.settings, &state.db_pool);
let token = app_settings
.get("BOT_TOKEN")
.and_then(|v| v.as_deref())
.unwrap_or("")
.to_string();
let channel = app_settings
.get("CHANNEL_NAME")
.and_then(|v| v.as_deref())
.unwrap_or("")
.to_string();
if token.is_empty() || channel.is_empty() {
return Err(http_error(
StatusCode::SERVICE_UNAVAILABLE,
"Bot 未配置",
"bot_not_configured",
));
}
Ok(TelegramService::new(
token,
channel,
state.http_client.clone(),
))
}
fn guess_content_type(filename: &str) -> String {
let mime = mime_guess::from_path(filename).first_or_octet_stream();
let mime_str = mime.to_string();
// Add charset for text types
if mime_str.starts_with("text/") && !mime_str.contains("charset") {
format!("{}; charset=utf-8", mime_str)
} else {
mime_str
}
}
fn content_disposition(filename: &str, force_download: bool) -> String {
// Allow-list of extensions that are safe to render inline on the
// download host. We deliberately EXCLUDE executable/active content such
// as `svg`, `html`, `htm`, `xml`, `js` and `css`, which browsers will
// execute scripts from. Those are served as attachments.
let preview_extensions = [
"jpg", "jpeg", "png", "gif", "webp", "bmp", "ico", "tiff", "mp4", "webm", "ogg", "mp3",
"wav", "flac", "pdf", "txt", "json", "csv", "md", "log",
];
let ext = filename.rsplit('.').next().unwrap_or("").to_lowercase();
let is_inline = !force_download && preview_extensions.contains(&ext.as_str());
let encoded_name =
percent_encoding::utf8_percent_encode(filename, percent_encoding::NON_ALPHANUMERIC)
.to_string();
if is_inline {
format!("inline; filename*=UTF-8''{}", encoded_name)
} else {
format!("attachment; filename*=UTF-8''{}", encoded_name)
}
}
fn chunk_download_failed_response(chunk_id: &str) -> Response {
// Log the chunk_id for operators; do NOT include it in the response body
// because it reveals internal manifest structure to clients.
tracing::error!("chunk download failed: {}", chunk_id);
http_error(
StatusCode::BAD_GATEWAY,
"文件下载失败",
"chunk_download_failed",
)
.into_response()
}
async fn serve_file(
state: &AppState,
tg_service: &TelegramService,
file_id: &str,
filename: &str,
headers: &HeaderMap,
force_download: bool,
is_head: bool,
) -> Response {
// Parse composite file_id "message_id:real_file_id"
let real_file_id = if let Some(pos) = file_id.find(':') {
&file_id[pos + 1..]
} else {
file_id
};
// Get download URL
let download_url = match tg_service.get_download_url(real_file_id).await {
Ok(Some(url)) => url,
Ok(None) => {
return http_error(StatusCode::NOT_FOUND, "文件未找到或链接已过期", "not_found")
.into_response()
}
Err(e) => {
tracing::error!("获取下载链接失败: {}", e);
return http_error(
StatusCode::SERVICE_UNAVAILABLE,
"无法连接到 Telegram",
"tg_error",
)
.into_response();
}
};
let client = &state.http_client;
// Peek first 128 bytes to check if manifest
let peek_resp = match client
.get(&download_url)
.header("Range", "bytes=0-127")
.send()
.await
{
Ok(r) => r,
Err(e) => {
tracing::error!("下载失败: {}", e);
return http_error(StatusCode::BAD_GATEWAY, "无法下载文件", "download_error")
.into_response();
}
};
let peek_bytes = match peek_resp.bytes().await {
Ok(b) => b,
Err(e) => {
tracing::error!("读取文件失败: {}", e);
return http_error(StatusCode::BAD_GATEWAY, "读取文件失败", "read_error")
.into_response();
}
};
// Check if manifest
if peek_bytes.starts_with(b"tgstate-blob\n") {
// Download full manifest
let full_resp = match client.get(&download_url).send().await {
Ok(r) => r,
Err(e) => {
tracing::error!("下载清单失败: {}", e);
return http_error(
StatusCode::BAD_GATEWAY,
"下载文件失败",
"download_error",
)
.into_response();
}
};
let manifest_bytes = match full_resp.bytes().await {
Ok(b) => b,
Err(e) => {
tracing::error!("读取清单失败: {}", e);
return http_error(
StatusCode::BAD_GATEWAY,
"读取文件失败",
"read_error",
)
.into_response();
}
};
let manifest_str = String::from_utf8_lossy(&manifest_bytes);
let lines: Vec<&str> = manifest_str.lines().collect();
if lines.len() < 3 {
return http_error(
StatusCode::INTERNAL_SERVER_ERROR,
"清单文件格式错误",
"manifest_error",
)
.into_response();
}
let original_filename = lines[1];
let chunk_ids: Vec<String> = lines[2..].iter().map(|s| s.to_string()).collect();
let ct = guess_content_type(original_filename);
let cd = content_disposition(original_filename, force_download);
if is_head {
return Response::builder()
.header("Content-Type", ct)
.header("Content-Disposition", cd)
.header("Accept-Ranges", "bytes")
.header("X-Content-Type-Options", "nosniff")
.body(Body::empty())
.unwrap();
}
let mut chunk_urls = Vec::with_capacity(chunk_ids.len());
for chunk_composite in &chunk_ids {
let real_id = if let Some(pos) = chunk_composite.find(':') {
chunk_composite[pos + 1..].to_string()
} else {
chunk_composite.clone()
};
let url = match tg_service.get_download_url(&real_id).await {
Ok(Some(u)) => u,
_ => return chunk_download_failed_response(chunk_composite),
};
chunk_urls.push((chunk_composite.clone(), url));
}
// Stream chunks with retry
let tg = tg_service.clone();
let http = client.clone();
let stream = async_stream::stream! {
for (chunk_composite, url) in chunk_urls {
let resp = match http.get(&url).send().await {
Ok(r) if r.status().is_success() => r,
_ => {
// Retry once after 1 second
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
let real_id = if let Some(pos) = chunk_composite.find(':') {
chunk_composite[pos + 1..].to_string()
} else {
chunk_composite.clone()
};
let retry_url = match tg.get_download_url(&real_id).await {
Ok(Some(u)) => u,
_ => {
yield Err::<bytes::Bytes, std::io::Error>(std::io::Error::other(format!(
"Failed to refresh chunk URL: {}",
chunk_composite
)));
return;
}
};
match http.get(&retry_url).send().await {
Ok(r) if r.status().is_success() => r,
_ => {
yield Err::<bytes::Bytes, std::io::Error>(std::io::Error::other(format!(
"Chunk download retry failed: {}",
chunk_composite
)));
return;
}
}
}
};
let mut stream = resp.bytes_stream();
while let Some(chunk) = stream.next().await {
match chunk {
Ok(bytes) => yield Ok::<_, std::io::Error>(bytes),
Err(e) => {
yield Err::<bytes::Bytes, std::io::Error>(std::io::Error::other(format!(
"Chunk stream error for {}: {}",
chunk_composite, e
)));
return;
}
}
}
}
};
return Response::builder()
.header("Content-Type", ct)
.header("Content-Disposition", cd)
.header("Accept-Ranges", "bytes")
.header("X-Content-Type-Options", "nosniff")
.body(Body::from_stream(stream))
.unwrap();
}
// Regular file - stream from Telegram
let ct = guess_content_type(filename);
let cd = content_disposition(filename, force_download);
// Handle Range request - proxy Range header to Telegram
if let Some(range_header) = headers.get("range").and_then(|v| v.to_str().ok()) {
let range_resp = match client
.get(&download_url)
.header("Range", range_header)
.send()
.await
{
Ok(r) => r,
Err(e) => {
tracing::error!("Range 请求失败: {}", e);
return http_error(StatusCode::BAD_GATEWAY, "无法下载文件", "download_error")
.into_response();
}
};
let status = range_resp.status();
let mut builder = Response::builder()
.status(status)
.header("Content-Type", &ct)
.header("Content-Disposition", &cd)
.header("Accept-Ranges", "bytes")
.header("X-Content-Type-Options", "nosniff");
// Forward Content-Range and Content-Length from upstream
if let Some(cr) = range_resp.headers().get("content-range") {
builder = builder.header("Content-Range", cr);
}
if let Some(cl) = range_resp.headers().get("content-length") {
builder = builder.header("Content-Length", cl);
}
let stream = range_resp.bytes_stream();
return builder
.body(Body::from_stream(stream.map(|r| {
r.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))
})))
.unwrap();
}
// Full file - stream
let full_resp = match client.get(&download_url).send().await {
Ok(r) => r,
Err(e) => {
tracing::error!("下载失败: {}", e);
return http_error(StatusCode::BAD_GATEWAY, "无法下载文件", "download_error")
.into_response();
}
};
let mut builder = Response::builder()
.header("Content-Type", ct)
.header("Content-Disposition", cd)
.header("Accept-Ranges", "bytes")
.header("X-Content-Type-Options", "nosniff");
if let Some(cl) = full_resp.headers().get("content-length") {
builder = builder.header("Content-Length", cl);
}
if is_head {
return builder.body(Body::empty()).unwrap();
}
let stream = full_resp.bytes_stream();
builder
.body(Body::from_stream(stream.map(|r| {
r.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))
})))
.unwrap()
}
async fn download_file_short(
State(state): State<Arc<AppState>>,
Path(identifier): Path<String>,
Query(query): Query<DownloadQuery>,
headers: HeaderMap,
) -> Response {
// Validate identifier format
if identifier.is_empty()
|| identifier.len() > 128
|| identifier.chars().any(|c| c.is_control() || c == '\0')
{
return http_error(StatusCode::BAD_REQUEST, "无效的文件标识", "invalid_id").into_response();
}
let tg_service = match get_telegram_service(&state) {
Ok(s) => s,
Err(e) => return e.into_response(),
};
let meta = database::get_file_by_id(&state.db_pool, &identifier);
match meta {
Ok(Some(f)) => {
let force_download = query
.download
.as_deref()
.map_or(false, |v| v == "1" || v == "true");
let is_head = false; // Will be handled by axum method routing
serve_file(
&state,
&tg_service,
&f.file_id,
&f.filename,
&headers,
force_download,
is_head,
)
.await
}
_ => http_error(StatusCode::NOT_FOUND, "文件未找到", "not_found").into_response(),
}
}
async fn download_file_short_head(
State(state): State<Arc<AppState>>,
Path(identifier): Path<String>,
Query(query): Query<DownloadQuery>,
headers: HeaderMap,
) -> Response {
let tg_service = match get_telegram_service(&state) {
Ok(s) => s,
Err(e) => return e.into_response(),
};
let meta = database::get_file_by_id(&state.db_pool, &identifier);
match meta {
Ok(Some(f)) => {
let force_download = query
.download
.as_deref()
.map_or(false, |v| v == "1" || v == "true");
serve_file(
&state,
&tg_service,
&f.file_id,
&f.filename,
&headers,
force_download,
true,
)
.await
}
_ => http_error(StatusCode::NOT_FOUND, "文件未找到", "not_found").into_response(),
}
}
async fn download_file_legacy(
State(state): State<Arc<AppState>>,
Path((file_id, filename)): Path<(String, String)>,
Query(query): Query<DownloadQuery>,
headers: HeaderMap,
) -> Response {
let tg_service = match get_telegram_service(&state) {
Ok(s) => s,
Err(e) => return e.into_response(),
};
let force_download = query
.download
.as_deref()
.map_or(false, |v| v == "1" || v == "true");
serve_file(
&state,
&tg_service,
&file_id,
&filename,
&headers,
force_download,
false,
)
.await
}
async fn download_file_legacy_head(
State(state): State<Arc<AppState>>,
Path((file_id, filename)): Path<(String, String)>,
Query(query): Query<DownloadQuery>,
headers: HeaderMap,
) -> Response {
let tg_service = match get_telegram_service(&state) {
Ok(s) => s,
Err(e) => return e.into_response(),
};
let force_download = query
.download
.as_deref()
.map_or(false, |v| v == "1" || v == "true");
serve_file(
&state,
&tg_service,
&file_id,
&filename,
&headers,
force_download,
true,
)
.await
}
async fn get_files_list(State(state): State<Arc<AppState>>) -> impl IntoResponse {
let files = database::get_all_files(&state.db_pool).unwrap_or_default();
Json(files)
}
async fn delete_file(
State(state): State<Arc<AppState>>,
Path(file_id): Path<String>,
) -> impl IntoResponse {
let tg_service = match get_telegram_service(&state) {
Ok(s) => s,
Err(e) => return e.into_response(),
};
tracing::info!("正在删除文件: {}", file_id);
let result = tg_service.delete_file_with_chunks(&file_id).await;
if result.main_message_deleted {
let db_deleted = database::delete_file_metadata(&state.db_pool, &file_id).unwrap_or(false);
let db_status = if db_deleted {
"deleted"
} else {
"not_found_in_db"
};
if result.failed_chunks.is_empty() {
return Json(serde_json::json!({
"status": "ok",
"message": format!("文件 {} 已删除。", file_id),
"details": {
"db": db_status,
"tg": result,
}
}))
.into_response();
} else {
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({
"status": "error",
"code": "partial_failure",
"message": "部分分块删除失败",
"details": result,
})),
)
.into_response();
}
}
// TG deletion failed, try force-delete from DB
let force_deleted = database::delete_file_metadata(&state.db_pool, &file_id).unwrap_or(false);
if force_deleted {
return Json(serde_json::json!({
"status": "ok",
"message": format!("文件 {} 已从数据库删除(Telegram 删除失败)。", file_id),
"details": result,
}))
.into_response();
}
(
StatusCode::BAD_REQUEST,
Json(serde_json::json!({
"status": "error",
"code": "delete_failed",
"message": "删除失败",
"details": result,
})),
)
.into_response()
}
async fn batch_delete_files(
State(state): State<Arc<AppState>>,
Json(payload): Json<BatchDeleteRequest>,
) -> impl IntoResponse {
// Cap the number of IDs per request so callers cannot abuse batch delete
// to issue an unbounded sequence of Telegram API requests.
if payload.file_ids.len() > crate::constants::BATCH_DELETE_MAX {
return http_error(
StatusCode::BAD_REQUEST,
"批量删除数量超过上限",
"too_many_items",
)
.into_response();
}
let tg_service = match get_telegram_service(&state) {
Ok(s) => s,
Err(e) => return e.into_response(),
};
let mut deleted = Vec::new();
let mut failed = Vec::new();
for fid in &payload.file_ids {
let result = tg_service.delete_file_with_chunks(fid).await;
if result.main_message_deleted {
database::delete_file_metadata(&state.db_pool, fid).ok();
deleted.push(fid.clone());
} else {
// Try force delete from DB
if database::delete_file_metadata(&state.db_pool, fid).unwrap_or(false) {
deleted.push(fid.clone());
} else {
failed.push(fid.clone());
}
}
}
Json(serde_json::json!({
"status": "completed",
"deleted": deleted,
"failed": failed,
}))
.into_response()
}
pub fn router() -> Router<Arc<AppState>> {
Router::new()
.route("/api/files", get(get_files_list))
.route("/api/files/:file_id", delete(delete_file))
.route("/api/batch_delete", post(batch_delete_files))
.route(
"/d/:file_id/:filename",
get(download_file_legacy).head(download_file_legacy_head),
)
.route(
"/d/:identifier",
get(download_file_short).head(download_file_short_head),
)
}
#[cfg(test)]
mod tests {
use super::chunk_download_failed_response;
use axum::http::StatusCode;
#[test]
fn manifest_chunk_failure_returns_bad_gateway() {
let response = chunk_download_failed_response("chunk-1");
assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
}
}