use super::HttpState; use crate::core::config::{self, AppConfig}; use crate::core::dispatch; use crate::core::message::{ChatInput, ChatRequest, ChatResponse, Destination, Envelope, Message, PlayerCommand, WifiCommand}; use bytes::Buf; use futures_util::{SinkExt, StreamExt, TryStreamExt}; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use serde_json::Value; use std::convert::Infallible; use std::path::{Path, PathBuf}; use std::sync::{mpsc, Arc}; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, Instant}; use tokio::io::AsyncWriteExt; use warp::http::StatusCode; use warp::multipart::FormData; use warp::{Filter, Reply}; const MAX_UPLOAD_FILE_SIZE: u64 = 100 * 1024 * 1024; static UPLOAD_TMP_COUNTER: AtomicU64 = AtomicU64::new(0); #[derive(Deserialize)] struct WifiConnectRequest { ssid: String, password: String, } #[derive(Default, Deserialize)] struct WifiApStartRequest { ssid: Option, password: Option, } #[derive(Default, Deserialize)] struct BleStartRequest { device_name: Option, } #[derive(Serialize)] struct ApiMessage<'a> { status: &'a str, message: String, } #[derive(Serialize)] struct VideoFileInfo { name: String, size: u64, } #[derive(Serialize)] struct FileEntry { name: String, is_dir: bool, size: u64, } #[derive(Deserialize)] struct FileMoveRequest { from_dir: String, from_path: String, to_dir: String, to_path: String, } #[derive(Serialize)] struct WifiStatusResponse { connected: bool, ssid: String, ip: String, } #[derive(Serialize)] struct BleStatusResponse { running: bool, embedded: bool, device_name: String, } #[derive(Serialize)] struct PlaylistSnapshot { playlist: Vec, current_index: usize, } #[derive(Serialize)] struct AppInfoResponse { version: String, apk_available: bool, apk_size: Option, download_url: &'static str, } pub(crate) fn build_routes( tx: mpsc::Sender, state: Arc, ) -> impl Filter + Clone { // 使用 boxed() 分段避免 warp 递归类型溢出 let core_api = status_route(Arc::clone(&state)) .or(play_route(tx.clone())) .or(pause_route(tx.clone())) .or(next_route(tx.clone())) .or(previous_route(tx.clone())) .or(goto_route(tx.clone(), Arc::clone(&state))) .or(playlist_route(Arc::clone(&state))) .or(scene_route(tx.clone())) .or(trigger_route(tx.clone())) .or(config_get_route(Arc::clone(&state))) .or(config_display_route(Arc::clone(&state))) .or(config_update_route(tx.clone(), Arc::clone(&state))) .or(config_available_route(Arc::clone(&state))) .or(config_switch_route(tx.clone(), Arc::clone(&state))) .boxed(); let media_api = video_list_route(Arc::clone(&state)) .or(video_upload_route(Arc::clone(&state))) .or(video_delete_route(Arc::clone(&state))) .or(app_info_route(Arc::clone(&state))) .or(wifi_status_route(tx.clone(), Arc::clone(&state))) .or(wifi_scan_route(tx.clone(), Arc::clone(&state))) .or(wifi_connect_route(tx.clone(), Arc::clone(&state))) .or(wifi_ap_start_route(tx.clone(), Arc::clone(&state))) .or(wifi_ap_stop_route(tx.clone(), Arc::clone(&state))) .or(ble_start_route(Arc::clone(&state))) .or(ble_stop_route()) .or(ble_status_route(Arc::clone(&state))) .boxed(); let plugin_api = plugins_list_route(Arc::clone(&state)) .or(plugin_detail_route(Arc::clone(&state))) .or(plugin_enable_route(tx.clone())) .or(plugin_disable_route(tx.clone())) .or(plugin_rollback_route(tx.clone())) .or(plugin_switch_route(tx.clone())) .or(plugin_install_route(tx.clone())) .or(plugin_check_updates_route(tx.clone())) .boxed(); let file_api = file_list_route(Arc::clone(&state)) .or(file_upload_route(Arc::clone(&state))) .or(file_download_route(Arc::clone(&state))) .or(file_delete_route(Arc::clone(&state))) .or(file_move_route(Arc::clone(&state))) .or(file_mkdir_route(Arc::clone(&state))) .boxed(); let ai_api = chat_text_route(Arc::clone(&state)) .or(chat_audio_route(Arc::clone(&state))) .or(models_list_route(Arc::clone(&state))) .or(model_download_route(Arc::clone(&state))) .or(model_switch_route(Arc::clone(&state))) .or(model_delete_route(Arc::clone(&state))) .boxed(); let api = core_api.or(media_api).or(plugin_api).or(file_api).or(ai_api); root_route() .or(download_route(Arc::clone(&state))) .or(ws_route(tx.clone(), Arc::clone(&state))) .or(api) .with( warp::cors() .allow_any_origin() .allow_headers(["content-type"]) .allow_methods(["GET", "POST", "DELETE", "OPTIONS"]), ) } fn root_route() -> impl Filter + Clone { warp::path::end() .and(warp::get()) .map(|| warp::reply::html(WEB_UI_HTML)) .or( warp::path("index.html") .and(warp::path::end()) .and(warp::get()) .map(|| warp::reply::html(WEB_UI_HTML)), ) } fn ws_route( tx: mpsc::Sender, state: Arc, ) -> impl Filter + Clone { warp::path("ws") .and(warp::path::end()) .and(warp::ws()) .and(with_tx(tx)) .and(with_state(state)) .map(|ws: warp::ws::Ws, tx: mpsc::Sender, state: Arc| { ws.on_upgrade(move |socket| websocket_session(socket, tx, state)) }) } fn download_route( state: Arc, ) -> impl Filter + Clone { warp::path!("download" / String) .and(warp::path::end()) .and(warp::get()) .and(with_state(state)) .and_then(|filename: String, state: Arc| async move { let safe_name = sanitize_filename(&filename); if safe_name.is_empty() || safe_name != filename { return Ok::<_, Infallible>(warp::reply::with_status( warp::reply::html("无效的文件名".to_string()), StatusCode::BAD_REQUEST, ) .into_response()); } let full = downloads_dir(state.config().as_ref()).join(&safe_name); if !full.is_file() { return Ok(warp::reply::with_status( warp::reply::html("文件不存在".to_string()), StatusCode::NOT_FOUND, ) .into_response()); } Ok(read_attachment_response(&full)) }) } fn status_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "status") .and(warp::get()) .and(with_state(state)) .and_then(|state: Arc| async move { Ok::<_, Infallible>(json_response(StatusCode::OK, &state.player_status())) }) } fn play_route( tx: mpsc::Sender, ) -> impl Filter + Clone { warp::path!("api" / "play") .and(warp::post()) .and(with_tx(tx)) .and_then(|tx| async move { send_video_command(tx, Message::PlayerCommand(PlayerCommand::Play), "开始播放").await }) } fn pause_route( tx: mpsc::Sender, ) -> impl Filter + Clone { warp::path!("api" / "pause") .and(warp::post()) .and(with_tx(tx)) .and_then(|tx| async move { send_video_command(tx, Message::PlayerCommand(PlayerCommand::Pause), "已暂停").await }) } fn next_route( tx: mpsc::Sender, ) -> impl Filter + Clone { warp::path!("api" / "next") .and(warp::post()) .and(with_tx(tx)) .and_then(|tx| async move { send_video_command( tx, Message::PlayerCommand(PlayerCommand::Next), "切换到下一个视频", ) .await }) } fn previous_route( tx: mpsc::Sender, ) -> impl Filter + Clone { warp::path!("api" / "previous") .and(warp::post()) .and(with_tx(tx)) .and_then(|tx| async move { send_video_command( tx, Message::PlayerCommand(PlayerCommand::Previous), "切换到上一个视频", ) .await }) } fn goto_route( tx: mpsc::Sender, state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "goto" / usize) .and(warp::post()) .and(with_tx(tx)) .and(with_state(state)) .and_then(|index, tx, state: Arc| async move { if index >= state.player_status().playlist_length { return Ok::<_, Infallible>(error_json(StatusCode::BAD_REQUEST, "无效的视频索引")); } send_video_command( tx, Message::PlayerCommand(PlayerCommand::Goto(index)), format!("跳转到视频 {index}"), ) .await }) } fn playlist_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "playlist") .and(warp::get()) .and(with_state(state)) .and_then(|state: Arc| async move { let config = state.config(); let player_status = state.player_status(); let snapshot = PlaylistSnapshot { playlist: config.playlist.clone(), current_index: player_status.current_index, }; Ok::<_, Infallible>(json_response(StatusCode::OK, &snapshot)) }) } fn scene_route( tx: mpsc::Sender, ) -> impl Filter + Clone { warp::path!("api" / "scene" / String) .and(warp::post()) .and(with_tx(tx)) .and_then(|name: String, tx| async move { send_video_command( tx, Message::PlayerCommand(PlayerCommand::ChangeScene(name.clone())), format!("切换到场景: {name}"), ) .await }) } fn trigger_route( tx: mpsc::Sender, ) -> impl Filter + Clone { let with_value = warp::path!("api" / "trigger" / String / String) .and(warp::post()) .and(with_tx(tx.clone())) .and_then(|name: String, value: String, tx| async move { send_video_command( tx, Message::Trigger { name: name.clone(), value: value.clone(), }, format!("触发器 '{name}' 已发送,值: {value}"), ) .await }); let without_value = warp::path!("api" / "trigger" / String) .and(warp::post()) .and(with_tx(tx)) .and_then(|name: String, tx| async move { send_video_command( tx, Message::Trigger { name: name.clone(), value: String::new(), }, format!("触发器 '{name}' 已发送"), ) .await }); with_value.or(without_value) } fn config_get_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "config") .and(warp::get()) .and(with_state(state)) .and_then(|state: Arc| async move { let config = state.config(); Ok::<_, Infallible>(json_response(StatusCode::OK, config.as_ref())) }) } fn config_display_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "config" / "display") .and(warp::get()) .and(with_state(state)) .and_then(|state: Arc| async move { let config = state.config(); Ok::<_, Infallible>(json_response(StatusCode::OK, &config.display)) }) } fn config_update_route( tx: mpsc::Sender, state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "config") .and(warp::post()) .and(warp::body::content_length_limit(1024 * 64)) .and(warp::body::bytes()) .and(with_tx(tx)) .and(with_state(state)) .and_then(handle_config_update) } fn config_available_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "config" / "available") .and(warp::get()) .and(with_state(state)) .and_then(|state: Arc| async move { let config = state.config(); let configs_dir = &config.source_dir; let mut files: Vec = Vec::new(); if let Ok(entries) = std::fs::read_dir(configs_dir) { for entry in entries.flatten() { let name = entry.file_name().to_string_lossy().into_owned(); if name.ends_with(".json") { files.push(name); } } } files.sort(); // 标记当前活跃的配置 let active = config.source_path .file_name() .map(|n| n.to_string_lossy().into_owned()) .unwrap_or_default(); let result = serde_json::json!({ "configs": files, "active": active, }); Ok::<_, Infallible>(json_response(StatusCode::OK, &result)) }) } #[derive(Deserialize)] struct ConfigSwitchRequest { filename: String, } fn config_switch_route( tx: mpsc::Sender, state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "config" / "switch") .and(warp::post()) .and(warp::body::json::()) .and(with_tx(tx)) .and(with_state(state)) .and_then(|req: ConfigSwitchRequest, tx: mpsc::Sender, state: Arc| async move { // 验证文件名安全 if req.filename.contains("..") || req.filename.contains('/') || req.filename.contains('\\') { return Ok::<_, Infallible>(error_json(StatusCode::BAD_REQUEST, "文件名不合法")); } if !req.filename.ends_with(".json") { return Ok::<_, Infallible>(error_json(StatusCode::BAD_REQUEST, "只支持 .json 配置文件")); } let config = state.config(); let target_path = config.source_dir.join(&req.filename); if !target_path.exists() { return Ok(error_json(StatusCode::NOT_FOUND, "配置文件不存在")); } // 读取目标配置文件内容 let raw = match std::fs::read_to_string(&target_path) { Ok(raw) => raw, Err(e) => return Ok(error_json(StatusCode::INTERNAL_SERVER_ERROR, &format!("读取失败: {e}"))), }; // 验证配置合法 if let Err(e) = config::parse_str(&raw, target_path.clone()) { return Ok(error_json(StatusCode::BAD_REQUEST, &format!("配置验证失败: {e}"))); } // 写入当前活跃配置文件路径 if let Err(e) = std::fs::write(&config.source_path, &raw) { return Ok(error_json(StatusCode::INTERNAL_SERVER_ERROR, &format!("写入失败: {e}"))); } // 触发热重载 if let Err(e) = tx.send(Envelope { from: "http".to_string(), to: Destination::Manager, message: Message::ConfigReloadRequest, }) { return Ok(error_json(StatusCode::INTERNAL_SERVER_ERROR, &format!("发送重载请求失败: {e}"))); } Ok(success_json(format!("已切换到配置: {}", req.filename))) }) } fn video_list_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "videos") .and(warp::get()) .and(with_state(state)) .and_then(|state: Arc| async move { let dir = video_dir(state.config().as_ref()); Ok::<_, Infallible>(json_response(StatusCode::OK, &list_video_files(&dir))) }) } fn app_info_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "app" / "info") .and(warp::get()) .and(with_state(state)) .and_then(|state: Arc| async move { let apk_path = downloads_dir(state.config().as_ref()).join("showen-app.apk"); let apk_size = std::fs::metadata(&apk_path) .ok() .filter(|meta| meta.is_file()) .map(|meta| meta.len()); Ok::<_, Infallible>(json_response( StatusCode::OK, &AppInfoResponse { version: env!("CARGO_PKG_VERSION").to_string(), apk_available: apk_size.is_some(), apk_size, download_url: "/download/showen-app.apk", }, )) }) } fn video_upload_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "videos" / "upload") .and(warp::post()) .and(warp::multipart::form().max_length(500 * 1024 * 1024)) .and(with_state(state)) .and_then(handle_video_upload) } fn video_delete_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "videos" / String) .and(warp::delete()) .and(with_state(state)) .and_then(handle_video_delete) } fn wifi_status_route( tx: mpsc::Sender, state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "wifi" / "status") .and(warp::get()) .and(with_tx(tx)) .and(with_state(state)) .and_then(handle_wifi_status) } fn wifi_scan_route( tx: mpsc::Sender, state: Arc, ) -> impl Filter + Clone { let tx_for_post = tx.clone(); let state_for_post = Arc::clone(&state); warp::path!("api" / "wifi" / "scan") .and(warp::get()) .and(with_tx(tx)) .and(with_state(Arc::clone(&state))) .and_then(handle_wifi_scan) .or( warp::path!("api" / "wifi" / "scan") .and(warp::post()) .and(with_tx(tx_for_post)) .and(with_state(state_for_post)) .and_then(handle_wifi_scan), ) } fn wifi_connect_route( tx: mpsc::Sender, state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "wifi" / "connect") .and(warp::post()) .and(warp::body::json()) .and(with_tx(tx)) .and(with_state(state)) .and_then(|req: WifiConnectRequest, tx, state| async move { wifi_action_reply( tx, state, WifiCommand::Connect { ssid: req.ssid, password: req.password, }, |payload| { let ssid = payload .get("ssid") .and_then(Value::as_str) .unwrap_or("未知网络"); format!("WiFi 连接成功: {ssid}") }, ) .await }) } fn wifi_ap_start_route( tx: mpsc::Sender, state: Arc, ) -> impl Filter + Clone { let tx_for_hotspot = tx.clone(); let state_for_hotspot = Arc::clone(&state); warp::path!("api" / "wifi" / "ap" / "start") .and(warp::post()) .and(warp::body::bytes()) .and(with_tx(tx)) .and(with_state(Arc::clone(&state))) .and_then(|body: bytes::Bytes, tx, state| async move { let req: WifiApStartRequest = match parse_optional_json(&body) { Ok(req) => req, Err(reply) => return Ok::<_, Infallible>(*reply), }; let ssid = req.ssid.unwrap_or_else(|| "showen".to_string()); let password = req.password.unwrap_or_else(|| "12345678".to_string()); let success_ssid = ssid.clone(); wifi_action_reply( tx, state, WifiCommand::ApStart { ssid, password }, move |_| format!("AP 热点已启动: SSID={success_ssid}"), ) .await }) .or( warp::path!("api" / "wifi" / "hotspot" / "start") .and(warp::post()) .and(warp::body::bytes()) .and(with_tx(tx_for_hotspot)) .and(with_state(state_for_hotspot)) .and_then(|body: bytes::Bytes, tx, state| async move { let req: WifiApStartRequest = match parse_optional_json(&body) { Ok(req) => req, Err(reply) => return Ok::<_, Infallible>(*reply), }; let ssid = req.ssid.unwrap_or_else(|| "showen".to_string()); let password = req.password.unwrap_or_else(|| "12345678".to_string()); let success_ssid = ssid.clone(); wifi_action_reply( tx, state, WifiCommand::ApStart { ssid, password }, move |_| format!("AP 热点已启动: SSID={success_ssid}"), ) .await }), ) } fn wifi_ap_stop_route( tx: mpsc::Sender, state: Arc, ) -> impl Filter + Clone { let tx_for_hotspot = tx.clone(); let state_for_hotspot = Arc::clone(&state); warp::path!("api" / "wifi" / "ap" / "stop") .and(warp::post()) .and(with_tx(tx)) .and(with_state(Arc::clone(&state))) .and_then(|tx, state| async move { wifi_action_reply(tx, state, WifiCommand::ApStop, |_| { "AP 热点已关闭".to_string() }) .await }) .or( warp::path!("api" / "wifi" / "hotspot" / "stop") .and(warp::post()) .and(with_tx(tx_for_hotspot)) .and(with_state(state_for_hotspot)) .and_then(|tx, state| async move { wifi_action_reply(tx, state, WifiCommand::ApStop, |_| { "AP 热点已关闭".to_string() }) .await }), ) } fn ble_start_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "ble" / "start") .and(warp::post()) .and(warp::body::bytes()) .and(with_state(state)) .and_then(|body: bytes::Bytes, state: Arc| async move { let req: BleStartRequest = match parse_optional_json(&body) { Ok(req) => req, Err(reply) => return Ok::<_, Infallible>(*reply), }; let config = state.config(); let device_name = req .device_name .unwrap_or_else(|| config.ble.device_name.clone()); Ok::<_, Infallible>(success_json(format!( "BLE 配网服务已内嵌运行中,设备名: {device_name}" ))) }) } fn ble_stop_route() -> impl Filter + Clone { warp::path!("api" / "ble" / "stop") .and(warp::post()) .map(|| success_json("BLE 配网服务随主进程运行,无需手动停止")) } fn ble_status_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "ble" / "status") .and(warp::get()) .and(with_state(state)) .and_then(|state: Arc| async move { let config = state.config(); Ok::<_, Infallible>(json_response( StatusCode::OK, &BleStatusResponse { running: state.ble_ready(), embedded: true, device_name: config.ble.device_name.clone(), }, )) }) } async fn handle_config_update( body: bytes::Bytes, tx: mpsc::Sender, state: Arc, ) -> Result { let raw = match std::str::from_utf8(&body) { Ok(raw) => raw, Err(_) => { return Ok(error_json( StatusCode::BAD_REQUEST, "请求体不是有效的 UTF-8", )) } }; let current = state.config(); if let Err(error) = config::parse_str(raw, current.source_path.clone()) { return Ok(error_json( StatusCode::BAD_REQUEST, &format!("配置验证失败: {error}"), )); } if let Err(error) = std::fs::write(¤t.source_path, raw) { return Ok(error_json( StatusCode::INTERNAL_SERVER_ERROR, &format!("写入配置文件失败: {error}"), )); } if let Err(error) = tx.send(Envelope { from: "http".to_string(), to: Destination::Manager, message: Message::ConfigReloadRequest, }) { return Ok(error_json( StatusCode::INTERNAL_SERVER_ERROR, &format!("发送配置重载请求失败: {error}"), )); } Ok(success_json("配置已保存,热重载将自动生效")) } async fn handle_video_upload( mut form: FormData, state: Arc, ) -> Result { let dir = video_dir(state.config().as_ref()); let mut uploaded = Vec::new(); while let Ok(Some(part)) = form.try_next().await { let Some(filename) = part.filename() else { continue; }; let safe_name = sanitize_filename(filename); if safe_name.is_empty() { continue; } if let Err(error) = stream_upload_part(part, &dir.join(&safe_name)).await { let status = if error.contains("文件大小超过限制") { StatusCode::PAYLOAD_TOO_LARGE } else { StatusCode::INTERNAL_SERVER_ERROR }; return Ok(error_json(status, &error)); } uploaded.push(safe_name); } if uploaded.is_empty() { Ok(error_json(StatusCode::BAD_REQUEST, "未找到上传文件")) } else { Ok(success_json(format!( "已上传 {} 个文件: {}", uploaded.len(), uploaded.join(", ") ))) } } async fn handle_video_delete( filename: String, state: Arc, ) -> Result { if filename.contains("..") { return Ok(error_json(StatusCode::BAD_REQUEST, "无效的文件名")); } let target = video_dir(state.config().as_ref()).join(&filename); if !target.exists() { return Ok(error_json(StatusCode::NOT_FOUND, "文件不存在")); } match std::fs::remove_file(&target) { Ok(()) => Ok(success_json(format!("已删除: {filename}"))), Err(error) => Ok(error_json( StatusCode::INTERNAL_SERVER_ERROR, &format!("删除失败: {error}"), )), } } async fn handle_wifi_status( tx: mpsc::Sender, state: Arc, ) -> Result { let payload = match wifi_request(tx, state, WifiCommand::Status).await { Ok(payload) => payload, Err(reply) => return Ok(reply), }; let device = payload .get("devices") .and_then(Value::as_array) .into_iter() .flatten() .find(|item| { item.get("device_type").and_then(Value::as_str) == Some("wifi") && item.get("state").and_then(Value::as_str) == Some("connected") }); let ssid = device .and_then(|item| item.get("connection")) .and_then(Value::as_str) .unwrap_or_default() .to_string(); let ip = device .and_then(|item| item.get("ip4_addresses")) .and_then(Value::as_array) .and_then(|ips| ips.first()) .and_then(Value::as_str) .map(strip_cidr) .unwrap_or_default(); Ok(json_response( StatusCode::OK, &WifiStatusResponse { connected: !ssid.is_empty(), ssid, ip, }, )) } async fn handle_wifi_scan( tx: mpsc::Sender, state: Arc, ) -> Result { let payload = match wifi_request(tx, state, WifiCommand::Scan).await { Ok(payload) => payload, Err(reply) => return Ok(reply), }; let networks = payload .get("networks") .cloned() .unwrap_or_else(|| Value::Array(Vec::new())); Ok(json_response(StatusCode::OK, &networks)) } async fn send_video_command( tx: mpsc::Sender, message: Message, success_message: impl Into, ) -> Result { match tx.send(Envelope { from: "http".to_string(), to: Destination::Plugin("video".to_string()), message, }) { Ok(()) => Ok(success_json(success_message.into())), Err(error) => Ok(error_json( StatusCode::INTERNAL_SERVER_ERROR, &format!("发送命令失败: {error}"), )), } } async fn wifi_action_reply( tx: mpsc::Sender, state: Arc, command: WifiCommand, build_message: F, ) -> Result where F: FnOnce(&Value) -> String, { let payload = match wifi_request(tx, state, command).await { Ok(payload) => payload, Err(reply) => return Ok(reply), }; Ok(success_json(build_message(&payload))) } async fn wifi_request( tx: mpsc::Sender, state: Arc, command: WifiCommand, ) -> Result { let _request_guard = state.wifi_request_lock.lock().await; let version = match state.wifi_response.lock() { Ok(guard) => guard.version, Err(_) => { return Err(error_json( StatusCode::INTERNAL_SERVER_ERROR, "WiFi 响应状态锁已损坏", )); } }; if let Err(error) = tx.send(Envelope { from: "http".to_string(), to: Destination::Plugin("wifi".to_string()), message: Message::WifiCommand(command), }) { return Err(error_json( StatusCode::INTERNAL_SERVER_ERROR, &format!("发送 WiFi 命令失败: {error}"), )); } let deadline = Instant::now() + Duration::from_secs(10); let mut guard = match state.wifi_response.lock() { Ok(guard) => guard, Err(_) => { return Err(error_json( StatusCode::INTERNAL_SERVER_ERROR, "WiFi 响应状态锁已损坏", )); } }; while guard.version == version { let now = Instant::now(); if now >= deadline { return Err(error_json( StatusCode::GATEWAY_TIMEOUT, "等待 WiFi 响应超时", )); } let result = state .wifi_response_cv .wait_timeout(guard, deadline.saturating_duration_since(now)); let (next_guard, wait_result) = match result { Ok(result) => result, Err(_) => { return Err(error_json( StatusCode::INTERNAL_SERVER_ERROR, "等待 WiFi 响应失败", )); } }; guard = next_guard; if wait_result.timed_out() && guard.version == version { return Err(error_json( StatusCode::GATEWAY_TIMEOUT, "等待 WiFi 响应超时", )); } } let raw = guard.payload.clone().unwrap_or_default(); let payload: Value = match serde_json::from_str(&raw) { Ok(payload) => payload, Err(error) => { return Err(error_json( StatusCode::BAD_GATEWAY, &format!("WiFi 返回了无效 JSON: {error}"), )); } }; if payload.get("ok").and_then(Value::as_bool) == Some(false) { let message = payload .get("error") .and_then(Value::as_str) .unwrap_or("WiFi 操作失败"); return Err(error_json(StatusCode::INTERNAL_SERVER_ERROR, message)); } Ok(payload) } async fn stream_upload_part( part: warp::multipart::Part, destination: &Path, ) -> Result<(), String> { let parent = destination .parent() .ok_or_else(|| "上传目标目录无效".to_string())?; let temp_path = parent.join(format!( ".upload-{}-{}.part", std::process::id(), UPLOAD_TMP_COUNTER.fetch_add(1, Ordering::Relaxed) )); let mut file = match tokio::fs::File::create(&temp_path).await { Ok(file) => file, Err(error) => return Err(format!("创建临时文件失败: {error}")), }; let mut total_size = 0u64; let mut stream = part.stream(); while let Some(chunk) = match stream.try_next().await { Ok(chunk) => chunk, Err(error) => { let _ = tokio::fs::remove_file(&temp_path).await; return Err(format!("读取文件失败: {error}")); } } { let chunk_size = chunk.remaining() as u64; total_size = total_size.saturating_add(chunk_size); if total_size > MAX_UPLOAD_FILE_SIZE { let _ = file.flush().await; drop(file); let _ = tokio::fs::remove_file(&temp_path).await; return Err(format!( "文件大小超过限制: 单文件最大 {} MB", MAX_UPLOAD_FILE_SIZE / 1024 / 1024 )); } if let Err(error) = file.write_all(chunk.chunk()).await { drop(file); let _ = tokio::fs::remove_file(&temp_path).await; return Err(format!("写入临时文件失败: {error}")); } } if let Err(error) = file.flush().await { drop(file); let _ = tokio::fs::remove_file(&temp_path).await; return Err(format!("刷新临时文件失败: {error}")); } drop(file); if let Err(error) = tokio::fs::rename(&temp_path, destination).await { let _ = tokio::fs::remove_file(&temp_path).await; return Err(format!("保存文件失败: {error}")); } Ok(()) } async fn websocket_session( ws: warp::ws::WebSocket, tx: mpsc::Sender, state: Arc, ) { let (mut sender, mut receiver) = ws.split(); let mut events = state.ws_subscribe(); for payload in state.ws_snapshots() { if sender.send(warp::ws::Message::text(payload)).await.is_err() { return; } } loop { tokio::select! { event = events.recv() => { match event { Ok(payload) => { if sender.send(warp::ws::Message::text(payload)).await.is_err() { break; } } Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {} Err(tokio::sync::broadcast::error::RecvError::Closed) => break, } } incoming = receiver.next() => { match incoming { Some(Ok(message)) => { if message.is_ping() { if sender.send(warp::ws::Message::pong(message.as_bytes())).await.is_err() { break; } } else if message.is_close() { break; } else if message.is_text() { let reply = handle_ws_command(message.to_str().unwrap_or(""), &tx); if sender.send(warp::ws::Message::text(reply)).await.is_err() { break; } } } Some(Err(_)) | None => break, } } } } } /// 解析 WebSocket 收到的 JSON 命令,返回 JSON 响应字符串。 /// /// 输入格式: `{"cmd":"play"}` 或 `{"cmd":"goto","index":3}` 或 /// `{"cmd":"connect","ssid":"x","password":"y"}` 等 fn handle_ws_command(text: &str, tx: &mpsc::Sender) -> String { let json: serde_json::Value = match serde_json::from_str(text) { Ok(v) => v, Err(_) => return r#"{"ok":false,"error":"invalid JSON"}"#.to_string(), }; let cmd = match json.get("cmd").and_then(|v| v.as_str()) { Some(c) => c, None => return r#"{"ok":false,"error":"missing cmd field"}"#.to_string(), }; // 将 JSON 字段组合为文本命令字符串 let command_str = build_command_string(cmd, &json); // 从 JSON 中提取 ssid/password 作为 hint(用于无参数的 connect/ap_start) let ssid_hint = json .get("ssid") .and_then(|v| v.as_str()) .unwrap_or(""); let password_hint = json .get("password") .and_then(|v| v.as_str()) .unwrap_or(""); match dispatch::parse_command(&command_str, "ws", ssid_hint, password_hint) { Ok(result) => { if tx.send(result.envelope).is_ok() { format!(r#"{{"ok":true,"cmd":"{}"}}"#, cmd) } else { r#"{"ok":false,"error":"channel closed"}"#.to_string() } } Err(error) => { format!( r#"{{"ok":false,"cmd":"{}","error":"{}"}}"#, cmd, error.replace('"', "\\\"") ) } } } /// 从 JSON 对象组装文本命令字符串。 /// 例: `{"cmd":"goto","index":3}` -> `"goto:3"` /// `{"cmd":"scene","name":"idle"}` -> `"scene:idle"` /// `{"cmd":"trigger","name":"voice","value":"hi"}` -> `"trigger:voice:hi"` /// `{"cmd":"connect","ssid":"x","password":"y"}` -> `"connect:x:y"` fn build_command_string(cmd: &str, json: &serde_json::Value) -> String { match cmd { "goto" => { if let Some(index) = json.get("index").and_then(|v| v.as_u64()) { format!("goto:{index}") } else { "goto".to_string() } } "scene" => { if let Some(name) = json.get("name").and_then(|v| v.as_str()) { format!("scene:{name}") } else { "scene".to_string() } } "trigger" => { let name = json.get("name").and_then(|v| v.as_str()).unwrap_or(""); let value = json.get("value").and_then(|v| v.as_str()).unwrap_or(""); if name.is_empty() { "trigger".to_string() } else { format!("trigger:{name}:{value}") } } "connect" => { let ssid = json.get("ssid").and_then(|v| v.as_str()).unwrap_or(""); let password = json.get("password").and_then(|v| v.as_str()).unwrap_or(""); if ssid.is_empty() { "connect".to_string() } else { format!("connect:{ssid}:{password}") } } "ap_start" => { let ssid = json.get("ssid").and_then(|v| v.as_str()).unwrap_or(""); let password = json.get("password").and_then(|v| v.as_str()).unwrap_or(""); if ssid.is_empty() { "ap_start".to_string() } else { format!("ap_start:{ssid}:{password}") } } _ => cmd.to_string(), } } fn parse_optional_json(body: &bytes::Bytes) -> Result> where T: DeserializeOwned + Default, { if body.is_empty() { return Ok(T::default()); } serde_json::from_slice(body).map_err(|error| { Box::new(error_json( StatusCode::BAD_REQUEST, &format!("JSON 格式错误: {error}"), )) }) } // ── 文件管理 API ── /// 根据 dir key 解析到实际文件系统路径。仅允许 videos/configs/plugins 三个目录。 fn resolve_managed_dir(dir_key: &str, config: &AppConfig) -> Option { match dir_key { "videos" => Some(video_dir(config)), "configs" => { // source_dir 就是 configs/ 目录 Some(config.source_dir.clone()) } "plugins" => { // plugin_store/ 在项目根目录下(source_dir 的父目录) let project_root = config.source_dir.parent()?; Some(project_root.join("plugin_store")) } _ => None, } } /// 验证相对路径不逃逸出基目录。返回规范化后的完整路径。 fn validate_managed_path(base: &Path, relative: &str) -> Option { if relative.contains("..") || relative.starts_with('/') || relative.starts_with('\\') { return None; } let full = base.join(relative); if let Ok(canonical_base) = base.canonicalize() { if full.exists() { let canonical = full.canonicalize().ok()?; if canonical.starts_with(&canonical_base) { return Some(canonical); } } else { // 新文件:确保父目录存在且在 base 内 let parent = full.parent()?; if parent.exists() { let canonical_parent = parent.canonicalize().ok()?; if canonical_parent.starts_with(&canonical_base) { return Some(full); } } } } None } fn list_dir_entries(dir: &Path) -> Vec { let mut entries = Vec::new(); if let Ok(rd) = std::fs::read_dir(dir) { for entry in rd.flatten() { if let Ok(meta) = entry.metadata() { entries.push(FileEntry { name: entry.file_name().to_string_lossy().into_owned(), is_dir: meta.is_dir(), size: if meta.is_file() { meta.len() } else { 0 }, }); } } } entries.sort_by(|a, b| { // 目录排前面 b.is_dir.cmp(&a.is_dir).then(a.name.cmp(&b.name)) }); entries } fn file_list_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "files" / String) .and(warp::get()) .and(warp::query::raw().or(warp::any().map(String::new)).unify()) .and(with_state(state)) .and_then(|dir_key: String, query: String, state: Arc| async move { let config = state.config(); let base = match resolve_managed_dir(&dir_key, config.as_ref()) { Some(d) => d, None => return Ok::<_, Infallible>(error_json( StatusCode::BAD_REQUEST, &format!("不支持的目录: {dir_key}(仅支持 videos/configs/plugins)"), )), }; // 支持 ?path=subdir 子目录浏览 let sub = query .split('&') .find_map(|kv| { let mut parts = kv.splitn(2, '='); if parts.next()? == "path" { Some( percent_decode(parts.next().unwrap_or("")) ) } else { None } }) .unwrap_or_default(); let target = if sub.is_empty() { base.clone() } else { match validate_managed_path(&base, &sub) { Some(p) => p, None => return Ok(error_json(StatusCode::FORBIDDEN, "路径不合法")), } }; if !target.is_dir() { return Ok(error_json(StatusCode::NOT_FOUND, "目录不存在")); } Ok(json_response(StatusCode::OK, &list_dir_entries(&target))) }) } fn file_upload_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "files" / String / "upload") .and(warp::post()) .and(warp::multipart::form().max_length(500 * 1024 * 1024)) .and(warp::query::raw().or(warp::any().map(String::new)).unify()) .and(with_state(state)) .and_then(|dir_key: String, mut form: FormData, query: String, state: Arc| async move { let config = state.config(); let base = match resolve_managed_dir(&dir_key, config.as_ref()) { Some(d) => d, None => return Ok::<_, Infallible>(error_json(StatusCode::BAD_REQUEST, "不支持的目录")), }; let sub = query .split('&') .find_map(|kv| { let mut parts = kv.splitn(2, '='); if parts.next()? == "path" { Some(percent_decode(parts.next().unwrap_or(""))) } else { None } }) .unwrap_or_default(); let target_dir = if sub.is_empty() { base.clone() } else { match validate_managed_path(&base, &sub) { Some(p) if p.is_dir() => p, _ => return Ok(error_json(StatusCode::FORBIDDEN, "目标目录不合法")), } }; let mut uploaded = Vec::new(); while let Ok(Some(part)) = form.try_next().await { let Some(filename) = part.filename() else { continue }; let safe_name = sanitize_filename(filename); if safe_name.is_empty() { continue } let dest = target_dir.join(&safe_name); if validate_managed_path(&base, &dest.strip_prefix(&base).unwrap_or(Path::new("")).to_string_lossy()).is_none() { continue; } if let Err(error) = stream_upload_part(part, &dest).await { let status = if error.contains("文件大小超过限制") { StatusCode::PAYLOAD_TOO_LARGE } else { StatusCode::INTERNAL_SERVER_ERROR }; return Ok(error_json(status, &error)); } uploaded.push(safe_name); } if uploaded.is_empty() { Ok(error_json(StatusCode::BAD_REQUEST, "未找到上传文件")) } else { Ok(success_json(format!("已上传 {} 个文件: {}", uploaded.len(), uploaded.join(", ")))) } }) } fn file_download_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "files" / String / "download") .and(warp::get()) .and(warp::query::raw().or(warp::any().map(String::new)).unify()) .and(with_state(state)) .and_then(|dir_key: String, query: String, state: Arc| async move { let config = state.config(); let base = match resolve_managed_dir(&dir_key, config.as_ref()) { Some(d) => d, None => return Ok::<_, Infallible>( warp::reply::with_status( warp::reply::html("不支持的目录".to_string()), StatusCode::BAD_REQUEST, ).into_response() ), }; let file_path = query .split('&') .find_map(|kv| { let mut parts = kv.splitn(2, '='); if parts.next()? == "path" { Some(percent_decode(parts.next().unwrap_or(""))) } else { None } }) .unwrap_or_default(); if file_path.is_empty() { return Ok(warp::reply::with_status( warp::reply::html("缺少 path 参数".to_string()), StatusCode::BAD_REQUEST, ).into_response()); } let full = match validate_managed_path(&base, &file_path) { Some(p) => p, None => return Ok(warp::reply::with_status( warp::reply::html("路径不合法".to_string()), StatusCode::FORBIDDEN, ).into_response()), }; if !full.is_file() { return Ok(warp::reply::with_status( warp::reply::html("文件不存在".to_string()), StatusCode::NOT_FOUND, ).into_response()); } Ok(read_attachment_response(&full)) }) } fn file_delete_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "files" / String / "delete") .and(warp::post()) .and(warp::body::json::()) .and(with_state(state)) .and_then(|dir_key: String, body: serde_json::Value, state: Arc| async move { let config = state.config(); let base = match resolve_managed_dir(&dir_key, config.as_ref()) { Some(d) => d, None => return Ok::<_, Infallible>(error_json(StatusCode::BAD_REQUEST, "不支持的目录")), }; let path = body.get("path").and_then(|v| v.as_str()).unwrap_or(""); if path.is_empty() { return Ok(error_json(StatusCode::BAD_REQUEST, "缺少 path")); } let full = match validate_managed_path(&base, path) { Some(p) => p, None => return Ok(error_json(StatusCode::FORBIDDEN, "路径不合法")), }; if !full.exists() { return Ok(error_json(StatusCode::NOT_FOUND, "文件不存在")); } let result = if full.is_dir() { std::fs::remove_dir_all(&full) } else { std::fs::remove_file(&full) }; match result { Ok(()) => Ok(success_json(format!("已删除: {path}"))), Err(e) => Ok(error_json(StatusCode::INTERNAL_SERVER_ERROR, &format!("删除失败: {e}"))), } }) } fn file_move_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "files" / "move") .and(warp::post()) .and(warp::body::json::()) .and(with_state(state)) .and_then(|req: FileMoveRequest, state: Arc| async move { let config = state.config(); let src_base = match resolve_managed_dir(&req.from_dir, config.as_ref()) { Some(d) => d, None => return Ok::<_, Infallible>(error_json(StatusCode::BAD_REQUEST, "源目录不支持")), }; let dst_base = match resolve_managed_dir(&req.to_dir, config.as_ref()) { Some(d) => d, None => return Ok::<_, Infallible>(error_json(StatusCode::BAD_REQUEST, "目标目录不支持")), }; let src = match validate_managed_path(&src_base, &req.from_path) { Some(p) => p, None => return Ok(error_json(StatusCode::FORBIDDEN, "源路径不合法")), }; let dst = match validate_managed_path(&dst_base, &req.to_path) { Some(p) => p, None => return Ok(error_json(StatusCode::FORBIDDEN, "目标路径不合法")), }; if !src.exists() { return Ok(error_json(StatusCode::NOT_FOUND, "源文件不存在")); } if dst.exists() { return Ok(error_json(StatusCode::CONFLICT, "目标路径已存在")); } // 先尝试 rename(同文件系统),失败则 copy+delete match std::fs::rename(&src, &dst) { Ok(()) => Ok(success_json(format!("已移动: {} → {}", req.from_path, req.to_path))), Err(_) => { // 跨文件系统:copy then delete match std::fs::copy(&src, &dst) { Ok(_) => { let _ = std::fs::remove_file(&src); Ok(success_json(format!("已移动: {} → {}", req.from_path, req.to_path))) } Err(e) => Ok(error_json(StatusCode::INTERNAL_SERVER_ERROR, &format!("移动失败: {e}"))), } } } }) } fn file_mkdir_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "files" / String / "mkdir") .and(warp::post()) .and(warp::body::json::()) .and(with_state(state)) .and_then(|dir_key: String, body: serde_json::Value, state: Arc| async move { let config = state.config(); let base = match resolve_managed_dir(&dir_key, config.as_ref()) { Some(d) => d, None => return Ok::<_, Infallible>(error_json(StatusCode::BAD_REQUEST, "不支持的目录")), }; let path = body.get("path").and_then(|v| v.as_str()).unwrap_or(""); if path.is_empty() { return Ok(error_json(StatusCode::BAD_REQUEST, "缺少 path")); } let full = match validate_managed_path(&base, path) { Some(p) => p, None => return Ok(error_json(StatusCode::FORBIDDEN, "路径不合法")), }; match std::fs::create_dir_all(&full) { Ok(()) => Ok(success_json(format!("已创建目录: {path}"))), Err(e) => Ok(error_json(StatusCode::INTERNAL_SERVER_ERROR, &format!("创建失败: {e}"))), } }) } fn percent_decode(s: &str) -> String { let mut bytes = Vec::with_capacity(s.len()); let mut chars = s.bytes(); while let Some(b) = chars.next() { if b == b'%' { let h = chars.next().unwrap_or(0); let l = chars.next().unwrap_or(0); let hex = [h, l]; if let Ok(hex_str) = std::str::from_utf8(&hex) { if let Ok(val) = u8::from_str_radix(hex_str, 16) { bytes.push(val); continue; } } bytes.push(b'%'); bytes.push(h); bytes.push(l); } else if b == b'+' { bytes.push(b' '); } else { bytes.push(b); } } String::from_utf8(bytes).unwrap_or_else(|e| String::from_utf8_lossy(e.as_bytes()).into_owned()) } fn list_video_files(dir: &Path) -> Vec { let mut files = Vec::new(); if let Ok(entries) = std::fs::read_dir(dir) { for entry in entries.flatten() { if let Ok(meta) = entry.metadata() { if meta.is_file() { files.push(VideoFileInfo { name: entry.file_name().to_string_lossy().into_owned(), size: meta.len(), }); } if meta.is_dir() { let prefix = entry.file_name().to_string_lossy().into_owned(); if let Ok(sub_entries) = std::fs::read_dir(entry.path()) { for sub_entry in sub_entries.flatten() { if let Ok(sub_meta) = sub_entry.metadata() { if sub_meta.is_file() { files.push(VideoFileInfo { name: format!( "{prefix}/{}", sub_entry.file_name().to_string_lossy() ), size: sub_meta.len(), }); } } } } } } } } files.sort_by(|left, right| left.name.cmp(&right.name)); files } fn sanitize_filename(name: &str) -> String { name.replace(['/', '\\'], "_").replace("..", "_") } fn video_dir(config: &AppConfig) -> PathBuf { if let Some(first) = config.playlist.first() { let resolved = config.resolve_media_path(first); return resolved .parent() .map(Path::to_path_buf) .unwrap_or_else(|| config.source_dir.clone()); } config.source_dir.clone() } fn downloads_dir(config: &AppConfig) -> PathBuf { config.source_dir.join("downloads") } fn read_attachment_response(path: &Path) -> warp::reply::Response { match std::fs::read(path) { Ok(data) => { let filename = path.file_name().unwrap_or_default().to_string_lossy(); // Sanitize filename for Content-Disposition header to prevent header injection let safe_filename = filename .replace('\\', "_") .replace('"', "_") .replace('\r', "") .replace('\n', ""); match warp::http::Response::builder() .header("Content-Type", "application/octet-stream") .header( "Content-Disposition", format!("attachment; filename=\"{safe_filename}\""), ) .header("Content-Length", data.len().to_string()) .body(data) { Ok(response) => response.into_response(), Err(_) => warp::reply::with_status( warp::reply::html("响应构建失败".to_string()), StatusCode::INTERNAL_SERVER_ERROR, ) .into_response(), } } Err(e) => warp::reply::with_status( warp::reply::html(format!("读取失败: {e}")), StatusCode::INTERNAL_SERVER_ERROR, ) .into_response(), } } fn strip_cidr(value: &str) -> String { value.split('/').next().unwrap_or_default().to_string() } fn with_tx( tx: mpsc::Sender, ) -> impl Filter,), Error = Infallible> + Clone { warp::any().map(move || tx.clone()) } fn with_state( state: Arc, ) -> impl Filter,), Error = Infallible> + Clone { warp::any().map(move || Arc::clone(&state)) } fn success_json(message: impl Into) -> warp::reply::Response { json_response( StatusCode::OK, &ApiMessage { status: "ok", message: message.into(), }, ) } fn error_json(status: StatusCode, message: &str) -> warp::reply::Response { json_response( status, &ApiMessage { status: "error", message: message.to_string(), }, ) } // ── 插件管理 API ── fn plugins_list_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "plugins") .and(warp::get()) .and(with_state(state)) .and_then(|state: Arc| async move { Ok::<_, Infallible>(json_response(StatusCode::OK, &state.plugin_states())) }) } fn plugin_detail_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "plugins" / String) .and(warp::get()) .and(with_state(state)) .and_then(|id: String, state: Arc| async move { let plugins = state.plugin_states(); match plugins.iter().find(|p| p.id == id) { Some(info) => Ok::<_, Infallible>(json_response(StatusCode::OK, info)), None => Ok(error_json(StatusCode::NOT_FOUND, &format!("plugin '{}' not found", id))), } }) } #[derive(Deserialize)] struct PluginSwitchRequest { version: String, } #[derive(Deserialize)] struct PluginInstallRequest { id: String, #[serde(default)] version: Option, } fn plugin_enable_route( tx: mpsc::Sender, ) -> impl Filter + Clone { warp::path!("api" / "plugins" / String / "enable") .and(warp::post()) .and(with_tx(tx)) .and_then(|id: String, tx: mpsc::Sender| async move { send_plugin_command(tx, "plugin_enable", &id).await }) } fn plugin_disable_route( tx: mpsc::Sender, ) -> impl Filter + Clone { warp::path!("api" / "plugins" / String / "disable") .and(warp::post()) .and(with_tx(tx)) .and_then(|id: String, tx: mpsc::Sender| async move { send_plugin_command(tx, "plugin_disable", &id).await }) } fn plugin_rollback_route( tx: mpsc::Sender, ) -> impl Filter + Clone { warp::path!("api" / "plugins" / String / "rollback") .and(warp::post()) .and(with_tx(tx)) .and_then(|id: String, tx: mpsc::Sender| async move { send_plugin_command(tx, "plugin_rollback", &id).await }) } fn plugin_switch_route( tx: mpsc::Sender, ) -> impl Filter + Clone { warp::path!("api" / "plugins" / String / "switch") .and(warp::post()) .and(warp::body::json::()) .and(with_tx(tx)) .and_then( |id: String, body: PluginSwitchRequest, tx: mpsc::Sender| async move { let payload = serde_json::json!({ "id": id, "version": body.version, }) .to_string(); match tx.send(Envelope { from: "http".to_string(), to: Destination::Manager, message: Message::Custom { kind: "plugin_switch".to_string(), payload, }, }) { Ok(()) => Ok::<_, Infallible>(success_json( format!("版本切换请求已发送: {} -> v{}", id, body.version), )), Err(e) => Ok(error_json( StatusCode::INTERNAL_SERVER_ERROR, &format!("发送失败: {e}"), )), } }, ) } fn plugin_install_route( tx: mpsc::Sender, ) -> impl Filter + Clone { warp::path!("api" / "plugins" / "install") .and(warp::post()) .and(warp::body::json::()) .and(with_tx(tx)) .and_then( |body: PluginInstallRequest, tx: mpsc::Sender| async move { let payload = serde_json::json!({ "id": body.id, "version": body.version, }) .to_string(); match tx.send(Envelope { from: "http".to_string(), to: Destination::Manager, message: Message::Custom { kind: "plugin_install".to_string(), payload, }, }) { Ok(()) => Ok::<_, Infallible>(success_json( format!("安装请求已发送: {}", body.id), )), Err(e) => Ok(error_json( StatusCode::INTERNAL_SERVER_ERROR, &format!("发送失败: {e}"), )), } }, ) } fn plugin_check_updates_route( tx: mpsc::Sender, ) -> impl Filter + Clone { warp::path!("api" / "plugins" / "check-updates") .and(warp::post()) .and(with_tx(tx)) .and_then(|tx: mpsc::Sender| async move { send_plugin_command(tx, "plugin_check_updates", "").await }) } async fn send_plugin_command( tx: mpsc::Sender, kind: &str, plugin_id: &str, ) -> Result { match tx.send(Envelope { from: "http".to_string(), to: Destination::Manager, message: Message::Custom { kind: kind.to_string(), payload: plugin_id.to_string(), }, }) { Ok(()) => Ok(success_json(format!("{kind} 命令已发送"))), Err(e) => Ok(error_json( StatusCode::INTERNAL_SERVER_ERROR, &format!("发送失败: {e}"), )), } } // ── AI 对话 API (M2.1) ── #[derive(Deserialize)] struct ChatTextRequest { #[serde(default)] session_id: Option, text: String, } /// POST /api/chat/text — 文字对话(Web 端主路径) fn chat_text_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "chat" / "text") .and(warp::post()) .and(warp::body::json::()) .and(with_state(state)) .and_then(|req: ChatTextRequest, state: Arc| async move { let pipeline = match state.ai_pipeline() { Some(p) => p, None => { return Ok::<_, Infallible>(error_json( StatusCode::SERVICE_UNAVAILABLE, "AI 插件未就绪", )) } }; let config = state.config(); let session_id = req .session_id .unwrap_or_else(|| format!("web_{}", std::process::id())); let chat_req = ChatRequest { session_id, input: ChatInput::Text { content: req.text, }, persona_prompt: config.character.persona_prompt.clone(), max_tokens: config.character.max_tokens, }; // AI 管线是同步阻塞调用(ASR/LLM/TTS 子进程),用 spawn_blocking 避免阻塞 tokio reactor let resp = tokio::task::spawn_blocking(move || pipeline.run(&chat_req)) .await .unwrap_or_else(|e| ChatResponse { session_id: String::new(), transcription: None, reply_text: String::new(), reply_audio_path: None, error: Some(format!("AI 管线执行失败: {e}")), }); if resp.error.is_some() { Ok(json_response(StatusCode::INTERNAL_SERVER_ERROR, &resp)) } else { Ok(json_response(StatusCode::OK, &resp)) } }) } /// POST /api/chat/audio — 语音对话(App 主路径,上传 wav) fn chat_audio_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "chat" / "audio") .and(warp::post()) .and(warp::body::content_length_limit(20 * 1024 * 1024)) .and(warp::body::bytes()) .and(with_state(state)) .and_then(|bytes: bytes::Bytes, state: Arc| async move { let pipeline = match state.ai_pipeline() { Some(p) => p, None => { return Ok::<_, Infallible>(error_json( StatusCode::SERVICE_UNAVAILABLE, "AI 插件未就绪", )) } }; let config = state.config(); let tmp_dir = config.ai.tmp_dir.clone(); let _ = std::fs::create_dir_all(&tmp_dir); let audio_path = tmp_dir.join(format!( "asr_{}.wav", std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_millis()) .unwrap_or(0) )); if let Err(e) = std::fs::write(&audio_path, &bytes) { return Ok(error_json( StatusCode::INTERNAL_SERVER_ERROR, &format!("保存音频失败: {e}"), )); } let session_id = format!("app_{}", std::process::id()); let chat_req = ChatRequest { session_id, input: ChatInput::Audio { path: audio_path.to_string_lossy().into_owned(), }, persona_prompt: config.character.persona_prompt.clone(), max_tokens: config.character.max_tokens, }; let audio_path_clone = audio_path.clone(); let resp = tokio::task::spawn_blocking(move || pipeline.run(&chat_req)) .await .unwrap_or_else(|e| ChatResponse { session_id: String::new(), transcription: None, reply_text: String::new(), reply_audio_path: None, error: Some(format!("AI 管线执行失败: {e}")), }); // 清理上传的临时音频 let _ = std::fs::remove_file(&audio_path_clone); if resp.error.is_some() { Ok(json_response(StatusCode::INTERNAL_SERVER_ERROR, &resp)) } else { Ok(json_response(StatusCode::OK, &resp)) } }) } // ── AI 模型管理 API (M2.1) ── /// GET /api/models — 列出所有模型 fn models_list_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "models") .and(warp::get()) .and(with_state(state)) .and_then(|state: Arc| async move { let models = match state.ai_models() { Some(m) => m, None => { return Ok::<_, Infallible>(error_json( StatusCode::SERVICE_UNAVAILABLE, "AI 插件未就绪", )) } }; let mgr = models.lock().unwrap(); let registry = mgr.registry(); let used = mgr.used_space(); let quota = mgr.quota(); let result = serde_json::json!({ "models": registry.models, "active": registry.active, "used_space": used, "quota": quota, }); Ok(json_response(StatusCode::OK, &result)) }) } #[derive(Deserialize)] struct ModelActionRequest { model_id: String, } /// POST /api/models/download — 下载模型 fn model_download_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "models" / "download") .and(warp::post()) .and(warp::body::json::()) .and(with_state(state)) .and_then(|req: ModelActionRequest, state: Arc| async move { let models = match state.ai_models() { Some(m) => m, None => { return Ok::<_, Infallible>(error_json( StatusCode::SERVICE_UNAVAILABLE, "AI 插件未就绪", )) } }; // 下载在独立线程执行,避免阻塞 HTTP let models_clone = models.clone(); let model_id = req.model_id.clone(); std::thread::spawn(move || { let mut mgr = models_clone.lock().unwrap(); if let Err(e) = mgr.download_model(&model_id) { eprintln!("[HttpPlugin] 模型下载失败 {model_id}: {e}"); } }); Ok::<_, Infallible>(success_json(format!("模型 {} 下载已启动", req.model_id))) }) } /// POST /api/models/switch — 切换激活模型 fn model_switch_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "models" / "switch") .and(warp::post()) .and(warp::body::json::()) .and(with_state(state)) .and_then(|body: serde_json::Value, state: Arc| async move { let models = match state.ai_models() { Some(m) => m, None => { return Ok::<_, Infallible>(error_json( StatusCode::SERVICE_UNAVAILABLE, "AI 插件未就绪", )) } }; let model_id = match body.get("model_id").and_then(|v| v.as_str()) { Some(s) => s.to_string(), None => return Ok(error_json(StatusCode::BAD_REQUEST, "缺少 model_id")), }; let kind_str = match body.get("kind").and_then(|v| v.as_str()) { Some(s) => s, None => return Ok(error_json(StatusCode::BAD_REQUEST, "缺少 kind")), }; let kind = match kind_str { "llm" => crate::plugins::ai::model_manager::ModelKind::Llm, "asr" => crate::plugins::ai::model_manager::ModelKind::Asr, "tts" => crate::plugins::ai::model_manager::ModelKind::Tts, _ => return Ok(error_json(StatusCode::BAD_REQUEST, "kind 必须为 llm/asr/tts")), }; let mut mgr = models.lock().unwrap(); match mgr.switch_model(kind, &model_id) { Ok(()) => Ok(success_json(format!("模型已切换为 {}", model_id))), Err(e) => Ok(error_json(StatusCode::BAD_REQUEST, &e.to_string())), } }) } /// POST /api/models/delete — 删除模型 fn model_delete_route( state: Arc, ) -> impl Filter + Clone { warp::path!("api" / "models" / "delete") .and(warp::post()) .and(warp::body::json::()) .and(with_state(state)) .and_then(|req: ModelActionRequest, state: Arc| async move { let models = match state.ai_models() { Some(m) => m, None => { return Ok::<_, Infallible>(error_json( StatusCode::SERVICE_UNAVAILABLE, "AI 插件未就绪", )) } }; let mut mgr = models.lock().unwrap(); match mgr.delete_model(&req.model_id) { Ok(()) => Ok(success_json(format!("模型 {} 已删除", req.model_id))), Err(e) => Ok(error_json(StatusCode::BAD_REQUEST, &e.to_string())), } }) } fn json_response(status: StatusCode, payload: &T) -> warp::reply::Response { warp::reply::with_status(warp::reply::json(payload), status).into_response() } const WEB_UI_HTML: &str = r##" Showen 控制台

Showen 控制台

WS 连接中...

播放状态

状态--
当前视频--
索引--
列表长度--

播放操作

触发器

上传视频

设备视频文件

加载中...

文件管理器

选择目录后加载...

网络状态

WiFi--
SSID--
IP--
BLE--

WiFi 扫描

点击扫描搜索附近网络

连接 WiFi

热点 / BLE

配置文件切换

显示设置

配置编辑器

"##;