"""术间配置: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))