85 lines
3.0 KiB
Python
85 lines
3.0 KiB
Python
|
"""术间配置：camera_ids 集合与语音桌面终端 ID 绑定。"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from dataclasses import dataclass
|
|||
|
|
from typing import Any
|
|||
|
|
|
|||
|
|
from loguru import logger
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass(frozen=True)
|
|||
|
|
class OrRoomBinding:
|
|||
|
|
or_room_id: str
|
|||
|
|
camera_ids: frozenset[str]
|
|||
|
|
voice_terminal_id: str
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class VoiceTerminalBindingIndex:
|
|||
|
|
"""由 ``or_site_config.voice_or_room_bindings`` 数组构建。"""
|
|||
|
|
|
|||
|
|
rooms: tuple[OrRoomBinding, ...]
|
|||
|
|
|
|||
|
|
def resolve_terminal(self, camera_ids: list[str]) -> str | None:
|
|||
|
|
"""精确匹配 camera 集合；否则开录路数为术间子集时匹配最小超集术间。"""
|
|||
|
|
key = frozenset(str(x).strip() for x in camera_ids if str(x).strip())
|
|||
|
|
if not key:
|
|||
|
|
return None
|
|||
|
|
for r in self.rooms:
|
|||
|
|
if r.camera_ids == key:
|
|||
|
|
return r.voice_terminal_id
|
|||
|
|
candidates = [r for r in self.rooms if key <= r.camera_ids]
|
|||
|
|
if not candidates:
|
|||
|
|
return None
|
|||
|
|
if len(candidates) == 1:
|
|||
|
|
return candidates[0].voice_terminal_id
|
|||
|
|
candidates.sort(key=lambda r: (len(r.camera_ids), r.or_room_id, r.voice_terminal_id))
|
|||
|
|
return candidates[0].voice_terminal_id
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def from_binding_list(data: list[Any]) -> VoiceTerminalBindingIndex | None:
|
|||
|
|
rows: list[OrRoomBinding] = []
|
|||
|
|
seen_terminals: set[str] = set()
|
|||
|
|
seen_camera_sets: set[frozenset[str]] = set()
|
|||
|
|
for i, item in enumerate(data):
|
|||
|
|
if not isinstance(item, dict):
|
|||
|
|
logger.warning("voice_or_room_bindings[{}] must be an object", i)
|
|||
|
|
return None
|
|||
|
|
rid = str(item.get("or_room_id") or "").strip()
|
|||
|
|
tid = str(item.get("voice_terminal_id") or "").strip()
|
|||
|
|
cams = item.get("camera_ids")
|
|||
|
|
if not rid or not tid or not isinstance(cams, list):
|
|||
|
|
logger.warning(
|
|||
|
|
"voice_or_room_bindings[{}] missing or invalid fields", i
|
|||
|
|
)
|
|||
|
|
return None
|
|||
|
|
cam_set = frozenset(str(x).strip() for x in cams if str(x).strip())
|
|||
|
|
if not cam_set:
|
|||
|
|
logger.warning(
|
|||
|
|
"voice_or_room_bindings[{}] camera_ids must be non-empty", i
|
|||
|
|
)
|
|||
|
|
return None
|
|||
|
|
if tid in seen_terminals:
|
|||
|
|
logger.warning(
|
|||
|
|
"voice_or_room_bindings: duplicate voice_terminal_id {!r}",
|
|||
|
|
tid,
|
|||
|
|
)
|
|||
|
|
return None
|
|||
|
|
if cam_set in seen_camera_sets:
|
|||
|
|
logger.warning(
|
|||
|
|
"voice_or_room_bindings: duplicate camera_ids set for room {!r}",
|
|||
|
|
rid,
|
|||
|
|
)
|
|||
|
|
return None
|
|||
|
|
seen_terminals.add(tid)
|
|||
|
|
seen_camera_sets.add(cam_set)
|
|||
|
|
rows.append(
|
|||
|
|
OrRoomBinding(
|
|||
|
|
or_room_id=rid,
|
|||
|
|
camera_ids=cam_set,
|
|||
|
|
voice_terminal_id=tid,
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
return VoiceTerminalBindingIndex(rooms=tuple(rows))
|