fix(alarm): 支持 app/stream 格式直接提取中文名称

问题:
- 警告日志:使用遗留格式 app/stream: 大堂吧台3/012
- app/stream 格式无法显示中文名称
- 旧逻辑返回 None 导致显示原始ID

根本原因:
对于 "大堂吧台3/012" 格式,app 部分本身就是中文名称,
但旧逻辑直接返回 None 不处理,完全没必要。

修复方案:
1. 新增 _parse_app_stream_format 方法
   - 直接解析 app/stream 格式
   - 构造虚拟 camera_info 对象
   - 无需查询 WVP API

2. 修改 get_camera_info 方法
   - camera_code 格式:查询 WVP
   - app/stream 格式:直接解析
   - 统一返回 camera_info

3. 修改 format_display_name 方法
   - app/stream 格式没有 camera_code
   - 直接返回 name,不使用模板
   - 避免字段缺失警告

4. 修改 get_camera_infos_batch 方法
   - 分类处理两种格式
   - camera_code:并发查询 WVP
   - app/stream:直接解析(无IO)

逻辑对比:
旧逻辑:
  cam_1f0e3dad9990 → 查询WVP → 大堂吧台3 ✓
  大堂吧台3/012 → 返回None → 大堂吧台3/012 ✗

新逻辑:
  cam_1f0e3dad9990 → 查询WVP → 大堂吧台3 ✓
  大堂吧台3/012 → 直接解析 → 大堂吧台3 ✓

测试结果:
✓ cam_1f0e3dad9990 → 大堂吧台3
✓ 大堂吧台3/012 → 大堂吧台3
✓ 一楼大堂吧台/008 → 一楼大堂吧台
✓ 无警告日志

性能提升:
- app/stream 格式无需 HTTP 查询
- 批量查询时性能更优
This commit is contained in:
2026-02-24 14:26:44 +08:00
parent 0c01e4c40d
commit 4bd369813e

View File

@@ -49,13 +49,35 @@ class CameraNameService:
if device_id.startswith("cam_"):
return await self._query_by_camera_code(device_id)
# app/stream 格式(遗留格式,需要认证
# app/stream 格式(遗留格式,直接解析
elif "/" in device_id:
logger.warning(f"使用遗留格式 app/stream: {device_id},建议使用 camera_code 格式")
return None # app/stream 格式需要token暂不支持无认证查询
return self._parse_app_stream_format(device_id)
return None
def _parse_app_stream_format(self, device_id: str) -> Optional[Dict]:
"""
解析 app/stream 格式的 device_id
Args:
device_id: 格式如 "大堂吧台3/012"
Returns:
虚拟的摄像头信息字典
"""
parts = device_id.split("/", 1)
if len(parts) != 2:
return None
app, stream = parts
# 构造虚拟的摄像头信息(兼容统一格式化逻辑)
return {
"app": app, # 中文名称
"stream": stream, # 流ID
"gbName": app, # 兼容字段
"cameraCode": None, # app/stream 格式没有 camera_code
}
async def _query_by_camera_code(self, camera_code: str) -> Optional[Dict]:
"""
通过 camera_code 查询摄像头信息
@@ -100,21 +122,25 @@ class CameraNameService:
# 去重
unique_ids = list(set(device_ids))
# 只查询 camera_code 格式
valid_ids = [did for did in unique_ids if did.startswith("cam_")]
# 分类:camera_code 和 app/stream 格式
camera_code_ids = [did for did in unique_ids if did.startswith("cam_")]
app_stream_ids = [did for did in unique_ids if "/" in did]
if not valid_ids:
return {did: None for did in device_ids}
# 初始化结果映射
info_map = {}
# 并发查询
import asyncio
tasks = [self.get_camera_info(did) for did in valid_ids]
results = await asyncio.gather(*tasks)
# 并发查询 camera_code 格式
if camera_code_ids:
import asyncio
tasks = [self.get_camera_info(did) for did in camera_code_ids]
results = await asyncio.gather(*tasks)
info_map.update(dict(zip(camera_code_ids, results)))
# 构建映射
info_map = dict(zip(valid_ids, results))
# 直接解析 app/stream 格式(无需查询)
for did in app_stream_ids:
info_map[did] = self._parse_app_stream_format(did)
# 补充非 camera_code 格式的
# 补充其他格式(未识别的)
for did in unique_ids:
if did not in info_map:
info_map[did] = None
@@ -168,6 +194,11 @@ class CameraNameService:
配置模板:"{name}"
返回: "大堂"
app/stream格式
device_id: "大堂吧台3/012"
camera_info: {app: "大堂吧台3", stream: "012", cameraCode: None}
返回: "大堂吧台3"(忽略模板,直接返回名称)
"""
# 如果没有摄像头信息直接返回device_id
if not camera_info:
@@ -178,24 +209,27 @@ class CameraNameService:
stream = camera_info.get("stream")
name = self.extract_name(camera_info)
# 检查模板需要的变量
template_vars = {
"{camera_code}": camera_code,
"{name}": name,
"{stream}": stream
}
# 如果没有提取到名称返回device_id
if not name:
logger.warning(f"无法提取摄像头名称: device_id={device_id}")
return device_id
# 如果模板只需要 {name},即使其他字段缺失也能返回
if self.config.display_format == "{name}" and name:
# 对于 app/stream 格式(没有 camera_code直接返回名称
if not camera_code:
return name
# 如果必需字段缺失返回device_id
if not all([camera_code, name, stream]):
# 对于 camera_code 格式,检查模板需要的变量
# 如果模板只需要 {name},直接返回名称
if self.config.display_format == "{name}":
return name
# 完整格式需要所有字段
if not stream:
logger.warning(
f"摄像头信息不完整: camera_code={camera_code}, "
f"name={name}, stream={stream}, 使用fallback"
)
return device_id
return name # 至少返回名称
# 按模板格式化
try:
@@ -206,7 +240,7 @@ class CameraNameService:
)
except KeyError as e:
logger.error(f"格式化模板变量错误: {e}, 模板={self.config.display_format}")
return device_id
return name # 出错时至少返回名称
async def get_display_name(self, device_id: str) -> str:
"""