diff --git a/app/constants.py b/app/constants.py new file mode 100644 index 0000000..617d08f --- /dev/null +++ b/app/constants.py @@ -0,0 +1,186 @@ +""" +全局术语注册中心（Single Source of Truth） + +所有业务术语、状态映射、类型映射集中在此文件定义。 +其他模块一律从这里 import，禁止各自重复定义。 + +修改告警类型、工单状态、保洁类型等业务术语时，只需改这一个文件。 +""" + +from enum import Enum +from typing import Dict + + +# ============================================================ +# 摄像头 ID 统一说明 +# ============================================================ +# 全系统的摄像头唯一标识统一叫 camera_code，格式为 cam_{hash} +# 不同系统的字段名映射关系： +# - security-ai-edge: camera_id → 实际是 camera_code +# - wvp-platform: camera_code → 数据库字段就叫这个 +# - IoT 工单表: camera_id → 实际存的是 camera_code +# - vsp-service: device_id → AlarmEvent 表，历史原因叫 device_id +# +# Python 代码中对外接口（API 参数、工具参数）统一使用 device_id， +# 因为 AlarmEvent 主表和 EdgeAlarmReport schema 已用此名。 +# 内部传递和显示时，用 camera_name_service 解析为人类可读名称。 + + +# ============================================================ +# 告警类型 (alarm_type) +# ============================================================ +class AlarmType(str, Enum): + LEAVE_POST = "leave_post" + INTRUSION = "intrusion" + ILLEGAL_PARKING = "illegal_parking" + VEHICLE_CONGESTION = "vehicle_congestion" + + +ALARM_TYPE_NAMES: Dict[str, str] = { + AlarmType.LEAVE_POST: "人员离岗", + AlarmType.INTRUSION: "周界入侵", + AlarmType.ILLEGAL_PARKING: "车辆违停", + AlarmType.VEHICLE_CONGESTION: "车辆拥堵", +} + +# VLM 场景下的简短名称（用于截图分析提示词，尽量精炼） +ALARM_TYPE_SHORT_NAMES: Dict[str, str] = { + AlarmType.LEAVE_POST: "离岗", + AlarmType.INTRUSION: "入侵", + AlarmType.ILLEGAL_PARKING: "违停", + AlarmType.VEHICLE_CONGESTION: "拥堵", +} + + +def get_alarm_type_name(code: str) -> str: + """获取告警类型中文名，未知类型原样返回""" + return ALARM_TYPE_NAMES.get(code, code) + + +# ============================================================ +# 告警状态 (alarm_status) — vsp-service 内部使用 +# ============================================================ +class AlarmStatus(str, Enum): + NEW = "NEW" # 新告警 + CONFIRMED = "CONFIRMED" # 已确认（已创建工单） + FALSE = "FALSE" # 误报 + CLOSED = "CLOSED" # 已关闭 + + +ALARM_STATUS_NAMES: Dict[str, str] = { + AlarmStatus.NEW: "待处理", + AlarmStatus.CONFIRMED: "处理中", + AlarmStatus.FALSE: "误报", + AlarmStatus.CLOSED: "已关闭", +} + +# 芋道前端兼容映射（前端期望的状态值） +ALARM_STATUS_TO_YUDAO: Dict[str, str] = { + AlarmStatus.NEW: "pending", + AlarmStatus.CONFIRMED: "processing", + AlarmStatus.FALSE: "false_alarm", + AlarmStatus.CLOSED: "completed", +} + +YUDAO_TO_ALARM_STATUS: Dict[str, str] = { + v: k for k, v in ALARM_STATUS_TO_YUDAO.items() +} + + +# ============================================================ +# 处理状态 (handle_status) — vsp-service 内部使用 +# ============================================================ +class HandleStatus(str, Enum): + UNHANDLED = "UNHANDLED" # 未处理 + HANDLING = "HANDLING" # 处理中 + DONE = "DONE" # 已完成 + IGNORED = "IGNORED" # 已忽略 + + +HANDLE_STATUS_NAMES: Dict[str, str] = { + HandleStatus.UNHANDLED: "未处理", + HandleStatus.HANDLING: "处理中", + HandleStatus.DONE: "已完成", + HandleStatus.IGNORED: "已忽略", +} + + +# ============================================================ +# IoT 工单状态 (order status) — 来自芋道 ops_order 表 +# ============================================================ +class OrderStatus(str, Enum): + PENDING = "PENDING" # 待处理 + ASSIGNED = "ASSIGNED" # 已派单 + ARRIVED = "ARRIVED" # 已到岗 + PAUSED = "PAUSED" # 已暂停 + COMPLETED = "COMPLETED" # 已完成 + CANCELLED = "CANCELLED" # 已取消 + + +ORDER_STATUS_NAMES: Dict[str, str] = { + OrderStatus.PENDING: "待处理", + OrderStatus.ASSIGNED: "已派单", + OrderStatus.ARRIVED: "已到岗", + OrderStatus.PAUSED: "已暂停", + OrderStatus.COMPLETED: "已完成", + OrderStatus.CANCELLED: "已取消", +} + +# 未完成状态集合（用于查询"待处理"工单） +ORDER_OPEN_STATUSES = frozenset({ + OrderStatus.PENDING, OrderStatus.ASSIGNED, + OrderStatus.ARRIVED, OrderStatus.PAUSED, +}) + + +# ============================================================ +# 告警等级 (alarm_level) +# ============================================================ +ALARM_LEVEL_NAMES: Dict[int, str] = { + 0: "紧急", + 1: "重要", + 2: "普通", + 3: "轻微", +} + +# 各算法的默认告警等级 +ALARM_TYPE_DEFAULT_LEVEL: Dict[str, int] = { + AlarmType.INTRUSION: 1, # 重要 + AlarmType.LEAVE_POST: 2, # 普通 + AlarmType.ILLEGAL_PARKING: 2, # 普通 + AlarmType.VEHICLE_CONGESTION: 2, # 普通 +} + + +# ============================================================ +# 工单优先级 (priority) +# ============================================================ +PRIORITY_NAMES: Dict[int, str] = { + 0: "低", + 1: "中", + 2: "高", +} + + +# ============================================================ +# 保洁类型 (cleaning_type) +# ============================================================ +class CleaningType(str, Enum): + ROUTINE = "ROUTINE" + DEEP = "DEEP" + SPOT = "SPOT" + EMERGENCY = "EMERGENCY" + + +CLEANING_TYPE_NAMES: Dict[str, str] = { + CleaningType.ROUTINE: "日常保洁", + CleaningType.DEEP: "深度保洁", + CleaningType.SPOT: "点状保洁", + CleaningType.EMERGENCY: "应急保洁", +} + + +# ============================================================ +# 通用常量 +# ============================================================ +WEEKDAY_NAMES = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"] diff --git a/app/services/agent/tools/order_query.py b/app/services/agent/tools/order_query.py index ccc518d..7b986b1 100644 --- a/app/services/agent/tools/order_query.py +++ b/app/services/agent/tools/order_query.py @@ -11,28 +11,10 @@ from langchain_core.runnables import RunnableConfig from app.utils.logger import logger from app.utils.timezone import beijing_now - - -# 告警类型中文映射 -ALARM_TYPE_NAMES = { - "leave_post": "人员离岗", "intrusion": "周界入侵", - "illegal_parking": "车辆违停", "vehicle_congestion": "车辆拥堵", -} - -# 工单状态映射 -ORDER_STATUS_NAMES = { - "PENDING": "待处理", "ASSIGNED": "已派单", "ARRIVED": "已到岗", - "PAUSED": "已暂停", "COMPLETED": "已完成", "CANCELLED": "已取消", -} - -# 工单优先级映射 -PRIORITY_NAMES = {0: "低", 1: "中", 2: "高"} - -# 保洁类型映射 -CLEANING_TYPE_NAMES = { - "ROUTINE": "日常保洁", "DEEP": "深度保洁", - "SPOT": "点状保洁", "EMERGENCY": "应急保洁", -} +from app.constants import ( + ALARM_TYPE_NAMES, ORDER_STATUS_NAMES, PRIORITY_NAMES, + CLEANING_TYPE_NAMES, ORDER_OPEN_STATUSES, +) def _parse_time_range(time_range: str): @@ -237,8 +219,7 @@ def list_orders( user_id = config.get("configurable", {}).get("user_id", "") # 查待处理工单时不限时间范围（待处理可能是之前创建的） - pending_statuses = {"PENDING", "ASSIGNED", "ARRIVED", "PAUSED"} - skip_time = status and status in pending_statuses + skip_time = status and status in ORDER_OPEN_STATUSES orders, total, sec_ext_map, clean_ext_map = _query_orders( order_type=order_type if order_type != "ALL" else None, diff --git a/app/services/daily_report_service.py b/app/services/daily_report_service.py index db4786e..7982203 100644 --- a/app/services/daily_report_service.py +++ b/app/services/daily_report_service.py @@ -12,22 +12,7 @@ from typing import Dict, List, Optional from app.config import settings from app.utils.logger import logger - -ALARM_TYPE_NAMES = { - "leave_post": "人员离岗", - "intrusion": "周界入侵", - "illegal_parking": "车辆违停", - "vehicle_congestion": "车辆拥堵", -} - -CLEANING_TYPE_NAMES = { - "ROUTINE": "日常保洁", - "DEEP": "深度保洁", - "SPOT": "点状保洁", - "EMERGENCY": "应急保洁", -} - -WEEKDAY_NAMES = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"] +from app.constants import ALARM_TYPE_NAMES, CLEANING_TYPE_NAMES, WEEKDAY_NAMES def _format_duration(minutes: float) -> str: