import asyncio from dataclasses import dataclass from typing import Dict, Optional VOICE_TARGET_KIND_DEVICE = "device" VOICE_TARGET_KIND_PARENT = "parent" @dataclass(frozen=True) class VoiceTarget: kind: str target_device_id: Optional[str] = None uuid: Optional[str] = None audio_cache_key: Optional[str] = None class DeviceTargetCache: def __init__(self): self.targets: Dict[str, VoiceTarget] = {} self.lock = asyncio.Lock() async def get_target(self, device_id: str) -> Optional[str]: async with self.lock: target = self.targets.get(device_id) if not target or target.kind != VOICE_TARGET_KIND_DEVICE: return None return target.target_device_id async def get_voice_target(self, device_id: str) -> Optional[VoiceTarget]: async with self.lock: return self.targets.get(device_id) async def set_target(self, device_id: str, target_device_id: str): async with self.lock: self.targets[device_id] = VoiceTarget( kind=VOICE_TARGET_KIND_DEVICE, target_device_id=target_device_id, audio_cache_key=target_device_id, ) async def set_parent_target(self, device_id: str, uuid: str): async with self.lock: self.targets[device_id] = VoiceTarget( kind=VOICE_TARGET_KIND_PARENT, uuid=uuid, audio_cache_key=f"parent:{device_id}", ) async def remove_target(self, device_id: str): async with self.lock: if device_id in self.targets: del self.targets[device_id] async def get_all_targets(self) -> list: async with self.lock: return list(self.targets.items()) # 单例实例 device_target_cache = DeviceTargetCache()