81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from config import ADMIN_ID, DATABASE_DIR
|
|
from db.storage import load_stats, save_stats
|
|
from models.user import UserProfile
|
|
|
|
|
|
def _user_dir(user_id: int) -> Path:
|
|
return DATABASE_DIR / str(user_id)
|
|
|
|
|
|
def path(user_id: int) -> Path:
|
|
return _user_dir(user_id) / "config.json"
|
|
|
|
|
|
def register(user_id: int, username: str | None = None, referrer: int | None = None) -> bool:
|
|
user_path = path(user_id)
|
|
is_new = False
|
|
|
|
if not user_path.parent.exists():
|
|
user_path.parent.mkdir(parents=True, exist_ok=True)
|
|
profile = UserProfile.default(
|
|
user_id=user_id,
|
|
username=username,
|
|
registered=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
referrer=referrer if referrer and referrer != user_id else None,
|
|
is_admin=(user_id == ADMIN_ID),
|
|
)
|
|
save(user_id, profile)
|
|
is_new = True
|
|
|
|
stats = load_stats()
|
|
stats.total_users += 1
|
|
save_stats(stats)
|
|
|
|
if referrer and referrer != user_id:
|
|
ref_profile = get(referrer)
|
|
if ref_profile and user_id not in ref_profile.referrals:
|
|
ref_profile.referrals.append(user_id)
|
|
save(referrer, ref_profile)
|
|
elif username:
|
|
profile = get(user_id)
|
|
if profile and profile.username != username:
|
|
profile.username = username
|
|
save(user_id, profile)
|
|
|
|
return is_new
|
|
|
|
|
|
def get(user_id: int) -> UserProfile | None:
|
|
user_path = path(user_id)
|
|
if not user_path.exists():
|
|
return None
|
|
with user_path.open("r", encoding="utf-8") as fh:
|
|
return UserProfile.from_dict(json.load(fh))
|
|
|
|
|
|
def save(user_id: int, data: UserProfile | dict) -> None:
|
|
profile = data if isinstance(data, UserProfile) else UserProfile.from_dict(data)
|
|
user_path = path(user_id)
|
|
user_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with user_path.open("w", encoding="utf-8") as fh:
|
|
json.dump(profile.to_dict(), fh, indent=4, ensure_ascii=False)
|
|
|
|
|
|
def all_users() -> list[int]:
|
|
if not DATABASE_DIR.exists():
|
|
return []
|
|
return [int(item.name) for item in DATABASE_DIR.iterdir() if item.is_dir() and item.name.isdigit()]
|
|
|
|
|
|
def find_by_username(username: str) -> int | None:
|
|
username = username.lstrip("@").lower()
|
|
for uid in all_users():
|
|
profile = get(uid)
|
|
if profile and (profile.username or "").lower() == username:
|
|
return uid
|
|
return None
|