mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-22 15:13:55 +08:00
39 lines
1.1 KiB
Python
39 lines
1.1 KiB
Python
import asyncio
|
|
from typing import Dict
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class FileLockManager:
|
|
_instance = None
|
|
_locks: Dict[str, asyncio.Lock] = {}
|
|
|
|
def __new__(cls):
|
|
if cls._instance is None:
|
|
cls._instance = super(FileLockManager, cls).__new__(cls)
|
|
return cls._instance
|
|
|
|
@classmethod
|
|
def get_lock(cls, file_path: str) -> asyncio.Lock:
|
|
"""获取指定文件的锁"""
|
|
if file_path not in cls._locks:
|
|
cls._locks[file_path] = asyncio.Lock()
|
|
return cls._locks[file_path]
|
|
|
|
@classmethod
|
|
async def acquire_lock(cls, file_path: str):
|
|
"""获取锁"""
|
|
lock = cls.get_lock(file_path)
|
|
await lock.acquire()
|
|
logger.debug(f"Acquired lock for {file_path}")
|
|
|
|
@classmethod
|
|
def release_lock(cls, file_path: str):
|
|
"""释放锁"""
|
|
if file_path in cls._locks:
|
|
try:
|
|
cls._locks[file_path].release()
|
|
logger.debug(f"Released lock for {file_path}")
|
|
except RuntimeError as e:
|
|
logger.warning(f"Failed to release lock for {file_path}: {e}")
|