Add modular Roblox buyer bot and documentation

db/storage.py

@@ -0,0 +1,62 @@
+import json
+from pathlib import Path
+from typing import Any
+
+from config import CONFIG_FILE, DEFAULT_CONFIG, DEFAULT_STATS, PROXIES_FILE, STATS_FILE
+from models.config import BotConfig
+from models.stats import BotStats
+
+
+def _load_json(path: Path) -> dict[str, Any] | None:
+    if not path.exists():
+        return None
+    try:
+        with path.open("r", encoding="utf-8") as fh:
+            return json.load(fh)
+    except Exception:
+        return None
+
+
+def _save_json(path: Path, data: dict[str, Any]) -> None:
+    with path.open("w", encoding="utf-8") as fh:
+        json.dump(data, fh, indent=4, ensure_ascii=False)
+
+
+def load_config() -> BotConfig:
+    data = _load_json(CONFIG_FILE)
+    if data is None:
+        config = DEFAULT_CONFIG
+        save_config(config)
+        return config
+    defaults = DEFAULT_CONFIG.to_dict()
+    defaults.update(data)
+    default_rates = defaults["rates"]
+    default_rates.update(data.get("rates") or {})
+    defaults["rates"] = default_rates
+    return BotConfig.from_dict(defaults)
+
+
+def save_config(cfg: BotConfig | dict[str, Any]) -> None:
+    config = cfg if isinstance(cfg, BotConfig) else BotConfig.from_dict(cfg)
+    _save_json(CONFIG_FILE, config.to_dict())
+
+
+def load_stats() -> BotStats:
+    data = _load_json(STATS_FILE)
+    if data is None:
+        stats = DEFAULT_STATS
+        save_stats(stats)
+        return stats
+    return BotStats.from_dict(data)
+
+
+def save_stats(stats: BotStats | dict[str, Any]) -> None:
+    model = stats if isinstance(stats, BotStats) else BotStats.from_dict(stats)
+    _save_json(STATS_FILE, model.to_dict())
+
+
+def load_proxies() -> list[str]:
+    if not PROXIES_FILE.exists():
+        return []
+    with PROXIES_FILE.open("r", encoding="utf-8", errors="ignore") as fh:
+        return [line.strip() for line in fh if line.strip()]

db/users.py

@@ -0,0 +1,80 @@
+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