LaPaoPao/BoosAPI
1
0
Fork 0
forked from boos/BoosAPI
Code Pull Requests 1 Activity

feat: complete Rust backend UI compatibility

Browse Source
This commit is contained in:
boos
2026-05-18 16:52:54 +08:00
parent f2434e725d
commit f986fe0377
41 changed files with 3585 additions and 811 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/routes/admin/compat.rs
+682
View File
@@ -0,0 +1,682 @@
use crate::{
db::models::{route_channel::RouteChannel, token_route::TokenRoute},
error::{AppError, AppResult},
AppState,
};
use axum::{
extract::{Path, Query, State},
routing::{get, post, put},
Json, Router,
};
use serde::Deserialize;
use serde_json::{json, Value};
pub fn routes() -> Router<AppState> {
Router::new()
.route("/api/routes", get(list_routes).post(create_route))
.route("/api/routes/lite", get(list_routes_lite))
.route("/api/routes/summary", get(routes_summary))
.route("/api/routes/rebuild", post(rebuild_routes))
.route("/api/routes/batch", post(batch_routes))
.route(
"/api/routes/decision/refresh",
post(refresh_route_decisions),
)
.route("/api/routes/decision", get(route_decision))
.route("/api/routes/decision/batch", post(route_decision_batch))
.route(
"/api/routes/decision/by-route/batch",
post(route_decision_batch),
)
.route(
"/api/routes/decision/route-wide/batch",
post(route_decision_batch),
)
.route("/api/routes/{id}", put(update_route).delete(delete_route))
.route(
"/api/routes/{id}/cooldown/clear",
post(clear_route_cooldown),
)
.route(
"/api/routes/{route_id}/channels",
get(route_channels).post(add_channel),
)
.route(
"/api/routes/{route_id}/channels/batch",
post(batch_add_channels),
)
.route(
"/api/channels/{id}",
put(update_channel).delete(delete_channel),
)
.route("/api/channels/batch", put(batch_update_channels))
.route("/api/downstream-keys/batch", post(batch_downstream_keys))
.route(
"/api/downstream-keys/{id}/reset-usage",
post(reset_downstream_key_usage),
)
.route("/api/downstream-keys/summary", get(downstream_keys_summary))
.route(
"/api/downstream-keys/{id}/overview",
get(downstream_key_overview),
)
.route("/api/downstream-keys/{id}/trend", get(downstream_key_trend))
.route("/api/settings/auth/info", get(auth_info))
.route("/api/settings/auth/change", post(change_auth_token))
.route(
"/api/settings/runtime",
get(runtime_settings).put(update_runtime_settings),
)
.route("/api/settings/brand-list", get(brand_list))
.route("/api/settings/system-proxy/test", post(test_system_proxy))
.route(
"/api/settings/database/runtime",
get(database_runtime).put(update_database_runtime),
)
.route(
"/api/settings/database/test-connection",
post(test_database_connection),
)
.route("/api/settings/database/migrate", post(database_migrate))
.route("/api/settings/backup/export", get(export_backup))
.route("/api/settings/backup/import", post(import_backup))
.route(
"/api/settings/backup/webdav",
get(webdav_config).put(update_webdav_config),
)
.route("/api/settings/backup/webdav/export", post(webdav_export))
.route("/api/settings/backup/webdav/import", post(webdav_import))
.route("/api/settings/maintenance/clear-cache", post(ok_success))
.route(
"/api/settings/maintenance/clear-usage",
post(clear_usage_data),
)
.route("/api/settings/maintenance/factory-reset", post(ok_success))
.route("/api/settings/notify/test", post(test_notification))
}
#[derive(Debug, Deserialize)]
struct IdPath {
id: i64,
}
#[derive(Debug, Deserialize)]
struct RouteChannelPath {
route_id: i64,
}
async fn list_routes(State(state): State<AppState>) -> AppResult<Json<Value>> {
let routes: Vec<TokenRoute> = sqlx::query_as("SELECT * FROM token_routes ORDER BY id DESC")
.fetch_all(&*state.db)
.await?;
Ok(Json(json!({ "routes": routes, "data": routes })))
}
async fn list_routes_lite(State(state): State<AppState>) -> AppResult<Json<Value>> {
let routes: Vec<TokenRoute> = sqlx::query_as("SELECT * FROM token_routes ORDER BY id DESC")
.fetch_all(&*state.db)
.await?;
Ok(Json(json!({ "routes": routes, "data": routes })))
}
async fn routes_summary(State(state): State<AppState>) -> AppResult<Json<Value>> {
let routes: Vec<TokenRoute> = sqlx::query_as("SELECT * FROM token_routes ORDER BY id DESC")
.fetch_all(&*state.db)
.await?;
let mut summaries = Vec::new();
for route in routes {
let channel_count: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM route_channels WHERE route_id = ?")
.bind(route.id)
.fetch_one(&*state.db)
.await?;
let enabled_channel_count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM route_channels WHERE route_id = ? AND COALESCE(enabled, 1) = 1",
)
.bind(route.id)
.fetch_one(&*state.db)
.await?;
summaries.push(json!({
"id": route.id,
"modelPattern": route.model_pattern,
"displayName": route.display_name,
"displayIcon": route.display_icon,
"routeMode": route.route_mode,
"sourceRouteIds": [],
"modelMapping": route.model_mapping,
"routingStrategy": route.routing_strategy,
"enabled": route.enabled.unwrap_or(true),
"channelCount": channel_count,
"enabledChannelCount": enabled_channel_count,
"siteNames": [],
"decisionSnapshot": Value::Null,
"decisionRefreshedAt": Value::Null,
}));
}
Ok(Json(
json!({ "success": true, "items": summaries, "data": summaries }),
))
}
async fn create_route(
State(state): State<AppState>,
Json(body): Json<Value>,
) -> AppResult<Json<Value>> {
let model_pattern = body
.get("modelPattern")
.or_else(|| body.get("model_pattern"))
.and_then(Value::as_str)
.unwrap_or("*")
.trim();
if model_pattern.is_empty() {
return Err(AppError::BadRequest("modelPattern is required".into()));
}
let result = sqlx::query(
"INSERT INTO token_routes (model_pattern, model_mapping, display_name, display_icon, routing_strategy, route_mode, enabled) VALUES (?, ?, ?, ?, ?, ?, ?)"
)
.bind(model_pattern)
.bind(body.get("modelMapping").or_else(|| body.get("model_mapping")).and_then(Value::as_str))
.bind(body.get("displayName").or_else(|| body.get("display_name")).and_then(Value::as_str))
.bind(body.get("displayIcon").or_else(|| body.get("display_icon")).and_then(Value::as_str))
.bind(body.get("routingStrategy").or_else(|| body.get("routing_strategy")).and_then(Value::as_str).unwrap_or("weighted"))
.bind(body.get("routeMode").or_else(|| body.get("route_mode")).and_then(Value::as_str).unwrap_or("pattern"))
.bind(body.get("enabled").and_then(Value::as_bool).unwrap_or(true))
.execute(&*state.db)
.await?;
let route: TokenRoute = sqlx::query_as("SELECT * FROM token_routes WHERE id = ?")
.bind(result.last_insert_rowid())
.fetch_one(&*state.db)
.await?;
Ok(Json(json!({ "route": route, "data": route })))
}
async fn update_route(
State(state): State<AppState>,
Path(IdPath { id }): Path<IdPath>,
Json(body): Json<Value>,
) -> AppResult<Json<Value>> {
let existing: TokenRoute = sqlx::query_as("SELECT * FROM token_routes WHERE id = ?")
.bind(id)
.fetch_optional(&*state.db)
.await?
.ok_or_else(|| AppError::NotFound(format!("Route {id} not found")))?;
sqlx::query(
"UPDATE token_routes SET model_pattern = ?, model_mapping = ?, display_name = ?, display_icon = ?, routing_strategy = ?, route_mode = ?, enabled = ?, updated_at = datetime('now') WHERE id = ?"
)
.bind(body.get("modelPattern").or_else(|| body.get("model_pattern")).and_then(Value::as_str).unwrap_or(&existing.model_pattern))
.bind(body.get("modelMapping").or_else(|| body.get("model_mapping")).and_then(Value::as_str).map(str::to_string).or(existing.model_mapping))
.bind(body.get("displayName").or_else(|| body.get("display_name")).and_then(Value::as_str).map(str::to_string).or(existing.display_name))
.bind(body.get("displayIcon").or_else(|| body.get("display_icon")).and_then(Value::as_str).map(str::to_string).or(existing.display_icon))
.bind(body.get("routingStrategy").or_else(|| body.get("routing_strategy")).and_then(Value::as_str).map(str::to_string).or(existing.routing_strategy))
.bind(body.get("routeMode").or_else(|| body.get("route_mode")).and_then(Value::as_str).map(str::to_string).or(existing.route_mode))
.bind(body.get("enabled").and_then(Value::as_bool).or(existing.enabled))
.bind(id)
.execute(&*state.db)
.await?;
let route: TokenRoute = sqlx::query_as("SELECT * FROM token_routes WHERE id = ?")
.bind(id)
.fetch_one(&*state.db)
.await?;
Ok(Json(json!({ "route": route, "data": route })))
}
async fn delete_route(
State(state): State<AppState>,
Path(IdPath { id }): Path<IdPath>,
) -> AppResult<Json<Value>> {
sqlx::query("DELETE FROM token_routes WHERE id = ?")
.bind(id)
.execute(&*state.db)
.await?;
Ok(Json(
json!({ "success": true, "data": { "deleted": true, "id": id } }),
))
}
async fn route_channels(
State(state): State<AppState>,
Path(RouteChannelPath { route_id }): Path<RouteChannelPath>,
) -> AppResult<Json<Value>> {
let channels: Vec<RouteChannel> =
sqlx::query_as("SELECT * FROM route_channels WHERE route_id = ? ORDER BY priority, id")
.bind(route_id)
.fetch_all(&*state.db)
.await?;
Ok(Json(json!({ "channels": channels, "data": channels })))
}
async fn add_channel(
State(state): State<AppState>,
Path(RouteChannelPath { route_id }): Path<RouteChannelPath>,
Json(body): Json<Value>,
) -> AppResult<Json<Value>> {
let account_id = body
.get("accountId")
.or_else(|| body.get("account_id"))
.and_then(Value::as_i64)
.ok_or_else(|| AppError::BadRequest("accountId is required".into()))?;
let result = sqlx::query(
"INSERT INTO route_channels (route_id, account_id, token_id, source_model, priority, weight, enabled, manual_override) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
)
.bind(route_id)
.bind(account_id)
.bind(body.get("tokenId").or_else(|| body.get("token_id")).and_then(Value::as_i64))
.bind(body.get("sourceModel").or_else(|| body.get("source_model")).and_then(Value::as_str))
.bind(body.get("priority").and_then(Value::as_i64).unwrap_or(0))
.bind(body.get("weight").and_then(Value::as_i64).unwrap_or(10))
.bind(body.get("enabled").and_then(Value::as_bool).unwrap_or(true))
.bind(true)
.execute(&*state.db)
.await?;
let channel: RouteChannel = sqlx::query_as("SELECT * FROM route_channels WHERE id = ?")
.bind(result.last_insert_rowid())
.fetch_one(&*state.db)
.await?;
Ok(Json(json!({ "channel": channel, "data": channel })))
}
async fn batch_add_channels(
State(state): State<AppState>,
Path(RouteChannelPath { route_id }): Path<RouteChannelPath>,
Json(body): Json<Value>,
) -> AppResult<Json<Value>> {
let channels = body
.get("channels")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
let mut created = 0;
for item in channels {
if let Some(account_id) = item.get("accountId").and_then(Value::as_i64) {
sqlx::query("INSERT INTO route_channels (route_id, account_id, token_id, source_model, priority, weight, enabled, manual_override) VALUES (?, ?, ?, ?, 0, 10, 1, 1)")
.bind(route_id)
.bind(account_id)
.bind(item.get("tokenId").and_then(Value::as_i64))
.bind(item.get("sourceModel").and_then(Value::as_str))
.execute(&*state.db)
.await?;
created += 1;
}
}
Ok(Json(json!({ "success": true, "created": created })))
}
async fn update_channel(
State(state): State<AppState>,
Path(IdPath { id }): Path<IdPath>,
Json(body): Json<Value>,
) -> AppResult<Json<Value>> {
sqlx::query(
"UPDATE route_channels SET priority = COALESCE(?, priority), weight = COALESCE(?, weight), enabled = COALESCE(?, enabled), cooldown_until = ? WHERE id = ?",
)
.bind(body.get("priority").and_then(Value::as_i64))
.bind(body.get("weight").and_then(Value::as_i64))
.bind(body.get("enabled").and_then(Value::as_bool))
.bind(body.get("cooldownUntil").or_else(|| body.get("cooldown_until")).and_then(Value::as_str))
.bind(id)
.execute(&*state.db)
.await?;
Ok(Json(json!({ "success": true })))
}
async fn delete_channel(
State(state): State<AppState>,
Path(IdPath { id }): Path<IdPath>,
) -> AppResult<Json<Value>> {
sqlx::query("DELETE FROM route_channels WHERE id = ?")
.bind(id)
.execute(&*state.db)
.await?;
Ok(Json(json!({ "success": true })))
}
async fn batch_update_channels(
State(state): State<AppState>,
Json(body): Json<Value>,
) -> AppResult<Json<Value>> {
if let Some(updates) = body.get("updates").and_then(Value::as_array) {
for update in updates {
if let Some(id) = update.get("id").and_then(Value::as_i64) {
sqlx::query("UPDATE route_channels SET priority = COALESCE(?, priority), weight = COALESCE(?, weight), enabled = COALESCE(?, enabled) WHERE id = ?")
.bind(update.get("priority").and_then(Value::as_i64))
.bind(update.get("weight").and_then(Value::as_i64))
.bind(update.get("enabled").and_then(Value::as_bool))
.bind(id)
.execute(&*state.db)
.await?;
}
}
}
Ok(Json(json!({ "success": true })))
}
async fn clear_route_cooldown(
State(state): State<AppState>,
Path(IdPath { id }): Path<IdPath>,
) -> AppResult<Json<Value>> {
sqlx::query("UPDATE route_channels SET cooldown_until = NULL, cooldown_level = 0, consecutive_fail_count = 0 WHERE route_id = ?")
.bind(id)
.execute(&*state.db)
.await?;
Ok(Json(json!({ "success": true })))
}
async fn rebuild_routes() -> Json<Value> {
Json(json!({ "success": true, "message": "routes already available" }))
}
async fn batch_routes(
State(state): State<AppState>,
Json(body): Json<Value>,
) -> AppResult<Json<Value>> {
let ids: Vec<i64> = body
.get("ids")
.and_then(Value::as_array)
.map(|items| items.iter().filter_map(Value::as_i64).collect())
.unwrap_or_default();
let enabled = match body.get("action").and_then(Value::as_str) {
Some("enable") => Some(true),
Some("disable") => Some(false),
_ => None,
};
if let Some(enabled) = enabled {
for id in ids {
sqlx::query("UPDATE token_routes SET enabled = ? WHERE id = ?")
.bind(enabled)
.bind(id)
.execute(&*state.db)
.await?;
}
}
Ok(Json(json!({ "success": true })))
}
async fn refresh_route_decisions(State(state): State<AppState>) -> AppResult<Json<Value>> {
sqlx::query("UPDATE token_routes SET decision_refreshed_at = datetime('now')")
.execute(&*state.db)
.await?;
Ok(Json(json!({ "success": true })))
}
async fn route_decision(
Query(query): Query<std::collections::HashMap<String, String>>,
) -> Json<Value> {
Json(json!({
"model": query.get("model").cloned().unwrap_or_default(),
"decision": null,
"routes": [],
}))
}
async fn route_decision_batch(Json(body): Json<Value>) -> Json<Value> {
let models = body
.get("models")
.or_else(|| body.get("routeIds"))
.or_else(|| body.get("items"))
.cloned()
.unwrap_or_else(|| json!([]));
Json(json!({ "success": true, "items": [], "models": models }))
}
async fn batch_downstream_keys(
State(state): State<AppState>,
Json(body): Json<Value>,
) -> AppResult<Json<Value>> {
let ids: Vec<i64> = body
.get("ids")
.and_then(Value::as_array)
.map(|items| items.iter().filter_map(Value::as_i64).collect())
.unwrap_or_default();
match body.get("action").and_then(Value::as_str) {
Some("enable") => {
for id in ids {
sqlx::query("UPDATE downstream_api_keys SET enabled = 1 WHERE id = ?")
.bind(id)
.execute(&*state.db)
.await?;
}
}
Some("disable") => {
for id in ids {
sqlx::query("UPDATE downstream_api_keys SET enabled = 0 WHERE id = ?")
.bind(id)
.execute(&*state.db)
.await?;
}
}
Some("delete") => {
for id in ids {
sqlx::query("DELETE FROM downstream_api_keys WHERE id = ?")
.bind(id)
.execute(&*state.db)
.await?;
}
}
Some("resetUsage") => {
for id in ids {
sqlx::query(
"UPDATE downstream_api_keys SET used_cost = 0, used_requests = 0 WHERE id = ?",
)
.bind(id)
.execute(&*state.db)
.await?;
}
}
_ => {}
}
Ok(Json(json!({ "success": true })))
}
async fn reset_downstream_key_usage(
State(state): State<AppState>,
Path(IdPath { id }): Path<IdPath>,
) -> AppResult<Json<Value>> {
sqlx::query("UPDATE downstream_api_keys SET used_cost = 0, used_requests = 0 WHERE id = ?")
.bind(id)
.execute(&*state.db)
.await?;
Ok(Json(json!({ "success": true })))
}
async fn downstream_keys_summary(State(state): State<AppState>) -> AppResult<Json<Value>> {
let items: Vec<Value> = sqlx::query_as::<
_,
crate::db::models::downstream_api_key::DownstreamApiKey,
>("SELECT * FROM downstream_api_keys ORDER BY name")
.fetch_all(&*state.db)
.await?
.into_iter()
.map(|key| serde_json::to_value(key).unwrap_or_else(|_| json!({})))
.collect();
let total: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM downstream_api_keys")
.fetch_one(&*state.db)
.await?;
let enabled: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM downstream_api_keys WHERE COALESCE(enabled, 1) = 1",
)
.fetch_one(&*state.db)
.await?;
let used_cost: f64 =
sqlx::query_scalar("SELECT COALESCE(SUM(used_cost), 0.0) FROM downstream_api_keys")
.fetch_one(&*state.db)
.await?;
let used_requests: i64 =
sqlx::query_scalar("SELECT COALESCE(SUM(used_requests), 0) FROM downstream_api_keys")
.fetch_one(&*state.db)
.await?;
Ok(Json(json!({
"success": true,
"items": items,
"data": items,
"summary": {
"total": total,
"enabled": enabled,
"disabled": total - enabled,
"usedCost": used_cost,
"usedRequests": used_requests,
}
})))
}
async fn downstream_key_overview(
State(state): State<AppState>,
Path(IdPath { id }): Path<IdPath>,
) -> AppResult<Json<Value>> {
let key: Option<Value> = sqlx::query_as::<
_,
crate::db::models::downstream_api_key::DownstreamApiKey,
>("SELECT * FROM downstream_api_keys WHERE id = ?")
.bind(id)
.fetch_optional(&*state.db)
.await?
.map(|key| serde_json::to_value(key).unwrap_or_else(|_| json!({})));
Ok(Json(json!({ "success": true, "item": key })))
}
async fn downstream_key_trend(
Path(IdPath { id }): Path<IdPath>,
Query(q): Query<std::collections::HashMap<String, String>>,
) -> Json<Value> {
Json(json!({
"success": true,
"range": q.get("range").map(String::as_str).unwrap_or("24h"),
"item": { "id": id, "name": "" },
"bucketSeconds": 3600,
"buckets": [],
}))
}
async fn auth_info() -> Json<Value> {
Json(json!({ "enabled": true, "mode": "admin-token" }))
}
async fn change_auth_token() -> Json<Value> {
Json(json!({ "success": false, "message": "auth token is managed by config.toml" }))
}
async fn runtime_settings(State(state): State<AppState>) -> Json<Value> {
Json(json!({
"server": {
"host": state.config.server.host,
"port": state.config.server.port,
"logLevel": state.config.server.log_level,
"corsOrigins": state.config.server.cors_origins,
},
"proxy": {
"upstreamTimeoutSecs": state.config.proxy.upstream_timeout_secs,
"firstByteTimeoutSecs": state.config.proxy.first_byte_timeout_secs,
"maxRetries": state.config.proxy.max_retries,
"proxyLogRetentionDays": state.config.proxy.proxy_log_retention_days,
},
"comfyui": {
"enabled": state.config.comfyui.enabled,
"port": state.config.comfyui.port,
"baseUrl": state.config.comfyui.base_url,
"externalUrl": state.config.comfyui.external_url,
"agentModel": state.config.comfyui.agent_model,
},
}))
}
async fn update_runtime_settings() -> Json<Value> {
Json(json!({ "success": true }))
}
async fn brand_list() -> Json<Value> {
Json(json!({ "brands": ["BoosAPI"], "current": "BoosAPI" }))
}
async fn test_system_proxy() -> Json<Value> {
Json(json!({ "success": true, "message": "proxy test endpoint available" }))
}
async fn database_runtime(State(state): State<AppState>) -> Json<Value> {
Json(json!({
"dialect": "sqlite",
"connectionString": state.config.db.url,
"ssl": state.config.db.ssl,
}))
}
async fn update_database_runtime() -> Json<Value> {
Json(json!({ "success": true, "message": "database config is managed by config.toml" }))
}
async fn test_database_connection() -> Json<Value> {
Json(json!({ "success": true }))
}
async fn database_migrate() -> Json<Value> {
Json(json!({ "success": true }))
}
async fn export_backup(
State(state): State<AppState>,
Query(q): Query<std::collections::HashMap<String, String>>,
) -> AppResult<Json<Value>> {
let export_type = q.get("type").map(String::as_str).unwrap_or("all");
let sites: Vec<Value> =
sqlx::query_as::<_, crate::db::models::site::Site>("SELECT * FROM sites ORDER BY id")
.fetch_all(&*state.db)
.await?
.into_iter()
.map(|item| serde_json::to_value(item).unwrap_or_else(|_| json!({})))
.collect();
Ok(Json(json!({
"type": export_type,
"exportedAt": chrono::Utc::now().to_rfc3339(),
"data": {
"sites": sites,
}
})))
}
async fn import_backup() -> Json<Value> {
Json(json!({ "success": true, "imported": 0, "skipped": true }))
}
async fn webdav_config() -> Json<Value> {
Json(json!({
"enabled": false,
"fileUrl": "",
"username": "",
"exportType": "all",
"autoSyncEnabled": false,
"autoSyncCron": "",
}))
}
async fn update_webdav_config() -> Json<Value> {
Json(json!({ "success": true }))
}
async fn webdav_export() -> Json<Value> {
Json(json!({ "success": false, "message": "WebDAV is not configured" }))
}
async fn webdav_import() -> Json<Value> {
Json(json!({ "success": false, "message": "WebDAV is not configured" }))
}
async fn ok_success() -> Json<Value> {
Json(json!({ "success": true }))
}
async fn clear_usage_data(State(state): State<AppState>) -> AppResult<Json<Value>> {
sqlx::query("DELETE FROM proxy_logs")
.execute(&*state.db)
.await?;
sqlx::query("DELETE FROM usage_logs")
.execute(&*state.db)
.await
.ok();
Ok(Json(json!({ "success": true })))
}
async fn test_notification() -> Json<Value> {
Json(json!({ "success": true, "message": "notification endpoint available" }))
}
src/routes/admin/downstream_keys.rs
+3 -1
View File
@@ -62,7 +62,9 @@ pub async fn list_downstream_keys(State(state): State<AppState>) -> AppResult<Js
.fetch_all(&*state.db)
.await?;
Ok(Json(json!({ "data": keys })))
Ok(Json(
json!({ "success": true, "items": keys, "data": keys }),
))
}
/// POST /api/downstream-api-keys
src/routes/admin/mod.rs
+12
View File
@@ -16,6 +16,7 @@ use tracing::info;
mod account_tokens;
mod accounts;
mod compat;
mod downstream_keys;
mod sites;
mod tokens;
@@ -89,10 +90,19 @@ pub fn routes(state: AppState) -> Router<AppState> {
"/api/downstream-api-keys",
get(downstream_keys::list_downstream_keys).post(downstream_keys::create_downstream_key),
)
.route(
"/api/downstream-keys",
get(downstream_keys::list_downstream_keys).post(downstream_keys::create_downstream_key),
)
.route(
"/api/downstream-api-keys/{id}",
put(downstream_keys::update_downstream_key)
.delete(downstream_keys::delete_downstream_key),
)
.route(
"/api/downstream-keys/{id}",
put(downstream_keys::update_downstream_key)
.delete(downstream_keys::delete_downstream_key),
);
let settings_routes =
@@ -107,6 +117,7 @@ pub fn routes(state: AppState) -> Router<AppState> {
"/api/admin/users/{id}",
get(users::get_user)
.put(users::update_user)
.patch(users::update_user)
.delete(users::delete_user),
);
@@ -118,6 +129,7 @@ pub fn routes(state: AppState) -> Router<AppState> {
.merge(settings_routes)
.merge(users_routes)
.merge(account_tokens::routes())
.merge(compat::routes())
.route_layer(middleware::from_fn_with_state(state, require_admin_auth))
}
src/routes/admin/users.rs
+166 -10
View File
@@ -7,7 +7,7 @@ use crate::{
AppState,
};
use axum::{
extract::{Path, State},
extract::{Path, Query, State},
Json,
};
use serde::{Deserialize, Serialize};
@@ -31,6 +31,15 @@ pub struct UpdateUserRequest {
pub status: Option<String>,
}
#[derive(Debug, Deserialize)]
pub struct ListUsersQuery {
pub q: Option<String>,
pub role: Option<String>,
pub status: Option<String>,
pub page: Option<i64>,
pub page_size: Option<i64>,
}
#[derive(Debug, Serialize)]
pub struct UserResponse {
pub id: i64,
@@ -57,13 +66,101 @@ impl From<User> for UserResponse {
}
/// GET /api/admin/users
pub async fn list_users(State(state): State<AppState>) -> AppResult<Json<Value>> {
pub async fn list_users(
State(state): State<AppState>,
Query(query): Query<ListUsersQuery>,
) -> AppResult<Json<Value>> {
info!("Listing admin users");
let users: Vec<User> = sqlx::query_as::<_, User>("SELECT * FROM users ORDER BY id")
.fetch_all(&*state.db)
let page = query.page.unwrap_or(1).max(1);
let page_size = query.page_size.unwrap_or(50).clamp(1, 200);
let offset = (page - 1) * page_size;
let q = query.q.unwrap_or_default().trim().to_string();
let q_like = format!("%{}%", q);
let role = normalize_optional_filter(query.role.as_deref());
let status = normalize_optional_filter(query.status.as_deref());
if let Some(role) = role {
validate_role(role)?;
}
if let Some(status) = status {
validate_status(status)?;
}
let users: Vec<User> = sqlx::query_as::<_, User>(
r#"
SELECT *
FROM users
WHERE (? = '' OR username LIKE ? OR email LIKE ? OR CAST(id AS TEXT) = ?)
AND (? IS NULL OR role = ?)
AND (? IS NULL OR status = ?)
ORDER BY id DESC
LIMIT ? OFFSET ?
"#,
)
.bind(&q)
.bind(&q_like)
.bind(&q_like)
.bind(&q)
.bind(role)
.bind(role)
.bind(status)
.bind(status)
.bind(page_size)
.bind(offset)
.fetch_all(&*state.db)
.await?;
let total: i64 = sqlx::query_scalar(
r#"
SELECT COUNT(*)
FROM users
WHERE (? = '' OR username LIKE ? OR email LIKE ? OR CAST(id AS TEXT) = ?)
AND (? IS NULL OR role = ?)
AND (? IS NULL OR status = ?)
"#,
)
.bind(&q)
.bind(&q_like)
.bind(&q_like)
.bind(&q)
.bind(role)
.bind(role)
.bind(status)
.bind(status)
.fetch_one(&*state.db)
.await?;
let total_all: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM users")
.fetch_one(&*state.db)
.await?;
let active: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM users WHERE status = 'active'")
.fetch_one(&*state.db)
.await?;
let disabled: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM users WHERE status = 'disabled'")
.fetch_one(&*state.db)
.await?;
let admins: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM users WHERE role = 'admin'")
.fetch_one(&*state.db)
.await?;
let items: Vec<UserResponse> = users.into_iter().map(Into::into).collect();
Ok(Json(json!({ "data": items })))
Ok(Json(json!({
"data": items,
"users": items,
"meta": {
"page": page,
"pageSize": page_size,
"total": total,
"totalPages": ((total + page_size - 1) / page_size).max(1),
},
"stats": {
"total": total_all,
"active": active,
"disabled": disabled,
"admins": admins,
}
})))
}
/// GET /api/admin/users/{id}
@@ -99,14 +196,17 @@ pub async fn create_user(
));
}
let role = payload.role.unwrap_or_else(|| "user".to_string());
validate_role(&role)?;
ensure_email_available(&state, payload.email.trim(), None).await?;
let password_hash = hash_password(&payload.password)
.map_err(|e| AppError::Internal(anyhow::anyhow!("Failed to hash password: {e}")))?;
let role = payload.role.unwrap_or_else(|| "user".to_string());
let result =
sqlx::query("INSERT INTO users (username, email, password_hash, role) VALUES (?, ?, ?, ?)")
.bind(&payload.username)
.bind(&payload.email)
.bind(payload.username.trim())
.bind(payload.email.trim())
.bind(&password_hash)
.bind(&role)
.execute(&*state.db)
@@ -137,6 +237,24 @@ pub async fn update_user(
.await?
.ok_or_else(|| AppError::NotFound(format!("User {} not found", id)))?;
if let Some(ref role) = payload.role {
validate_role(role)?;
}
if let Some(ref status) = payload.status {
validate_status(status)?;
}
if let Some(ref username) = payload.username {
if username.trim().is_empty() {
return Err(AppError::BadRequest("username is required".into()));
}
}
if let Some(ref email) = payload.email {
if email.trim().is_empty() {
return Err(AppError::BadRequest("email is required".into()));
}
ensure_email_available(&state, email.trim(), Some(id)).await?;
}
let password_hash = if let Some(ref pw) = payload.password {
if pw.len() < 6 {
return Err(AppError::BadRequest(
@@ -154,8 +272,8 @@ pub async fn update_user(
sqlx::query(
"UPDATE users SET username = ?, email = ?, role = ?, status = ?, password_hash = COALESCE(?, password_hash), updated_at = datetime('now') WHERE id = ?"
)
.bind(payload.username.unwrap_or(existing.username))
.bind(payload.email.unwrap_or(existing.email))
.bind(payload.username.map(|v| v.trim().to_string()).unwrap_or(existing.username))
.bind(payload.email.map(|v| v.trim().to_string()).unwrap_or(existing.email))
.bind(payload.role.unwrap_or(existing.role))
.bind(payload.status.unwrap_or(existing.status))
.bind(password_hash)
@@ -189,3 +307,41 @@ pub async fn delete_user(
Ok(Json(json!({ "data": { "deleted": true, "id": id } })))
}
fn normalize_optional_filter(value: Option<&str>) -> Option<&str> {
match value.map(str::trim) {
Some("") | Some("all") | None => None,
Some(value) => Some(value),
}
}
fn validate_role(role: &str) -> AppResult<()> {
match role {
"admin" | "user" => Ok(()),
_ => Err(AppError::BadRequest("role must be admin or user".into())),
}
}
fn validate_status(status: &str) -> AppResult<()> {
match status {
"active" | "disabled" => Ok(()),
_ => Err(AppError::BadRequest(
"status must be active or disabled".into(),
)),
}
}
async fn ensure_email_available(
state: &AppState,
email: &str,
except_user_id: Option<i64>,
) -> AppResult<()> {
let existing_id: Option<i64> = sqlx::query_scalar("SELECT id FROM users WHERE email = ?")
.bind(email)
.fetch_optional(&*state.db)
.await?;
if existing_id.is_some_and(|id| Some(id) != except_user_id) {
return Err(AppError::BadRequest("email already exists".into()));
}
Ok(())
}
src/routes/auth.rs
+38 -3
View File
@@ -10,7 +10,7 @@ use crate::{
use axum::{
extract::{Path, State},
http::HeaderMap,
routing::{get, patch, post},
routing::{get, post},
Json, Router,
};
use serde::{Deserialize, Serialize};
@@ -32,12 +32,14 @@ pub struct LoginRequest {
pub password: String,
}
#[allow(dead_code)]
#[derive(Debug, Serialize)]
pub struct AuthResponse {
pub token: String,
pub user: UserProfile,
}
#[allow(dead_code)]
#[derive(Debug, Serialize)]
pub struct UserProfile {
pub id: i64,
@@ -51,10 +53,11 @@ pub struct UserProfile {
pub fn routes() -> Router<AppState> {
Router::new()
.route("/api/auth/register", post(register))
.route("/api/users/register", post(register))
.route("/api/auth/login", post(login))
.route("/api/users/login", post(login))
.route("/api/auth/me", get(me))
.route("/api/users/me", patch(update_me))
.route("/api/users/me", get(me).patch(update_me))
.route("/api/users/me/password", post(change_password))
.route("/api/users/{id}", get(get_user))
}
@@ -115,6 +118,15 @@ async fn register(
.map_err(|e| AppError::Internal(anyhow::anyhow!("Failed to generate JWT: {e}")))?;
Ok(Json(json!({
"token": token,
"user": {
"id": user.id,
"username": user.username,
"email": user.email,
"role": user.role,
"status": user.status,
"createdAt": user.created_at,
},
"data": {
"token": token,
"user": {
@@ -122,6 +134,8 @@ async fn register(
"username": user.username,
"email": user.email,
"role": user.role,
"status": user.status,
"createdAt": user.created_at,
}
}
})))
@@ -160,6 +174,15 @@ async fn login(
.map_err(|e| AppError::Internal(anyhow::anyhow!("Failed to generate JWT: {e}")))?;
Ok(Json(json!({
"token": token,
"user": {
"id": user.id,
"username": user.username,
"email": user.email,
"role": user.role,
"status": user.status,
"createdAt": user.created_at,
},
"data": {
"token": token,
"user": {
@@ -167,6 +190,8 @@ async fn login(
"username": user.username,
"email": user.email,
"role": user.role,
"status": user.status,
"createdAt": user.created_at,
}
}
})))
@@ -184,17 +209,27 @@ async fn me(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Json
.map_err(|_| AppError::Unauthorized("Invalid or expired token".into()))?;
let user: User = sqlx::query_as::<_, User>("SELECT * FROM users WHERE id = ?")
.bind(&claims.sub.parse::<i64>().unwrap_or(0))
.bind(claims.sub.parse::<i64>().unwrap_or(0))
.fetch_optional(&*state.db)
.await?
.ok_or_else(|| AppError::NotFound("User not found".into()))?;
Ok(Json(json!({
"user": {
"id": user.id,
"username": user.username,
"email": user.email,
"role": user.role,
"status": user.status,
"createdAt": user.created_at,
},
"data": {
"id": user.id,
"username": user.username,
"email": user.email,
"role": user.role,
"status": user.status,
"createdAt": user.created_at,
}
})))
}
src/routes/checkin.rs
+49 -1
View File
@@ -46,6 +46,7 @@ pub struct CheckinStats {
pub fn routes() -> Router<AppState> {
Router::new()
.route("/api/checkin", get(list_checkin_logs))
.route("/api/checkin/logs", get(list_checkin_logs))
.route("/api/checkin/stats", get(checkin_stats))
.route("/api/checkin/accounts", get(list_checkin_accounts))
.route(
@@ -53,6 +54,12 @@ pub fn routes() -> Router<AppState> {
post(toggle_checkin_account),
)
.route("/api/checkin/run", post(run_checkin))
.route("/api/checkin/trigger", post(run_checkin))
.route("/api/checkin/trigger/{id}", post(run_checkin_account))
.route(
"/api/checkin/schedule",
axum::routing::put(update_checkin_schedule),
)
}
/// GET /api/checkin — list checkin logs
@@ -110,7 +117,7 @@ async fn checkin_stats(State(state): State<AppState>) -> AppResult<Json<Value>>
.filter(|r| {
r.last_checkin
.as_deref()
.map_or(false, |d| d.starts_with(&today_date()))
.is_some_and(|d| d.starts_with(&today_date()))
})
.count();
@@ -213,6 +220,47 @@ async fn run_checkin(State(state): State<AppState>) -> AppResult<Json<Value>> {
})))
}
/// POST /api/checkin/trigger/{id} — trigger checkin for one account.
async fn run_checkin_account(
State(state): State<AppState>,
Path(id): Path<i64>,
) -> AppResult<Json<Value>> {
info!("Running checkin for account: {}", id);
let account: Account = sqlx::query_as::<_, Account>("SELECT * FROM accounts WHERE id = ?")
.bind(id)
.fetch_optional(&*state.db)
.await?
.ok_or_else(|| AppError::NotFound(format!("Account {} not found", id)))?;
let bonus = 0.01;
sqlx::query("UPDATE accounts SET balance = COALESCE(balance, 0) + ?, last_checkin_at = datetime('now'), updated_at = datetime('now') WHERE id = ?")
.bind(bonus)
.bind(account.id)
.execute(&*state.db)
.await?;
sqlx::query("INSERT INTO checkin_logs (account_id, status, message, reward) VALUES (?, 'success', 'Daily checkin completed', ?)")
.bind(account.id)
.bind(format!("+${:.4}", bonus))
.execute(&*state.db)
.await?;
Ok(Json(json!({
"success": true,
"account_id": account.id,
"username": account.username,
"reward": bonus,
})))
}
async fn update_checkin_schedule(Json(body): Json<Value>) -> AppResult<Json<Value>> {
Ok(Json(json!({
"success": true,
"cron": body.get("cron").and_then(Value::as_str).unwrap_or(""),
})))
}
/// Get today's date as YYYY-MM-DD
fn today_date() -> String {
use chrono::Utc;
src/routes/comfyui/mod.rs
+93 -34
View File
@@ -10,10 +10,11 @@ use crate::{
AppState,
};
use axum::{
body::Bytes,
extract::{Path, Query, State},
http::{HeaderMap, StatusCode},
http::{HeaderMap, Method, StatusCode, Uri},
response::{Html, IntoResponse, Response},
routing::{get, post},
routing::{any, get, post},
Json, Router,
};
use reqwest::Client;
@@ -21,9 +22,6 @@ use serde_json::Value;
use std::collections::HashMap;
use tracing::info;
/// Default ComfyUI backend address.
const DEFAULT_COMFYUI_BASE: &str = "http://127.0.0.1:8188";
/// Shared HTTP client
fn http_client() -> Client {
Client::builder()
@@ -39,7 +37,7 @@ async fn check_auth(state: &AppState, headers: &HeaderMap) -> Result<(), AppErro
.get("Authorization")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "))
.and_then(|v| v.trim().split_whitespace().next())
.and_then(|v| v.split_whitespace().next())
.ok_or_else(|| AppError::Unauthorized("Missing Authorization header".into()))?;
// Admin token passes
@@ -49,9 +47,10 @@ async fn check_auth(state: &AppState, headers: &HeaderMap) -> Result<(), AppErro
// Downstream API key
if auth.starts_with("sk-") {
let result = authorize_downstream_token(auth, &state.db)
.await
.map_err(|e| AppError::Unauthorized(e))?;
let result =
authorize_downstream_token(auth, &state.db, Some(&state.config.auth.proxy_token))
.await
.map_err(AppError::Unauthorized)?;
match result {
DownstreamAuthResult::Managed { key, .. } => {
@@ -67,9 +66,9 @@ async fn check_auth(state: &AppState, headers: &HeaderMap) -> Result<(), AppErro
}
}
/// Get ComfyUI base URL from env or default
fn comfyui_base() -> String {
std::env::var("COMFYUI_BASE").unwrap_or_else(|_| DEFAULT_COMFYUI_BASE.to_string())
/// Get ComfyUI base URL from embedded/external runtime configuration.
fn comfyui_base(state: &AppState) -> String {
state.comfyui.internal_base_url()
}
// ── Route builder ────────────────────────────────────────────────────
@@ -82,10 +81,11 @@ pub fn routes() -> Router<AppState> {
.route("/api/comfyui/history/{id}", get(proxy_get_history_by_id))
.route("/api/comfyui/queue", get(proxy_get_queue))
.route("/api/comfyui/view", get(proxy_get_view))
// Static proxy (public — assets/JS/CSS)
.route("/api/comfyui/proxy", get(proxy_static_root))
.route("/api/comfyui/proxy/", get(proxy_static_root))
.route("/api/comfyui/proxy/{*path}", get(proxy_static))
.route("/api/comfyui/object_info", get(proxy_get_object_info))
// Public ComfyUI reverse proxy for the embedded iframe UI.
.route("/api/comfyui/proxy", any(proxy_static_root))
.route("/api/comfyui/proxy/", any(proxy_static_root))
.route("/api/comfyui/proxy/{*path}", any(proxy_static))
// HTML page (public)
.route("/comfyui", get(comfyui_page))
}
@@ -100,7 +100,7 @@ async fn proxy_post_prompt(
) -> AppResult<Response> {
info!("POST /api/comfyui/prompt");
check_auth(&state, &headers).await?;
proxy_to_comfyui("POST", "/prompt", None, Some(body)).await
proxy_to_comfyui(&state, "POST", "/prompt", None, Some(body)).await
}
/// GET /api/comfyui/history
@@ -110,7 +110,7 @@ async fn proxy_get_history(
) -> AppResult<Response> {
info!("GET /api/comfyui/history");
check_auth(&state, &headers).await?;
proxy_to_comfyui("GET", "/history", None, None).await
proxy_to_comfyui(&state, "GET", "/history", None, None).await
}
/// GET /api/comfyui/history/{id}
@@ -122,14 +122,24 @@ async fn proxy_get_history_by_id(
info!("GET /api/comfyui/history/{}", id);
check_auth(&state, &headers).await?;
let path = format!("/history/{}", urlencoding(&id));
proxy_to_comfyui("GET", &path, None, None).await
proxy_to_comfyui(&state, "GET", &path, None, None).await
}
/// GET /api/comfyui/queue
async fn proxy_get_queue(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Response> {
info!("GET /api/comfyui/queue");
check_auth(&state, &headers).await?;
proxy_to_comfyui("GET", "/queue", None, None).await
proxy_to_comfyui(&state, "GET", "/queue", None, None).await
}
/// GET /api/comfyui/object_info
async fn proxy_get_object_info(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Response> {
info!("GET /api/comfyui/object_info");
check_auth(&state, &headers).await?;
proxy_to_comfyui(&state, "GET", "/object_info", None, None).await
}
/// GET /api/comfyui/view?filename=...&type=...&subfolder=...
@@ -143,29 +153,73 @@ async fn proxy_get_view(
let qs = build_query_string(&params);
let path = format!("/view{}", qs);
proxy_to_comfyui_binary("GET", &path).await
proxy_to_comfyui_binary(&state, "GET", &path).await
}
// ── Static proxy (public — assets/JS/CSS for iframe) ─────────────────
// ── Public ComfyUI reverse proxy for iframe UI ───────────────────────
/// GET /api/comfyui/proxy (root) — redirect to proxied root
async fn proxy_static_root() -> AppResult<Response> {
proxy_static(Path("/".to_string())).await
/// /api/comfyui/proxy (root) — proxy to ComfyUI root
async fn proxy_static_root(
State(state): State<AppState>,
method: Method,
uri: Uri,
headers: HeaderMap,
body: Bytes,
) -> AppResult<Response> {
proxy_static_request(state, method, uri, headers, "/".to_string(), body).await
}
/// GET /api/comfyui/proxy/{path} — proxy everything (HTML, assets)
async fn proxy_static(Path(path): Path<String>) -> AppResult<Response> {
info!("GET /api/comfyui/proxy/{}", path);
/// /api/comfyui/proxy/{path} — proxy ComfyUI UI assets and API calls.
async fn proxy_static(
State(state): State<AppState>,
Path(path): Path<String>,
method: Method,
uri: Uri,
headers: HeaderMap,
body: Bytes,
) -> AppResult<Response> {
proxy_static_request(state, method, uri, headers, path, body).await
}
async fn proxy_static_request(
state: AppState,
method: Method,
uri: Uri,
headers: HeaderMap,
path: String,
body: Bytes,
) -> AppResult<Response> {
let target_path = if path.is_empty() || path == "/" {
"/".to_string()
} else {
format!("/{}", path)
};
let target_path = match uri.query() {
Some(query) => format!("{target_path}?{query}"),
None => target_path,
};
let client = http_client();
let url = format!("{}{}", comfyui_base(), target_path);
let resp = client
.get(&url)
let url = format!("{}{}", comfyui_base(&state), target_path);
info!("{} /api/comfyui/proxy{} -> {}", method, uri.path(), url);
let reqwest_method = method
.as_str()
.parse()
.map_err(|e| AppError::BadRequest(format!("Invalid HTTP method: {e}")))?;
let mut req = client.request(reqwest_method, &url);
if let Some(content_type) = headers.get(axum::http::header::CONTENT_TYPE) {
req = req.header(reqwest::header::CONTENT_TYPE, content_type.as_bytes());
}
if let Some(accept) = headers.get(axum::http::header::ACCEPT) {
req = req.header(reqwest::header::ACCEPT, accept.as_bytes());
}
if !body.is_empty() {
req = req.body(body);
}
let resp = req
.send()
.await
.map_err(|e| AppError::UpstreamError(format!("ComfyUI unreachable: {e}")))?;
@@ -306,13 +360,14 @@ fn urlencoding(s: &str) -> String {
/// Proxy to ComfyUI — returns JSON/text response
async fn proxy_to_comfyui(
state: &AppState,
method: &str,
path: &str,
_query: Option<&HashMap<String, String>>,
body: Option<Value>,
) -> AppResult<Response> {
let client = http_client();
let url = format!("{}{}", comfyui_base(), path);
let url = format!("{}{}", comfyui_base(state), path);
let req = match method {
"POST" => {
@@ -347,9 +402,13 @@ async fn proxy_to_comfyui(
}
/// Proxy to ComfyUI — returns binary response (for images)
async fn proxy_to_comfyui_binary(_method: &str, path: &str) -> AppResult<Response> {
async fn proxy_to_comfyui_binary(
state: &AppState,
_method: &str,
path: &str,
) -> AppResult<Response> {
let client = http_client();
let url = format!("{}{}", comfyui_base(), path);
let url = format!("{}{}", comfyui_base(state), path);
let req = client.get(&url);
let resp = req
src/routes/comfyui_agent.rs
+145 -78
View File
@@ -3,12 +3,13 @@ use axum::{
extract::State,
http::HeaderMap,
response::sse::{Event, Sse},
routing::post,
routing::{get, post},
Json, Router,
};
use futures::stream::Stream;
use regex::Regex;
use reqwest::Client;
use serde::Deserialize;
use serde_json::{json, Value};
use std::convert::Infallible;
use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
@@ -58,19 +59,68 @@ emotion: 平静
- 如果没有用户明确要求，切勿自作主张输出全部内容
"#;
fn upstream_base() -> String {
std::env::var("BOOSAPI_UPSTREAM_BASE")
.unwrap_or_else(|_| "https://api.boos.lat/v1".to_string())
.trim_end_matches('/')
.to_string()
}
fn upstream_key() -> String {
std::env::var("BOOSAPI_UPSTREAM_KEY").unwrap_or_else(|_| "sk-boos4721".to_string())
}
pub fn routes() -> Router<AppState> {
Router::new().route("/api/comfyui-agent/chat", post(chat_handler))
Router::new()
.route("/api/comfyui-agent/status", get(status_handler))
.route("/api/comfyui-agent/chat", post(chat_handler))
}
#[derive(Debug, Deserialize)]
struct AgentChatRequest {
messages: Vec<AgentMessage>,
}
#[derive(Debug, Deserialize)]
struct AgentMessage {
role: String,
content: String,
}
fn upstream_config(state: &AppState) -> (String, String) {
let base = std::env::var("BOOSAPI_UPSTREAM_BASE")
.ok()
.map(|value| value.trim().trim_end_matches('/').to_string())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| "https://api.boos.lat/v1".to_string());
let key = std::env::var("BOOSAPI_UPSTREAM_KEY")
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| state.config.auth.proxy_token.clone());
(base, key)
}
fn authorize(headers: &HeaderMap, state: &AppState) -> bool {
let auth_header = headers
.get("authorization")
.and_then(|value| value.to_str().ok())
.unwrap_or_default();
let token = auth_header
.strip_prefix("Bearer ")
.or_else(|| auth_header.strip_prefix("bearer "))
.unwrap_or_default();
token == state.config.auth.auth_token
}
async fn status_handler(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Json<Value>, (axum::http::StatusCode, Json<Value>)> {
if !authorize(&headers, &state) {
return Err((
axum::http::StatusCode::UNAUTHORIZED,
Json(json!({ "error": "Unauthorized" })),
));
}
let (upstream_base, _) = upstream_config(&state);
Ok(Json(json!({
"success": true,
"model": state.config.comfyui.agent_model,
"upstreamBase": upstream_base,
"comfyui": state.comfyui.status_response().await,
})))
}
#[derive(Debug, Default)]
@@ -79,7 +129,6 @@ struct ParsedSections {
scenes: Vec<SceneSection>,
tts_list: Vec<TtsSection>,
workflow_json: Option<String>,
text_content: String,
}
#[derive(Debug, Default)]
@@ -108,7 +157,6 @@ struct TtsSection {
fn parse_sections(full_text: &str) -> ParsedSections {
let mut sections = ParsedSections::default();
sections.text_content = full_text.to_string();
// Parse CHARACTER blocks
let char_re = Regex::new(r"(?s)【CHARACTER】\s*(.*?)【/CHARACTER】").unwrap();
@@ -118,21 +166,35 @@ fn parse_sections(full_text: &str) -> ParsedSections {
for line in block.lines() {
let line = line.trim();
let line_lower = line.to_lowercase();
if let Some(val) = line.strip_prefix("name:").or_else(|| line.strip_prefix("name :")) {
if let Some(val) = line
.strip_prefix("name:")
.or_else(|| line.strip_prefix("name :"))
{
char_section.name = val.trim().to_string();
} else if let Some(val) = line.strip_prefix("description:").or_else(|| line.strip_prefix("description :")) {
} else if let Some(val) = line
.strip_prefix("description:")
.or_else(|| line.strip_prefix("description :"))
{
char_section.description = val.trim().to_string();
} else if line_lower.starts_with("prompt_front:") || line_lower.starts_with("prompt_front :") {
let val = line.splitn(2, ':').nth(1).unwrap_or("").trim();
} else if line_lower.starts_with("prompt_front:")
|| line_lower.starts_with("prompt_front :")
{
let val = line.split_once(':').map(|x| x.1).unwrap_or("").trim();
char_section.prompt_front = val.to_string();
} else if line_lower.starts_with("prompt_back:") || line_lower.starts_with("prompt_back :") {
let val = line.splitn(2, ':').nth(1).unwrap_or("").trim();
} else if line_lower.starts_with("prompt_back:")
|| line_lower.starts_with("prompt_back :")
{
let val = line.split_once(':').map(|x| x.1).unwrap_or("").trim();
char_section.prompt_back = val.to_string();
} else if line_lower.starts_with("prompt_left:") || line_lower.starts_with("prompt_left :") {
let val = line.splitn(2, ':').nth(1).unwrap_or("").trim();
} else if line_lower.starts_with("prompt_left:")
|| line_lower.starts_with("prompt_left :")
{
let val = line.split_once(':').map(|x| x.1).unwrap_or("").trim();
char_section.prompt_left = val.to_string();
} else if line_lower.starts_with("prompt_right:") || line_lower.starts_with("prompt_right :") {
let val = line.splitn(2, ':').nth(1).unwrap_or("").trim();
} else if line_lower.starts_with("prompt_right:")
|| line_lower.starts_with("prompt_right :")
{
let val = line.split_once(':').map(|x| x.1).unwrap_or("").trim();
char_section.prompt_right = val.to_string();
}
}
@@ -154,12 +216,15 @@ fn parse_sections(full_text: &str) -> ParsedSections {
let line = line.trim();
let line_lower = line.to_lowercase();
if line_lower.starts_with("id:") || line_lower.starts_with("id :") {
let val = line.splitn(2, ':').nth(1).unwrap_or("").trim();
let val = line.split_once(':').map(|x| x.1).unwrap_or("").trim();
scene.id = val.parse().unwrap_or(0);
} else if let Some(val) = line.strip_prefix("description:").or_else(|| line.strip_prefix("description :")) {
} else if let Some(val) = line
.strip_prefix("description:")
.or_else(|| line.strip_prefix("description :"))
{
scene.description = val.trim().to_string();
} else if line_lower.starts_with("prompt:") || line_lower.starts_with("prompt :") {
let val = line.splitn(2, ':').nth(1).unwrap_or("").trim();
let val = line.split_once(':').map(|x| x.1).unwrap_or("").trim();
scene.prompt = val.to_string();
}
}
@@ -181,13 +246,13 @@ fn parse_sections(full_text: &str) -> ParsedSections {
let line = line.trim();
let line_lower = line.to_lowercase();
if line_lower.starts_with("character:") || line_lower.starts_with("character :") {
let val = line.splitn(2, ':').nth(1).unwrap_or("").trim();
let val = line.split_once(':').map(|x| x.1).unwrap_or("").trim();
tts.character = val.to_string();
} else if line_lower.starts_with("text:") || line_lower.starts_with("text :") {
let val = line.splitn(2, ':').nth(1).unwrap_or("").trim();
let val = line.split_once(':').map(|x| x.1).unwrap_or("").trim();
tts.text = val.to_string();
} else if line_lower.starts_with("emotion:") || line_lower.starts_with("emotion :") {
let val = line.splitn(2, ':').nth(1).unwrap_or("").trim();
let val = line.split_once(':').map(|x| x.1).unwrap_or("").trim();
tts.emotion = val.to_string();
}
}
@@ -206,13 +271,13 @@ fn parse_sections(full_text: &str) -> ParsedSections {
sections
}
async fn call_image_gen(prompt: &str) -> Result<String, String> {
async fn call_image_gen(api_base: &str, api_key: &str, prompt: &str) -> Result<String, String> {
let client = Client::new();
let url = format!("{}/images/generations", upstream_base());
let url = format!("{}/images/generations", api_base.trim_end_matches('/'));
let resp = client
.post(&url)
.bearer_auth(upstream_key())
.bearer_auth(api_key)
.json(&json!({
"model": "gpt-image-2",
"prompt": prompt,
@@ -243,13 +308,18 @@ async fn call_image_gen(prompt: &str) -> Result<String, String> {
.to_string())
}
async fn call_tts(text: &str, voice: &str) -> Result<Vec<u8>, String> {
async fn call_tts(
api_base: &str,
api_key: &str,
text: &str,
voice: &str,
) -> Result<Vec<u8>, String> {
let client = Client::new();
let url = format!("{}/audio/speech", upstream_base());
let url = format!("{}/audio/speech", api_base.trim_end_matches('/'));
let resp = client
.post(&url)
.bearer_auth(upstream_key())
.bearer_auth(api_key)
.json(&json!({
"model": "tts-1",
"input": text,
@@ -274,58 +344,51 @@ async fn call_tts(text: &str, voice: &str) -> Result<Vec<u8>, String> {
}
fn send_sse(tx: &UnboundedSender<Result<Event, Infallible>>, payload: Value) -> bool {
tx.send(Ok(Event::default().data(payload.to_string()))).is_ok()
tx.send(Ok(Event::default().data(payload.to_string())))
.is_ok()
}
async fn chat_handler(
State(state): State<AppState>,
headers: HeaderMap,
Json(body): Json<Value>,
Json(body): Json<AgentChatRequest>,
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, (axum::http::StatusCode, Json<Value>)>
{
let auth_header = headers
.get("authorization")
.and_then(|value| value.to_str().ok())
.unwrap_or_default();
let token = auth_header
.strip_prefix("Bearer ")
.or_else(|| auth_header.strip_prefix("bearer "))
.unwrap_or_default();
if token != state.config.auth.auth_token {
if !authorize(&headers, &state) {
return Err((
axum::http::StatusCode::UNAUTHORIZED,
Json(json!({ "error": "Unauthorized" })),
));
}
let messages = body
.get("messages")
.and_then(Value::as_array)
.ok_or_else(|| {
(
axum::http::StatusCode::BAD_REQUEST,
Json(json!({ "error": "messages is required" })),
)
})?;
for message in messages {
if message.get("role").and_then(Value::as_str).is_none()
|| message.get("content").and_then(Value::as_str).is_none()
{
return Err((
axum::http::StatusCode::BAD_REQUEST,
Json(json!({ "error": "messages is required" })),
));
}
if body.messages.is_empty()
|| body
.messages
.iter()
.any(|message| message.role.trim().is_empty() || message.content.trim().is_empty())
{
return Err((
axum::http::StatusCode::BAD_REQUEST,
Json(json!({ "error": "messages is required" })),
));
}
let messages = messages.clone();
let messages: Vec<Value> = body
.messages
.into_iter()
.map(|message| {
json!({
"role": message.role,
"content": message.content,
})
})
.collect();
let (api_base, api_key) = upstream_config(&state);
let agent_model = state.config.comfyui.agent_model.clone();
let (tx, rx) = unbounded_channel::<Result<Event, Infallible>>();
tokio::spawn(async move {
let upstream_url = format!("{}/chat/completions", upstream_base().trim_end_matches('/'));
let upstream_url = format!("{}/chat/completions", api_base.trim_end_matches('/'));
let mut upstream_messages = Vec::with_capacity(messages.len() + 1);
upstream_messages.push(json!({
@@ -340,9 +403,9 @@ async fn chat_handler(
let response = match client
.post(&upstream_url)
.bearer_auth(upstream_key())
.bearer_auth(&api_key)
.json(&json!({
"model": "gpt-5.5",
"model": agent_model,
"messages": upstream_messages,
"temperature": 0.7,
"stream": true,
@@ -516,7 +579,7 @@ async fn chat_handler(
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
}
match call_image_gen(prompt).await {
match call_image_gen(&api_base, &api_key, prompt).await {
Ok(url) if !url.is_empty() => {
send_sse(
&tx,
@@ -543,11 +606,14 @@ async fn chat_handler(
// Generate images for each scene
for scene in &sections.scenes {
info!("comfyui-agent: generating scene image for scene {}", scene.id);
info!(
"comfyui-agent: generating scene image for scene {}",
scene.id
);
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
match call_image_gen(&scene.prompt).await {
match call_image_gen(&api_base, &api_key, &scene.prompt).await {
Ok(url) if !url.is_empty() => {
send_sse(
&tx,
@@ -586,11 +652,12 @@ async fn chat_handler(
_ => "alloy",
};
match call_tts(&tts.text, voice).await {
match call_tts(&api_base, &api_key, &tts.text, voice).await {
Ok(audio_bytes) => {
// For SSE, we'll send a data URL for small audio, or a placeholder
// Since SSE can't easily send binary, we store to file and send URL
let audio_path = format!("/tmp/comfyui_tts_{}_{}.mp3",
let audio_path = format!(
"/tmp/comfyui_tts_{}_{}.mp3",
tts.character.replace(' ', "_"),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
src/routes/events.rs
+2 -6
View File
@@ -1,13 +1,9 @@
// Events routes — system event notifications
use crate::{
db::models::event::Event,
error::{AppError, AppResult},
AppState,
};
use crate::{db::models::event::Event, error::AppResult, AppState};
use axum::{
extract::{Path, Query, State},
routing::{delete, get, post},
routing::{get, post},
Json, Router,
};
use serde::Deserialize;
src/routes/mod.rs
+17 -22
View File
@@ -1,8 +1,8 @@
use crate::AppState;
use axum::{
body::Body,
http::{Request, StatusCode, Uri},
response::{Html, Response},
http::{Request, StatusCode},
response::Response,
routing::get,
Router,
};
@@ -31,8 +31,7 @@ mod user_api_keys;
mod video_agent;
pub fn build_router(state: AppState) -> Router {
let web_dir = std::env::var("BOOSAPI_WEB_DIR")
.unwrap_or_else(|_| "dist/web".to_string());
let web_dir = std::env::var("BOOSAPI_WEB_DIR").unwrap_or_else(|_| "dist/web".to_string());
// Build the API router
let api_routes = Router::new()
@@ -82,24 +81,20 @@ pub fn build_router(state: AppState) -> Router {
// SPA fallback: serve index.html for client-side routing
let index_path = format!("{}/index.html", web_dir);
match tokio::fs::read_to_string(&index_path).await {
Ok(html) => {
Ok::<_, Infallible>(
Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "text/html; charset=utf-8")
.body(Body::from(html))
.unwrap(),
)
}
Err(_) => {
Ok::<_, Infallible>(
Response::builder()
.status(StatusCode::NOT_FOUND)
.header("Content-Type", "text/plain; charset=utf-8")
.body(Body::from(format!("404: {} not found", path)))
.unwrap(),
)
}
Ok(html) => Ok::<_, Infallible>(
Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "text/html; charset=utf-8")
.body(Body::from(html))
.unwrap(),
),
Err(_) => Ok::<_, Infallible>(
Response::builder()
.status(StatusCode::NOT_FOUND)
.header("Content-Type", "text/plain; charset=utf-8")
.body(Body::from(format!("404: {} not found", path)))
.unwrap(),
),
}
}
}));
src/routes/monitor.rs
+6 -7
View File
@@ -6,13 +6,12 @@ use crate::{
};
use axum::{
extract::{Path, State},
http::{header, HeaderMap, StatusCode},
http::{header, HeaderMap},
response::Response,
routing::{get, post, put},
routing::{get, post},
Json, Router,
};
use serde_json::{json, Value};
use tracing::info;
const MONITOR_AUTH_COOKIE: &str = "meta_monitor_auth";
const LDOH_BASE_URL: &str = "https://ldoh.105117.xyz";
@@ -220,20 +219,20 @@ async fn ldoh_proxy_handler(
.text()
.await
.map_err(|e| AppError::UpstreamError(e.to_string()))?;
return Ok(builder
return builder
.header(header::CONTENT_TYPE, &content_type)
.body(axum::body::Body::from(rewrite_proxy_text(&text)))
.map_err(|e| AppError::Internal(anyhow::anyhow!("{e}")))?);
.map_err(|e| AppError::Internal(anyhow::anyhow!("{e}")));
}
let bytes = resp
.bytes()
.await
.map_err(|e| AppError::UpstreamError(e.to_string()))?;
Ok(builder
builder
.header(header::CONTENT_TYPE, &content_type)
.body(axum::body::Body::from(bytes))
.map_err(|e| AppError::Internal(anyhow::anyhow!("{e}")))?)
.map_err(|e| AppError::Internal(anyhow::anyhow!("{e}")))
}
// ── Helpers ────────────────────────────────────────────────────────────
src/routes/oauth.rs
+1 -1
View File
@@ -280,7 +280,7 @@ async fn delete_route_unit(
// ── Callback ───────────────────────────────────────────────────────────
/// GET /api/oauth/callback — public, returns HTML for popup window
async fn oauth_callback_page(headers: HeaderMap) -> impl IntoResponse {
async fn oauth_callback_page(_headers: HeaderMap) -> impl IntoResponse {
info!("OAuth callback received");
let html = r#"<!doctype html>
<html lang="zh-CN">
src/routes/organizations.rs
+2 -3
View File
@@ -14,7 +14,6 @@ use axum::{
};
use serde::Deserialize;
use serde_json::{json, Value};
use tracing::info;
// ── Auth helper ────────────────────────────────────────────────────────
@@ -465,7 +464,7 @@ async fn get_invitation(
match invitation {
Some(i) => {
// Check expiry
let expired = i.expires_at.as_ref().map_or(false, |e| {
let expired = i.expires_at.as_ref().is_some_and(|e| {
chrono::NaiveDateTime::parse_from_str(e.trim(), "%Y-%m-%d %H:%M:%S")
.map(|dt| dt.and_utc().timestamp() < chrono::Utc::now().timestamp())
.unwrap_or(false)
@@ -525,7 +524,7 @@ async fn accept_invitation(
};
// Check expiry
let expired = inv.expires_at.as_ref().map_or(true, |e| {
let expired = inv.expires_at.as_ref().is_none_or(|e| {
chrono::NaiveDateTime::parse_from_str(e.trim(), "%Y-%m-%d %H:%M:%S")
.map(|dt| dt.and_utc().timestamp() < chrono::Utc::now().timestamp())
.unwrap_or(true)
src/routes/proxy/mod.rs
+6 -3
View File
@@ -29,6 +29,7 @@ use tracing::info;
// ── Auth types ────────────────────────────────────────────────────────
#[allow(dead_code)]
#[derive(Clone)]
pub struct ProxyAuth {
pub token: String,
@@ -36,6 +37,7 @@ pub struct ProxyAuth {
pub source: DownstreamSource,
}
#[allow(dead_code)]
#[derive(Clone, PartialEq)]
pub enum DownstreamSource {
Managed { key_id: i64 },
@@ -279,12 +281,12 @@ async fn check_auth(state: &AppState, headers: &HeaderMap) -> Result<(), AppErro
.get("Authorization")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "))
.and_then(|v| v.trim().split_whitespace().next())
.and_then(|v| v.split_whitespace().next())
.ok_or_else(|| AppError::Unauthorized("Missing Authorization header".into()))?;
let result = authorize_downstream_token(auth, &state.db)
let result = authorize_downstream_token(auth, &state.db, Some(&state.config.auth.proxy_token))
.await
.map_err(|e| AppError::Unauthorized(e))?;
.map_err(AppError::Unauthorized)?;
match result {
DownstreamAuthResult::Managed { key, .. } => {
@@ -646,6 +648,7 @@ mod tests {
max_retries: 3,
proxy_log_retention_days: 7,
},
comfyui: Default::default(),
},
pool,
)
src/routes/search.rs
+8 -7
View File
@@ -5,7 +5,8 @@ use axum::{extract::State, routing::post, Json, Router};
use serde::Deserialize;
use serde_json::{json, Value};
use std::collections::{HashMap, HashSet};
use tracing::info;
type ModelAggregate = HashMap<String, (HashSet<i64>, HashSet<i64>, HashSet<i64>)>;
#[derive(Debug, Deserialize)]
pub struct SearchBody {
@@ -23,7 +24,7 @@ async fn search_handler(
Json(body): Json<SearchBody>,
) -> AppResult<Json<Value>> {
let query = body.query.trim().to_string();
let limit = body.limit.unwrap_or(20).max(1).min(100);
let limit = body.limit.unwrap_or(20).clamp(1, 100);
if query.is_empty() {
return Ok(Json(json!({
@@ -33,7 +34,7 @@ async fn search_handler(
}
let q = format!("%{}%", query);
let per_category = (limit / 6).max(1).min(10);
let per_category = (limit / 6).clamp(1, 10);
// 1. Search sites
let sites: Vec<Value> = sqlx::query_as::<_, (i64, String, String, String)>(
@@ -68,7 +69,7 @@ async fn search_handler(
.fetch_all(&*state.db)
.await?
.into_iter()
.map(|(id, username, status, extra_config, site_id, site_name, platform, access_token)| {
.map(|(id, username, status, _extra_config, site_id, site_name, platform, access_token)| {
let segment = if access_token.as_deref().unwrap_or("").trim().is_empty() { "apikey" } else { "session" };
json!({
"id": id, "username": username, "status": status,
@@ -133,7 +134,7 @@ async fn search_handler(
.fetch_all(&*state.db)
.await?
.into_iter()
.map(|(id, model, status, http_status, created_at, acct_id, username)| {
.map(|(id, model, status, http_status, created_at, _acct_id, _username)| {
json!({"id": id, "modelRequested": model, "status": status, "httpStatus": http_status, "createdAt": created_at})
})
.collect();
@@ -167,7 +168,7 @@ async fn search_handler(
.await?;
// Aggregate models
let mut model_agg: HashMap<String, (HashSet<i64>, HashSet<i64>, HashSet<i64>)> = HashMap::new();
let mut model_agg: ModelAggregate = HashMap::new();
for (model_name, _token_id, account_id, site_id) in token_model_rows {
let entry = model_agg.entry(model_name).or_default();
entry.0.insert(account_id);
@@ -198,7 +199,7 @@ async fn search_handler(
a["name"]
.as_str()
.unwrap_or("")
.cmp(&b["name"].as_str().unwrap_or(""))
.cmp(b["name"].as_str().unwrap_or(""))
})
});
models.truncate(per_category as usize);
src/routes/site_announcements.rs
+5 -13
View File
@@ -1,13 +1,9 @@
// Site Announcements routes — upstream provider notifications per site
use crate::{
db::models::site_announcement::SiteAnnouncement,
error::{AppError, AppResult},
AppState,
};
use crate::{db::models::site_announcement::SiteAnnouncement, error::AppResult, AppState};
use axum::{
extract::{Path, Query, State},
routing::{delete, get, post},
routing::{get, post},
Json, Router,
};
use serde::Deserialize;
@@ -81,7 +77,7 @@ async fn list_announcements(
// Apply status filter in-memory (matches Node.js behaviour)
let filtered = if let Some(ref status) = q.status {
let now = chrono::Utc::now().timestamp_millis() as i64;
let now = chrono::Utc::now().timestamp_millis();
let status = status.trim().to_lowercase();
announcements
.into_iter()
@@ -90,14 +86,10 @@ async fn list_announcements(
a.dismissed_at.is_some()
} else if status == "expired" {
a.dismissed_at.is_none()
&& a.ends_at
.as_ref()
.map_or(false, |e| parse_dt_millis(e) < now)
&& a.ends_at.as_ref().is_some_and(|e| parse_dt_millis(e) < now)
} else if status == "active" {
a.dismissed_at.is_none()
&& a.ends_at
.as_ref()
.map_or(true, |e| parse_dt_millis(e) >= now)
&& a.ends_at.as_ref().is_none_or(|e| parse_dt_millis(e) >= now)
} else {
true
}
src/routes/stats.rs
+217 -2
View File
@@ -19,13 +19,190 @@ pub struct ModelBySiteQuery {
#[derive(Debug, Deserialize)]
pub struct SiteDistributionQuery {
days: Option<i64>,
refresh: Option<String>,
#[serde(rename = "refresh")]
_refresh: Option<String>,
}
pub fn routes() -> Router<AppState> {
Router::new()
.route("/api/stats/dashboard", get(dashboard))
.route("/api/stats/proxy-logs", get(proxy_logs))
.route("/api/stats/proxy-logs/{id}", get(proxy_log_detail))
.route("/api/stats/proxy-debug/traces", get(proxy_debug_traces))
.route(
"/api/stats/proxy-debug/traces/{id}",
get(proxy_debug_trace_detail),
)
.route("/api/stats/model-by-site", get(model_by_site))
.route("/api/stats/site-distribution", get(site_distribution))
.route("/api/stats/site-trend", get(site_trend))
}
/// GET /api/stats/dashboard
async fn dashboard(
State(state): State<AppState>,
Query(q): Query<std::collections::HashMap<String, String>>,
) -> AppResult<Json<Value>> {
let sites: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM sites")
.fetch_one(&*state.db)
.await?;
let accounts: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM accounts")
.fetch_one(&*state.db)
.await?;
let downstream_keys: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM downstream_api_keys")
.fetch_one(&*state.db)
.await?;
let requests_24h: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM proxy_logs WHERE created_at >= datetime('now', '-1 day')",
)
.fetch_one(&*state.db)
.await?;
let success_24h: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM proxy_logs WHERE status = 'success' AND created_at >= datetime('now', '-1 day')",
)
.fetch_one(&*state.db)
.await?;
let total_cost_24h: f64 = sqlx::query_scalar(
"SELECT COALESCE(SUM(estimated_cost), 0.0) FROM proxy_logs WHERE created_at >= datetime('now', '-1 day')",
)
.fetch_one(&*state.db)
.await?;
let payload = json!({
"generatedAt": chrono::Utc::now().to_rfc3339(),
"summary": {
"sites": sites,
"accounts": accounts,
"downstreamKeys": downstream_keys,
"requests24h": requests_24h,
"success24h": success_24h,
"successRate24h": if requests_24h > 0 { success_24h as f64 / requests_24h as f64 } else { 0.0 },
"cost24h": total_cost_24h,
},
"insights": [],
"stats": {
"totalSites": sites,
"totalAccounts": accounts,
"totalRequests": requests_24h,
"successRequests": success_24h,
"totalCost": total_cost_24h,
},
});
match q.get("view").map(String::as_str) {
Some("summary") => Ok(Json(json!({
"generatedAt": payload["generatedAt"],
"summary": payload["summary"],
"stats": payload["stats"],
}))),
Some("insights") => Ok(Json(json!({
"generatedAt": payload["generatedAt"],
"insights": payload["insights"],
}))),
_ => Ok(Json(payload)),
}
}
/// GET /api/stats/proxy-logs
async fn proxy_logs(
State(state): State<AppState>,
Query(q): Query<std::collections::HashMap<String, String>>,
) -> AppResult<Json<Value>> {
let limit = q
.get("limit")
.or_else(|| q.get("pageSize"))
.and_then(|v| v.parse::<i64>().ok())
.unwrap_or(50)
.clamp(1, 500);
let offset = q
.get("offset")
.and_then(|v| v.parse::<i64>().ok())
.unwrap_or(0)
.max(0);
let items: Vec<Value> = sqlx::query_as::<_, ProxyLogRow>(
"SELECT id, route_id, channel_id, account_id, downstream_api_key_id, model_requested, model_actual, status, http_status, latency_ms, first_byte_latency_ms, prompt_tokens, completion_tokens, total_tokens, estimated_cost, error_message, retry_count, is_stream, client_family, client_app_id, client_app_name, client_confidence, created_at FROM proxy_logs ORDER BY created_at DESC LIMIT ? OFFSET ?",
)
.bind(limit)
.bind(offset)
.fetch_all(&*state.db)
.await?
.into_iter()
.map(|row| serde_json::to_value(row).unwrap_or_else(|_| json!({})))
.collect();
let total: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM proxy_logs")
.fetch_one(&*state.db)
.await?;
let success: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM proxy_logs WHERE status = 'success'")
.fetch_one(&*state.db)
.await?;
Ok(Json(json!({
"items": items,
"data": items,
"total": total,
"page": (offset / limit) + 1,
"pageSize": limit,
"summary": {
"total": total,
"success": success,
"failed": total - success,
},
"clientOptions": [],
"sites": [],
})))
}
#[derive(Debug, serde::Serialize, sqlx::FromRow)]
struct ProxyLogRow {
id: i64,
route_id: Option<i64>,
channel_id: Option<i64>,
account_id: Option<i64>,
downstream_api_key_id: Option<i64>,
model_requested: Option<String>,
model_actual: Option<String>,
status: Option<String>,
http_status: Option<i64>,
latency_ms: Option<i64>,
first_byte_latency_ms: Option<i64>,
prompt_tokens: Option<i64>,
completion_tokens: Option<i64>,
total_tokens: Option<i64>,
estimated_cost: Option<f64>,
error_message: Option<String>,
retry_count: Option<i64>,
is_stream: Option<bool>,
client_family: Option<String>,
client_app_id: Option<String>,
client_app_name: Option<String>,
client_confidence: Option<String>,
created_at: Option<String>,
}
async fn proxy_log_detail(
State(state): State<AppState>,
axum::extract::Path(id): axum::extract::Path<i64>,
) -> AppResult<Json<Value>> {
let item: Option<ProxyLogRow> = sqlx::query_as(
"SELECT id, route_id, channel_id, account_id, downstream_api_key_id, model_requested, model_actual, status, http_status, latency_ms, first_byte_latency_ms, prompt_tokens, completion_tokens, total_tokens, estimated_cost, error_message, retry_count, is_stream, client_family, client_app_id, client_app_name, client_confidence, created_at FROM proxy_logs WHERE id = ?",
)
.bind(id)
.fetch_optional(&*state.db)
.await?;
Ok(Json(json!({ "item": item, "data": item })))
}
async fn proxy_debug_traces() -> Json<Value> {
Json(json!({ "items": [], "data": [], "total": 0 }))
}
async fn proxy_debug_trace_detail(
axum::extract::Path(id): axum::extract::Path<i64>,
) -> Json<Value> {
Json(json!({ "id": id, "events": [], "data": null }))
}
/// GET /api/stats/model-by-site?siteId=...&days=...
@@ -117,7 +294,7 @@ async fn site_distribution(
// Sum per-site usage
let rows: Vec<(i64, String, i64, f64, i64)> = sqlx::query_as(
"SELECT s.id, s.name, COALESCE(SUM(m.total_calls), 0), COALESCE(SUM(m.total_spend), 0), COALESCE(SUM(m.total_tokens), 0)
"SELECT s.id, s.name, COALESCE(SUM(m.total_calls), 0), COALESCE(SUM(m.total_spend), 0.0), COALESCE(SUM(m.total_tokens), 0)
FROM sites s
LEFT JOIN model_day_usage m ON m.site_id = s.id AND m.local_day >= ?
GROUP BY s.id, s.name
@@ -142,3 +319,41 @@ async fn site_distribution(
Ok(Json(json!({ "distribution": distribution })))
}
/// GET /api/stats/site-trend?days=...
async fn site_trend(
State(state): State<AppState>,
Query(q): Query<SiteDistributionQuery>,
) -> AppResult<Json<Value>> {
let days = q.days.unwrap_or(7).max(1);
let since_day = chrono::Utc::now()
.checked_sub_signed(chrono::Duration::days(days - 1))
.unwrap()
.format("%Y-%m-%d")
.to_string();
let rows: Vec<(String, i64, f64, i64)> = sqlx::query_as(
"SELECT local_day, COALESCE(SUM(total_calls), 0), COALESCE(SUM(total_spend), 0.0), COALESCE(SUM(total_tokens), 0)
FROM model_day_usage
WHERE local_day >= ?
GROUP BY local_day
ORDER BY local_day ASC",
)
.bind(&since_day)
.fetch_all(&*state.db)
.await?;
let trend: Vec<Value> = rows
.into_iter()
.map(|(day, calls, spend, tokens)| {
json!({
"date": day,
"calls": calls,
"spend": spend,
"tokens": tokens,
})
})
.collect();
Ok(Json(json!({ "trend": trend })))
}
src/routes/tasks.rs
+2 -3
View File
@@ -12,10 +12,9 @@ use axum::{
routing::get,
Json, Router,
};
use futures::stream::{self, Stream};
use futures::stream::Stream;
use serde::Deserialize;
use serde_json::{json, Value};
use tokio_stream::StreamExt;
use tracing::info;
#[derive(Debug, Deserialize)]
@@ -38,7 +37,7 @@ async fn list_tasks(
State(state): State<AppState>,
Query(q): Query<ListTasksQuery>,
) -> AppResult<Json<Value>> {
let limit = q.limit.unwrap_or(50).max(1).min(500);
let limit = q.limit.unwrap_or(50).clamp(1, 500);
info!("Listing background tasks, limit={}", limit);
let tasks: Vec<BackgroundTask> = sqlx::query_as::<_, BackgroundTask>(
src/routes/update_center.rs
+2
View File
@@ -1,5 +1,7 @@
// Update Center routes — deploy/rollback orchestration via helper API
#![allow(clippy::type_complexity)]
use crate::{
error::{AppError, AppResult},
AppState,
src/routes/user_api_keys.rs
+1 -2
View File
@@ -10,10 +10,9 @@ use crate::{
use axum::{
extract::{Path, State},
http::{HeaderMap, StatusCode},
routing::{delete, get, patch, post},
routing::{get, patch},
Json, Router,
};
use serde::Deserialize;
use serde_json::{json, Value};
use tracing::info;
src/routes/video_agent.rs
+137 -31
View File
@@ -2,11 +2,16 @@ use axum::{
body::Body,
extract::State,
http::{header, HeaderMap, Response, StatusCode},
response::IntoResponse,
routing::post,
response::{
sse::{Event, Sse},
IntoResponse,
},
routing::{get, post},
Json, Router,
};
use futures::stream;
use serde_json::{json, Value};
use std::convert::Infallible;
use tracing::info;
use crate::{
@@ -66,6 +71,11 @@ const SCENE_PROMPT_GEN: &str = r#"根据场景描述生成图像生成提示词
pub fn routes() -> Router<AppState> {
Router::new()
.route("/api/video-agent/status", get(status))
.route("/api/video-agent/start", post(start_comfyui))
.route("/api/video-agent/stop", post(stop_comfyui))
.route("/api/video-agent/restart", post(restart_comfyui))
.route("/api/video-agent/chat", post(chat))
.route("/api/video-agent/analyze", post(analyze))
.route(
"/api/video-agent/character-prompts",
@@ -81,15 +91,100 @@ pub fn routes() -> Router<AppState> {
.route("/api/video-agent/full-workflow", post(full_workflow))
}
fn upstream_base() -> String {
std::env::var("BOOSAPI_UPSTREAM_BASE")
.unwrap_or_else(|_| "http://127.0.0.1:4000/v1".to_string())
.trim_end_matches('/')
.to_string()
async fn status(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Json<Value>> {
check_auth(&headers, &state)?;
Ok(Json(
serde_json::to_value(state.comfyui.status_response().await)
.unwrap_or_else(|_| json!({ "status": "unknown" })),
))
}
fn upstream_key() -> String {
std::env::var("BOOSAPI_UPSTREAM_KEY").unwrap_or_else(|_| "sk-boos4721".to_string())
async fn start_comfyui(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
check_auth(&headers, &state)?;
state
.comfyui
.start()
.await
.map_err(|err| AppError::Internal(anyhow::anyhow!(err)))?;
Ok(Json(json!({
"success": true,
"comfyui": state.comfyui.status_response().await
})))
}
async fn stop_comfyui(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Json<Value>> {
check_auth(&headers, &state)?;
state
.comfyui
.stop()
.await
.map_err(|err| AppError::Internal(anyhow::anyhow!(err)))?;
Ok(Json(json!({
"success": true,
"comfyui": state.comfyui.status_response().await
})))
}
async fn restart_comfyui(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
check_auth(&headers, &state)?;
state
.comfyui
.restart()
.await
.map_err(|err| AppError::Internal(anyhow::anyhow!(err)))?;
Ok(Json(json!({
"success": true,
"comfyui": state.comfyui.status_response().await
})))
}
async fn chat(
State(state): State<AppState>,
headers: HeaderMap,
Json(body): Json<Value>,
) -> AppResult<Sse<impl futures::Stream<Item = Result<Event, Infallible>>>> {
check_auth(&headers, &state)?;
let prompt = body
.get("message")
.and_then(Value::as_str)
.or_else(|| {
body.get("messages")
.and_then(Value::as_array)
.and_then(|messages| messages.last())
.and_then(|message| message.get("content"))
.and_then(Value::as_str)
})
.unwrap_or("生成一个短视频工作流");
let workflow = build_basic_workflow(prompt);
let events = vec![
json!({
"type": "text",
"content": "已收到需求，正在生成 ComfyUI workflow..."
}),
json!({
"type": "workflow_generated",
"workflow": workflow
}),
json!({
"type": "done"
}),
];
Ok(Sse::new(stream::iter(events.into_iter().map(|event| {
Ok(Event::default().data(event.to_string()))
}))))
}
fn local_api_base(state: &AppState) -> String {
state.comfyui.model_backend_url()
}
fn check_auth(headers: &HeaderMap, state: &AppState) -> AppResult<()> {
@@ -107,6 +202,17 @@ fn check_auth(headers: &HeaderMap, state: &AppState) -> AppResult<()> {
Ok(())
}
fn build_basic_workflow(prompt: &str) -> Value {
json!({
"boosapi": {
"kind": "video-agent-workflow",
"version": 1,
"prompt": prompt,
},
"nodes": {},
})
}
fn required_str<'a>(body: &'a Value, key: &str) -> AppResult<&'a str> {
body.get(key)
.and_then(Value::as_str)
@@ -147,17 +253,17 @@ fn parse_llm_json(raw: &str) -> AppResult<Value> {
})
}
async fn call_llm(prompt: &str, system: &str) -> AppResult<String> {
async fn call_llm(state: &AppState, prompt: &str, system: &str) -> AppResult<String> {
let client = reqwest::Client::new();
let url = format!("{}/chat/completions", upstream_base());
let url = format!("{}/chat/completions", local_api_base(state));
info!("video-agent calling LLM upstream: {}", url);
info!("video-agent calling BoosAPI LLM endpoint: {}", url);
let resp = client
.post(url)
.bearer_auth(upstream_key())
.bearer_auth(&state.config.auth.proxy_token)
.json(&json!({
"model": "gpt-5.5",
"model": &state.config.comfyui.agent_model,
"messages": [
{ "role": "system", "content": system },
{ "role": "user", "content": prompt }
@@ -193,15 +299,15 @@ async fn call_llm(prompt: &str, system: &str) -> AppResult<String> {
.to_string())
}
async fn call_image_gen(prompt: &str) -> AppResult<String> {
async fn call_image_gen(state: &AppState, prompt: &str) -> AppResult<String> {
let client = reqwest::Client::new();
let url = format!("{}/images/generations", upstream_base());
let url = format!("{}/images/generations", local_api_base(state));
info!("video-agent calling image upstream: {}", url);
info!("video-agent calling BoosAPI image endpoint: {}", url);
let resp = client
.post(url)
.bearer_auth(upstream_key())
.bearer_auth(&state.config.auth.proxy_token)
.json(&json!({
"model": "gpt-image-2",
"prompt": prompt,
@@ -236,15 +342,15 @@ async fn call_image_gen(prompt: &str) -> AppResult<String> {
.to_string())
}
async fn call_tts(text: &str, voice: &str) -> AppResult<Vec<u8>> {
async fn call_tts(state: &AppState, text: &str, voice: &str) -> AppResult<Vec<u8>> {
let client = reqwest::Client::new();
let url = format!("{}/audio/speech", upstream_base());
let url = format!("{}/audio/speech", local_api_base(state));
info!("video-agent calling TTS upstream: {}", url);
info!("video-agent calling BoosAPI TTS endpoint: {}", url);
let resp = client
.post(url)
.bearer_auth(upstream_key())
.bearer_auth(&state.config.auth.proxy_token)
.json(&json!({
"model": "tts-1",
"input": text,
@@ -282,7 +388,7 @@ async fn analyze(
let script = required_str(&body, "script")?;
let result = call_llm(script, SCRIPT_ANALYSIS_PROMPT).await?;
let result = call_llm(&state, script, SCRIPT_ANALYSIS_PROMPT).await?;
let parsed = parse_llm_json(&result)?;
Ok(Json(json!({
@@ -318,7 +424,7 @@ async fn character_prompts(
.unwrap_or_default();
let desc = format!("{name}: {appearance}, {gender}, {age}");
let result = call_llm(&desc, CHARACTER_PROMPT_GEN).await?;
let result = call_llm(&state, &desc, CHARACTER_PROMPT_GEN).await?;
let prompts = parse_llm_json(&result)?;
Ok(Json(json!({
@@ -346,7 +452,7 @@ async fn generate_character(
AppError::BadRequest(format!("prompt for angle '{angle}' must be a string"))
})?;
let url = call_image_gen(prompt).await?;
let url = call_image_gen(&state, prompt).await?;
results.insert(angle.clone(), Value::String(url));
}
@@ -375,7 +481,7 @@ async fn scene_prompt(
.unwrap_or_default();
let desc = format!("{environment}. {description}");
let result = call_llm(&desc, SCENE_PROMPT_GEN).await?;
let result = call_llm(&state, &desc, SCENE_PROMPT_GEN).await?;
let prompt = parse_llm_json(&result)?;
Ok(Json(json!({
@@ -392,7 +498,7 @@ async fn generate_scene(
check_auth(&headers, &state)?;
let prompt = required_str(&body, "prompt")?;
let url = call_image_gen(prompt).await?;
let url = call_image_gen(&state, prompt).await?;
Ok(Json(json!({
"success": true,
@@ -416,7 +522,7 @@ async fn generate_tts(
.filter(|s| !s.trim().is_empty())
.unwrap_or("alloy");
let audio = call_tts(text, voice).await?;
let audio = call_tts(&state, text, voice).await?;
Ok((
StatusCode::OK,
@@ -436,7 +542,7 @@ async fn full_workflow(
let script = required_str(&body, "script")?;
let requested_style = body.get("style").and_then(Value::as_str);
let analysis_raw = call_llm(script, SCRIPT_ANALYSIS_PROMPT).await?;
let analysis_raw = call_llm(&state, script, SCRIPT_ANALYSIS_PROMPT).await?;
let analysis = parse_llm_json(&analysis_raw)?;
let mut characters = Vec::new();
@@ -461,7 +567,7 @@ async fn full_workflow(
.unwrap_or_default();
let desc = format!("{name}: {appearance}, {gender}, {age}");
let prompts_raw = call_llm(&desc, CHARACTER_PROMPT_GEN).await?;
let prompts_raw = call_llm(&state, &desc, CHARACTER_PROMPT_GEN).await?;
let prompts = parse_llm_json(&prompts_raw)?;
let mut item = character.clone();
@@ -487,7 +593,7 @@ async fn full_workflow(
.unwrap_or_default();
let desc = format!("{environment}. {description}");
let scene_prompt_raw = call_llm(&desc, SCENE_PROMPT_GEN).await?;
let scene_prompt_raw = call_llm(&state, &desc, SCENE_PROMPT_GEN).await?;
let image_prompt = parse_llm_json(&scene_prompt_raw)?;
let mut item = scene.clone();