//! 使用统计服务 //! //! 提供使用量数据的聚合查询功能 use crate::database::{lock_conn, Database}; use crate::error::AppError; use chrono::{Local, NaiveDate, TimeZone, Timelike}; 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)" ) } #[derive(Debug, Clone, Default)] struct RollupDateBounds { start: Option, end: Option, is_empty: bool, } fn local_datetime_from_timestamp(ts: i64) -> Result, AppError> { Local .timestamp_opt(ts, 0) .single() .ok_or_else(|| AppError::Database(format!("无法解析本地时间戳: {ts}"))) } fn compute_rollup_date_bounds( start_ts: Option, end_ts: Option, ) -> Result { let start = match start_ts { Some(ts) => { let local = local_datetime_from_timestamp(ts)?; let day = local.date_naive(); if local.time().num_seconds_from_midnight() == 0 { Some(day.format("%Y-%m-%d").to_string()) } else { day.succ_opt() .map(|next| next.format("%Y-%m-%d").to_string()) } } None => None, }; let end = match end_ts { Some(ts) => { let local = local_datetime_from_timestamp(ts)?; let day = local.date_naive(); if local.time().hour() == 23 && local.time().minute() == 59 { Some(day.format("%Y-%m-%d").to_string()) } else { day.pred_opt() .map(|prev| prev.format("%Y-%m-%d").to_string()) } } None => None, }; let is_empty = matches!((&start, &end), (Some(start), Some(end)) if start > end); Ok(RollupDateBounds { start, end, is_empty, }) } fn push_rollup_date_filters( conditions: &mut Vec, params: &mut Vec>, column: &str, bounds: &RollupDateBounds, ) { if bounds.is_empty { conditions.push("1 = 0".to_string()); return; } if let Some(start) = &bounds.start { conditions.push(format!("{column} >= ?")); params.push(Box::new(start.clone())); } if let Some(end) = &bounds.end { conditions.push(format!("{column} <= ?")); params.push(Box::new(end.clone())); } } fn local_day_start_rfc3339(day: NaiveDate) -> String { let local_midnight = day .and_hms_opt(0, 0, 0) .and_then(|naive| match Local.from_local_datetime(&naive) { chrono::LocalResult::Single(dt) => Some(dt), chrono::LocalResult::Ambiguous(earliest, _) => Some(earliest), chrono::LocalResult::None => None, }) .unwrap_or_else(Local::now); local_midnight.to_rfc3339() } 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 ")) }; // Only include rolled-up rows for full local days that are fully covered by the range. let mut rollup_conditions: Vec = Vec::new(); let mut rollup_params: Vec> = Vec::new(); let rollup_bounds = compute_rollup_date_bounds(start_date, end_date)?; push_rollup_date_filters( &mut rollup_conditions, &mut rollup_params, "date", &rollup_bounds, ); 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; if duration <= 24 * 60 * 60 { let bucket_seconds: i64 = 60 * 60; let mut bucket_count: i64 = if duration <= 0 { 1 } else { (duration + bucket_seconds - 1) / bucket_seconds }; if bucket_count < 1 { bucket_count = 1; } let app_type_filter = if app_type.is_some() { "AND app_type = ?4" } else { "" }; 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(); 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); } 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_datetime_from_timestamp(bucket_start_ts)?; 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, }); } } return Ok(stats); } let start_day = local_datetime_from_timestamp(start_ts)?.date_naive(); let end_day = local_datetime_from_timestamp(end_ts)?.date_naive(); let bucket_count = (end_day.signed_duration_since(start_day).num_days() + 1) as usize; let app_type_filter = if app_type.is_some() { "AND app_type = ?3" } else { "" }; let detail_sql = format!( "SELECT date(created_at, 'unixepoch', 'localtime') as bucket_date, 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_date ORDER BY bucket_date ASC" ); let mut detail_stmt = conn.prepare(&detail_sql)?; let detail_row_mapper = |row: &rusqlite::Row| { Ok(( row.get::<_, String>(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(); let detail_rows = if let Some(at) = app_type { detail_stmt.query_map(params![start_ts, end_ts, at], detail_row_mapper)? } else { detail_stmt.query_map(params![start_ts, end_ts], detail_row_mapper)? }; for row in detail_rows { let (bucket_date, stat) = row?; let date = NaiveDate::parse_from_str(&bucket_date, "%Y-%m-%d") .map_err(|err| AppError::Database(format!("解析趋势日期失败: {err}")))?; map.insert(date, stat); } let rollup_bounds = compute_rollup_date_bounds(Some(start_ts), Some(end_ts))?; let mut rollup_conditions = Vec::new(); let mut rollup_params: Vec> = Vec::new(); push_rollup_date_filters( &mut rollup_conditions, &mut rollup_params, "date", &rollup_bounds, ); 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 rollup_sql = format!( "SELECT date, 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 {rollup_where} GROUP BY date ORDER BY date ASC" ); let mut rollup_stmt = conn.prepare(&rollup_sql)?; let rollup_row_mapper = |row: &rusqlite::Row| { Ok(( row.get::<_, String>(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 rollup_param_refs: Vec<&dyn rusqlite::ToSql> = rollup_params.iter().map(|param| param.as_ref()).collect(); let rollup_rows = rollup_stmt.query_map(rollup_param_refs.as_slice(), rollup_row_mapper)?; for row in rollup_rows { let (bucket_date, (req, cost, tok, inp, out, cc, cr)) = row?; let date = NaiveDate::parse_from_str(&bucket_date, "%Y-%m-%d") .map_err(|err| AppError::Database(format!("解析 rollup 趋势日期失败: {err}")))?; let entry = map.entry(date).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); let mut current_day = start_day; for _ in 0..bucket_count { let date = local_day_start_rfc3339(current_day); if let Some(mut stat) = map.remove(¤t_day) { 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, }); } current_day = current_day.succ_opt().unwrap_or(current_day); } Ok(stats) } /// 获取 Provider 统计 pub fn get_provider_stats( &self, start_date: Option, end_date: Option, app_type: Option<&str>, ) -> Result, AppError> { let conn = lock_conn!(self.conn); let mut detail_conditions = Vec::new(); let mut detail_params: Vec> = Vec::new(); if let Some(start) = start_date { detail_conditions.push("l.created_at >= ?"); detail_params.push(Box::new(start)); } if let Some(end) = end_date { detail_conditions.push("l.created_at <= ?"); detail_params.push(Box::new(end)); } if let Some(at) = app_type { detail_conditions.push("l.app_type = ?"); detail_params.push(Box::new(at.to_string())); } let detail_where = if detail_conditions.is_empty() { String::new() } else { format!("WHERE {}", detail_conditions.join(" AND ")) }; let mut rollup_conditions = Vec::new(); let mut rollup_params: Vec> = Vec::new(); let rollup_bounds = compute_rollup_date_bounds(start_date, end_date)?; push_rollup_date_filters( &mut rollup_conditions, &mut rollup_params, "r.date", &rollup_bounds, ); if let Some(at) = app_type { rollup_conditions.push("r.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 ")) }; // 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 mut params: Vec> = detail_params; params.extend(rollup_params); let param_refs: Vec<&dyn rusqlite::ToSql> = params.iter().map(|p| p.as_ref()).collect(); 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 = stmt.query_map(param_refs.as_slice(), row_mapper)?; let mut stats = Vec::new(); for row in rows { stats.push(row?); } Ok(stats) } /// 获取模型统计 pub fn get_model_stats( &self, start_date: Option, end_date: Option, app_type: Option<&str>, ) -> Result, AppError> { let conn = lock_conn!(self.conn); let mut detail_conditions = Vec::new(); let mut detail_params: Vec> = Vec::new(); if let Some(start) = start_date { detail_conditions.push("l.created_at >= ?"); detail_params.push(Box::new(start)); } if let Some(end) = end_date { detail_conditions.push("l.created_at <= ?"); detail_params.push(Box::new(end)); } if let Some(at) = app_type { detail_conditions.push("l.app_type = ?"); detail_params.push(Box::new(at.to_string())); } let detail_where = if detail_conditions.is_empty() { String::new() } else { format!("WHERE {}", detail_conditions.join(" AND ")) }; let mut rollup_conditions = Vec::new(); let mut rollup_params: Vec> = Vec::new(); let rollup_bounds = compute_rollup_date_bounds(start_date, end_date)?; push_rollup_date_filters( &mut rollup_conditions, &mut rollup_params, "r.date", &rollup_bounds, ); if let Some(at) = app_type { rollup_conditions.push("r.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 ")) }; // 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 l.model, 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 FROM proxy_request_logs l {detail_where} GROUP BY l.model UNION ALL SELECT r.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 r {rollup_where} GROUP BY r.model ) GROUP BY model ORDER BY total_cost DESC" ); let mut stmt = conn.prepare(&sql)?; let mut params: Vec> = detail_params; params.extend(rollup_params); let param_refs: Vec<&dyn rusqlite::ToSql> = params.iter().map(|p| p.as_ref()).collect(); 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 = stmt.query_map(param_refs.as_slice(), 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::*; fn local_ts(year: i32, month: u32, day: u32, hour: u32, minute: u32, second: u32) -> i64 { match Local.with_ymd_and_hms(year, month, day, hour, minute, second) { chrono::LocalResult::Single(dt) => dt.timestamp(), chrono::LocalResult::Ambiguous(earliest, _) => earliest.timestamp(), chrono::LocalResult::None => panic!("valid local datetime"), } } #[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_usage_summary_excludes_partial_rollup_boundary_days() -> Result<(), AppError> { let db = Database::memory()?; let start = local_ts(2024, 1, 1, 12, 0, 0); let end = local_ts(2024, 1, 3, 12, 0, 0); { let conn = lock_conn!(db.conn); conn.execute( "INSERT INTO usage_daily_rollups ( date, app_type, provider_id, model, request_count, success_count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "2024-01-01", "claude", "p1", "claude-3", 10, 10, 1000, 500, 0, 0, "1.00", 100 ], )?; conn.execute( "INSERT INTO usage_daily_rollups ( date, app_type, provider_id, model, request_count, success_count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "2024-01-02", "claude", "p1", "claude-3", 20, 19, 2000, 1000, 0, 0, "2.00", 120 ], )?; conn.execute( "INSERT INTO usage_daily_rollups ( date, app_type, provider_id, model, request_count, success_count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "2024-01-03", "claude", "p1", "claude-3", 30, 29, 3000, 1500, 0, 0, "3.00", 140 ], )?; } let summary = db.get_usage_summary(Some(start), Some(end), Some("claude"))?; assert_eq!(summary.total_requests, 20); assert_eq!(summary.total_input_tokens, 2000); assert_eq!(summary.total_output_tokens, 1000); Ok(()) } #[test] fn test_get_usage_summary_includes_end_day_rollup_for_minute_precision_end_time( ) -> Result<(), AppError> { let db = Database::memory()?; let start = local_ts(2024, 1, 1, 0, 0, 0); let end = local_ts(2024, 1, 2, 23, 59, 0); { let conn = lock_conn!(db.conn); conn.execute( "INSERT INTO usage_daily_rollups ( date, app_type, provider_id, model, request_count, success_count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "2024-01-01", "claude", "p1", "claude-3", 10, 10, 1000, 500, 0, 0, "1.00", 100 ], )?; conn.execute( "INSERT INTO usage_daily_rollups ( date, app_type, provider_id, model, request_count, success_count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "2024-01-02", "claude", "p1", "claude-3", 20, 19, 2000, 1000, 0, 0, "2.00", 120 ], )?; } let summary = db.get_usage_summary(Some(start), Some(end), Some("claude"))?; assert_eq!(summary.total_requests, 30); assert_eq!(summary.total_input_tokens, 3000); assert_eq!(summary.total_output_tokens, 1500); 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, None, 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_get_provider_stats_with_time_filter() -> 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!["old", "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!["new", "p1", "claude", "claude-3", 200, 75, "0.02", 120, 200, 2000], )?; } let stats = db.get_provider_stats(Some(1500), Some(2500), Some("claude"))?; assert_eq!(stats.len(), 1); assert_eq!(stats[0].provider_id, "p1"); assert_eq!(stats[0].request_count, 1); assert_eq!(stats[0].total_tokens, 275); Ok(()) } #[test] fn test_get_provider_stats_excludes_partial_rollup_boundary_days() -> Result<(), AppError> { let db = Database::memory()?; let start = local_ts(2024, 2, 1, 12, 0, 0); let end = local_ts(2024, 2, 3, 12, 0, 0); { let conn = lock_conn!(db.conn); conn.execute( "INSERT INTO usage_daily_rollups ( date, app_type, provider_id, model, request_count, success_count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "2024-02-01", "claude", "p-rollup", "claude-3", 5, 5, 500, 250, 0, 0, "0.50", 100 ], )?; conn.execute( "INSERT INTO usage_daily_rollups ( date, app_type, provider_id, model, request_count, success_count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "2024-02-02", "claude", "p-rollup", "claude-3", 8, 7, 800, 400, 0, 0, "0.80", 120 ], )?; conn.execute( "INSERT INTO usage_daily_rollups ( date, app_type, provider_id, model, request_count, success_count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "2024-02-03", "claude", "p-rollup", "claude-3", 12, 11, 1200, 600, 0, 0, "1.20", 140 ], )?; } let stats = db.get_provider_stats(Some(start), Some(end), Some("claude"))?; assert_eq!(stats.len(), 1); assert_eq!(stats[0].provider_id, "p-rollup"); assert_eq!(stats[0].request_count, 8); assert_eq!(stats[0].total_tokens, 1200); Ok(()) } #[test] fn test_get_daily_trends_respects_shorter_than_24_hours() -> 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![ "req-short", "p1", "claude", "claude-3", 100, 50, "0.01", 100, 200, 10_800 ], )?; } let stats = db.get_daily_trends(Some(0), Some(15 * 60 * 60), Some("claude"))?; assert_eq!(stats.len(), 15); assert_eq!(stats[3].request_count, 1); Ok(()) } #[test] fn test_get_daily_trends_groups_ranges_longer_than_24_hours_by_local_day( ) -> Result<(), AppError> { let db = Database::memory()?; let start = local_ts(2024, 3, 1, 12, 0, 0); let end = local_ts(2024, 3, 3, 12, 0, 0); { 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![ "day-1-detail", "p1", "claude", "claude-3", 100, 50, "0.01", 100, 200, local_ts(2024, 3, 1, 13, 0, 0) ], )?; 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![ "day-3-detail", "p1", "claude", "claude-3", 200, 75, "0.02", 110, 200, local_ts(2024, 3, 3, 10, 0, 0) ], )?; conn.execute( "INSERT INTO usage_daily_rollups ( date, app_type, provider_id, model, request_count, success_count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "2024-03-02", "claude", "p1", "claude-3", 4, 4, 400, 200, 0, 0, "0.40", 120 ], )?; } let stats = db.get_daily_trends(Some(start), Some(end), Some("claude"))?; assert_eq!(stats.len(), 3); assert_eq!(stats[0].request_count, 1); assert_eq!(stats[0].total_tokens, 150); assert_eq!(stats[1].request_count, 4); assert_eq!(stats[1].total_tokens, 600); assert_eq!(stats[2].request_count, 1); assert_eq!(stats[2].total_tokens, 275); Ok(()) } #[test] fn test_get_model_stats_excludes_partial_rollup_boundary_days() -> Result<(), AppError> { let db = Database::memory()?; let start = local_ts(2024, 4, 1, 12, 0, 0); let end = local_ts(2024, 4, 3, 12, 0, 0); { let conn = lock_conn!(db.conn); conn.execute( "INSERT INTO usage_daily_rollups ( date, app_type, provider_id, model, request_count, success_count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "2024-04-01", "claude", "p1", "claude-3-haiku", 6, 6, 600, 300, 0, 0, "0.60", 100 ], )?; conn.execute( "INSERT INTO usage_daily_rollups ( date, app_type, provider_id, model, request_count, success_count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "2024-04-02", "claude", "p1", "claude-3-haiku", 9, 8, 900, 450, 0, 0, "0.90", 110 ], )?; conn.execute( "INSERT INTO usage_daily_rollups ( date, app_type, provider_id, model, request_count, success_count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, total_cost_usd, avg_latency_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params![ "2024-04-03", "claude", "p1", "claude-3-haiku", 12, 11, 1200, 600, 0, 0, "1.20", 130 ], )?; } let stats = db.get_model_stats(Some(start), Some(end), Some("claude"))?; assert_eq!(stats.len(), 1); assert_eq!(stats[0].model, "claude-3-haiku"); assert_eq!(stats[0].request_count, 9); assert_eq!(stats[0].total_tokens, 1350); 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(()) } }