//! 能耗管理 API — 总览、排行、趋势、导出 //! //! 基于 energy_stats 表按日期聚合的能耗数据进行统计分析。 //! 电费估算按 ¥0.80/kWh 计算(可配置),energy 单位为 Wh。 use axum::extract::{Query, State}; use axum::Json; use serde::Deserialize; use serde_json::{json, Value}; use crate::error::AppError; use crate::middleware::auth::{self, CurrentUser}; /// 电费单价,0.80 元/kWh = 80 分 / 1000 Wh const COST_PER_WH_CENT: i64 = 80; /// 趋势查询参数 #[derive(Debug, Deserialize)] pub struct TrendQuery { /// day | week | month #[serde(default = "default_granularity")] pub granularity: String, /// 天数范围,默认30天 #[serde(default = "default_range")] pub range: i64, } /// 导出查询参数 #[derive(Debug, Deserialize)] pub struct ExportQuery { pub cabinet_id: Option, pub project_id: Option, pub start_date: Option, pub end_date: Option, } /// 柜子排行行 #[derive(Debug, sqlx::FromRow)] struct CabinetRankRow { id: i64, abstract_id: String, name: Option, today_energy: Option, month_energy: Option, } /// 项目统计行 #[derive(Debug, sqlx::FromRow)] struct ProjectStatRow { id: i64, name: String, cabinet_count: i64, today_energy: Option, month_energy: Option, } /// 趋势数据行 #[derive(Debug, sqlx::FromRow)] struct TrendRow { period: String, // 日期或周/月标识 energy: Option, charge_count: Option, } /// 能耗明细行(用于导出) #[derive(Debug, sqlx::FromRow)] struct EnergyDetailRow { stat_date: chrono::NaiveDate, cabinet_id: i64, cabinet_name: Option, project_name: Option, total_energy: Option, charge_count: Option, } fn default_granularity() -> String { "day".to_string() } fn default_range() -> i64 { 30 } /// GET /api/energy-stats/summary — 总览 /// /// 返回今日/本月能耗、电费估算、柜子/在线数量。 pub async fn get_summary( user: CurrentUser, State(state): State, ) -> Result, AppError> { auth::check_permission(&user, "energy:view")?; let pool = &state.mysql; // 组织过滤子查询 let (org_filter, org_binds): (String, Vec) = if user.role_level >= 2 { (String::new(), vec![]) } else if let Some(oid) = user.organization_id { ( " AND es.cabinet_id IN (SELECT c.id FROM cabinets c JOIN projects p ON c.project_id = p.id WHERE p.organization_id = ?)".to_string(), vec![oid.to_string()], ) } else { (" AND 1=0".to_string(), vec![]) }; // 今日总能耗 let today_sql = format!( "SELECT CAST(COALESCE(SUM(total_energy), 0) AS SIGNED) FROM energy_stats es WHERE stat_date = CURDATE(){}", org_filter ); let mut today_q = sqlx::query_scalar::<_, i64>(&today_sql); for v in &org_binds { today_q = today_q.bind(v); } let today_energy: i64 = today_q.fetch_one(pool).await?; // 本月总能耗 let month_sql = format!( "SELECT CAST(COALESCE(SUM(total_energy), 0) AS SIGNED) FROM energy_stats es \ WHERE stat_date >= DATE_FORMAT(CURDATE(), '%Y-%m-01'){}", org_filter ); let mut month_q = sqlx::query_scalar::<_, i64>(&month_sql); for v in &org_binds { month_q = month_q.bind(v); } let month_energy: i64 = month_q.fetch_one(pool).await?; // 柜子总数(含组织过滤) let cabinet_count: i64 = if user.role_level >= 2 { sqlx::query_scalar("SELECT COUNT(*) FROM cabinets") .fetch_one(pool).await? } else if let Some(oid) = user.organization_id { sqlx::query_scalar( "SELECT COUNT(*) FROM cabinets c JOIN projects p ON c.project_id = p.id WHERE p.organization_id = ?" ) .bind(oid) .fetch_one(pool).await? } else { 0 }; // 在线柜子数 let online_count: i64 = if user.role_level >= 2 { sqlx::query_scalar("SELECT COUNT(*) FROM cabinets WHERE status = 1") .fetch_one(pool).await? } else if let Some(oid) = user.organization_id { sqlx::query_scalar( "SELECT COUNT(*) FROM cabinets c JOIN projects p ON c.project_id = p.id WHERE c.status = 1 AND p.organization_id = ?" ) .bind(oid) .fetch_one(pool).await? } else { 0 }; Ok(Json(json!({ "code": 0, "data": { "today_energy": today_energy, "month_energy": month_energy, "estimated_cost": (month_energy * COST_PER_WH_CENT / 1000) as f64 / 100.0, "cabinet_count": cabinet_count, "online_count": online_count, }, "message": "ok" }))) } /// GET /api/energy-stats/cabinets — 柜子能耗排行 /// /// 按今日能耗降序排列,返回柜子列表及今日/本月能耗。 pub async fn get_cabinet_ranking( user: CurrentUser, State(state): State, ) -> Result, AppError> { auth::check_permission(&user, "energy:view")?; let pool = &state.mysql; let (org_join, org_binds): (String, Vec) = if user.role_level >= 2 { (String::new(), vec![]) } else if let Some(oid) = user.organization_id { ( " JOIN projects p ON c.project_id = p.id AND p.organization_id = ?".to_string(), vec![oid.to_string()], ) } else { (" JOIN projects p ON 1=0".to_string(), vec![]) }; let sql = format!( "SELECT c.id, c.abstract_id, c.name, \ CAST(COALESCE(es_today.total_energy, 0) AS SIGNED) AS today_energy, \ CAST(COALESCE(es_month.month_energy, 0) AS SIGNED) AS month_energy \ FROM cabinets c \ LEFT JOIN energy_stats es_today \ ON c.id = es_today.cabinet_id AND es_today.stat_date = CURDATE() \ LEFT JOIN ( \ SELECT cabinet_id, SUM(total_energy) AS month_energy \ FROM energy_stats \ WHERE stat_date >= DATE_FORMAT(CURDATE(), '%Y-%m-01') \ GROUP BY cabinet_id \ ) es_month ON c.id = es_month.cabinet_id \ {} \ ORDER BY today_energy DESC, c.id ASC", org_join ); let mut query = sqlx::query_as::<_, CabinetRankRow>(&sql); for v in &org_binds { query = query.bind(v); } let rows = query.fetch_all(pool).await?; let list: Vec = rows.iter().map(|r| { json!({ "cabinet_id": r.id, "abstract_id": r.abstract_id, "name": r.name, "today_energy": r.today_energy.unwrap_or(0), "month_energy": r.month_energy.unwrap_or(0), }) }).collect(); Ok(Json(json!({ "code": 0, "data": list, "message": "ok" }))) } /// GET /api/energy-stats/projects — 项目能耗统计 /// /// 按今日能耗降序排列,返回项目级聚合数据。 pub async fn get_project_stats( user: CurrentUser, State(state): State, ) -> Result, AppError> { auth::check_permission(&user, "energy:view")?; let pool = &state.mysql; let (org_filter, org_binds): (String, Vec) = if user.role_level >= 2 { (String::new(), vec![]) } else if let Some(oid) = user.organization_id { (" WHERE p.organization_id = ?".to_string(), vec![oid.to_string()]) } else { (" WHERE 1=0".to_string(), vec![]) }; let sql = format!( "SELECT p.id, p.name, \ COUNT(DISTINCT c.id) AS cabinet_count, \ CAST(COALESCE(SUM(es_today.total_energy), 0) AS SIGNED) AS today_energy, \ CAST(COALESCE(SUM(es_month.month_energy), 0) AS SIGNED) AS month_energy \ FROM projects p \ LEFT JOIN cabinets c ON p.id = c.project_id \ LEFT JOIN energy_stats es_today \ ON c.id = es_today.cabinet_id AND es_today.stat_date = CURDATE() \ LEFT JOIN ( \ SELECT cabinet_id, SUM(total_energy) AS month_energy \ FROM energy_stats \ WHERE stat_date >= DATE_FORMAT(CURDATE(), '%Y-%m-01') \ GROUP BY cabinet_id \ ) es_month ON c.id = es_month.cabinet_id \ {} \ GROUP BY p.id, p.name \ ORDER BY today_energy DESC, p.id ASC", org_filter ); let mut query = sqlx::query_as::<_, ProjectStatRow>(&sql); for v in &org_binds { query = query.bind(v); } let rows = query.fetch_all(pool).await?; let list: Vec = rows.iter().map(|r| { json!({ "project_id": r.id, "name": r.name, "cabinet_count": r.cabinet_count, "today_energy": r.today_energy.unwrap_or(0), "month_energy": r.month_energy.unwrap_or(0), }) }).collect(); Ok(Json(json!({ "code": 0, "data": list, "message": "ok" }))) } /// GET /api/energy-stats/trend — 趋势数据 /// /// granularity: day | week | month, 默认 day /// range: 天数范围, 默认 30 pub async fn get_trend( user: CurrentUser, State(state): State, Query(params): Query, ) -> Result, AppError> { auth::check_permission(&user, "energy:view")?; let pool = &state.mysql; // 组织过滤:es.cabinet_id → cabinets.project_id → projects.organization_id let (org_filter, org_binds): (String, Vec) = if user.role_level >= 2 { (String::new(), vec![]) } else if let Some(oid) = user.organization_id { ( " AND es.cabinet_id IN (SELECT c.id FROM cabinets c JOIN projects p ON c.project_id = p.id WHERE p.organization_id = ?)".to_string(), vec![oid.to_string()], ) } else { (" AND 1=0".to_string(), vec![]) }; let rows = match params.granularity.as_str() { "week" => { let sql = format!( "SELECT DATE_FORMAT(stat_date, '%Y-%u') AS period, \ CAST(SUM(total_energy) AS SIGNED) AS energy, CAST(SUM(charge_count) AS SIGNED) AS charge_count \ FROM energy_stats es \ WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL ? DAY){} \ GROUP BY period ORDER BY period ASC", org_filter ); let mut q = sqlx::query_as::<_, TrendRow>(&sql).bind(params.range); for v in &org_binds { q = q.bind(v); } q.fetch_all(pool).await? } "month" => { let sql = format!( "SELECT DATE_FORMAT(stat_date, '%Y-%m') AS period, \ CAST(SUM(total_energy) AS SIGNED) AS energy, CAST(SUM(charge_count) AS SIGNED) AS charge_count \ FROM energy_stats es \ WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL ? DAY){} \ GROUP BY period ORDER BY period ASC", org_filter ); let mut q = sqlx::query_as::<_, TrendRow>(&sql).bind(params.range); for v in &org_binds { q = q.bind(v); } q.fetch_all(pool).await? } _ => { // day let sql = format!( "SELECT stat_date AS period, \ total_energy AS energy, charge_count AS charge_count \ FROM energy_stats es \ WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL ? DAY){} \ ORDER BY stat_date ASC", org_filter ); let mut q = sqlx::query_as::<_, TrendRow>(&sql).bind(params.range); for v in &org_binds { q = q.bind(v); } q.fetch_all(pool).await? } }; let list: Vec = rows.iter().map(|r| { json!({ "period": r.period, "energy": r.energy.unwrap_or(0), "charge_count": r.charge_count.unwrap_or(0), }) }).collect(); Ok(Json(json!({ "code": 0, "data": { "granularity": params.granularity, "list": list, }, "message": "ok" }))) } /// GET /api/energy-stats/export — 异步导出能耗报表 pub async fn export_energy_stats( user: CurrentUser, State(state): State, Query(params): Query, ) -> Result, AppError> { auth::check_permission(&user, "energy:export")?; let user_id = user.user_id; let file_name = generate_export_filename(¶ms); // 序列化查询参数(含组织过滤信息) let query_json = serde_json::to_string(&json!({ "cabinet_id": params.cabinet_id, "project_id": params.project_id, "start_date": params.start_date, "end_date": params.end_date, "org_id": user.organization_id, "role_level": user.role_level, })) .unwrap_or_default(); // 插入下载任务 let result = sqlx::query( "INSERT INTO download_tasks (user_id, task_type, status, file_name, error_message) \ VALUES (?, ?, 0, ?, ?)" ) .bind(user_id) .bind("energy_stats") .bind(&file_name) .bind(&query_json) .execute(&state.mysql) .await?; let task_id = result.last_insert_id() as i64; // 后台异步生成 Excel let pool = state.mysql.clone(); tokio::spawn(async move { if let Err(e) = generate_energy_excel(pool, task_id).await { tracing::error!("导出能耗统计失败 task_id={}: {}", task_id, e); } }); Ok(Json(json!({ "code": 0, "data": { "task_id": task_id }, "message": "导出任务已创建" }))) } /// 生成导出文件名 fn generate_export_filename(params: &ExportQuery) -> String { let start = params .start_date .as_ref() .map(|s| s.replace('-', "")) .unwrap_or_default(); let end = params .end_date .as_ref() .map(|s| s.replace('-', "")) .unwrap_or_default(); format!("能耗统计_{}-{}.xlsx", start, end) } /// 构建导出条件 fn build_export_clause( cabinet_id: Option, project_id: Option, start_date: Option<&str>, end_date: Option<&str>, ) -> (String, Vec) { let mut conditions = Vec::new(); let mut bind_values = Vec::new(); if let Some(id) = cabinet_id { conditions.push("es.cabinet_id = ?"); bind_values.push(id.to_string()); } if let Some(id) = project_id { conditions.push("c.project_id = ?"); bind_values.push(id.to_string()); } if let Some(d) = start_date { conditions.push("es.stat_date >= ?"); bind_values.push(d.to_string()); } if let Some(d) = end_date { conditions.push("es.stat_date <= ?"); bind_values.push(d.to_string()); } let clause = if conditions.is_empty() { String::new() } else { format!("WHERE {}", conditions.join(" AND ")) }; (clause, bind_values) } /// 后台异步生成能耗 Excel 文件 async fn generate_energy_excel( pool: sqlx::MySqlPool, task_id: i64, ) -> Result<(), Box> { // 更新状态为处理中 sqlx::query("UPDATE download_tasks SET status = 1 WHERE id = ?") .bind(task_id) .execute(&pool) .await?; // 获取查询参数 let row: Option<(String, String)> = sqlx::query_as( "SELECT file_name, error_message FROM download_tasks WHERE id = ?" ) .bind(task_id) .fetch_optional(&pool) .await?; let (file_name, query_json) = row.ok_or("任务不存在")?; let params: Value = serde_json::from_str(&query_json).unwrap_or(json!({})); let (mut where_clause, mut bind_values) = build_export_clause( params.get("cabinet_id").and_then(|v| v.as_i64()), params.get("project_id").and_then(|v| v.as_i64()), params.get("start_date").and_then(|v| v.as_str()), params.get("end_date").and_then(|v| v.as_str()), ); // 组织过滤(非总管理员) let role_level = params.get("role_level").and_then(|v| v.as_i64()).unwrap_or(0); let org_id = params.get("org_id").and_then(|v| v.as_i64()); if role_level < 2 { if let Some(oid) = org_id { let prefix = if where_clause.is_empty() { "WHERE " } else { " AND " }; where_clause.push_str(&format!("{}es.cabinet_id IN (SELECT c.id FROM cabinets c JOIN projects p ON c.project_id = p.id WHERE p.organization_id = ?)", prefix)); bind_values.push(oid.to_string()); } else { let prefix = if where_clause.is_empty() { "WHERE " } else { " AND " }; where_clause.push_str(&format!("{}1=0", prefix)); } } // 查询能耗明细 let sql = format!( "SELECT es.stat_date, es.cabinet_id, c.name AS cabinet_name, \ p.name AS project_name, \ es.total_energy, \ es.charge_count \ FROM energy_stats es \ LEFT JOIN cabinets c ON es.cabinet_id = c.id \ LEFT JOIN projects p ON c.project_id = p.id \ {} ORDER BY es.stat_date DESC, es.cabinet_id ASC", where_clause ); let mut query = sqlx::query_as::<_, EnergyDetailRow>(&sql); for val in &bind_values { query = query.bind(val); } let rows = query.fetch_all(&pool).await?; // 更新进度 50% sqlx::query("UPDATE download_tasks SET progress = 50 WHERE id = ?") .bind(task_id) .execute(&pool) .await?; // 生成 Excel let mut workbook = rust_xlsxwriter::Workbook::new(); let sheet = workbook.add_worksheet(); sheet.set_name("能耗统计")?; let headers = [ "日期", "柜子ID", "柜子名称", "所属项目", "能耗(Wh)", "充电次数", ]; for (col, header) in headers.iter().enumerate() { sheet.write_string(0, col as u16, *header)?; } for (row_idx, record) in rows.iter().enumerate() { let row = (row_idx + 1) as u32; sheet.write_string(row, 0, record.stat_date.to_string())?; sheet.write_number(row, 1, record.cabinet_id as f64)?; sheet.write_string(row, 2, record.cabinet_name.as_deref().unwrap_or("-"))?; sheet.write_string(row, 3, record.project_name.as_deref().unwrap_or("-"))?; sheet.write_number(row, 4, record.total_energy.unwrap_or(0) as f64)?; sheet.write_number(row, 5, record.charge_count.unwrap_or(0) as f64)?; } // 保存文件 let file_dir = "./exports"; tokio::fs::create_dir_all(file_dir).await?; let file_path = format!("{}/{}", file_dir, file_name); workbook.save(&file_path)?; // 更新任务状态为完成 sqlx::query( "UPDATE download_tasks SET status = 2, progress = 100, file_path = ?, \ error_message = NULL, completed_at = NOW() WHERE id = ?" ) .bind(&file_path) .bind(task_id) .execute(&pool) .await?; tracing::info!("能耗统计导出完成 task_id={} file={}", task_id, file_path); Ok(()) }