feat: 新建工单体系三表(security_work_order/security_user/work_order_log)

- SecurityWorkOrder: 安保工单表,关联alarm_event一对一,支持告警生成和手动创建
  含派发人信息冗余(责任追溯)、告警合并索引
- SecurityUser: 安保人员表,含企微uid、角色、班组
- WorkOrderLog: 工单操作记录,追踪CREATE/DISPATCH/ACCEPT/FINISH/CLOSE

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-10 09:24:24 +08:00
parent 9143022ee8
commit 26f1512a1c

View File

@@ -368,6 +368,138 @@ class AlarmLlmAnalysis(Base):
} }
# ==================== 工单体系 ====================
class SecurityUser(Base):
"""安保人员表"""
__tablename__ = "security_user"
id = Column(Integer, primary_key=True, autoincrement=True)
user_id = Column(String(64), unique=True, nullable=False, index=True, comment="人员唯一ID")
name = Column(String(100), nullable=False, comment="姓名")
phone = Column(String(20), nullable=True, comment="手机号")
wechat_uid = Column(String(100), nullable=True, comment="企微userid")
role = Column(String(50), default="guard", comment="角色: guard/supervisor/manager")
team_id = Column(String(64), nullable=True, index=True, comment="班组ID")
status = Column(String(20), default="active", comment="状态: active/inactive")
created_at = Column(DateTime, default=lambda: beijing_now())
updated_at = Column(DateTime, default=lambda: beijing_now(),
onupdate=lambda: beijing_now())
def to_dict(self) -> dict:
fmt = '%Y-%m-%d %H:%M:%S'
return {
"id": self.id,
"user_id": self.user_id,
"name": self.name,
"phone": self.phone,
"wechat_uid": self.wechat_uid,
"role": self.role,
"team_id": self.team_id,
"status": self.status,
"created_at": self.created_at.strftime(fmt) if self.created_at else None,
"updated_at": self.updated_at.strftime(fmt) if self.updated_at else None,
}
class SecurityWorkOrder(Base):
"""安保工单表"""
__tablename__ = "security_work_order"
order_id = Column(String(64), primary_key=True, comment="工单ID: WO + 时间戳 + uuid")
# 来源告警(告警工单必填,手动工单可空)
alarm_id = Column(String(64), nullable=True, unique=True, comment="关联 alarm_event.alarm_id")
# 工单内容
title = Column(String(200), nullable=False, comment="工单标题")
description = Column(Text, nullable=True, comment="工单描述")
priority = Column(SmallInteger, default=2, comment="优先级: 1低 2中 3高 4紧急")
# 设备/区域信息(冗余,方便查询和责任追溯)
camera_id = Column(String(64), nullable=True, comment="摄像头ID")
roi_id = Column(String(64), nullable=True, comment="ROI区域ID")
alarm_type = Column(String(50), nullable=True, comment="告警类型")
image_url = Column(String(512), nullable=True, comment="截图URL")
# 派发信息(写入时确定,不依赖实时查询,确保责任可追溯)
assigned_user_id = Column(String(64), nullable=True, index=True, comment="处理人user_id")
assigned_user_name = Column(String(100), nullable=True, comment="处理人姓名")
assigned_team_id = Column(String(64), nullable=True, comment="班组ID")
# 状态: PENDING / DISPATCHED / PROCESSING / DONE / CLOSED
status = Column(String(20), default="PENDING", nullable=False, index=True, comment="工单状态")
# 处理结果
result = Column(Text, nullable=True, comment="处理结果描述")
# 创建人
created_by = Column(String(64), nullable=True, comment="创建人")
# 时间
created_at = Column(DateTime, default=lambda: beijing_now())
updated_at = Column(DateTime, default=lambda: beijing_now(),
onupdate=lambda: beijing_now())
dispatch_time = Column(DateTime, nullable=True, comment="派单时间")
finish_time = Column(DateTime, nullable=True, comment="完成时间")
__table_args__ = (
Index('idx_swo_status', 'status'),
Index('idx_swo_assigned', 'assigned_user_id'),
Index('idx_swo_created_at', 'created_at'),
Index('idx_swo_camera_roi_type', 'camera_id', 'roi_id', 'alarm_type'),
)
def to_dict(self) -> dict:
fmt = '%Y-%m-%d %H:%M:%S'
return {
"order_id": self.order_id,
"alarm_id": self.alarm_id,
"title": self.title,
"description": self.description,
"priority": self.priority,
"camera_id": self.camera_id,
"roi_id": self.roi_id,
"alarm_type": self.alarm_type,
"image_url": self.image_url,
"assigned_user_id": self.assigned_user_id,
"assigned_user_name": self.assigned_user_name,
"assigned_team_id": self.assigned_team_id,
"status": self.status,
"result": self.result,
"created_by": self.created_by,
"created_at": self.created_at.strftime(fmt) if self.created_at else None,
"updated_at": self.updated_at.strftime(fmt) if self.updated_at else None,
"dispatch_time": self.dispatch_time.strftime(fmt) if self.dispatch_time else None,
"finish_time": self.finish_time.strftime(fmt) if self.finish_time else None,
}
class WorkOrderLog(Base):
"""工单操作记录表"""
__tablename__ = "work_order_log"
id = Column(Integer, primary_key=True, autoincrement=True)
order_id = Column(String(64), nullable=False, index=True, comment="关联工单ID")
action = Column(String(50), nullable=False,
comment="操作: CREATE/DISPATCH/ACCEPT/FINISH/CLOSE")
operator_id = Column(String(64), nullable=True, comment="操作人ID")
operator_name = Column(String(100), nullable=True, comment="操作人姓名")
remark = Column(Text, nullable=True, comment="备注")
created_at = Column(DateTime, default=lambda: beijing_now())
def to_dict(self) -> dict:
return {
"id": self.id,
"order_id": self.order_id,
"action": self.action,
"operator_id": self.operator_id,
"operator_name": self.operator_name,
"remark": self.remark,
"created_at": self.created_at.strftime('%Y-%m-%d %H:%M:%S') if self.created_at else None,
}
# ==================== 数据库管理 ==================== # ==================== 数据库管理 ====================
_engine = None _engine = None