重构：建立术语注册中心 constants.py（阶段一）
新建 app/constants.py 作为全局术语单一真相源，包含： - AlarmType/AlarmStatus/HandleStatus/OrderStatus/CleaningType 枚举 - 所有中文映射字典（ALARM_TYPE_NAMES 等） - 芋道前端兼容状态映射 - 告警等级、优先级、保洁类型等常量 - ORDER_OPEN_STATUSES 未完成状态集合 已替换 daily_report_service.py 和 order_query.py 中的重复定义。 其余文件（wechat_service/vlm_service/yudao_*等）待下一阶段替换。
This commit is contained in:
186
app/constants.py
Normal file
186
app/constants.py
Normal file
@@ -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 = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
|
||||
@@ -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,
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user