重构(report_generator): TYPE/LEVEL/STATUS_NAMES 改用 constants 统一定义
This commit is contained in:
@@ -11,19 +11,12 @@ from typing import Optional, Tuple
|
||||
from openpyxl import Workbook
|
||||
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
|
||||
|
||||
from app.constants import ALARM_TYPE_NAMES, ALARM_LEVEL_NAMES, ALARM_STATUS_NAMES
|
||||
from app.models import AlarmEvent, get_session
|
||||
from app.utils.logger import logger
|
||||
from app.utils.timezone import beijing_now
|
||||
|
||||
|
||||
TYPE_NAMES = {"leave_post": "人员离岗", "intrusion": "周界入侵"}
|
||||
LEVEL_NAMES = {0: "紧急", 1: "重要", 2: "普通", 3: "轻微"}
|
||||
STATUS_NAMES = {
|
||||
"NEW": "待处理", "CONFIRMED": "已确认",
|
||||
"FALSE": "误报", "CLOSED": "已关闭",
|
||||
}
|
||||
|
||||
|
||||
def generate_alarm_report(
|
||||
time_range: str = "week",
|
||||
) -> Optional[Tuple[str, io.BytesIO]]:
|
||||
@@ -91,11 +84,11 @@ def generate_alarm_report(
|
||||
for row, a in enumerate(alarms, 2):
|
||||
values = [
|
||||
a.alarm_id,
|
||||
TYPE_NAMES.get(a.alarm_type, a.alarm_type),
|
||||
ALARM_TYPE_NAMES.get(a.alarm_type, a.alarm_type),
|
||||
a.device_id,
|
||||
a.scene_id or "",
|
||||
LEVEL_NAMES.get(a.alarm_level, str(a.alarm_level or "")),
|
||||
STATUS_NAMES.get(a.alarm_status, a.alarm_status or ""),
|
||||
ALARM_LEVEL_NAMES.get(a.alarm_level, str(a.alarm_level or "")),
|
||||
ALARM_STATUS_NAMES.get(a.alarm_status, a.alarm_status or ""),
|
||||
a.handle_status or "",
|
||||
f"{a.confidence_score:.2f}" if a.confidence_score else "",
|
||||
a.event_time.strftime("%Y-%m-%d %H:%M:%S") if a.event_time else "",
|
||||
@@ -129,7 +122,7 @@ def generate_alarm_report(
|
||||
ws2.cell(row=2, column=1, value="类型")
|
||||
ws2.cell(row=2, column=2, value="数量")
|
||||
for i, (t, c) in enumerate(type_count.items(), 3):
|
||||
ws2.cell(row=i, column=1, value=TYPE_NAMES.get(t, t))
|
||||
ws2.cell(row=i, column=1, value=ALARM_TYPE_NAMES.get(t, t))
|
||||
ws2.cell(row=i, column=2, value=c)
|
||||
|
||||
# 状态统计
|
||||
@@ -138,7 +131,7 @@ def generate_alarm_report(
|
||||
ws2.cell(row=offset + 1, column=1, value="状态")
|
||||
ws2.cell(row=offset + 1, column=2, value="数量")
|
||||
for i, (s, c) in enumerate(status_count.items(), offset + 2):
|
||||
ws2.cell(row=i, column=1, value=STATUS_NAMES.get(s, s))
|
||||
ws2.cell(row=i, column=1, value=ALARM_STATUS_NAMES.get(s, s))
|
||||
ws2.cell(row=i, column=2, value=c)
|
||||
|
||||
# 级别统计
|
||||
@@ -147,7 +140,7 @@ def generate_alarm_report(
|
||||
ws2.cell(row=offset2 + 1, column=1, value="级别")
|
||||
ws2.cell(row=offset2 + 1, column=2, value="数量")
|
||||
for i, (lv, c) in enumerate(level_count.items(), offset2 + 2):
|
||||
ws2.cell(row=i, column=1, value=LEVEL_NAMES.get(lv, str(lv)))
|
||||
ws2.cell(row=i, column=1, value=ALARM_LEVEL_NAMES.get(lv, str(lv)))
|
||||
ws2.cell(row=i, column=2, value=c)
|
||||
|
||||
# 输出到内存
|
||||
|
||||
Reference in New Issue
Block a user