mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-25 00:23:53 +08:00
增加通过认证码绑定设备和管理账户的功能,需要开启私有设备配置
增加登录用户鉴权
This commit is contained in:
@@ -0,0 +1,38 @@
|
||||
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}")
|
||||
Reference in New Issue
Block a user