//! 使用统计服务 //! //! 提供使用量数据的聚合查询功能 use crate::database::{lock_conn, Database}; use crate::error::AppError; use chrono::{Local, TimeZone}; use rusqlite::{params, Connection, OptionalExtension}; use serde::{Deserialize, Serialize}; use serde_json::Value; use std::collections::HashMap; use std::str::FromStr; /// 使用量汇总 #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct UsageSummary { pub total_requests: u64, pub total_cost: String, pub total_input_tokens: u64, pub total_output_tokens: u64, pub total_cache_creation_tokens: u64, pub total_cache_read_tokens: u64, pub success_rate: f32, } /// 每日统计 #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct DailyStats { pub date: String, pub request_count: u64, pub total_cost: String, pub total_tokens: u64, pub total_input_tokens: u64, pub total_output_tokens: u64, pub total_cache_creation_tokens: u64, pub total_cache_read_tokens: u64, } /// Provider 统计 #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct ProviderStats { pub provider_id: String, pub provider_name: String, pub request_count: u64, pub total_tokens: u64, pub total_cost: String, pub success_rate: f32, pub avg_latency_ms: u64, } /// 模型统计 #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct ModelStats { pub model: String, pub request_count: u64, pub total_tokens: u64, pub total_cost: String, pub avg_cost_per_request: String, } /// 请求日志过滤器 #[derive(Debug, Clone, Default, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LogFilters { pub app_type: Option, pub provider_name: Option, pub model: Option, pub status_code: Option, pub start_date: Option, pub end_date: Option, } /// 分页请求日志响应 #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] pub struct PaginatedLogs { pub data: Vec, pub total: u32, pub page: u32, pub page_size: u32, } /// 请求日志详情 #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct RequestLogDetail { pub request_id: String, pub provider_id: String, #[serde(skip_serializing_if = "Option::is_none")] pub provider_name: Option, pub app_type: String, pub model: String, #[serde(skip_serializing_if = "Option::is_none")] pub request_model: Option, pub cost_multiplier: String, pub input_tokens: u32, pub output_tokens: u32, pub cache_read_tokens: u32, pub cache_creation_tokens: u32, pub input_cost_usd: String, pub output_cost_usd: String, pub cache_read_cost_usd: String, pub cache_creation_cost_usd: String, pub total_cost_usd: String, pub is_streaming: bool, pub latency_ms: u64, pub first_token_ms: Option, pub duration_ms: Option, pub status_code: u16, pub error_message: Option, pub created_at: i64, #[serde(skip_serializing_if = "Option::is_none")] pub data_source: Option, } /// SQL fragment: resolve provider_name with fallback for session-based entries. /// Session logs use placeholder provider_ids (_session, _codex_session, _gemini_session) /// that don't exist in the providers table — this COALESCE gives them readable names. fn provider_name_coalesce(log_alias: &str, provider_alias: &str) -> String { format!( "COALESCE({provider_alias}.name, CASE {log_alias}.provider_id \ WHEN '_session' THEN 'Claude (Session)' \ WHEN '_codex_session' THEN 'Codex (Session)' \ WHEN '_gemini_session' THEN 'Gemini (Session)' \ ELSE {log_alias}.provider_id END)" ) } impl Database { /// 获取使用量汇总 pub fn get_usage_summary( &self, start_date: Option, end_date: Option, app_type: Option<&str>, ) -> Result { let conn = lock_conn!(self.conn); // Build detail WHERE clause let mut conditions = Vec::new(); let mut params_vec: Vec> = Vec::new(); if let Some(start) = start_date { conditions.push("created_at >= ?"); params_vec.push(Box::new(start)); } if let Some(end) = end_date { conditions.push("created_at <= ?"); params_vec.push(Box::new(end)); } if let Some(at) = app_type { conditions.push("app_type = ?"); params_vec.push(Box::new(at.to_string())); } let where_clause = if conditions.is_empty() { String::new() } else { format!("WHERE {}", conditions.join(" AND ")) }; // Build rollup WHERE clause using date strings let mut rollup_conditions: Vec = Vec::new(); let mut rollup_params: Vec> = Vec::new(); if let Some(start) = start_date { rollup_conditions.push("date >= date(?, 'unixepoch', 'localtime')".to_string()); rollup_params.push(Box::new(start)); } if let Some(end) = end_date { rollup_conditions.push("date <= date(?, 'unixepoch', 'localtime')".to_string()); rollup_params.push(Box::new(end)); } if let Some(at) = app_type { rollup_conditions.push("app_type = ?".to_string()); rollup_params.push(Box::new(at.to_string())); } let rollup_where = if rollup_conditions.is_empty() { String::new() } else { format!("WHERE {}", rollup_conditions.join(" AND ")) }; let sql = format!( "SELECT COALESCE(d.total_requests, 0) + COALESCE(r.total_requests, 0), COALESCE(d.total_cost, 0) + COALESCE(r.total_cost, 0), COALESCE(d.total_input_tokens, 0) + COALESCE(r.total_input_tokens, 0), COALESCE(d.total_output_tokens, 0) + COALESCE(r.total_output_tokens, 0), COALESCE(d.total_cache_creation_tokens, 0) + COALESCE(r.total_cache_creation_tokens, 0), COALESCE(d.total_cache_read_tokens, 0) + COALESCE(r.total_cache_read_tokens, 0), COALESCE(d.success_count, 0) + COALESCE(r.success_count, 0) FROM (SELECT COUNT(*) as total_requests, COALESCE(SUM(CAST(total_cost_usd AS REAL)), 0) as total_cost, COALESCE(SUM(input_tokens), 0) as total_input_tokens, COALESCE(SUM(output_tokens), 0) as total_output_tokens, COALESCE(SUM(cache_creation_tokens), 0) as total_cache_creation_tokens, COALESCE(SUM(cache_read_tokens), 0) as total_cache_read_tokens, COALESCE(SUM(CASE WHEN status_code >= 200 AND status_code < 300 THEN 1 ELSE 0 END), 0) as success_count FROM proxy_request_logs {where_clause}) d, (SELECT COALESCE(SUM(request_count), 0) as total_requests, COALESCE(SUM(CAST(total_cost_usd AS REAL)), 0) as total_cost, COALESCE(SUM(input_tokens), 0) as total_input_tokens, COALESCE(SUM(output_tokens), 0) as total_output_tokens, COALESCE(SUM(cache_creation_tokens), 0) as total_cache_creation_tokens, COALESCE(SUM(cache_read_tokens), 0) as total_cache_read_tokens, COALESCE(SUM(success_count), 0) as success_count FROM usage_daily_rollups {rollup_where}) r" ); // Combine params: detail params first, then rollup params let mut all_params: Vec> = params_vec; all_params.extend(rollup_params); let param_refs: Vec<&dyn rusqlite::ToSql> = all_params.iter().map(|p| p.as_ref()).collect(); let result = conn.query_row(&sql, param_refs.as_slice(), |row| { let total_requests: i64 = row.get(0)?; let total_cost: f64 = row.get(1)?; let total_input_tokens: i64 = row.get(2)?; let total_output_tokens: i64 = row.get(3)?; let total_cache_creation_tokens: i64 = row.get(4)?; let total_cache_read_tokens: i64 = row.get(5)?; let success_count: i64 = row.get(6)?; let success_rate = if total_requests > 0 { (success_count as f32 / total_requests as f32) * 100.0 } else { 0.0 }; Ok(UsageSummary { total_requests: total_requests as u64, total_cost: format!("{total_cost:.6}"), total_input_tokens: total_input_tokens as u64, total_output_tokens: total_output_tokens as u64, total_cache_creation_tokens: total_cache_creation_tokens as u64, total_cache_read_tokens: total_cache_read_tokens as u64, success_rate, }) })?; Ok(result) } /// 获取每日趋势(滑动窗口,<=24h 按小时,>24h 按天,窗口与汇总一致) pub fn get_daily_trends( &self, start_date: Option, end_date: Option, app_type: Option<&str>, ) -> Result, AppError> { let conn = lock_conn!(self.conn); let end_ts = end_date.unwrap_or_else(|| Local::now().timestamp()); let mut start_ts = start_date.unwrap_or_else(|| end_ts - 24 * 60 * 60); if start_ts >= end_ts { start_ts = end_ts - 24 * 60 * 60; } let duration = end_ts - start_ts; let bucket_seconds: i64 = if duration <= 24 * 60 * 60 { 60 * 60 } else { 24 * 60 * 60 }; let mut bucket_count: i64 = if duration <= 0 { 1 } else { ((duration as f64) / bucket_seconds as f64).ceil() as i64 }; // 固定 24 小时窗口为 24 个小时桶,避免浮点误差 if bucket_seconds == 60 * 60 { bucket_count = 24; } if bucket_count < 1 { bucket_count = 1; } let app_type_filter = if app_type.is_some() { "AND app_type = ?4" } else { "" }; // Query detail logs let sql = format!( "SELECT CAST((created_at - ?1) / ?3 AS INTEGER) as bucket_idx, COUNT(*) as request_count, COALESCE(SUM(CAST(total_cost_usd AS REAL)), 0) as total_cost, COALESCE(SUM(input_tokens + output_tokens), 0) as total_tokens, COALESCE(SUM(input_tokens), 0) as total_input_tokens, COALESCE(SUM(output_tokens), 0) as total_output_tokens, COALESCE(SUM(cache_creation_tokens), 0) as total_cache_creation_tokens, COALESCE(SUM(cache_read_tokens), 0) as total_cache_read_tokens FROM proxy_request_logs WHERE created_at >= ?1 AND created_at <= ?2 {app_type_filter} GROUP BY bucket_idx ORDER BY bucket_idx ASC" ); let mut stmt = conn.prepare(&sql)?; let row_mapper = |row: &rusqlite::Row| { Ok(( row.get::<_, i64>(0)?, DailyStats { date: String::new(), request_count: row.get::<_, i64>(1)? as u64, total_cost: format!("{:.6}", row.get::<_, f64>(2)?), total_tokens: row.get::<_, i64>(3)? as u64, total_input_tokens: row.get::<_, i64>(4)? as u64, total_output_tokens: row.get::<_, i64>(5)? as u64, total_cache_creation_tokens: row.get::<_, i64>(6)? as u64, total_cache_read_tokens: row.get::<_, i64>(7)? as u64, }, )) }; let mut map: HashMap = HashMap::new(); // Collect rows into map (need to handle both param variants) { let rows = if let Some(at) = app_type { stmt.query_map(params![start_ts, end_ts, bucket_seconds, at], row_mapper)? } else { stmt.query_map(params![start_ts, end_ts, bucket_seconds], row_mapper)? }; for row in rows { let (mut bucket_idx, stat) = row?; if bucket_idx < 0 { continue; } if bucket_idx >= bucket_count { bucket_idx = bucket_count - 1; } map.insert(bucket_idx, stat); } } // Also query rollup data (daily granularity, only useful for daily buckets) if bucket_seconds >= 86400 { let rollup_sql = format!( "SELECT CAST((CAST(strftime('%s', date) AS INTEGER) - ?1) / ?3 AS INTEGER) as bucket_idx, COALESCE(SUM(request_count), 0), COALESCE(SUM(CAST(total_cost_usd AS REAL)), 0), COALESCE(SUM(input_tokens + output_tokens), 0), COALESCE(SUM(input_tokens), 0), COALESCE(SUM(output_tokens), 0), COALESCE(SUM(cache_creation_tokens), 0), COALESCE(SUM(cache_read_tokens), 0) FROM usage_daily_rollups WHERE date >= date(?1, 'unixepoch', 'localtime') AND date <= date(?2, 'unixepoch', 'localtime') {app_type_filter} GROUP BY bucket_idx ORDER BY bucket_idx ASC" ); let rollup_row_mapper = |row: &rusqlite::Row| { Ok(( row.get::<_, i64>(0)?, ( row.get::<_, i64>(1)? as u64, row.get::<_, f64>(2)?, row.get::<_, i64>(3)? as u64, row.get::<_, i64>(4)? as u64, row.get::<_, i64>(5)? as u64, row.get::<_, i64>(6)? as u64, row.get::<_, i64>(7)? as u64, ), )) }; let mut rstmt = conn.prepare(&rollup_sql)?; let rrows = if let Some(at) = app_type { rstmt.query_map( params![start_ts, end_ts, bucket_seconds, at], rollup_row_mapper, )? } else { rstmt.query_map(params![start_ts, end_ts, bucket_seconds], rollup_row_mapper)? }; for row in rrows { let (mut bucket_idx, (req, cost, tok, inp, out, cc, cr)) = row?; if bucket_idx < 0 { continue; } if bucket_idx >= bucket_count { bucket_idx = bucket_count - 1; } let entry = map.entry(bucket_idx).or_insert_with(|| DailyStats { date: String::new(), request_count: 0, total_cost: "0.000000".to_string(), total_tokens: 0, total_input_tokens: 0, total_output_tokens: 0, total_cache_creation_tokens: 0, total_cache_read_tokens: 0, }); entry.request_count += req; let existing_cost: f64 = entry.total_cost.parse().unwrap_or(0.0); entry.total_cost = format!("{:.6}", existing_cost + cost); entry.total_tokens += tok; entry.total_input_tokens += inp; entry.total_output_tokens += out; entry.total_cache_creation_tokens += cc; entry.total_cache_read_tokens += cr; } } let mut stats = Vec::with_capacity(bucket_count as usize); for i in 0..bucket_count { let bucket_start_ts = start_ts + i * bucket_seconds; let bucket_start = Local .timestamp_opt(bucket_start_ts, 0) .single() .unwrap_or_else(Local::now); let date = bucket_start.to_rfc3339(); if let Some(mut stat) = map.remove(&i) { stat.date = date; stats.push(stat); } else { stats.push(DailyStats { date, request_count: 0, total_cost: "0.000000".to_string(), total_tokens: 0, total_input_tokens: 0, total_output_tokens: 0, total_cache_creation_tokens: 0, total_cache_read_tokens: 0, }); } } Ok(stats) } /// 获取 Provider 统计 pub fn get_provider_stats( &self, app_type: Option<&str>, ) -> Result, AppError> { let conn = lock_conn!(self.conn); let (detail_where, rollup_where) = if app_type.is_some() { ("WHERE l.app_type = ?1", "WHERE r.app_type = ?2") } else { ("", "") }; // UNION detail logs + rollup data, then aggregate let detail_pname = provider_name_coalesce("l", "p"); let rollup_pname = provider_name_coalesce("r", "p2"); let sql = format!( "SELECT provider_id, app_type, provider_name, SUM(request_count) as request_count, SUM(total_tokens) as total_tokens, SUM(total_cost) as total_cost, SUM(success_count) as success_count, CASE WHEN SUM(request_count) > 0 THEN SUM(latency_sum) / SUM(request_count) ELSE 0 END as avg_latency FROM ( SELECT l.provider_id, l.app_type, {detail_pname} as provider_name, COUNT(*) as request_count, COALESCE(SUM(l.input_tokens + l.output_tokens), 0) as total_tokens, COALESCE(SUM(CAST(l.total_cost_usd AS REAL)), 0) as total_cost, COALESCE(SUM(CASE WHEN l.status_code >= 200 AND l.status_code < 300 THEN 1 ELSE 0 END), 0) as success_count, COALESCE(SUM(l.latency_ms), 0) as latency_sum FROM proxy_request_logs l LEFT JOIN providers p ON l.provider_id = p.id AND l.app_type = p.app_type {detail_where} GROUP BY l.provider_id, l.app_type UNION ALL SELECT r.provider_id, r.app_type, {rollup_pname} as provider_name, COALESCE(SUM(r.request_count), 0), COALESCE(SUM(r.input_tokens + r.output_tokens), 0), COALESCE(SUM(CAST(r.total_cost_usd AS REAL)), 0), COALESCE(SUM(r.success_count), 0), COALESCE(SUM(r.avg_latency_ms * r.request_count), 0) FROM usage_daily_rollups r LEFT JOIN providers p2 ON r.provider_id = p2.id AND r.app_type = p2.app_type {rollup_where} GROUP BY r.provider_id, r.app_type ) GROUP BY provider_id, app_type ORDER BY total_cost DESC" ); let mut stmt = conn.prepare(&sql)?; let row_mapper = |row: &rusqlite::Row| { let request_count: i64 = row.get(3)?; let success_count: i64 = row.get(6)?; let success_rate = if request_count > 0 { (success_count as f32 / request_count as f32) * 100.0 } else { 0.0 }; Ok(ProviderStats { provider_id: row.get(0)?, provider_name: row.get(2)?, request_count: request_count as u64, total_tokens: row.get::<_, i64>(4)? as u64, total_cost: format!("{:.6}", row.get::<_, f64>(5)?), success_rate, avg_latency_ms: row.get::<_, f64>(7)? as u64, }) }; let rows = if let Some(at) = app_type { stmt.query_map(params![at, at], row_mapper)? } else { stmt.query_map([], row_mapper)? }; let mut stats = Vec::new(); for row in rows { stats.push(row?); } Ok(stats) } /// 获取模型统计 pub fn get_model_stats(&self, app_type: Option<&str>) -> Result, AppError> { let conn = lock_conn!(self.conn); let (detail_where, rollup_where) = if app_type.is_some() { ("WHERE app_type = ?1", "WHERE app_type = ?2") } else { ("", "") }; // UNION detail logs + rollup data let sql = format!( "SELECT model, SUM(request_count) as request_count, SUM(total_tokens) as total_tokens, SUM(total_cost) as total_cost FROM ( SELECT model, COUNT(*) as request_count, COALESCE(SUM(input_tokens + output_tokens), 0) as total_tokens, COALESCE(SUM(CAST(total_cost_usd AS REAL)), 0) as total_cost FROM proxy_request_logs {detail_where} GROUP BY model UNION ALL SELECT model, COALESCE(SUM(request_count), 0), COALESCE(SUM(input_tokens + output_tokens), 0), COALESCE(SUM(CAST(total_cost_usd AS REAL)), 0) FROM usage_daily_rollups {rollup_where} GROUP BY model ) GROUP BY model ORDER BY total_cost DESC" ); let mut stmt = conn.prepare(&sql)?; let row_mapper = |row: &rusqlite::Row| { let request_count: i64 = row.get(1)?; let total_cost: f64 = row.get(3)?; let avg_cost = if request_count > 0 { total_cost / request_count as f64 } else { 0.0 }; Ok(ModelStats { model: row.get(0)?, request_count: request_count as u64, total_tokens: row.get::<_, i64>(2)? as u64, total_cost: format!("{total_cost:.6}"), avg_cost_per_request: format!("{avg_cost:.6}"), }) }; let rows = if let Some(at) = app_type { stmt.query_map(params![at, at], row_mapper)? } else { stmt.query_map([], row_mapper)? }; let mut stats = Vec::new(); for row in rows { stats.push(row?); } Ok(stats) } /// 获取请求日志列表(分页) pub fn get_request_logs( &self, filters: &LogFilters, page: u32, page_size: u32, ) -> Result { let conn = lock_conn!(self.conn); let mut conditions = Vec::new(); let mut params: Vec> = Vec::new(); if let Some(ref app_type) = filters.app_type { conditions.push("l.app_type = ?"); params.push(Box::new(app_type.clone())); } if let Some(ref provider_name) = filters.provider_name { conditions.push("p.name LIKE ?"); params.push(Box::new(format!("%{provider_name}%"))); } if let Some(ref model) = filters.model { conditions.push("l.model LIKE ?"); params.push(Box::new(format!("%{model}%"))); } if let Some(status) = filters.status_code { conditions.push("l.status_code = ?"); params.push(Box::new(status as i64)); } if let Some(start) = filters.start_date { conditions.push("l.created_at >= ?"); params.push(Box::new(start)); } if let Some(end) = filters.end_date { conditions.push("l.created_at <= ?"); params.push(Box::new(end)); } let where_clause = if conditions.is_empty() { String::new() } else { format!("WHERE {}", conditions.join(" AND ")) }; // 获取总数 let count_sql = format!( "SELECT COUNT(*) FROM proxy_request_logs l LEFT JOIN providers p ON l.provider_id = p.id AND l.app_type = p.app_type {where_clause}" ); let count_params: Vec<&dyn rusqlite::ToSql> = params.iter().map(|p| p.as_ref()).collect(); let total: u32 = conn.query_row(&count_sql, count_params.as_slice(), |row| { row.get::<_, i64>(0).map(|v| v as u32) })?; // 获取数据 let offset = page * page_size; params.push(Box::new(page_size as i64)); params.push(Box::new(offset as i64)); let logs_pname = provider_name_coalesce("l", "p"); let sql = format!( "SELECT l.request_id, l.provider_id, {logs_pname} as provider_name, l.app_type, l.model, l.request_model, l.cost_multiplier, l.input_tokens, l.output_tokens, l.cache_read_tokens, l.cache_creation_tokens, l.input_cost_usd, l.output_cost_usd, l.cache_read_cost_usd, l.cache_creation_cost_usd, l.total_cost_usd, l.is_streaming, l.latency_ms, l.first_token_ms, l.duration_ms, l.status_code, l.error_message, l.created_at, l.data_source FROM proxy_request_logs l LEFT JOIN providers p ON l.provider_id = p.id AND l.app_type = p.app_type {where_clause} ORDER BY l.created_at DESC LIMIT ? OFFSET ?" ); let mut stmt = conn.prepare(&sql)?; let params_refs: Vec<&dyn rusqlite::ToSql> = params.iter().map(|p| p.as_ref()).collect(); let rows = stmt.query_map(params_refs.as_slice(), |row| { Ok(RequestLogDetail { request_id: row.get(0)?, provider_id: row.get(1)?, provider_name: row.get(2)?, app_type: row.get(3)?, model: row.get(4)?, request_model: row.get(5)?, cost_multiplier: row .get::<_, Option>(6)? .unwrap_or_else(|| "1".to_string()), input_tokens: row.get::<_, i64>(7)? as u32, output_tokens: row.get::<_, i64>(8)? as u32, cache_read_tokens: row.get::<_, i64>(9)? as u32, cache_creation_tokens: row.get::<_, i64>(10)? as u32, input_cost_usd: row.get(11)?, output_cost_usd: row.get(12)?, cache_read_cost_usd: row.get(13)?, cache_creation_cost_usd: row.get(14)?, total_cost_usd: row.get(15)?, is_streaming: row.get::<_, i64>(16)? != 0, latency_ms: row.get::<_, i64>(17)? as u64, first_token_ms: row.get::<_, Option>(18)?.map(|v| v as u64), duration_ms: row.get::<_, Option>(19)?.map(|v| v as u64), status_code: row.get::<_, i64>(20)? as u16, error_message: row.get(21)?, created_at: row.get(22)?, data_source: row.get(23)?, }) })?; let mut logs = Vec::new(); let mut provider_cache = HashMap::new(); let mut pricing_cache = HashMap::new(); for row in rows { let mut log = row?; Self::maybe_backfill_log_costs( &conn, &mut log, &mut provider_cache, &mut pricing_cache, )?; logs.push(log); } Ok(PaginatedLogs { data: logs, total, page, page_size, }) } /// 获取单个请求详情 pub fn get_request_detail( &self, request_id: &str, ) -> Result, AppError> { let conn = lock_conn!(self.conn); let detail_pname = provider_name_coalesce("l", "p"); let detail_sql = format!( "SELECT l.request_id, l.provider_id, {detail_pname} as provider_name, l.app_type, l.model, l.request_model, l.cost_multiplier, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, input_cost_usd, output_cost_usd, cache_read_cost_usd, cache_creation_cost_usd, total_cost_usd, is_streaming, latency_ms, first_token_ms, duration_ms, status_code, error_message, created_at, l.data_source FROM proxy_request_logs l LEFT JOIN providers p ON l.provider_id = p.id AND l.app_type = p.app_type WHERE l.request_id = ?" ); let result = conn.query_row(&detail_sql, [request_id], |row| { Ok(RequestLogDetail { request_id: row.get(0)?, provider_id: row.get(1)?, provider_name: row.get(2)?, app_type: row.get(3)?, model: row.get(4)?, request_model: row.get(5)?, cost_multiplier: row .get::<_, Option>(6)? .unwrap_or_else(|| "1".to_string()), input_tokens: row.get::<_, i64>(7)? as u32, output_tokens: row.get::<_, i64>(8)? as u32, cache_read_tokens: row.get::<_, i64>(9)? as u32, cache_creation_tokens: row.get::<_, i64>(10)? as u32, input_cost_usd: row.get(11)?, output_cost_usd: row.get(12)?, cache_read_cost_usd: row.get(13)?, cache_creation_cost_usd: row.get(14)?, total_cost_usd: row.get(15)?, is_streaming: row.get::<_, i64>(16)? != 0, latency_ms: row.get::<_, i64>(17)? as u64, first_token_ms: row.get::<_, Option>(18)?.map(|v| v as u64), duration_ms: row.get::<_, Option>(19)?.map(|v| v as u64), status_code: row.get::<_, i64>(20)? as u16, error_message: row.get(21)?, created_at: row.get(22)?, data_source: row.get(23)?, }) }); match result { Ok(mut detail) => { let mut provider_cache = HashMap::new(); let mut pricing_cache = HashMap::new(); Self::maybe_backfill_log_costs( &conn, &mut detail, &mut provider_cache, &mut pricing_cache, )?; Ok(Some(detail)) } Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None), Err(e) => Err(AppError::Database(e.to_string())), } } /// 检查 Provider 使用限额 pub fn check_provider_limits( &self, provider_id: &str, app_type: &str, ) -> Result { let conn = lock_conn!(self.conn); // 获取 provider 的限额设置 let (limit_daily, limit_monthly) = conn .query_row( "SELECT meta FROM providers WHERE id = ? AND app_type = ?", params![provider_id, app_type], |row| { let meta_str: String = row.get(0)?; Ok(meta_str) }, ) .ok() .and_then(|meta_str| serde_json::from_str::(&meta_str).ok()) .map(|meta| { let daily = meta .get("limitDailyUsd") .and_then(|v| v.as_str()) .and_then(|s| s.parse::().ok()); let monthly = meta .get("limitMonthlyUsd") .and_then(|v| v.as_str()) .and_then(|s| s.parse::().ok()); (daily, monthly) }) .unwrap_or((None, None)); // 计算今日使用量 (detail logs + rollup) let daily_usage: f64 = conn .query_row( "SELECT COALESCE(SUM(cost), 0) FROM ( SELECT CAST(total_cost_usd AS REAL) as cost FROM proxy_request_logs WHERE provider_id = ? AND app_type = ? AND date(datetime(created_at, 'unixepoch', 'localtime')) = date('now', 'localtime') UNION ALL SELECT CAST(total_cost_usd AS REAL) FROM usage_daily_rollups WHERE provider_id = ? AND app_type = ? AND date = date('now', 'localtime') )", params![provider_id, app_type, provider_id, app_type], |row| row.get(0), ) .unwrap_or(0.0); // 计算本月使用量 (detail logs + rollup) let monthly_usage: f64 = conn .query_row( "SELECT COALESCE(SUM(cost), 0) FROM ( SELECT CAST(total_cost_usd AS REAL) as cost FROM proxy_request_logs WHERE provider_id = ? AND app_type = ? AND strftime('%Y-%m', datetime(created_at, 'unixepoch', 'localtime')) = strftime('%Y-%m', 'now', 'localtime') UNION ALL SELECT CAST(total_cost_usd AS REAL) FROM usage_daily_rollups WHERE provider_id = ? AND app_type = ? AND strftime('%Y-%m', date) = strftime('%Y-%m', 'now', 'localtime') )", params![provider_id, app_type, provider_id, app_type], |row| row.get(0), ) .unwrap_or(0.0); let daily_exceeded = limit_daily .map(|limit| daily_usage >= limit) .unwrap_or(false); let monthly_exceeded = limit_monthly .map(|limit| monthly_usage >= limit) .unwrap_or(false); Ok(ProviderLimitStatus { provider_id: provider_id.to_string(), daily_usage: format!("{daily_usage:.6}"), daily_limit: limit_daily.map(|l| format!("{l:.2}")), daily_exceeded, monthly_usage: format!("{monthly_usage:.6}"), monthly_limit: limit_monthly.map(|l| format!("{l:.2}")), monthly_exceeded, }) } } /// Provider 限额状态 #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct ProviderLimitStatus { pub provider_id: String, pub daily_usage: String, pub daily_limit: Option, pub daily_exceeded: bool, pub monthly_usage: String, pub monthly_limit: Option, pub monthly_exceeded: bool, } #[derive(Clone)] struct PricingInfo { input: rust_decimal::Decimal, output: rust_decimal::Decimal, cache_read: rust_decimal::Decimal, cache_creation: rust_decimal::Decimal, } impl Database { fn maybe_backfill_log_costs( conn: &Connection, log: &mut RequestLogDetail, provider_cache: &mut HashMap<(String, String), rust_decimal::Decimal>, pricing_cache: &mut HashMap, ) -> Result<(), AppError> { let total_cost = rust_decimal::Decimal::from_str(&log.total_cost_usd) .unwrap_or(rust_decimal::Decimal::ZERO); let has_cost = total_cost > rust_decimal::Decimal::ZERO; let has_usage = log.input_tokens > 0 || log.output_tokens > 0 || log.cache_read_tokens > 0 || log.cache_creation_tokens > 0; if has_cost || !has_usage { return Ok(()); } let pricing = match Self::get_model_pricing_cached(conn, pricing_cache, &log.model)? { Some(info) => info, None => return Ok(()), }; let multiplier = Self::get_cost_multiplier_cached( conn, provider_cache, &log.provider_id, &log.app_type, )?; let million = rust_decimal::Decimal::from(1_000_000u64); // 与 CostCalculator::calculate 保持一致的计算逻辑: // 1. input_cost 需要扣除 cache_read_tokens(避免缓存部分被重复计费) // 2. 各项成本是基础成本(不含倍率) // 3. 倍率只作用于最终总价 let billable_input_tokens = (log.input_tokens as u64).saturating_sub(log.cache_read_tokens as u64); let input_cost = rust_decimal::Decimal::from(billable_input_tokens) * pricing.input / million; let output_cost = rust_decimal::Decimal::from(log.output_tokens as u64) * pricing.output / million; let cache_read_cost = rust_decimal::Decimal::from(log.cache_read_tokens as u64) * pricing.cache_read / million; let cache_creation_cost = rust_decimal::Decimal::from(log.cache_creation_tokens as u64) * pricing.cache_creation / million; // 总成本 = 基础成本之和 × 倍率 let base_total = input_cost + output_cost + cache_read_cost + cache_creation_cost; let total_cost = base_total * multiplier; log.input_cost_usd = format!("{input_cost:.6}"); log.output_cost_usd = format!("{output_cost:.6}"); log.cache_read_cost_usd = format!("{cache_read_cost:.6}"); log.cache_creation_cost_usd = format!("{cache_creation_cost:.6}"); log.total_cost_usd = format!("{total_cost:.6}"); conn.execute( "UPDATE proxy_request_logs SET input_cost_usd = ?1, output_cost_usd = ?2, cache_read_cost_usd = ?3, cache_creation_cost_usd = ?4, total_cost_usd = ?5 WHERE request_id = ?6", params![ log.input_cost_usd, log.output_cost_usd, log.cache_read_cost_usd, log.cache_creation_cost_usd, log.total_cost_usd, log.request_id ], ) .map_err(|e| AppError::Database(format!("更新请求成本失败: {e}")))?; Ok(()) } fn get_cost_multiplier_cached( conn: &Connection, cache: &mut HashMap<(String, String), rust_decimal::Decimal>, provider_id: &str, app_type: &str, ) -> Result { let key = (provider_id.to_string(), app_type.to_string()); if let Some(multiplier) = cache.get(&key) { return Ok(*multiplier); } let meta_json: Option = conn .query_row( "SELECT meta FROM providers WHERE id = ? AND app_type = ?", params![provider_id, app_type], |row| row.get(0), ) .optional() .map_err(|e| AppError::Database(format!("查询 provider meta 失败: {e}")))?; let multiplier = meta_json .and_then(|meta| serde_json::from_str::(&meta).ok()) .and_then(|value| value.get("costMultiplier").cloned()) .and_then(|val| { val.as_str() .and_then(|s| rust_decimal::Decimal::from_str(s).ok()) }) .unwrap_or(rust_decimal::Decimal::ONE); cache.insert(key, multiplier); Ok(multiplier) } fn get_model_pricing_cached( conn: &Connection, cache: &mut HashMap, model: &str, ) -> Result, AppError> { if let Some(info) = cache.get(model) { return Ok(Some(info.clone())); } let row = find_model_pricing_row(conn, model)?; let Some((input, output, cache_read, cache_creation)) = row else { return Ok(None); }; let pricing = PricingInfo { input: rust_decimal::Decimal::from_str(&input) .map_err(|e| AppError::Database(format!("解析输入价格失败: {e}")))?, output: rust_decimal::Decimal::from_str(&output) .map_err(|e| AppError::Database(format!("解析输出价格失败: {e}")))?, cache_read: rust_decimal::Decimal::from_str(&cache_read) .map_err(|e| AppError::Database(format!("解析缓存读取价格失败: {e}")))?, cache_creation: rust_decimal::Decimal::from_str(&cache_creation) .map_err(|e| AppError::Database(format!("解析缓存写入价格失败: {e}")))?, }; cache.insert(model.to_string(), pricing.clone()); Ok(Some(pricing)) } } pub(crate) fn find_model_pricing_row( conn: &Connection, model_id: &str, ) -> Result, AppError> { // 清洗模型名称:去前缀(/)、去后缀(:)、@ 替换为 - // 例如 moonshotai/gpt-5.2-codex@low:v2 → gpt-5.2-codex-low let cleaned = model_id .rsplit_once('/') .map_or(model_id, |(_, r)| r) .split(':') .next() .unwrap_or(model_id) .trim() .replace('@', "-"); // 精确匹配清洗后的名称 let exact = conn .query_row( "SELECT input_cost_per_million, output_cost_per_million, cache_read_cost_per_million, cache_creation_cost_per_million FROM model_pricing WHERE model_id = ?1", [&cleaned], |row| { Ok(( row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, row.get::<_, String>(3)?, )) }, ) .optional() .map_err(|e| AppError::Database(format!("查询模型定价失败: {e}")))?; if exact.is_none() { log::warn!("模型 {model_id}(清洗后: {cleaned})未找到定价信息,成本将记录为 0"); } Ok(exact) } #[cfg(test)] mod tests { use super::*; #[test] fn test_get_usage_summary() -> Result<(), AppError> { let db = Database::memory()?; // 插入测试数据 { let conn = lock_conn!(db.conn); conn.execute( "INSERT INTO proxy_request_logs ( request_id, provider_id, app_type, model, input_tokens, output_tokens, total_cost_usd, latency_ms, status_code, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params!["req1", "p1", "claude", "claude-3", 100, 50, "0.01", 100, 200, 1000], )?; conn.execute( "INSERT INTO proxy_request_logs ( request_id, provider_id, app_type, model, input_tokens, output_tokens, total_cost_usd, latency_ms, status_code, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params!["req2", "p1", "claude", "claude-3", 200, 100, "0.02", 150, 200, 2000], )?; } let summary = db.get_usage_summary(None, None, None)?; assert_eq!(summary.total_requests, 2); assert_eq!(summary.success_rate, 100.0); Ok(()) } #[test] fn test_get_model_stats() -> Result<(), AppError> { let db = Database::memory()?; // 插入测试数据 { let conn = lock_conn!(db.conn); conn.execute( "INSERT INTO proxy_request_logs ( request_id, provider_id, app_type, model, input_tokens, output_tokens, total_cost_usd, latency_ms, status_code, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "req1", "p1", "claude", "claude-3-sonnet", 100, 50, "0.01", 100, 200, 1000 ], )?; } let stats = db.get_model_stats(None)?; assert_eq!(stats.len(), 1); assert_eq!(stats[0].model, "claude-3-sonnet"); assert_eq!(stats[0].request_count, 1); Ok(()) } #[test] fn test_model_pricing_matching() -> Result<(), AppError> { let db = Database::memory()?; let conn = lock_conn!(db.conn); // 准备额外定价数据,覆盖前缀/后缀清洗场景 conn.execute( "INSERT OR REPLACE INTO model_pricing ( model_id, display_name, input_cost_per_million, output_cost_per_million, cache_read_cost_per_million, cache_creation_cost_per_million ) VALUES (?, ?, ?, ?, ?, ?)", params![ "claude-haiku-4.5", "Claude Haiku 4.5", "1.0", "2.0", "0.0", "0.0" ], )?; // 测试精确匹配(seed_model_pricing 已预置 claude-sonnet-4-5-20250929) let result = find_model_pricing_row(&conn, "claude-sonnet-4-5-20250929")?; assert!( result.is_some(), "应该能精确匹配 claude-sonnet-4-5-20250929" ); // 清洗:去除前缀和冒号后缀 let result = find_model_pricing_row(&conn, "anthropic/claude-haiku-4.5")?; assert!( result.is_some(), "带前缀的模型 anthropic/claude-haiku-4.5 应能匹配到 claude-haiku-4.5" ); let result = find_model_pricing_row(&conn, "moonshotai/kimi-k2-0905:exa")?; assert!( result.is_some(), "带前缀+冒号后缀的模型应清洗后匹配到 kimi-k2-0905" ); // 清洗:@ 替换为 -(seed_model_pricing 已预置 gpt-5.2-codex-low) let result = find_model_pricing_row(&conn, "gpt-5.2-codex@low")?; assert!( result.is_some(), "带 @ 分隔符的模型 gpt-5.2-codex@low 应能匹配到 gpt-5.2-codex-low" ); // 测试不存在的模型 let result = find_model_pricing_row(&conn, "unknown-model-123")?; assert!(result.is_none(), "不应该匹配不存在的模型"); Ok(()) } }