"""设备详情、微信摘要、删除和单机命令的专用服务。""" from __future__ import annotations from typing import Any, Optional import uuid from services.device_manager import device_manager from services.device_transport import device_transport from services.device_id_util import sanitize_sensitive_fields from services.ws_hub import ws_hub from services.workbench import ( ControlPlaneStorageUnavailable, IdempotencyConflict, control_plane_store, now_iso, record_audit, reserve_idempotency, save_idempotency_result, ) class DeviceOnlineError(RuntimeError): def __init__(self, result: dict): self.result = result class DeviceDeleteConfirmationRequired(RuntimeError): def __init__(self, result: dict): self.result = result class DeviceOfflineError(RuntimeError): def __init__(self, result: dict): self.result = result class DeviceCommandVerificationError(RuntimeError): def __init__(self, result: dict): self.result = result class DeviceConsoleService: def __init__(self, store: Any = control_plane_store) -> None: self.store = store async def _log_command_history(self, *, device_id: str, action: str, payload: dict, result: dict) -> None: """保存命令历史；只留下脱敏后的请求和结果必要字段。""" history_result = sanitize_sensitive_fields({ "command_id": result.get("command_id"), "status": result.get("status"), "trace_id": result.get("trace_id"), "readback": result.get("readback"), "raw_receipt": result.get("raw_receipt"), "verified": result.get("status") == "succeeded", }) await device_manager.log_command( device_id, action, sanitize_sensitive_fields(payload), history_result, ) async def _lookup(self, device_id: str) -> Optional[dict]: online = ws_hub.get_device_info(device_id) stored = None try: stored = await device_manager.get_device(device_id) except Exception: stored = None if not online and not stored: return None merged = {**(stored or {}), **(online or {})} merged["device_id"] = device_id # Mongo 仅保存历史登记；控制台“在线”只由本进程 WSS 会话判定。 is_online = ws_hub.is_online(device_id) merged["status"] = "online" if is_online else "offline" merged["connection_type"] = "websocket" if is_online else "offline" return merged async def detail(self, device_id: str) -> Optional[dict]: item = await self._lookup(device_id) if item is None: return None return sanitize_sensitive_fields({ "device_id": device_id, "name": item.get("name") or item.get("model") or device_id, "status": item.get("status", "offline"), "project_id": item.get("project_id"), "groups": await self._groups_for(device_id), "capabilities": item.get("capabilities") or [], "connection": { "type": item.get("connection_type", "offline"), "online": item.get("status") == "online", "last_heartbeat": item.get("last_heartbeat"), "source": "wss_heartbeat" if item.get("status") == "online" else "mongo.devices", }, "version": { "agent_version": item.get("agent_version"), "wechat_version": self._wechat_value(item, "wechat_version"), "android_version": item.get("android_version"), "source_at": self._source_at(item), }, "wechat": await self.wechat_summary(device_id, item=item), "health": { "battery_level": item.get("battery_level"), "memory_usage_pct": item.get("memory_usage_pct"), "storage_free_mb": item.get("storage_free_mb"), "last_heartbeat": item.get("last_heartbeat"), "source_at": self._source_at(item), }, "source_at": self._source_at(item), }) async def wechat_summary(self, device_id: str, *, item: Optional[dict] = None) -> dict: item = item or await self._lookup(device_id) or {} quick = item.get("quick_status") or item.get("last_status") or {} logged_in = item.get("wechat_logged_in") if logged_in is None: logged_in = quick.get("logged_in", quick.get("wechat_logged_in")) wechat_id = item.get("wechat_id") or item.get("wxid") or quick.get("wechat_id") or quick.get("wxid") friend_count = item.get("friend_count") if friend_count is None: friend_count = quick.get("friend_count") version = item.get("wechat_version") or quick.get("wechat_version") source_at = item.get("wechat_source_at") or quick.get("source_at") or self._source_at(item) return sanitize_sensitive_fields({ "device_id": device_id, "logged_in": logged_in if isinstance(logged_in, bool) else None, "wechat_id": wechat_id, "friend_count": friend_count if isinstance(friend_count, int) else None, "wechat_version": version, "source_at": source_at, "state": "ready" if logged_in is True and wechat_id else ("not_bound" if logged_in is False else "not_collected"), }) async def delete_device(self, *, device_id: str, actor: str, key: str, trace_id: str, confirm: bool = False) -> dict: payload = {"device_id": device_id, "confirm": bool(confirm)} item = await self._lookup(device_id) # 在线设备必须先拦截，不能因为控制面存储暂不可用而落入删除分支。 if ws_hub.is_online(device_id): result = { "deleted": False, "device_id": device_id, "trace_id": trace_id, "protection": "online_device_delete_blocked", "phone_command_increment": 0, } try: idem = await reserve_idempotency(self.store, scope="device-delete", actor=actor, key=key, payload=payload) if idem.replay: return idem.result or {} audit = await record_audit(self.store, actor=actor, action="device.delete", resource={"device_id": device_id}, request_payload=payload, result_code="device_online", trace_id=trace_id) result["audit_id"] = audit["audit_id"] await save_idempotency_result(self.store, scope="device-delete", actor=actor, key=key, result=result) except ControlPlaneStorageUnavailable: result["audit_id"] = None result["audit_persisted"] = False result["audit_error"] = "control_plane_storage_unavailable" raise DeviceOnlineError(result) if item is not None and not confirm: result = { "deleted": False, "device_id": device_id, "trace_id": trace_id, "precheck": True, "confirmation_required": True, "protection": "offline_device_delete_confirmation_required", "phone_command_increment": 0, } audit = await record_audit( self.store, actor=actor, action="device.delete.precheck", resource={"device_id": device_id}, request_payload=payload, result_code="confirmation_required", trace_id=trace_id, risk_level="high", ) result["audit_id"] = audit["audit_id"] raise DeviceDeleteConfirmationRequired(result) idem = await reserve_idempotency(self.store, scope="device-delete", actor=actor, key=key, payload=payload) if idem.replay: return idem.result or {} if item is None: result = {"deleted": False, "device_id": device_id, "trace_id": trace_id} await save_idempotency_result(self.store, scope="device-delete", actor=actor, key=key, result=result) return result deletion = await device_manager.delete_device(device_id) if not deletion.get("deleted"): raise ControlPlaneStorageUnavailable("设备目录删除未产生持久化变更") audit = await record_audit(self.store, actor=actor, action="device.delete", resource={"device_id": device_id}, request_payload=payload, result_code="succeeded", trace_id=trace_id) result = { "deleted": True, "device_id": device_id, "audit_id": audit["audit_id"], "trace_id": trace_id, "confirmed": True, "precheck": False, } await save_idempotency_result(self.store, scope="device-delete", actor=actor, key=key, result=result) return result async def command(self, *, device_id: str, action: str, parameters: dict, timeout_ms: int, confirm: bool, actor: str, key: str, trace_id: str) -> dict: payload = {"device_id": device_id, "action": action, "parameters": parameters, "timeout_ms": timeout_ms, "confirm": confirm} idem = await reserve_idempotency(self.store, scope="device-command", actor=actor, key=key, payload=payload) if idem.replay: return idem.result or {} if not ws_hub.is_online(device_id): raw = {"operation": action, "device_id": device_id, "channel": "websocket/offline", "trace_id": trace_id} audit = await record_audit(self.store, actor=actor, action="device.command", resource={"device_id": device_id}, request_payload=payload, result_code="device_offline", trace_id=trace_id, channel_used="websocket/offline", raw_receipt=raw, risk_level="high" if not confirm else "medium") result = {"command_id": None, "status": "device_offline", "device_id": device_id, "raw_receipt": raw, "readback": None, "audit_id": audit["audit_id"], "trace_id": trace_id} await self._log_command_history(device_id=device_id, action=action, payload=payload, result=result) await save_idempotency_result(self.store, scope="device-command", actor=actor, key=key, result=result) raise DeviceOfflineError(result) raw = await device_transport.execute_via_ws(device_id, "system", action, {**parameters, "trace_id": trace_id}, timeout=max(1, int(timeout_ms / 1000))) raw = dict(raw or {}) data = raw.get("data") if isinstance(raw.get("data"), dict) else {} readback = data.get("readback") or data.get("db_readback") or raw.get("readback") command_id = raw.get("command_id") or data.get("command_id") rpc_success = raw.get("code") == 200 and data.get("success") is True verified = rpc_success and bool(command_id) and isinstance(readback, dict) and bool(readback.get("verified", True)) result = { "command_id": command_id, "device_id": device_id, "status": "succeeded" if verified else ("rpc_succeeded" if rpc_success else "failed"), "raw_receipt": raw, "readback": readback, "trace_id": trace_id, } await self._log_command_history(device_id=device_id, action=action, payload=payload, result=result) audit = await record_audit(self.store, actor=actor, action="device.command", resource={"device_id": device_id, "command_id": command_id}, request_payload=payload, result_code="succeeded" if verified else "business_verification_failed", trace_id=trace_id, readback=readback, channel_used="websocket/agent", raw_receipt=raw, risk_level="high" if not confirm else "medium") result["audit_id"] = audit["audit_id"] await save_idempotency_result(self.store, scope="device-command", actor=actor, key=key, result=result) if not verified: raise DeviceCommandVerificationError(result) return result async def _groups_for(self, device_id: str) -> list[dict]: try: groups = await self.store.list("device_groups") except ControlPlaneStorageUnavailable: return [] return [{"group_id": item.get("group_id"), "name": item.get("name"), "version": item.get("version", 1)} for item in groups if device_id in (item.get("device_ids") or [])] @staticmethod def _wechat_value(item: dict, key: str) -> Any: quick = item.get("quick_status") or item.get("last_status") or {} return item.get(key) or quick.get(key) @staticmethod def _source_at(item: dict) -> Optional[str]: return item.get("source_at") or item.get("updated_at") or item.get("last_heartbeat") device_console_service = DeviceConsoleService()