diff --git a/server.py b/server.py index 6cfa1a3..8e71c9b 100644 --- a/server.py +++ b/server.py @@ -37,6 +37,7 @@ class TTSService: self._shutdown_event: asyncio.Event | None = None self._worker_task: asyncio.Task | None = None self._cookies: dict[str, str] | None = None + self._lifecycle_lock = asyncio.Lock() self.running = False self.start_time: float = 0 self.messages_processed = 0 @@ -48,49 +49,51 @@ class TTSService: self._cookies = cookies async def start(self, config: dict) -> None: - if self.running: - raise RuntimeError("Already running") - room_id = config["room_id"] - voice = config.get("voice", "f1") - speed = config.get("speed", 100) - self._volume = config.get("volume", 100) - message_format = config.get("format", "{uname}\u8bf4\u3001 {msg}") - convert_numbers = config.get("convert_numbers", True) - self.current_config = config - self.recent_messages = [] - self.messages_processed = 0 - self.start_time = time.time() - self._queue = asyncio.Queue(maxsize=256) - self._shutdown_event = asyncio.Event() - self._bridge = TTSBridge(voice=voice, speed=speed) - await self._bridge.start() - self._danmaku_client = DanmakuClient( - room_id=room_id, queue=self._queue, - cookies=self._cookies, message_format=message_format, - ) - await self._danmaku_client.start() - self.running = True - self._worker_task = asyncio.create_task(self._tts_worker(convert_numbers)) + async with self._lifecycle_lock: + if self.running: + raise RuntimeError("Already running") + room_id = config["room_id"] + voice = config.get("voice", "f1") + speed = config.get("speed", 100) + self._volume = config.get("volume", 100) + message_format = config.get("format", "{uname}\u8bf4\u3001 {msg}") + convert_numbers = config.get("convert_numbers", True) + self.current_config = config + self.recent_messages = [] + self.messages_processed = 0 + self.start_time = time.time() + self._queue = asyncio.Queue(maxsize=256) + self._shutdown_event = asyncio.Event() + self._bridge = TTSBridge(voice=voice, speed=speed) + await self._bridge.start() + self._danmaku_client = DanmakuClient( + room_id=room_id, queue=self._queue, + cookies=self._cookies, message_format=message_format, + ) + await self._danmaku_client.start() + self.running = True + self._worker_task = asyncio.create_task(self._tts_worker(convert_numbers)) async def stop(self) -> None: - if not self.running: - return - self.running = False - if self._shutdown_event: - self._shutdown_event.set() - if self._worker_task: - self._worker_task.cancel() - try: - await self._worker_task - except asyncio.CancelledError: - pass - if self._danmaku_client: - await self._danmaku_client.stop() - if self._bridge: - await self._bridge.stop() - self._queue = None - self._bridge = None - self._danmaku_client = None + async with self._lifecycle_lock: + if not self.running: + return + self.running = False + if self._shutdown_event: + self._shutdown_event.set() + if self._worker_task: + self._worker_task.cancel() + try: + await self._worker_task + except asyncio.CancelledError: + pass + if self._danmaku_client: + await self._danmaku_client.stop() + if self._bridge: + await self._bridge.stop() + self._queue = None + self._bridge = None + self._danmaku_client = None async def _tts_worker(self, convert_numbers: bool) -> None: total_in = 0 diff --git a/tts.py b/tts.py index 533fab0..bfa6dc9 100644 --- a/tts.py +++ b/tts.py @@ -46,6 +46,7 @@ class TTSBridge: self._bridge_script = bridge_script or _resolve_path("tts_bridge.js") self._node_exe = _get_node_exe() self._stderr_task: asyncio.Task | None = None + self._io_lock = asyncio.Lock() async def start(self) -> None: if not pathlib.Path(self._bridge_script).exists(): @@ -96,39 +97,40 @@ class TTSBridge: if self._process is None or self._process.stdin is None: raise RuntimeError("Bridge not started. Call start() first.") - tmpdir = pathlib.Path(tempfile.mkdtemp(prefix="tts_")) - try: - input_path = tmpdir / "input.txt" - output_path = tmpdir / "output.wav" + async with self._io_lock: + tmpdir = pathlib.Path(tempfile.mkdtemp(prefix="tts_")) + try: + input_path = tmpdir / "input.txt" + output_path = tmpdir / "output.wav" - input_path.write_text(kana_text, encoding="utf-8") + input_path.write_text(kana_text, encoding="utf-8") - cmd_line = f"{input_path}|{output_path}\n" - self._process.stdin.write(cmd_line.encode("utf-8")) - await self._process.stdin.drain() + cmd_line = f"{input_path}|{output_path}\n" + self._process.stdin.write(cmd_line.encode("utf-8")) + await self._process.stdin.drain() - line = await asyncio.wait_for(self._process.stdout.readline(), timeout=120) - line_str = line.decode("utf-8").strip() + line = await asyncio.wait_for(self._process.stdout.readline(), timeout=120) + line_str = line.decode("utf-8").strip() - if line_str.startswith("ERR:"): - raise RuntimeError(f"TTS synthesis error: {line_str[4:]}") + if line_str.startswith("ERR:"): + raise RuntimeError(f"TTS synthesis error: {line_str[4:]}") - if not line_str.startswith("OK:") or line_str[3:] != str(output_path): - raise RuntimeError(f"Unexpected bridge response: {line_str}") + if not line_str.startswith("OK:") or line_str[3:] != str(output_path): + raise RuntimeError(f"Unexpected bridge response: {line_str}") - wav_data = output_path.read_bytes() - return wav_data + wav_data = output_path.read_bytes() + return wav_data - finally: - for f in tmpdir.iterdir(): + finally: + for f in tmpdir.iterdir(): + try: + f.unlink() + except OSError: + pass try: - f.unlink() + tmpdir.rmdir() except OSError: pass - try: - tmpdir.rmdir() - except OSError: - pass async def stop(self) -> None: if self._process is not None: