Yuriy Yuriev
110 lines
4.1 KiB
Python
110 lines
4.1 KiB
Python
"""
|
|
Хранилище зарегистрированных пользователей.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AuthStorage:
|
|
"""Реестр пользователей бота в JSON-файле.
|
|
|
|
Паролей не хранит: личность подтверждает Telegram, идентификатором
|
|
служит user_id.
|
|
"""
|
|
|
|
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, username: str = None, full_name: str = None
|
|
) -> bool:
|
|
"""Регистрирует пользователя и освежает его профиль.
|
|
|
|
Профиль перезаписывается и для уже известных пользователей — ник в
|
|
Telegram может смениться, а взять его неоткуда, кроме входящего
|
|
апдейта.
|
|
"""
|
|
async with self._lock:
|
|
data = await asyncio.to_thread(self._load_sync)
|
|
key = str(user_id)
|
|
is_new = key not in data
|
|
record = data.get(key, {})
|
|
if is_new:
|
|
record["registered_at"] = datetime.now().isoformat()
|
|
if username is not None:
|
|
record["username"] = username
|
|
if full_name is not None:
|
|
record["full_name"] = full_name
|
|
data[key] = record
|
|
await asyncio.to_thread(self._save_sync, data)
|
|
if is_new:
|
|
logger.info(f"User {user_id} registered ({username or full_name or 'no name'})")
|
|
return is_new
|
|
|
|
async def get_profile(self, user_id: int) -> dict:
|
|
async with self._lock:
|
|
data = await asyncio.to_thread(self._load_sync)
|
|
return data.get(str(user_id), {})
|
|
|
|
async def get_all_profiles(self) -> dict:
|
|
"""Возвращает {user_id: профиль} для всех пользователей."""
|
|
async with self._lock:
|
|
data = await asyncio.to_thread(self._load_sync)
|
|
return {int(k): v for k, v in data.items()}
|
|
|
|
@staticmethod
|
|
def display_name(profile: dict) -> str:
|
|
"""Человекочитаемое имя: @ник, иначе имя, иначе пусто."""
|
|
if not profile:
|
|
return ""
|
|
username = profile.get("username")
|
|
if username:
|
|
return f"@{username}"
|
|
return profile.get("full_name") or ""
|
|
|
|
async def delete_user(self, user_id: int) -> bool:
|
|
async with self._lock:
|
|
data = await asyncio.to_thread(self._load_sync)
|
|
user_str = str(user_id)
|
|
if user_str not in data:
|
|
return False
|
|
del data[user_str]
|
|
await asyncio.to_thread(self._save_sync, data)
|
|
logger.info(f"User {user_id} deleted")
|
|
return True
|
|
|
|
async def user_exists(self, user_id: int) -> bool:
|
|
async with self._lock:
|
|
data = await asyncio.to_thread(self._load_sync)
|
|
return str(user_id) in data
|
|
|
|
async def get_all_user_ids(self) -> list[int]:
|
|
async with self._lock:
|
|
data = await asyncio.to_thread(self._load_sync)
|
|
return [int(k) for k in data.keys()]
|
|
|
|
def _load_sync(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_sync(self, data: dict) -> None:
|
|
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}")
|