Yuriy Yuriev
88 lines
3.1 KiB
Python
88 lines
3.1 KiB
Python
"""
|
|
Хранилище учётных данных пользователей.
|
|
"""
|
|
|
|
import json
|
|
import asyncio
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Optional, Dict
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AuthStorage:
|
|
"""Хранилище паролей пользователей в JSON-файле."""
|
|
|
|
def __init__(self, file_path: str = "data/users.json"):
|
|
self.file_path = Path(file_path)
|
|
self.file_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._lock = asyncio.Lock()
|
|
|
|
async def register(self, user_id: int, password_hash: str, salt: str) -> bool:
|
|
"""Регистрирует нового пользователя. Возвращает True если успешно."""
|
|
async with self._lock:
|
|
data = self._load()
|
|
if str(user_id) in data:
|
|
return False
|
|
data[str(user_id)] = {
|
|
"password_hash": password_hash,
|
|
"salt": salt,
|
|
}
|
|
self._save(data)
|
|
logger.info(f"User {user_id} registered")
|
|
return True
|
|
|
|
async def get_user(self, user_id: int) -> Optional[Dict]:
|
|
"""Получает данные пользователя по ID."""
|
|
async with self._lock:
|
|
data = self._load()
|
|
return data.get(str(user_id))
|
|
|
|
async def change_password(self, user_id: int, new_hash: str, new_salt: str) -> bool:
|
|
"""Сменить пароль пользователя."""
|
|
async with self._lock:
|
|
data = self._load()
|
|
user_str = str(user_id)
|
|
if user_str not in data:
|
|
return False
|
|
data[user_str]["password_hash"] = new_hash
|
|
data[user_str]["salt"] = new_salt
|
|
self._save(data)
|
|
logger.info(f"User {user_id} changed password")
|
|
return True
|
|
|
|
async def delete_user(self, user_id: int) -> bool:
|
|
"""Удалить учётную запись пользователя."""
|
|
async with self._lock:
|
|
data = self._load()
|
|
user_str = str(user_id)
|
|
if user_str not in data:
|
|
return False
|
|
del data[user_str]
|
|
self._save(data)
|
|
logger.info(f"User {user_id} deleted")
|
|
return True
|
|
|
|
async def user_exists(self, user_id: int) -> bool:
|
|
"""Проверяет, существует ли пользователь."""
|
|
async with self._lock:
|
|
data = self._load()
|
|
return str(user_id) in data
|
|
|
|
def _load(self) -> dict:
|
|
if not self.file_path.exists():
|
|
return {}
|
|
try:
|
|
with open(self.file_path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, IOError) as e:
|
|
logger.warning(f"Failed to load auth storage: {e}")
|
|
return {}
|
|
|
|
def _save(self, data: dict):
|
|
try:
|
|
with open(self.file_path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
except IOError as e:
|
|
logger.error(f"Failed to save auth storage: {e}") |