//! H5 用户端 API //! //! 提供微信扫码充电的 H5 页面所需的全部后端接口: //! - POST /api/h5/auth/login — H5 手机号登录 //! - GET /api/h5/dashboard — 首页仪表盘数据 //! - GET /api/h5/projects — 项目列表 //! - GET /api/h5/cabinets — 设备列表(按项目过滤) //! - GET /api/h5/cabinets/:id — 设备详情(含仓板+仓体状态) //! - GET /api/h5/cabinets/abstract/:aid — 通过抽象ID查设备(扫码入口) //! - GET /api/h5/compartments/:id — 仓体/通道详情 //! - POST /api/h5/charge/start — 开始充电 //! - POST /api/h5/charge/stop — 停止充电 //! - POST /api/h5/door/open — 开门 use axum::extract::{Path, Query, State}; use axum::Json; use serde::Deserialize; use serde_json::{json, Value}; use crate::commands::{AppState, DeviceCommand}; use crate::error::AppError; use crate::middleware::auth::{self, CurrentUser}; // --------------------------------------------------------------------------- // 请求体定义 // --------------------------------------------------------------------------- /// H5 登录请求体 #[derive(Debug, Deserialize)] pub(crate) struct H5LoginInput { phone: String, password: String, } /// 充电控制请求体(开始/停止充电、开门共用) #[derive(Debug, Deserialize)] pub(crate) struct CompartmentActionInput { compartment_id: i64, } /// 设备列表查询参数 #[derive(Debug, Deserialize)] pub(crate) struct CabinetListQuery { project_id: Option, } // --------------------------------------------------------------------------- // 行映射结构体 // --------------------------------------------------------------------------- /// 项目简要信息行 #[derive(Debug, sqlx::FromRow)] struct ProjectRow { id: i64, name: String, } /// 柜子列表行 #[derive(Debug, sqlx::FromRow)] struct H5CabinetRow { id: i64, abstract_id: String, name: Option, status: i8, project_id: Option, } /// 仓控板行 #[derive(Debug, sqlx::FromRow)] struct H5CabinBoardRow { id: i64, board_id: i8, status: i8, } /// 仓体行 #[derive(Debug, sqlx::FromRow)] struct H5CompartmentRow { id: i64, channel_id: i8, status: i8, } /// 充电记录简要行 #[derive(Debug, sqlx::FromRow)] struct ChargeRecordBriefRow { id: i64, start_time: Option, end_time: Option, energy: Option, } /// 柜子 IMEI 查询行(用于指令下发) #[derive(Debug, sqlx::FromRow)] struct CabinetImeiRow { imei: String, } // --------------------------------------------------------------------------- // H5 登录 // --------------------------------------------------------------------------- /// POST /api/h5/auth/login — H5 用户登录(复用 users 表) pub async fn h5_login( State(state): State, Json(input): Json, ) -> Result, AppError> { let pool = &state.mysql; if input.phone.is_empty() || input.password.is_empty() { return Err(AppError::BadRequest("手机号和密码不能为空".into())); } // 查询用户 let user_row: Option<(i64, Option, String, String, i32)> = sqlx::query_as( "SELECT id, name, phone, password, status FROM users WHERE phone = ?", ) .bind(&input.phone) .fetch_optional(pool) .await?; let (id, name, phone, stored_password, status) = user_row.ok_or_else(|| AppError::Unauthorized("手机号或密码错误".into()))?; if status == 0 { return Err(AppError::Forbidden("账户已被禁用".into())); } let valid = auth::verify_password(&input.password, &stored_password) .map_err(|_| AppError::Internal("密码验证失败".into()))?; if !valid { return Err(AppError::Unauthorized("手机号或密码错误".into())); } let token = auth::generate_token(id, &phone)?; Ok(Json(json!({ "code": 0, "data": { "token": token, "user": { "id": id, "phone": phone, "name": name, } }, "message": "登录成功" }))) } // --------------------------------------------------------------------------- // 仪表盘 // --------------------------------------------------------------------------- /// GET /api/h5/dashboard — H5 首页仪表盘数据(带组织隔离) pub async fn get_dashboard( user: CurrentUser, State(state): State, ) -> Result, AppError> { let pool = &state.mysql; // 根据角色确定组织过滤条件 let org_id = user.organization_id; let is_super = user.role_level >= 2; // 统计在线柜子数(带组织隔离) let online_cabinets = count_online_cabinets(pool, is_super, org_id).await?; // 统计充电中仓体数(带组织隔离) let charging_count = count_compartments_by_status(pool, is_super, org_id, 1).await?; // 统计空闲仓体数(带组织隔离) let idle_channels = count_compartments_by_status(pool, is_super, org_id, 0).await?; // 统计故障仓体数(带组织隔离) let fault_channels = count_compartments_by_status(pool, is_super, org_id, 2).await?; // 今日充电次数(带组织隔离) let today_charge_count = count_today_charges(pool, is_super, org_id).await?; // 今日充电量(kWh,带组织隔离) let today_energy = sum_today_energy(pool, is_super, org_id).await?; // 今日充电时长(小时,带组织隔离) let today_charge_hours = sum_today_charge_hours(pool, is_super, org_id).await?; // 故障告警(取最近10条故障仓体,返回项目名+柜名) let alerts = fetch_fault_alerts(pool, is_super, org_id).await?; // 项目摘要 let projects = fetch_project_summaries(pool, &user).await?; Ok(Json(json!({ "code": 0, "data": { "online_cabinets": online_cabinets, "charging_count": charging_count, "idle_channels": idle_channels, "fault_channels": fault_channels, "today_charge_count": today_charge_count, "today_energy_kwh": today_energy, "today_charge_hours": today_charge_hours, "alerts": alerts, "projects": projects, }, "message": "ok" }))) } /// 统计在线柜子数(带组织隔离) async fn count_online_cabinets( pool: &sqlx::MySqlPool, is_super: bool, org_id: Option, ) -> Result { if !is_super && org_id.is_none() { return Ok(0); } let count: i64 = if is_super { sqlx::query_scalar("SELECT COUNT(*) FROM cabinets WHERE status = 1") .fetch_one(pool) .await? } else { sqlx::query_scalar( r#"SELECT COUNT(*) FROM cabinets cab INNER JOIN projects p ON cab.project_id = p.id WHERE cab.status = 1 AND p.organization_id = ?"#, ) .bind(org_id.unwrap()) .fetch_one(pool) .await? }; Ok(count) } /// 按状态统计仓体数(带组织隔离,通过 cabin_boards → cabinets → projects 链路) async fn count_compartments_by_status( pool: &sqlx::MySqlPool, is_super: bool, org_id: Option, status: i8, ) -> Result { if !is_super && org_id.is_none() { return Ok(0); } let count: i64 = if is_super { sqlx::query_scalar("SELECT COUNT(*) FROM compartments WHERE status = ?") .bind(status) .fetch_one(pool) .await? } else { sqlx::query_scalar( r#"SELECT COUNT(*) FROM compartments comp INNER JOIN cabin_boards cb ON comp.cabin_board_id = cb.id INNER JOIN cabinets cab ON cb.cabinet_id = cab.id INNER JOIN projects p ON cab.project_id = p.id WHERE comp.status = ? AND p.organization_id = ?"#, ) .bind(status) .bind(org_id.unwrap()) .fetch_one(pool) .await? }; Ok(count) } /// 今日充电次数(带组织隔离) async fn count_today_charges( pool: &sqlx::MySqlPool, is_super: bool, org_id: Option, ) -> Result { if !is_super && org_id.is_none() { return Ok(0); } let count: i64 = if is_super { sqlx::query_scalar( "SELECT COUNT(*) FROM charge_records WHERE DATE(start_time) = CURDATE()", ) .fetch_one(pool) .await? } else { sqlx::query_scalar( r#"SELECT COUNT(*) FROM charge_records cr INNER JOIN cabinets cab ON cr.cabinet_id = cab.id INNER JOIN projects p ON cab.project_id = p.id WHERE DATE(cr.start_time) = CURDATE() AND p.organization_id = ?"#, ) .bind(org_id.unwrap()) .fetch_one(pool) .await? }; Ok(count) } /// 今日充电量(kWh,带组织隔离) async fn sum_today_energy( pool: &sqlx::MySqlPool, is_super: bool, org_id: Option, ) -> Result { if !is_super && org_id.is_none() { return Ok(0.0); } let energy: Option = if is_super { sqlx::query_scalar( "SELECT CAST(SUM(energy) AS DECIMAL(10,2)) FROM charge_records WHERE DATE(start_time) = CURDATE()", ) .fetch_one(pool) .await? } else { sqlx::query_scalar( r#"SELECT CAST(SUM(cr.energy) AS DECIMAL(10,2)) FROM charge_records cr INNER JOIN cabinets cab ON cr.cabinet_id = cab.id INNER JOIN projects p ON cab.project_id = p.id WHERE DATE(cr.start_time) = CURDATE() AND p.organization_id = ?"#, ) .bind(org_id.unwrap()) .fetch_one(pool) .await? }; Ok(energy.unwrap_or(0.0)) } /// 今日充电时长(小时,带组织隔离) async fn sum_today_charge_hours( pool: &sqlx::MySqlPool, is_super: bool, org_id: Option, ) -> Result { if !is_super && org_id.is_none() { return Ok(0.0); } // 计算总秒数(未结束的记录用 NOW() 代替 end_time) let total_seconds: f64 = if is_super { sqlx::query_scalar( r#"SELECT COALESCE(SUM( TIMESTAMPDIFF(SECOND, start_time, COALESCE(end_time, NOW())) ), 0) FROM charge_records WHERE DATE(start_time) = CURDATE()"#, ) .fetch_one(pool) .await? } else { sqlx::query_scalar( r#"SELECT COALESCE(SUM( TIMESTAMPDIFF(SECOND, cr.start_time, COALESCE(cr.end_time, NOW())) ), 0) FROM charge_records cr INNER JOIN cabinets cab ON cr.cabinet_id = cab.id INNER JOIN projects p ON cab.project_id = p.id WHERE DATE(cr.start_time) = CURDATE() AND p.organization_id = ?"#, ) .bind(org_id.unwrap()) .fetch_one(pool) .await? }; // 秒转小时,保留1位小数 Ok((total_seconds / 3600.0 * 10.0).round() / 10.0) } /// 获取故障告警列表(返回项目名+柜名+通道号,带组织隔离) async fn fetch_fault_alerts( pool: &sqlx::MySqlPool, is_super: bool, org_id: Option, ) -> Result, AppError> { if !is_super && org_id.is_none() { return Ok(vec![]); } let base_sql = if is_super { r#"SELECT IFNULL(p.name, '') AS project, IFNULL(cab.name, cab.abstract_id) AS cabinet, comp.channel_id FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id JOIN cabinets cab ON cb.cabinet_id = cab.id LEFT JOIN projects p ON cab.project_id = p.id WHERE comp.status = 2 ORDER BY comp.updated_at DESC LIMIT 10"# } else { r#"SELECT IFNULL(p.name, '') AS project, IFNULL(cab.name, cab.abstract_id) AS cabinet, comp.channel_id FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id JOIN cabinets cab ON cb.cabinet_id = cab.id INNER JOIN projects p ON cab.project_id = p.id WHERE comp.status = 2 AND p.organization_id = ? ORDER BY comp.updated_at DESC LIMIT 10"# }; let rows: Vec<(String, String, i8)> = if is_super { sqlx::query_as(base_sql).fetch_all(pool).await? } else { sqlx::query_as(base_sql) .bind(org_id.unwrap()) .fetch_all(pool) .await? }; Ok(rows .into_iter() .map(|(project, cabinet, channel)| { json!({ "project": project, "cabinet": cabinet, "channel": channel, "alert_type": "fault", }) }) .collect()) } /// 查询用户可见的项目摘要(含柜子/充电/空闲/故障统计) async fn fetch_project_summaries( pool: &sqlx::MySqlPool, user: &CurrentUser, ) -> Result, AppError> { // 获取用户可见的项目 let projects: Vec = if user.role_level >= 2 { sqlx::query_as::<_, ProjectRow>("SELECT id, name FROM projects ORDER BY id") .fetch_all(pool) .await? } else if let Some(org_id) = user.organization_id { sqlx::query_as::<_, ProjectRow>( "SELECT id, name FROM projects WHERE organization_id = ? ORDER BY id", ) .bind(org_id) .fetch_all(pool) .await? } else { vec![] }; let mut result = Vec::with_capacity(projects.len()); for p in &projects { let cabinet_count: i64 = sqlx::query_scalar( "SELECT COUNT(*) FROM cabinets WHERE project_id = ?", ) .bind(p.id) .fetch_one(pool) .await?; let charging: i64 = sqlx::query_scalar( r#"SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id JOIN cabinets cab ON cb.cabinet_id = cab.id WHERE cab.project_id = ? AND comp.status = 1"#, ) .bind(p.id) .fetch_one(pool) .await?; let idle: i64 = sqlx::query_scalar( r#"SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id JOIN cabinets cab ON cb.cabinet_id = cab.id WHERE cab.project_id = ? AND comp.status = 0"#, ) .bind(p.id) .fetch_one(pool) .await?; let fault: i64 = sqlx::query_scalar( r#"SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id JOIN cabinets cab ON cb.cabinet_id = cab.id WHERE cab.project_id = ? AND comp.status = 2"#, ) .bind(p.id) .fetch_one(pool) .await?; result.push(json!({ "id": p.id, "name": p.name, "cabinet_count": cabinet_count, "charging": charging, "idle": idle, "fault": fault, })); } Ok(result) } // --------------------------------------------------------------------------- // 项目列表 // --------------------------------------------------------------------------- /// GET /api/h5/projects — H5 项目列表(简要) pub async fn get_projects( user: CurrentUser, State(state): State, ) -> Result, AppError> { let pool = &state.mysql; let projects: Vec = if user.role_level >= 2 { sqlx::query_as::<_, ProjectRow>("SELECT id, name FROM projects ORDER BY id") .fetch_all(pool) .await? } else if let Some(org_id) = user.organization_id { sqlx::query_as::<_, ProjectRow>( "SELECT id, name FROM projects WHERE organization_id = ? ORDER BY id", ) .bind(org_id) .fetch_all(pool) .await? } else { vec![] }; let list: Vec = projects .into_iter() .map(|p| json!({ "id": p.id, "name": p.name })) .collect(); Ok(Json(json!({ "code": 0, "data": list, "message": "ok" }))) } // --------------------------------------------------------------------------- // 设备列表 // --------------------------------------------------------------------------- /// GET /api/h5/cabinets — H5 设备列表(可按项目过滤,带组织隔离) pub async fn get_cabinets( user: CurrentUser, State(state): State, Query(query): Query, ) -> Result, AppError> { let pool = &state.mysql; // 权限校验 auth::check_permission(&user, "device:view")?; let rows: Vec = if let Some(pid) = query.project_id { // 按项目过滤(同时验证组织权限) if user.role_level >= 2 { sqlx::query_as::<_, H5CabinetRow>( "SELECT id, abstract_id, name, status, project_id FROM cabinets WHERE project_id = ?", ) .bind(pid) .fetch_all(pool) .await? } else if let Some(org_id) = user.organization_id { sqlx::query_as::<_, H5CabinetRow>( r#"SELECT id, abstract_id, name, status, project_id FROM cabinets cab INNER JOIN projects p ON cab.project_id = p.id WHERE cab.project_id = ? AND p.organization_id = ?"#, ) .bind(pid) .bind(org_id) .fetch_all(pool) .await? } else { vec![] } } else if user.role_level >= 2 { // 总管理员看全部 sqlx::query_as::<_, H5CabinetRow>( "SELECT id, abstract_id, name, status, project_id FROM cabinets", ) .fetch_all(pool) .await? } else if let Some(org_id) = user.organization_id { // 企业管理员只看本组织 sqlx::query_as::<_, H5CabinetRow>( r#"SELECT id, abstract_id, name, status, project_id FROM cabinets cab INNER JOIN projects p ON cab.project_id = p.id WHERE p.organization_id = ?"#, ) .bind(org_id) .fetch_all(pool) .await? } else { vec![] }; let list: Vec = rows .into_iter() .map(|r| { json!({ "id": r.id, "abstract_id": r.abstract_id, "name": r.name, "status": r.status, "project_id": r.project_id, }) }) .collect(); Ok(Json(json!({ "code": 0, "data": list, "message": "ok" }))) } // --------------------------------------------------------------------------- // 设备详情 // --------------------------------------------------------------------------- /// GET /api/h5/cabinets/:id — H5 设备详情(含仓控板+仓体,带权限校验) pub async fn get_cabinet_detail( user: CurrentUser, State(state): State, Path(id): Path, ) -> Result, AppError> { let pool = &state.mysql; // 权限校验 auth::check_permission(&user, "device:view")?; // 组织隔离:验证用户是否有权访问该设备 verify_cabinet_access(&user, id, pool).await?; // 查询柜子基本信息 let cab: Option<(i64, String, String, Option, i8)> = sqlx::query_as( "SELECT id, abstract_id, imei, name, status FROM cabinets WHERE id = ?", ) .bind(id) .fetch_optional(pool) .await?; let (cab_id, abstract_id, imei, name, status) = cab.ok_or_else(|| AppError::NotFound("设备不存在".into()))?; // 查询仓控板 let boards: Vec = sqlx::query_as::<_, H5CabinBoardRow>( "SELECT id, board_id, status FROM cabin_boards WHERE cabinet_id = ? ORDER BY board_id", ) .bind(cab_id) .fetch_all(pool) .await?; // 查询每个仓控板下的仓体 let mut cabin_boards = Vec::with_capacity(boards.len()); for board in &boards { let comps: Vec = sqlx::query_as::<_, H5CompartmentRow>( "SELECT id, channel_id, status FROM compartments WHERE cabin_board_id = ? ORDER BY channel_id", ) .bind(board.id) .fetch_all(pool) .await?; let compartments: Vec = comps .into_iter() .map(|c| { json!({ "id": c.id, "channel_id": c.channel_id, "status": c.status, }) }) .collect(); cabin_boards.push(json!({ "id": board.id, "board_id": board.board_id, "status": board.status, "compartments": compartments, })); } Ok(Json(json!({ "code": 0, "data": { "id": cab_id, "abstract_id": abstract_id, "imei": imei, "name": name, "status": status, "cabin_boards": cabin_boards, }, "message": "ok" }))) } /// GET /api/h5/cabinets/abstract/:abstract_id — 通过抽象ID查设备(扫码入口,免认证) pub async fn get_cabinet_by_abstract_id( State(state): State, Path(abstract_id): Path, ) -> Result, AppError> { let pool = &state.mysql; let cab: Option<(i64, String, String, Option, i8)> = sqlx::query_as( "SELECT id, abstract_id, imei, name, status FROM cabinets WHERE abstract_id = ?", ) .bind(&abstract_id) .fetch_optional(pool) .await?; let (cab_id, abs_id, imei, name, status) = cab.ok_or_else(|| AppError::NotFound("设备不存在".into()))?; // 查询仓控板+仓体 let boards: Vec = sqlx::query_as::<_, H5CabinBoardRow>( "SELECT id, board_id, status FROM cabin_boards WHERE cabinet_id = ? ORDER BY board_id", ) .bind(cab_id) .fetch_all(pool) .await?; let mut cabin_boards = Vec::with_capacity(boards.len()); for board in &boards { let comps: Vec = sqlx::query_as::<_, H5CompartmentRow>( "SELECT id, channel_id, status FROM compartments WHERE cabin_board_id = ? ORDER BY channel_id", ) .bind(board.id) .fetch_all(pool) .await?; let compartments: Vec = comps .into_iter() .map(|c| json!({ "id": c.id, "channel_id": c.channel_id, "status": c.status })) .collect(); cabin_boards.push(json!({ "id": board.id, "board_id": board.board_id, "status": board.status, "compartments": compartments, })); } Ok(Json(json!({ "code": 0, "data": { "id": cab_id, "abstract_id": abs_id, "imei": imei, "name": name, "status": status, "cabin_boards": cabin_boards, }, "message": "ok" }))) } // --------------------------------------------------------------------------- // 仓体/通道详情 // --------------------------------------------------------------------------- /// GET /api/h5/compartments/:id — 仓体详情(含充电记录,带权限校验) pub async fn get_compartment_detail( user: CurrentUser, State(state): State, Path(id): Path, ) -> Result, AppError> { let pool = &state.mysql; // 权限校验 auth::check_permission(&user, "device:view")?; // 查询仓体基本信息 let comp: Option<(i64, i8, i8, i64)> = sqlx::query_as( "SELECT id, channel_id, status, cabin_board_id FROM compartments WHERE id = ?", ) .bind(id) .fetch_optional(pool) .await?; let (comp_id, channel_id, status, cabin_board_id) = comp.ok_or_else(|| AppError::NotFound("仓体不存在".into()))?; // 组织隔离:通过 cabin_board → cabinet → project 链路验证访问权限 let cabinet_id: Option<(i64,)> = sqlx::query_as( "SELECT cabinet_id FROM cabin_boards WHERE id = ?", ) .bind(cabin_board_id) .fetch_optional(pool) .await?; if let Some((cab_id,)) = cabinet_id { verify_cabinet_access(&user, cab_id, pool).await?; } // 从 Redis 获取实时数据 let redis_key = format!("compartment:status:{}", comp_id); let mut conn = state.redis.clone(); let realtime_data: Option = redis::cmd("GET") .arg(&redis_key) .query_async(&mut conn) .await .unwrap_or(None); let realtime: Value = realtime_data .and_then(|s| serde_json::from_str(&s).ok()) .unwrap_or(json!({})); // 查询最近充电记录 let records: Vec = sqlx::query_as::<_, ChargeRecordBriefRow>( r#"SELECT id, start_time, end_time, CAST(energy AS DOUBLE) AS energy FROM charge_records WHERE compartment_id = ? ORDER BY start_time DESC LIMIT 10"#, ) .bind(comp_id) .fetch_all(pool) .await?; let charge_records: Vec = records .into_iter() .map(|r| { json!({ "id": r.id, "start_time": r.start_time.map(|t| t.to_string()), "end_time": r.end_time.map(|t| t.to_string()), "energy_kwh": r.energy.unwrap_or(0.0), "cost": 0, }) }) .collect(); Ok(Json(json!({ "code": 0, "data": { "id": comp_id, "channel_id": channel_id, "status": status, "cabin_board_id": cabin_board_id, "voltage": realtime.get("voltage").and_then(|v| v.as_f64()).unwrap_or(0.0), "current": realtime.get("current").and_then(|v| v.as_f64()).unwrap_or(0.0), "power": realtime.get("power").and_then(|v| v.as_f64()).unwrap_or(0.0), "energy": realtime.get("energy").and_then(|v| v.as_f64()).unwrap_or(0.0), "soc": realtime.get("soc").and_then(|v| v.as_i64()).unwrap_or(0), "soh": realtime.get("soh").and_then(|v| v.as_i64()).unwrap_or(0), "cell_voltages": realtime.get("cell_voltages").cloned().unwrap_or(json!([])), "cell_temps": realtime.get("cell_temps").cloned().unwrap_or(json!([])), "bms_alerts": realtime.get("bms_alerts").cloned().unwrap_or(json!([])), "charge_records": charge_records, }, "message": "ok" }))) } // --------------------------------------------------------------------------- // 充电控制 // --------------------------------------------------------------------------- /// POST /api/h5/charge/start — 开始充电 pub async fn start_charge( user: CurrentUser, State(state): State, Json(body): Json, ) -> Result, AppError> { auth::check_permission(&user, "charge:start")?; send_compartment_command(&state, body.compartment_id, "on").await } /// POST /api/h5/charge/stop — 停止充电 pub async fn stop_charge( user: CurrentUser, State(state): State, Json(body): Json, ) -> Result, AppError> { auth::check_permission(&user, "charge:stop")?; send_compartment_command(&state, body.compartment_id, "off").await } /// POST /api/h5/door/open — 开门 pub async fn open_door( user: CurrentUser, State(state): State, Json(body): Json, ) -> Result, AppError> { auth::check_permission(&user, "device:door:unlock")?; send_compartment_command(&state, body.compartment_id, "open").await } // --------------------------------------------------------------------------- // 辅助函数 // --------------------------------------------------------------------------- /// 根据仓体ID查找所属柜子的IMEI,并通过 Redis 下发指令 async fn send_compartment_command( state: &AppState, compartment_id: i64, action: &str, ) -> Result, AppError> { let pool = &state.mysql; // 通过 compartment → cabin_board → cabinet 链路查找 IMEI 和 sub_device_id let row: Option = sqlx::query_as( r#"SELECT cab.imei FROM cabinets cab JOIN cabin_boards cb ON cb.cabinet_id = cab.id JOIN compartments comp ON comp.cabin_board_id = cb.id WHERE comp.id = ?"#, ) .bind(compartment_id) .fetch_optional(pool) .await?; let cab_row = row.ok_or_else(|| AppError::NotFound("仓体对应的设备不存在".into()))?; let imei = &cab_row.imei; // 查询 sub_device_id(仓控板 ID) let sub_row: Option<(i8,)> = sqlx::query_as( r#"SELECT cb.board_id FROM cabin_boards cb JOIN compartments comp ON comp.cabin_board_id = cb.id WHERE comp.id = ?"#, ) .bind(compartment_id) .fetch_optional(pool) .await?; let sub_device_id = sub_row.map(|(bid,)| bid as u8); // 检查是否有设备节点在线 if !state.nodes.is_device_online(imei).await { return Err(AppError::BadRequest("设备不在线".into())); } // 构建指令并通过 Redis 下发 let cmd = DeviceCommand { act: action.to_string(), msg_id: 0, dev_id: Some(imei.clone()), sub_device_id, params: Some(json!({})), }; let json_str = cmd.to_json(); let mut conn = state.redis.clone(); redis::cmd("LPUSH") .arg("device:commands") .arg(&json_str) .query_async::<()>(&mut conn) .await .map_err(|e| AppError::Internal(format!("Redis 推入失败: {}", e)))?; tracing::info!( "[h5] 指令已推入队列 imei={} action={} compartment_id={}", imei, action, compartment_id ); Ok(Json(json!({ "code": 0, "data": { "success": true }, "message": "指令已下发" }))) } /// 验证用户是否有权访问指定柜子(组织数据隔离) /// /// 总管理员(role_level >= 2)可访问所有柜子; /// 企业管理员只能访问本组织下项目关联的柜子; /// 无组织关联的用户无权访问任何柜子。 async fn verify_cabinet_access( user: &CurrentUser, cabinet_id: i64, pool: &sqlx::MySqlPool, ) -> Result<(), AppError> { // 总管理员无需校验 if user.role_level >= 2 { return Ok(()); } let org_id = user .organization_id .ok_or_else(|| AppError::Forbidden("无组织关联,无权访问设备".into()))?; // 通过 cabinet → project → organization 链路验证归属 let access: Option<(i64,)> = sqlx::query_as( r#"SELECT cab.id FROM cabinets cab INNER JOIN projects p ON cab.project_id = p.id WHERE cab.id = ? AND p.organization_id = ?"#, ) .bind(cabinet_id) .bind(org_id) .fetch_optional(pool) .await?; if access.is_none() { return Err(AppError::Forbidden("无权访问该设备".into())); } Ok(()) }