#!/usr/bin/env python3 """ SQLite Database Manager for Plants vs. Zombies Game (ai.qpid.pro) Zero external dependencies. Manages users, game progress profiles, leaderboards, sessions, login audit logs, campaign levels, bug reports, admin settings, and game balance configuration. """ import sqlite3 import json import os import re import secrets import hashlib import hmac from datetime import datetime, timedelta BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DATA_DIR = os.path.join(BASE_DIR, "data") DB_PATH = os.path.join(DATA_DIR, "pvz.db") os.makedirs(DATA_DIR, exist_ok=True) DEFAULT_ADMIN_PASSWORDS = [ os.environ.get("ADMIN_PASSWORD", "").strip(), "admin123", "admin" ] DEFAULT_ADMIN_PASSWORDS = [p for p in DEFAULT_ADMIN_PASSWORDS if p] DEFAULT_GAME_CONFIG = { "preset": "Classic 2009", "economy": { "sunMultiplier": 1.0, "startingSun": 50, "coinDropRate": 1.0, "sunDropRate": 1.0, "plantCostMultiplier": 1.0, "upgradeCostMultiplier": 1.0, "plantPrices": { "sunflower": 0, "peashooter": 0, "wallnut": 0, "potatomine": 150, "snowpea": 350, "repeater": 500, "chomper": 450, "cherrybomb": 400, "jalapeno": 450, "sunshroom": 250, "fumeshroom": 300, "lilypad": 200, "gravebuster": 250, "flowerpot": 200, "cabbagepult": 400, "melonpult": 750 }, "plantSunCosts": { "sunflower": 50, "peashooter": 100, "wallnut": 50, "potatomine": 25, "snowpea": 175, "repeater": 200, "chomper": 150, "cherrybomb": 150, "jalapeno": 125, "sunshroom": 25, "fumeshroom": 75, "lilypad": 25, "gravebuster": 75, "flowerpot": 25, "cabbagepult": 100, "melonpult": 300 } }, "zombies": { "hpMultiplier": 1.0, "speedMultiplier": 1.0, "damageMultiplier": 1.0, "waveDensityMultiplier": 1.0, "types": { "normal": {"hp": 200, "speed": 22, "damage": 30}, "conehead": {"hp": 300, "speed": 22, "damage": 30}, "buckethead": {"hp": 600, "speed": 22, "damage": 30}, "newspaper": {"hp": 300, "speed": 18, "damage": 30, "angrySpeed": 48}, "screendoor": {"hp": 650, "speed": 20, "damage": 30}, "football": {"hp": 600, "speed": 42, "damage": 35}, "dancing": {"hp": 350, "speed": 25, "damage": 30}, "gargantuar": {"hp": 3000, "speed": 14, "damage": 300}, "zomboss": {"hp": 10000, "speed": 0, "damage": 500}, "polevaulter": {"hp": 180, "speed": 44, "damage": 30}, "duckytube": {"hp": 120, "speed": 24, "damage": 30}, "flag": {"hp": 200, "speed": 28, "damage": 30} } }, "plants": { "damageMultiplier": 1.0, "hpMultiplier": 1.0, "cooldownMultiplier": 1.0, "sunProduceIntervalMultiplier": 1.0 }, "zenGarden": { "dropRateMultiplier": 1.0, "growthRateMultiplier": 1.0, "fertilizerMultiplier": 1.0, "waterIntervalMinutes": 30 }, "events": { "doubleGold": False, "fastCooldowns": False, "infiniteSun": False, "turboZombies": False, "activeEventName": "none" }, "presets": { "Classic 2009": { "description": "Классический баланс оригинальной Plants vs. Zombies 2009 года", "economy": {"sunMultiplier": 1.0, "startingSun": 50, "coinDropRate": 1.0, "sunDropRate": 1.0, "plantCostMultiplier": 1.0, "upgradeCostMultiplier": 1.0}, "zombies": {"hpMultiplier": 1.0, "speedMultiplier": 1.0, "damageMultiplier": 1.0, "waveDensityMultiplier": 1.0}, "plants": {"damageMultiplier": 1.0, "hpMultiplier": 1.0, "cooldownMultiplier": 1.0, "sunProduceIntervalMultiplier": 1.0}, "zenGarden": {"dropRateMultiplier": 1.0, "growthRateMultiplier": 1.0, "fertilizerMultiplier": 1.0, "waterIntervalMinutes": 30}, "events": {"doubleGold": False, "fastCooldowns": False, "infiniteSun": False, "turboZombies": False, "activeEventName": "none"} }, "Hardcore": { "description": "Сложный режим для ветеранов: живучие и быстрые зомби, меньше ресурсов, плотные волны", "economy": {"sunMultiplier": 0.8, "startingSun": 25, "coinDropRate": 0.8, "sunDropRate": 0.8, "plantCostMultiplier": 1.2, "upgradeCostMultiplier": 1.3}, "zombies": {"hpMultiplier": 1.6, "speedMultiplier": 1.3, "damageMultiplier": 1.5, "waveDensityMultiplier": 1.5}, "plants": {"damageMultiplier": 0.85, "hpMultiplier": 0.85, "cooldownMultiplier": 1.2, "sunProduceIntervalMultiplier": 1.2}, "zenGarden": {"dropRateMultiplier": 0.8, "growthRateMultiplier": 0.8, "fertilizerMultiplier": 1.0, "waterIntervalMinutes": 45}, "events": {"doubleGold": False, "fastCooldowns": False, "infiniteSun": False, "turboZombies": True, "activeEventName": "hardcore"} }, "Turbo": { "description": "Ураганный экшен: супер-быстрая перезарядка, море солнца и стремительные зомби", "economy": {"sunMultiplier": 2.0, "startingSun": 150, "coinDropRate": 1.5, "sunDropRate": 1.8, "plantCostMultiplier": 0.8, "upgradeCostMultiplier": 0.8}, "zombies": {"hpMultiplier": 1.1, "speedMultiplier": 1.6, "damageMultiplier": 1.2, "waveDensityMultiplier": 1.3}, "plants": {"damageMultiplier": 1.3, "hpMultiplier": 1.2, "cooldownMultiplier": 0.4, "sunProduceIntervalMultiplier": 0.6}, "zenGarden": {"dropRateMultiplier": 1.5, "growthRateMultiplier": 2.0, "fertilizerMultiplier": 1.5, "waterIntervalMinutes": 15}, "events": {"doubleGold": True, "fastCooldowns": True, "infiniteSun": False, "turboZombies": True, "activeEventName": "turbo"} }, "Casual": { "description": "Расслабленный режим: увеличенный урон растений, много солнца и золота", "economy": {"sunMultiplier": 1.5, "startingSun": 200, "coinDropRate": 2.0, "sunDropRate": 1.5, "plantCostMultiplier": 0.7, "upgradeCostMultiplier": 0.7}, "zombies": {"hpMultiplier": 0.65, "speedMultiplier": 0.8, "damageMultiplier": 0.7, "waveDensityMultiplier": 0.8}, "plants": {"damageMultiplier": 1.7, "hpMultiplier": 1.5, "cooldownMultiplier": 0.7, "sunProduceIntervalMultiplier": 0.7}, "zenGarden": {"dropRateMultiplier": 2.0, "growthRateMultiplier": 2.0, "fertilizerMultiplier": 2.0, "waterIntervalMinutes": 10}, "events": {"doubleGold": True, "fastCooldowns": False, "infiniteSun": False, "turboZombies": False, "activeEventName": "casual"} }, "Custom": { "description": "Пользовательские настройки баланса", "economy": {"sunMultiplier": 1.0, "startingSun": 50, "coinDropRate": 1.0, "sunDropRate": 1.0, "plantCostMultiplier": 1.0, "upgradeCostMultiplier": 1.0}, "zombies": {"hpMultiplier": 1.0, "speedMultiplier": 1.0, "damageMultiplier": 1.0, "waveDensityMultiplier": 1.0}, "plants": {"damageMultiplier": 1.0, "hpMultiplier": 1.0, "cooldownMultiplier": 1.0, "sunProduceIntervalMultiplier": 1.0}, "zenGarden": {"dropRateMultiplier": 1.0, "growthRateMultiplier": 1.0, "fertilizerMultiplier": 1.0, "waterIntervalMinutes": 30}, "events": {"doubleGold": False, "fastCooldowns": False, "infiniteSun": False, "turboZombies": False, "activeEventName": "custom"} } } } DEFAULT_CAMPAIGN_LEVELS = [ # World 1: Day (1-1 .. 1-5) { "level": 1, "world": 1, "stage": 1, "code": "1-1", "title": "Первые шаги", "desc": "Знакомство с садом и простыми зомби", "waves": {"count": 3, "types": ["normal"]}, "biome": "day", "reward": 100, "is_boss": 0, "modifiers": {}, "dialogue": None }, { "level": 2, "world": 1, "stage": 2, "code": "1-2", "title": "Опасные конусы", "desc": "Появление стойких зомби с дорожными конусами", "waves": {"count": 3, "types": ["normal", "conehead"]}, "biome": "day", "reward": 120, "is_boss": 0, "modifiers": {}, "dialogue": None }, { "level": 3, "world": 1, "stage": 3, "code": "1-3", "title": "Картофельный рубеж", "desc": "Тактическая оборона с картофельными минами", "waves": {"count": 4, "types": ["normal", "conehead"]}, "biome": "day", "reward": 140, "is_boss": 0, "modifiers": {}, "dialogue": None }, { "level": 4, "world": 1, "stage": 4, "code": "1-4", "title": "Прыгуны с шестом", "desc": "Быстрые зомби легко перепрыгивают первые растения", "waves": {"count": 4, "types": ["normal", "conehead", "polevaulter"]}, "biome": "day", "reward": 160, "is_boss": 0, "modifiers": {}, "dialogue": None }, { "level": 5, "world": 1, "stage": 5, "code": "1-5", "title": "Штурм Дня", "desc": "Финальная массивная волна зомби на солнечном газоне", "waves": {"count": 4, "types": ["normal", "conehead", "buckethead", "polevaulter"]}, "biome": "day", "reward": 200, "is_boss": 0, "modifiers": {}, "dialogue": None }, # World 2: Night (2-1 .. 2-5) { "level": 6, "world": 2, "stage": 1, "code": "2-1", "title": "Сгущаются сумерки", "desc": "Ночь без падающего солнца: солнцегрибы и надгробия", "waves": {"count": 3, "types": ["normal", "conehead"]}, "biome": "night", "reward": 150, "is_boss": 0, "modifiers": {"noSkySun": True, "graves": True}, "dialogue": { "title": "🌙 Безумный Дэйв предупреждает:", "text": "Эй, сосед! Ночью солнце с неба не падает! Обязательно возьми Солнцегриб!", "btnText": "ДАЛЕЕ / В БОЙ ⚔️" } }, { "level": 7, "world": 2, "stage": 2, "code": "2-2", "title": "Ярость газетчика", "desc": "Разъярённые зомби с газетами ускоряются в бешенстве", "waves": {"count": 4, "types": ["normal", "newspaper"]}, "biome": "night", "reward": 170, "is_boss": 0, "modifiers": {"noSkySun": True}, "dialogue": None }, { "level": 8, "world": 2, "stage": 3, "code": "2-3", "title": "Споровый шквал", "desc": "Дымогрибы пробивают плотные ряды врагов", "waves": {"count": 4, "types": ["normal", "newspaper", "conehead"]}, "biome": "night", "reward": 190, "is_boss": 0, "modifiers": {"noSkySun": True}, "dialogue": None }, { "level": 9, "world": 2, "stage": 4, "code": "2-4", "title": "Бронекаски", "desc": "Зомби с вёдрами штурмуют тёмный двор", "waves": {"count": 4, "types": ["normal", "screendoor", "buckethead"]}, "biome": "night", "reward": 210, "is_boss": 0, "modifiers": {"noSkySun": True}, "dialogue": None }, { "level": 10, "world": 2, "stage": 5, "code": "2-5", "title": "Полночный кошмар", "desc": "Зомби восстают прямо из могил в финальной волне", "waves": {"count": 5, "types": ["normal", "newspaper", "buckethead", "dancing"]}, "biome": "night", "reward": 250, "is_boss": 0, "modifiers": {"noSkySun": True, "graveSpawns": True}, "dialogue": None }, # World 3: Pool (3-1 .. 3-5) { "level": 11, "world": 3, "stage": 1, "code": "3-1", "title": "Водная гладь", "desc": "6 дорожек газона, кувшинки и зомби на утках", "waves": {"count": 3, "types": ["normal", "duckytube"]}, "biome": "pool", "reward": 200, "is_boss": 0, "modifiers": {"hasPool": True, "lanes": 6}, "dialogue": { "title": "🏊 Безумный Дэйв спешит на помощь:", "text": "Там бассейн! На воде растения тонут — сначала сажай Кувшинки (Lily Pad)!", "btnText": "ДАЛЕЕ / В БОЙ ⚔️" } }, { "level": 12, "world": 3, "stage": 2, "code": "3-2", "title": "Заплыв с шестом", "desc": "Прыгуны на суше и быстрые пловцы на воде", "waves": {"count": 4, "types": ["normal", "duckytube", "polevaulter"]}, "biome": "pool", "reward": 220, "is_boss": 0, "modifiers": {"hasPool": True, "lanes": 6}, "dialogue": None }, { "level": 13, "world": 3, "stage": 3, "code": "3-3", "title": "Зубастики у воды", "desc": "Хищные зубастики против бронированных уток", "waves": {"count": 4, "types": ["normal", "duckytube", "conehead"]}, "biome": "pool", "reward": 240, "is_boss": 0, "modifiers": {"hasPool": True, "lanes": 6}, "dialogue": None }, { "level": 14, "world": 3, "stage": 4, "code": "3-4", "title": "Футбольный матч", "desc": "Мощные футболисты прорывают береговую линию", "waves": {"count": 4, "types": ["normal", "football", "duckytube"]}, "biome": "pool", "reward": 260, "is_boss": 0, "modifiers": {"hasPool": True, "lanes": 6}, "dialogue": None }, { "level": 15, "world": 3, "stage": 5, "code": "3-5", "title": "Водный шквал", "desc": "Штурм всех 6 дорожек с очистителями бассейна", "waves": {"count": 5, "types": ["normal", "buckethead", "football", "duckytube"]}, "biome": "pool", "reward": 300, "is_boss": 0, "modifiers": {"hasPool": True, "lanes": 6}, "dialogue": None }, # World 4: Fog (4-1 .. 4-5) { "level": 16, "world": 4, "stage": 1, "code": "4-1", "title": "Плотная мгла", "desc": "Густой туман скрывает правую половину поля", "waves": {"count": 4, "types": ["normal", "conehead", "duckytube"]}, "biome": "fog", "reward": 250, "is_boss": 0, "modifiers": {"hasPool": True, "hasFog": True, "noSkySun": True}, "dialogue": { "title": "🌫️ Безумный Дэйв щурится во мгле:", "text": "Ого, какой густой туман! Зомби крадутся в темноте, используй Плантерн и Вздуватель!", "btnText": "ДАЛЕЕ / В БОЙ ⚔️" } }, { "level": 17, "world": 4, "stage": 2, "code": "4-2", "title": "Невидимая угроза", "desc": "Внезапные появления врагов прямо из белой пелены", "waves": {"count": 4, "types": ["normal", "newspaper", "screendoor"]}, "biome": "fog", "reward": 270, "is_boss": 0, "modifiers": {"hasPool": True, "hasFog": True, "noSkySun": True}, "dialogue": None }, { "level": 18, "world": 4, "stage": 3, "code": "4-3", "title": "Туманный бассейн", "desc": "Водные линии скрыты во мраке ночного тумана", "waves": {"count": 4, "types": ["normal", "duckytube", "buckethead"]}, "biome": "fog", "reward": 290, "is_boss": 0, "modifiers": {"hasPool": True, "hasFog": True, "noSkySun": True}, "dialogue": None }, { "level": 19, "world": 4, "stage": 4, "code": "4-4", "title": "Спринт сквозь мглу", "desc": "Футболисты и прыгуны выскакивают из тумана", "waves": {"count": 5, "types": ["normal", "football", "polevaulter", "duckytube"]}, "biome": "fog", "reward": 310, "is_boss": 0, "modifiers": {"hasPool": True, "hasFog": True, "noSkySun": True}, "dialogue": None }, { "level": 20, "world": 4, "stage": 5, "code": "4-5", "title": "Штормовой туман", "desc": "Финальный прорыв орды через непроглядную пелену", "waves": {"count": 5, "types": ["normal", "football", "dancing", "buckethead"]}, "biome": "fog", "reward": 350, "is_boss": 0, "modifiers": {"hasPool": True, "hasFog": True, "noSkySun": True}, "dialogue": None }, # World 5: Roof (5-1 .. 5-5) { "level": 21, "world": 5, "stage": 1, "code": "5-1", "title": "Черепичный скат", "desc": "Крыша дома, цветочные горшки и капустопульты", "waves": {"count": 4, "types": ["normal", "conehead"]}, "biome": "roof", "reward": 300, "is_boss": 0, "modifiers": {"isRoof": True, "requiresPots": True}, "dialogue": { "title": "🏠 Безумный Дэйв лезет на скат:", "text": "Мы на крыше! Скаты мешают прямой стрельбе — используй Горшки и Катапульты!", "btnText": "ДАЛЕЕ / В БОЙ ⚔️" } }, { "level": 22, "world": 5, "stage": 2, "code": "5-2", "title": "Арбузный гром", "desc": "Тяжёлые арбузопульты со сплэш-уроном по площади", "waves": {"count": 4, "types": ["normal", "buckethead", "conehead"]}, "biome": "roof", "reward": 320, "is_boss": 0, "modifiers": {"isRoof": True, "requiresPots": True}, "dialogue": None }, { "level": 23, "world": 5, "stage": 3, "code": "5-3", "title": "Опасные верхолазы", "desc": "Бронезомби карабкаются по скользкому скату крыши", "waves": {"count": 4, "types": ["normal", "screendoor", "football"]}, "biome": "roof", "reward": 350, "is_boss": 0, "modifiers": {"isRoof": True, "requiresPots": True}, "dialogue": None }, { "level": 24, "world": 5, "stage": 4, "code": "5-4", "title": "Осада дымохода", "desc": "Генеральный штурм перед чердаком дома", "waves": {"count": 5, "types": ["normal", "gargantuar", "football", "buckethead"]}, "biome": "roof", "reward": 400, "is_boss": 0, "modifiers": {"isRoof": True, "requiresPots": True}, "dialogue": None }, { "level": 25, "world": 5, "stage": 5, "code": "5-5", "title": "Доктор Зомбосс!", "desc": "Финальная битва против шагающего робота Zombot!", "waves": {"count": 1, "types": ["zomboss"]}, "biome": "roof", "reward": 1000, "is_boss": 1, "modifiers": {"isRoof": True, "isBoss": True}, "dialogue": { "title": "🤖 БЕЗУМИЕ ПОЛНОГО МАСШТАБА!", "text": "БЕЗУМИЕ! Доктор Зомбосс выкатил своего гигантского робота! Держи оборону!", "btnText": "ДАЛЕЕ / В БОЙ ⚔️" } } ] def parse_device_info(user_agent): """Extract friendly device and browser name from User-Agent string.""" if not user_agent or not isinstance(user_agent, str): return "Unknown Device" ua = user_agent.strip() if not ua: return "Unknown Device" # OS detection if "Windows NT 10.0" in ua or "Windows NT 11.0" in ua: os_name = "Windows 10/11" elif "Windows NT 6.3" in ua: os_name = "Windows 8.1" elif "Windows NT 6.1" in ua: os_name = "Windows 7" elif "Windows NT" in ua or "Windows" in ua: os_name = "Windows" elif "iPhone" in ua: os_name = "iPhone" elif "iPad" in ua: os_name = "iPad" elif "Mac OS X" in ua or "Macintosh" in ua: os_name = "macOS" elif "Android" in ua: os_name = "Android" elif "CrOS" in ua: os_name = "ChromeOS" elif "Linux" in ua: os_name = "Linux" else: os_name = "Unknown OS" # Browser detection if "Edg/" in ua or "Edge/" in ua: m = re.search(r"Edg[e]?/(\d+)", ua) browser = f"Edge {m.group(1)}" if m else "Edge" elif "OPR/" in ua or "Opera/" in ua: m = re.search(r"(?:OPR|Opera)/(\d+)", ua) browser = f"Opera {m.group(1)}" if m else "Opera" elif "YaBrowser/" in ua: m = re.search(r"YaBrowser/(\d+)", ua) browser = f"Yandex {m.group(1)}" if m else "Yandex" elif "Chrome/" in ua or "CriOS/" in ua: m = re.search(r"(?:Chrome|CriOS)/(\d+)", ua) browser = f"Chrome {m.group(1)}" if m else "Chrome" elif "Firefox/" in ua or "FxiOS/" in ua: m = re.search(r"(?:Firefox|FxiOS)/(\d+)", ua) browser = f"Firefox {m.group(1)}" if m else "Firefox" elif "Safari/" in ua and "Version/" in ua: m = re.search(r"Version/(\d+)", ua) browser = f"Safari {m.group(1)}" if m else "Safari" elif "PostmanRuntime" in ua: browser = "Postman" elif "curl" in ua: browser = "curl" elif "python" in ua.lower(): browser = "Python Client" else: browser = "Browser" return f"{browser} ({os_name})" def get_connection(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def hash_password(password, salt=None): """Secure server-side salted PBKDF2 password hashing.""" if not password: return "" salt = salt or secrets.token_hex(16) iterations = 100_000 derived = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), iterations).hex() return f"pbkdf2:sha256:{iterations}:{salt}:{derived}" def verify_password(password, stored_hash): """Verifies a password against stored hash with backward compatibility for legacy hashes.""" if not password or not stored_hash: return False if stored_hash.startswith("pbkdf2:sha256:"): try: parts = stored_hash.split(":") if len(parts) == 5: iterations = int(parts[2]) salt = parts[3] expected = parts[4] derived = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), iterations).hex() return hmac.compare_digest(derived, expected) except Exception: return False # Backward compatibility for plain/legacy hashes return hmac.compare_digest(str(password), str(stored_hash)) def init_db(): conn = get_connection() cursor = conn.cursor() # 1. Users Table cursor.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL COLLATE NOCASE, password_hash TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_login TIMESTAMP DEFAULT CURRENT_TIMESTAMP, is_banned INTEGER DEFAULT 0, ban_reason TEXT DEFAULT '', is_admin INTEGER DEFAULT 0 ) """) cursor.execute("PRAGMA table_info(users)") cols = [c["name"] for c in cursor.fetchall()] if "is_banned" not in cols: cursor.execute("ALTER TABLE users ADD COLUMN is_banned INTEGER DEFAULT 0") if "ban_reason" not in cols: cursor.execute("ALTER TABLE users ADD COLUMN ban_reason TEXT DEFAULT ''") if "is_admin" not in cols: cursor.execute("ALTER TABLE users ADD COLUMN is_admin INTEGER DEFAULT 0") # 2. Sessions Table cursor.execute(""" CREATE TABLE IF NOT EXISTS sessions ( token TEXT PRIMARY KEY, user_id INTEGER NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, expires_at TIMESTAMP NOT NULL, ip_address TEXT DEFAULT '', user_agent TEXT DEFAULT '', device_name TEXT DEFAULT '', last_active TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_sessions_user_id ON sessions(user_id)") cursor.execute("PRAGMA table_info(sessions)") sess_cols = [c["name"] for c in cursor.fetchall()] if "ip_address" not in sess_cols: cursor.execute("ALTER TABLE sessions ADD COLUMN ip_address TEXT DEFAULT ''") if "user_agent" not in sess_cols: cursor.execute("ALTER TABLE sessions ADD COLUMN user_agent TEXT DEFAULT ''") if "device_name" not in sess_cols: cursor.execute("ALTER TABLE sessions ADD COLUMN device_name TEXT DEFAULT ''") if "last_active" not in sess_cols: cursor.execute("ALTER TABLE sessions ADD COLUMN last_active TIMESTAMP DEFAULT ''") # 3. Login Audit Logs Table cursor.execute(""" CREATE TABLE IF NOT EXISTS login_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, username TEXT, ip_address TEXT, user_agent TEXT, device_name TEXT, status TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_login_logs_user_id ON login_logs(user_id)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_login_logs_created_at ON login_logs(created_at)") # 4. Profiles Table cursor.execute(""" CREATE TABLE IF NOT EXISTS profiles ( user_id INTEGER PRIMARY KEY, coins INTEGER DEFAULT 150, level INTEGER DEFAULT 1, max_unlocked_level INTEGER DEFAULT 1, high_score INTEGER DEFAULT 0, theme TEXT DEFAULT 'classic', upgrades TEXT DEFAULT '{}', materials TEXT DEFAULT '{}', plant_levels TEXT DEFAULT '{}', perks TEXT DEFAULT '{}', zen_garden TEXT DEFAULT '{}', tree_of_wisdom TEXT DEFAULT '{}', achievements TEXT DEFAULT '{}', stats TEXT DEFAULT '{}', casino TEXT DEFAULT '{}', zombie_lab TEXT DEFAULT '{}', admin_updated_at TIMESTAMP DEFAULT NULL, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) """) # 5. Leaderboards Table cursor.execute(""" CREATE TABLE IF NOT EXISTS leaderboards ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, player_name TEXT NOT NULL, mode TEXT NOT NULL, score INTEGER NOT NULL, detail TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # 6. Bug Reports Table cursor.execute(""" CREATE TABLE IF NOT EXISTS bugs ( id INTEGER PRIMARY KEY AUTOINCREMENT, author TEXT NOT NULL, category TEXT NOT NULL, text TEXT NOT NULL, status TEXT DEFAULT 'новый', timestamp TEXT NOT NULL, admin_reply TEXT ) """) # 7. Game Balance Config Table cursor.execute(""" CREATE TABLE IF NOT EXISTS game_config ( id INTEGER PRIMARY KEY AUTOINCREMENT, preset_name TEXT DEFAULT 'Classic 2009', config_data TEXT NOT NULL, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # 8. Campaign Levels Table cursor.execute(""" CREATE TABLE IF NOT EXISTS campaign_levels ( level INTEGER PRIMARY KEY, world INTEGER, stage INTEGER, code TEXT, title TEXT, desc TEXT, reward INTEGER, biome TEXT, is_boss INTEGER DEFAULT 0, waves_json TEXT DEFAULT '{}', modifiers_json TEXT DEFAULT '{}', dialogue_json TEXT DEFAULT '{}', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # 9. Admin Settings Table cursor.execute(""" CREATE TABLE IF NOT EXISTS admin_settings ( key TEXT PRIMARY KEY, value TEXT, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Seed default game config if not present cursor.execute("SELECT COUNT(*) as cnt FROM game_config") row = cursor.fetchone() if row["cnt"] == 0: cursor.execute( "INSERT INTO game_config (preset_name, config_data) VALUES (?, ?)", ("Classic 2009", json.dumps(DEFAULT_GAME_CONFIG, ensure_ascii=False)) ) # Seed default campaign levels if empty cursor.execute("SELECT COUNT(*) as cnt FROM campaign_levels") lvl_row = cursor.fetchone() if lvl_row["cnt"] == 0: for lvl in DEFAULT_CAMPAIGN_LEVELS: cursor.execute(""" INSERT INTO campaign_levels ( level, world, stage, code, title, desc, reward, biome, is_boss, waves_json, modifiers_json, dialogue_json, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) """, ( lvl["level"], lvl["world"], lvl["stage"], lvl["code"], lvl["title"], lvl["desc"], lvl["reward"], lvl["biome"], lvl.get("is_boss", 0), json.dumps(lvl.get("waves", {}), ensure_ascii=False), json.dumps(lvl.get("modifiers", {}), ensure_ascii=False), json.dumps(lvl.get("dialogue") or {}, ensure_ascii=False) )) # Ensure column migrations for profiles cursor.execute("PRAGMA table_info(profiles)") prof_cols = [col["name"] for col in cursor.fetchall()] if "stats" not in prof_cols: cursor.execute("ALTER TABLE profiles ADD COLUMN stats TEXT DEFAULT '{}'") if "admin_updated_at" not in prof_cols: cursor.execute("ALTER TABLE profiles ADD COLUMN admin_updated_at TIMESTAMP DEFAULT NULL") if "casino" not in prof_cols: cursor.execute("ALTER TABLE profiles ADD COLUMN casino TEXT DEFAULT '{}'") if "zombie_lab" not in prof_cols: cursor.execute("ALTER TABLE profiles ADD COLUMN zombie_lab TEXT DEFAULT '{}'") conn.commit() conn.close() # ========================================== # Admin Settings & Master Password # ========================================== def set_admin_password(new_password): """Securely updates admin master password hash in admin_settings.""" if not new_password or not isinstance(new_password, str): return {"success": False, "message": "Пароль не может быть пустым"} new_password = new_password.strip() if len(new_password) < 4: return {"success": False, "message": "Пароль должен содержать не менее 4 символов"} sec_hash = hash_password(new_password) now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") conn = get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO admin_settings (key, value, updated_at) VALUES ('admin_password_hash', ?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at """, (sec_hash, now_str)) conn.commit() conn.close() return {"success": True, "message": "Мастер-пароль администратора успешно обновлен"} def verify_admin_password(password): """Verifies password against admin_settings custom hash or default admin passwords.""" if not password: return False password = str(password).strip() if not password: return False conn = get_connection() cursor = conn.cursor() cursor.execute("SELECT value FROM admin_settings WHERE key = 'admin_password_hash'") row = cursor.fetchone() conn.close() if row and row["value"]: stored = row["value"] if verify_password(password, stored): return True # Fallback to default admin passwords or env variable for default_p in DEFAULT_ADMIN_PASSWORDS: if hmac.compare_digest(password, default_p): return True return False # ========================================== # Session Management Methods # ========================================== def create_session(user_id, duration_days=30, ip_address="", user_agent="", device_name=""): """Generates a secure 256-bit token and creates an extended session record.""" token = secrets.token_hex(32) now = datetime.now() now_str = now.strftime("%Y-%m-%d %H:%M:%S") expires_at = (now + timedelta(days=duration_days)).strftime("%Y-%m-%d %H:%M:%S") if not device_name and user_agent: device_name = parse_device_info(user_agent) conn = get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO sessions ( token, user_id, created_at, expires_at, ip_address, user_agent, device_name, last_active ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, (token, user_id, now_str, expires_at, ip_address or "", user_agent or "", device_name or "", now_str)) conn.commit() conn.close() return token def update_session_activity(token, ip_address=None): """Updates the last_active timestamp and optionally IP for an active session.""" if not token or not isinstance(token, str): return False token = token.strip() now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") conn = get_connection() cursor = conn.cursor() if ip_address: cursor.execute("UPDATE sessions SET last_active = ?, ip_address = ? WHERE token = ?", (now_str, str(ip_address).strip(), token)) else: cursor.execute("UPDATE sessions SET last_active = ? WHERE token = ?", (now_str, token)) conn.commit() conn.close() return True def validate_session(token): """Validates session token. Returns user dict if valid, else None.""" if not token or not isinstance(token, str): return None token = token.strip() if not token: return None conn = get_connection() cursor = conn.cursor() cursor.execute(""" SELECT u.id, u.username, u.created_at, u.last_login, u.is_banned, u.ban_reason, u.is_admin, s.expires_at, s.token, s.ip_address, s.user_agent, s.device_name, s.last_active FROM sessions s JOIN users u ON s.user_id = u.id WHERE s.token = ? """, (token,)) row = cursor.fetchone() if not row: conn.close() return None # Check if user was banned after session creation if row["is_banned"]: cursor.execute("DELETE FROM sessions WHERE token = ?", (token,)) conn.commit() conn.close() return None # Check expiration try: expires_at = datetime.strptime(row["expires_at"], "%Y-%m-%d %H:%M:%S") if datetime.now() > expires_at: cursor.execute("DELETE FROM sessions WHERE token = ?", (token,)) conn.commit() conn.close() return None except Exception: conn.close() return None # Update last active timestamp now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") cursor.execute("UPDATE sessions SET last_active = ? WHERE token = ?", (now_str, token)) conn.commit() user_info = { "id": row["id"], "user_id": row["id"], "username": row["username"], "createdAt": row["created_at"], "lastLogin": row["last_login"], "isBanned": bool(row["is_banned"]), "banReason": row["ban_reason"] or "", "isAdmin": bool(row["is_admin"]), "token": row["token"], "ipAddress": row["ip_address"] or "", "deviceName": row["device_name"] or "", "lastActive": now_str } conn.close() return user_info def delete_session(token): """Deletes/invalidates a single session by token.""" return terminate_session(token) def terminate_session(token, user_id=None): """Terminates a specific session by token (optional user_id verification).""" if not token: return True conn = get_connection() cursor = conn.cursor() if user_id is not None: cursor.execute("DELETE FROM sessions WHERE token = ? AND user_id = ?", (token.strip(), user_id)) else: cursor.execute("DELETE FROM sessions WHERE token = ?", (token.strip(),)) conn.commit() conn.close() return True def terminate_all_user_sessions(user_id, except_token=None): """Terminates all sessions for a user, optionally preserving the current active session token.""" conn = get_connection() cursor = conn.cursor() if except_token: cursor.execute("DELETE FROM sessions WHERE user_id = ? AND token != ?", (user_id, except_token.strip())) else: cursor.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,)) conn.commit() conn.close() return True def get_user_sessions(user_id): """Returns all active unexpired sessions for a specific user.""" conn = get_connection() cursor = conn.cursor() now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") cursor.execute(""" SELECT token, user_id, created_at, expires_at, ip_address, user_agent, device_name, last_active FROM sessions WHERE user_id = ? AND expires_at > ? ORDER BY last_active DESC """, (user_id, now_str)) rows = cursor.fetchall() conn.close() return [{ "token": r["token"], "userId": r["user_id"], "ipAddress": r["ip_address"] or "", "userAgent": r["user_agent"] or "", "deviceName": r["device_name"] or "Unknown Device", "createdAt": r["created_at"], "expiresAt": r["expires_at"], "lastActive": r["last_active"] or r["created_at"] } for r in rows] def get_all_active_sessions(): """Returns all active unexpired sessions across all users for admin dashboard.""" conn = get_connection() cursor = conn.cursor() now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") cursor.execute(""" SELECT s.token, s.user_id, u.username, u.is_admin, s.created_at, s.expires_at, s.ip_address, s.user_agent, s.device_name, s.last_active FROM sessions s JOIN users u ON s.user_id = u.id WHERE s.expires_at > ? ORDER BY COALESCE(NULLIF(s.last_active, ''), s.created_at) DESC """, (now_str,)) rows = cursor.fetchall() conn.close() return [{ "token": r["token"], "userId": r["user_id"], "username": r["username"], "isAdmin": bool(r["is_admin"]), "ipAddress": r["ip_address"] if r["ip_address"] else "127.0.0.1", "userAgent": r["user_agent"] or "", "deviceName": r["device_name"] if r["device_name"] and r["device_name"] != "Unknown Device" else "Браузер", "createdAt": r["created_at"], "expiresAt": r["expires_at"], "lastActive": r["last_active"] or r["created_at"] } for r in rows] def cleanup_expired_sessions(days=30): """Deletes inactive/expired sessions older than days days from sessions, returns count of deleted rows.""" try: days = int(days) except (ValueError, TypeError): days = 30 conn = get_connection() cursor = conn.cursor() now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") cutoff_str = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S") cursor.execute(""" DELETE FROM sessions WHERE expires_at < ? OR (last_active != '' AND last_active IS NOT NULL AND last_active < ?) OR ((last_active = '' OR last_active IS NULL) AND created_at < ?) """, (now_str, cutoff_str, cutoff_str)) deleted_count = cursor.rowcount conn.commit() conn.close() return deleted_count # ========================================== # Login Audit Logs # ========================================== def record_login_log(user_id, username, ip_address="", user_agent="", status="success", device_name=None): """Records authentication attempt in audit logs.""" if not device_name and user_agent: device_name = parse_device_info(user_agent) conn = get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO login_logs (user_id, username, ip_address, user_agent, device_name, status, created_at) VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) """, ( user_id, str(username or "").strip(), str(ip_address or "").strip(), str(user_agent or "").strip(), str(device_name or "Unknown Device").strip(), str(status or "success").strip() )) conn.commit() conn.close() def get_login_logs(limit=100, username=None, ip=None): """Retrieves login audit logs with optional filtering by username or IP.""" conn = get_connection() cursor = conn.cursor() query = "SELECT id, user_id, username, ip_address, user_agent, device_name, status, created_at FROM login_logs WHERE 1=1" params = [] if username: query += " AND username = ? COLLATE NOCASE" params.append(username.strip()) if ip: query += " AND ip_address = ?" params.append(ip.strip()) query += " ORDER BY id DESC LIMIT ?" params.append(limit) cursor.execute(query, params) rows = cursor.fetchall() conn.close() return [{ "id": r["id"], "userId": r["user_id"], "username": r["username"], "ipAddress": r["ip_address"] or "", "userAgent": r["user_agent"] or "", "deviceName": r["device_name"] or "Unknown Device", "status": r["status"] or "unknown", "createdAt": r["created_at"] } for r in rows] def cleanup_old_audit_logs(days=30): """Deletes records from login_logs older than days days, returns count of deleted rows.""" try: days = int(days) except (ValueError, TypeError): days = 30 conn = get_connection() cursor = conn.cursor() cutoff_str = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S") cursor.execute("DELETE FROM login_logs WHERE created_at < ?", (cutoff_str,)) deleted_count = cursor.rowcount conn.commit() conn.close() return deleted_count def perform_system_maintenance(audit_days=30, session_days=30): """ Performs combined system maintenance: - Cleans up old audit logs (older than audit_days) - Cleans up expired/inactive sessions (older than session_days) - Executes VACUUM on database to optimize storage Returns a maintenance summary dictionary. """ deleted_logs = cleanup_old_audit_logs(days=audit_days) deleted_sessions = cleanup_expired_sessions(days=session_days) vacuumed = False try: conn = get_connection() conn.execute("PRAGMA busy_timeout = 5000") conn.execute("VACUUM") conn.close() vacuumed = True except Exception: vacuumed = False return { "success": True, "deletedLogs": deleted_logs, "deletedSessions": deleted_sessions, "deleted_audit_logs": deleted_logs, "deleted_sessions": deleted_sessions, "vacuumed": vacuumed, "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } # ========================================== # Campaign Levels Configuration # ========================================== def get_campaign_levels(level_num=None): """Retrieves list of all campaign levels or single level by index with parsed JSON details.""" conn = get_connection() cursor = conn.cursor() if level_num is not None: cursor.execute("SELECT * FROM campaign_levels WHERE level = ?", (int(level_num),)) row = cursor.fetchone() conn.close() if not row: return None return _format_level_row(row) cursor.execute("SELECT * FROM campaign_levels ORDER BY level ASC") rows = cursor.fetchall() conn.close() return [_format_level_row(r) for r in rows] def _format_level_row(row): return { "level": row["level"], "world": row["world"], "stage": row["stage"], "code": row["code"], "title": row["title"], "desc": row["desc"], "reward": row["reward"], "biome": row["biome"], "isBoss": bool(row["is_boss"]), "waves": json.loads(row["waves_json"] or "{}"), "modifiers": json.loads(row["modifiers_json"] or "{}"), "dialogue": json.loads(row["dialogue_json"] or "{}"), "updatedAt": row["updated_at"] } def save_campaign_level(level_data): """Creates or updates a campaign level in campaign_levels table.""" if not isinstance(level_data, dict): raise ValueError("Level data must be a dictionary") level_num = int(level_data.get("level", 1)) world = int(level_data.get("world", (level_num - 1) // 5 + 1)) stage = int(level_data.get("stage", (level_num - 1) % 5 + 1)) code = str(level_data.get("code") or f"{world}-{stage}") title = str(level_data.get("title") or f"Уровень {code}") desc = str(level_data.get("desc") or "") reward = int(level_data.get("reward", 100)) biome = str(level_data.get("biome") or "day") is_boss = 1 if (level_data.get("isBoss") or level_data.get("is_boss")) else 0 waves = level_data.get("waves") or level_data.get("waves_json") or {} waves_json = json.dumps(waves, ensure_ascii=False) if isinstance(waves, (dict, list)) else str(waves) modifiers = level_data.get("modifiers") or level_data.get("modifiers_json") or {} modifiers_json = json.dumps(modifiers, ensure_ascii=False) if isinstance(modifiers, (dict, list)) else str(modifiers) dialogue = level_data.get("dialogue") or level_data.get("dialogue_json") or {} dialogue_json = json.dumps(dialogue, ensure_ascii=False) if isinstance(dialogue, (dict, list)) else str(dialogue) now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") conn = get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO campaign_levels ( level, world, stage, code, title, desc, reward, biome, is_boss, waves_json, modifiers_json, dialogue_json, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(level) DO UPDATE SET world = excluded.world, stage = excluded.stage, code = excluded.code, title = excluded.title, desc = excluded.desc, reward = excluded.reward, biome = excluded.biome, is_boss = excluded.is_boss, waves_json = excluded.waves_json, modifiers_json = excluded.modifiers_json, dialogue_json = excluded.dialogue_json, updated_at = excluded.updated_at """, ( level_num, world, stage, code, title, desc, reward, biome, is_boss, waves_json, modifiers_json, dialogue_json, now_str )) conn.commit() conn.close() return get_campaign_levels(level_num) def reset_campaign_levels(level_num=None): """Resets one or all campaign levels to default seed configurations.""" conn = get_connection() cursor = conn.cursor() if level_num is not None: target_lvl = int(level_num) seed = next((lvl for lvl in DEFAULT_CAMPAIGN_LEVELS if lvl["level"] == target_lvl), None) if not seed: conn.close() return {"success": False, "message": f"Дефолтная конфигурация для уровня {target_lvl} не найдена"} cursor.execute(""" INSERT INTO campaign_levels ( level, world, stage, code, title, desc, reward, biome, is_boss, waves_json, modifiers_json, dialogue_json, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(level) DO UPDATE SET world = excluded.world, stage = excluded.stage, code = excluded.code, title = excluded.title, desc = excluded.desc, reward = excluded.reward, biome = excluded.biome, is_boss = excluded.is_boss, waves_json = excluded.waves_json, modifiers_json = excluded.modifiers_json, dialogue_json = excluded.dialogue_json, updated_at = CURRENT_TIMESTAMP """, ( seed["level"], seed["world"], seed["stage"], seed["code"], seed["title"], seed["desc"], seed["reward"], seed["biome"], seed.get("is_boss", 0), json.dumps(seed.get("waves", {}), ensure_ascii=False), json.dumps(seed.get("modifiers", {}), ensure_ascii=False), json.dumps(seed.get("dialogue") or {}, ensure_ascii=False) )) conn.commit() conn.close() return {"success": True, "message": f"Уровень {seed['code']} сброшен к заводским настройкам", "level": get_campaign_levels(target_lvl)} # Reset all 25 levels cursor.execute("DELETE FROM campaign_levels") for lvl in DEFAULT_CAMPAIGN_LEVELS: cursor.execute(""" INSERT INTO campaign_levels ( level, world, stage, code, title, desc, reward, biome, is_boss, waves_json, modifiers_json, dialogue_json, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) """, ( lvl["level"], lvl["world"], lvl["stage"], lvl["code"], lvl["title"], lvl["desc"], lvl["reward"], lvl["biome"], lvl.get("is_boss", 0), json.dumps(lvl.get("waves", {}), ensure_ascii=False), json.dumps(lvl.get("modifiers", {}), ensure_ascii=False), json.dumps(lvl.get("dialogue") or {}, ensure_ascii=False) )) conn.commit() conn.close() return {"success": True, "message": "Все 25 уровней кампании успешно сброшены к начальным настройкам"} # ========================================== # User Profiles & Game Progress # ========================================== def get_default_profile_dict(): return { "coins": 150, "level": 1, "maxUnlockedLevel": 1, "highScore": 0, "theme": "classic", "upgrades": { "extraSlot": False, "hasRake": False, "unlockedPlants": ["sunflower", "peashooter", "wallnut"], "selectedDeck": ["sunflower", "peashooter", "wallnut"] }, "unlockedPlants": ["sunflower", "peashooter", "wallnut"], "selectedDeck": ["sunflower", "peashooter", "wallnut"], "materials": {"bioHumus": 10, "scrapMetal": 5, "sunEssence": 20, "crystals": 0}, "plantLevels": {"peashooter": 1, "sunflower": 1, "wallnut": 1, "potatomine": 1, "repeater": 1, "snowpea": 1, "chomper": 1}, "perks": {"magnetism": 0, "startSun": 0, "fastCool": 0, "hardyRoots": 0, "luckyDrop": 0}, "zenGarden": { "pots": [ {"id": 1, "plantType": "sunflower", "stage": 2, "watered": True, "fertilized": True, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0}, {"id": 2, "plantType": "peashooter", "stage": 1, "watered": True, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0}, {"id": 3, "plantType": None, "stage": 0, "watered": False, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0}, {"id": 4, "plantType": None, "stage": 0, "watered": False, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0}, {"id": 5, "plantType": None, "stage": 0, "watered": False, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0}, {"id": 6, "plantType": None, "stage": 0, "watered": False, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0}, {"id": 7, "plantType": None, "stage": 0, "watered": False, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0}, {"id": 8, "plantType": None, "stage": 0, "watered": False, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0} ], "fertilizerCount": 5, "chocolateCount": 2, "snail": {"name": "Стинки", "active": False, "speed": 60, "boostTimer": 0, "x": 50, "direction": 1}, "lastVisit": int(datetime.now().timestamp() * 1000) }, "treeOfWisdom": {"height": 0, "cheats": {}}, "achievements": {}, "stats": { "zombiesKilled": 0, "sunProduced": 0, "plantsPlanted": 0, "levelsCompleted": 0, "coinsEarned": 0 }, "casino": { "deposit": 0, "depositStartTime": 0, "loan": 0, "spinsTotal": 0, "totalWon": 0, "jackpotsWon": 0 }, "zombieLab": { "unlockedStage": 1, "mutations": {}, "levels": {"normal": 1, "conehead": 1, "buckethead": 1, "football": 1, "dancing": 1} } } def get_user_by_username(username): conn = get_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM users WHERE username = ? COLLATE NOCASE", (username.strip(),)) user = cursor.fetchone() conn.close() return user def register_user(username, password_hash, initial_data=None, ip_address="", user_agent="", device_name=""): username = username.strip() conn = get_connection() cursor = conn.cursor() try: now = datetime.now() now_str = now.strftime("%Y-%m-%d %H:%M:%S") secure_hash = hash_password(password_hash) cursor.execute( "INSERT INTO users (username, password_hash, created_at, last_login, is_banned, ban_reason, is_admin) VALUES (?, ?, ?, ?, 0, '', 0)", (username, secure_hash, now_str, now_str) ) user_id = cursor.lastrowid profile_data = get_default_profile_dict() if initial_data and isinstance(initial_data, dict): for k, v in initial_data.items(): if v is not None: profile_data[k] = v cursor.execute(""" INSERT INTO profiles ( user_id, coins, level, max_unlocked_level, high_score, theme, upgrades, materials, plant_levels, perks, zen_garden, tree_of_wisdom, achievements, stats, casino, zombie_lab, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( user_id, profile_data.get("coins", 150), profile_data.get("level", 1), profile_data.get("maxUnlockedLevel", 1), profile_data.get("highScore", 0), profile_data.get("theme", "classic"), json.dumps(profile_data.get("upgrades", {})), json.dumps(profile_data.get("materials", {})), json.dumps(profile_data.get("plantLevels", {})), json.dumps(profile_data.get("perks", {})), json.dumps(profile_data.get("zenGarden", {})), json.dumps(profile_data.get("treeOfWisdom", {})), json.dumps(profile_data.get("achievements", {})), json.dumps(profile_data.get("stats", {})), json.dumps(profile_data.get("casino", {})), json.dumps(profile_data.get("zombieLab", {})), now_str )) conn.commit() conn.close() # Create session token with full device / ip tracking token = create_session(user_id, duration_days=30, ip_address=ip_address, user_agent=user_agent, device_name=device_name) record_login_log(user_id, username, ip_address, user_agent, "registered", device_name) return {"success": True, "token": token, "user_id": user_id, "username": username, "data": profile_data} except sqlite3.IntegrityError: conn.close() return {"success": False, "message": "Пользователь с таким именем уже существует"} except Exception as e: conn.close() return {"success": False, "message": str(e)} def authenticate_user(username, password_hash, ip_address="", user_agent="", device_name=""): username = username.strip() conn = get_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM users WHERE username = ? COLLATE NOCASE", (username,)) user = cursor.fetchone() if not user: conn.close() record_login_log(None, username, ip_address, user_agent, "user_not_found", device_name) return None # Check if user is banned if user["is_banned"]: conn.close() reason = user["ban_reason"] or "Нарушение правил сообщества" record_login_log(user["id"], username, ip_address, user_agent, "banned_login_attempt", device_name) return { "success": False, "message": f"Ваш аккаунт заблокирован администратором. Причина: {reason}", "is_banned": True, "ban_reason": reason } if not verify_password(password_hash, user["password_hash"]): conn.close() record_login_log(user["id"], username, ip_address, user_agent, "wrong_password", device_name) return {"success": False, "message": "Неверный логин или пароль"} now = datetime.now() now_str = now.strftime("%Y-%m-%d %H:%M:%S") # Seamless upgrade from legacy plaintext/djb2 hash to salted PBKDF2 if not user["password_hash"].startswith("pbkdf2:"): new_secure_hash = hash_password(password_hash) cursor.execute("UPDATE users SET password_hash = ?, last_login = ? WHERE id = ?", (new_secure_hash, now_str, user["id"])) else: cursor.execute("UPDATE users SET last_login = ? WHERE id = ?", (now_str, user["id"])) conn.commit() # Fetch profile cursor.execute("SELECT * FROM profiles WHERE user_id = ?", (user["id"],)) p = cursor.fetchone() conn.close() # Create session token with full device / ip tracking token = create_session(user["id"], duration_days=30, ip_address=ip_address, user_agent=user_agent, device_name=device_name) record_login_log(user["id"], user["username"], ip_address, user_agent, "success", device_name) if not p: profile_dict = get_default_profile_dict() else: upg_data = json.loads(p["upgrades"] or "{}") unlocked_plants = upg_data.get("unlockedPlants", ["sunflower", "peashooter", "wallnut"]) selected_deck = upg_data.get("selectedDeck", ["sunflower", "peashooter", "wallnut"]) profile_dict = { "coins": p["coins"], "level": p["level"], "maxUnlockedLevel": p["max_unlocked_level"], "highScore": p["high_score"], "theme": p["theme"], "upgrades": upg_data, "unlockedPlants": unlocked_plants, "selectedDeck": selected_deck, "materials": json.loads(p["materials"] or "{}"), "plantLevels": json.loads(p["plant_levels"] or "{}"), "perks": json.loads(p["perks"] or "{}"), "zenGarden": json.loads(p["zen_garden"] or "{}"), "treeOfWisdom": json.loads(p["tree_of_wisdom"] or "{}"), "achievements": json.loads(p["achievements"] or "{}"), "stats": json.loads(p["stats"] or "{}") if "stats" in p.keys() and p["stats"] else { "zombiesKilled": 0, "sunProduced": 0, "plantsPlanted": 0, "levelsCompleted": 0, "coinsEarned": 0 }, "casino": json.loads(p["casino"] or "{}") if "casino" in p.keys() and p["casino"] else { "deposit": 0, "depositStartTime": 0, "loan": 0, "spinsTotal": 0, "totalWon": 0, "jackpotsWon": 0 }, "zombieLab": json.loads(p["zombie_lab"] or "{}") if "zombie_lab" in p.keys() and p["zombie_lab"] else { "unlockedStage": 1, "mutations": {}, "levels": {"normal": 1, "conehead": 1, "buckethead": 1, "football": 1, "dancing": 1} } } return { "success": True, "token": token, "user_id": user["id"], "username": user["username"], "isAdmin": bool(user["is_admin"]), "data": profile_dict } def get_user_profile(username): """Returns the full parsed profile dictionary for a given username.""" conn = get_connection() cursor = conn.cursor() cursor.execute("SELECT id, username, is_admin FROM users WHERE username = ? COLLATE NOCASE", (username.strip(),)) user = cursor.fetchone() if not user: conn.close() return None cursor.execute("SELECT * FROM profiles WHERE user_id = ?", (user["id"],)) p = cursor.fetchone() conn.close() if not p: return get_default_profile_dict() upg_data = json.loads(p["upgrades"] or "{}") unlocked_plants = upg_data.get("unlockedPlants", ["sunflower", "peashooter", "wallnut"]) selected_deck = upg_data.get("selectedDeck", ["sunflower", "peashooter", "wallnut"]) return { "coins": p["coins"], "level": p["level"], "maxUnlockedLevel": p["max_unlocked_level"], "highScore": p["high_score"], "theme": p["theme"] or "classic", "upgrades": upg_data, "unlockedPlants": unlocked_plants, "selectedDeck": selected_deck, "materials": json.loads(p["materials"] or "{}"), "plantLevels": json.loads(p["plant_levels"] or "{}"), "perks": json.loads(p["perks"] or "{}"), "zenGarden": json.loads(p["zen_garden"] or "{}"), "treeOfWisdom": json.loads(p["tree_of_wisdom"] or "{}"), "achievements": json.loads(p["achievements"] or "{}"), "stats": json.loads(p["stats"] or "{}") if "stats" in p.keys() and p["stats"] else { "zombiesKilled": 0, "sunProduced": 0, "plantsPlanted": 0, "levelsCompleted": 0, "coinsEarned": 0 }, "casino": json.loads(p["casino"] or "{}") if "casino" in p.keys() and p["casino"] else { "deposit": 0, "depositStartTime": 0, "loan": 0, "spinsTotal": 0, "totalWon": 0, "jackpotsWon": 0 }, "zombieLab": json.loads(p["zombie_lab"] or "{}") if "zombie_lab" in p.keys() and p["zombie_lab"] else { "unlockedStage": 1, "mutations": {}, "levels": {"normal": 1, "conehead": 1, "buckethead": 1, "football": 1, "dancing": 1} } } def save_user_profile(username, data, token=None, is_admin_override=False, ip_address=None): """Persists user progress profile with strict server-authoritative progress protection.""" username = username.strip() # If token is provided, verify it belongs to username and update its activity if token is not None: session_user = validate_session(token) if not session_user or session_user["username"].lower() != username.lower(): return {"success": False, "message": "Недействительный токен авторизации или доступ запрещен"} if ip_address: update_session_activity(token, ip_address) conn = get_connection() cursor = conn.cursor() cursor.execute("SELECT id, is_banned, ban_reason FROM users WHERE username = ? COLLATE NOCASE", (username,)) user = cursor.fetchone() if not user: conn.close() return {"success": False, "message": "Пользователь не найден"} if user["is_banned"]: conn.close() return {"success": False, "message": f"Аккаунт заблокирован: {user['ban_reason'] or 'Нарушение правил'}"} user_id = user["id"] cursor.execute("SELECT * FROM profiles WHERE user_id = ?", (user_id,)) current = cursor.fetchone() now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # Safe progress progression: never downgrade unlocked level or high score from regular client saves cur_coins = current["coins"] if current else 150 cur_lvl = current["level"] if current else 1 cur_max_lvl = current["max_unlocked_level"] if current else 1 cur_high_score = current["high_score"] if current else 0 cur_theme = current["theme"] if current else "classic" has_admin_updated = bool(current and "admin_updated_at" in current.keys() and current["admin_updated_at"]) if is_admin_override: coins = int(data.get("coins", cur_coins)) level = int(data.get("level", cur_lvl)) max_unlocked = int(data.get("maxUnlockedLevel", data.get("max_unlocked_level", cur_max_lvl))) high_score = int(data.get("highScore", data.get("high_score", cur_high_score))) theme = str(data.get("theme", cur_theme)) admin_updated_val = now_str else: req_coins = int(data.get("coins", cur_coins)) # Protect admin-granted coins from being overwritten by stale client auto-saves on refresh if has_admin_updated and req_coins < cur_coins: coins = cur_coins else: coins = req_coins level = int(data.get("level", cur_lvl)) req_max = int(data.get("maxUnlockedLevel", data.get("max_unlocked_level", cur_max_lvl))) max_unlocked = max(req_max, cur_max_lvl, level) req_score = int(data.get("highScore", data.get("high_score", cur_high_score))) high_score = max(req_score, cur_high_score) theme = str(data.get("theme", cur_theme)) admin_updated_val = current["admin_updated_at"] if current and "admin_updated_at" in current.keys() else None # Merge upgrades & plants upg_obj = json.loads(current["upgrades"] or "{}") if current and current["upgrades"] else {} if "upgrades" in data and isinstance(data["upgrades"], dict): upg_obj.update(data["upgrades"]) if "unlockedPlants" in data and isinstance(data["unlockedPlants"], list): cur_unlocked = upg_obj.get("unlockedPlants", ["sunflower", "peashooter", "wallnut"]) upg_obj["unlockedPlants"] = list(dict.fromkeys(cur_unlocked + data["unlockedPlants"])) if "selectedDeck" in data and isinstance(data["selectedDeck"], list): upg_obj["selectedDeck"] = data["selectedDeck"] upgrades_json = json.dumps(upg_obj) # Materials merge (protect admin-credited materials from stale client downgrade) cur_materials = json.loads(current["materials"] or "{}") if current and current["materials"] else {"bioHumus": 10, "scrapMetal": 5, "sunEssence": 20, "crystals": 0} if "materials" in data and isinstance(data["materials"], dict): for mk, mv in data["materials"].items(): val = int(mv) if is_admin_override: cur_materials[mk] = val else: if has_admin_updated and val < cur_materials.get(mk, 0): pass else: cur_materials[mk] = val materials_json = json.dumps(cur_materials) # Plant Levels merge (never downgrade plant levels) cur_plant_levels = json.loads(current["plant_levels"] or "{}") if current and current["plant_levels"] else {"peashooter": 1, "sunflower": 1, "wallnut": 1, "potatomine": 1, "repeater": 1, "snowpea": 1, "chomper": 1} new_plant_levels = data.get("plantLevels") or data.get("plant_levels") if new_plant_levels and isinstance(new_plant_levels, dict): for plant_k, plant_lvl in new_plant_levels.items(): if is_admin_override: cur_plant_levels[plant_k] = int(plant_lvl) else: cur_plant_levels[plant_k] = max(cur_plant_levels.get(plant_k, 1), int(plant_lvl)) plant_levels_json = json.dumps(cur_plant_levels) # Perks merge (never downgrade perks for client) cur_perks = json.loads(current["perks"] or "{}") if current and current["perks"] else {"magnetism": 0, "startSun": 0, "fastCool": 0, "hardyRoots": 0, "luckyDrop": 0} new_perks = data.get("perks") if new_perks and isinstance(new_perks, dict): for perk_k, perk_v in new_perks.items(): if is_admin_override: cur_perks[perk_k] = int(perk_v) else: cur_perks[perk_k] = max(cur_perks.get(perk_k, 0), int(perk_v)) perks_json = json.dumps(cur_perks) # Zen Garden deep merge (never wipe pots or snail when admin edits items) cur_zen = json.loads(current["zen_garden"] or "{}") if current and current["zen_garden"] else {} if "zenGarden" in data and isinstance(data["zenGarden"], dict): new_zen = data["zenGarden"] for zk, zv in new_zen.items(): if zk == "snail" and isinstance(zv, dict) and isinstance(cur_zen.get("snail"), dict): cur_zen["snail"].update(zv) else: cur_zen[zk] = zv zen_garden_json = json.dumps(cur_zen) else: zen_garden_json = current["zen_garden"] if current else "{}" tree_json = json.dumps(data["treeOfWisdom"]) if "treeOfWisdom" in data else (current["tree_of_wisdom"] if current else "{}") achievements_json = json.dumps(data["achievements"]) if "achievements" in data else (current["achievements"] if current else "{}") # Stats merge cur_stats = json.loads(current["stats"] or "{}") if current and "stats" in current.keys() and current["stats"] else { "zombiesKilled": 0, "sunProduced": 0, "plantsPlanted": 0, "levelsCompleted": 0, "coinsEarned": 0 } if "stats" in data and isinstance(data["stats"], dict): for sk, sv in data["stats"].items(): if is_admin_override: cur_stats[sk] = int(sv) else: cur_stats[sk] = max(cur_stats.get(sk, 0), int(sv)) stats_json = json.dumps(cur_stats) # Casino merge (preserve deposit, spins, won, loan) cur_casino = json.loads(current["casino"] or "{}") if current and "casino" in current.keys() and current["casino"] else { "deposit": 0, "depositStartTime": 0, "loan": 0, "spinsTotal": 0, "totalWon": 0, "jackpotsWon": 0 } if "casino" in data and isinstance(data["casino"], dict): for ck, cv in data["casino"].items(): if ck in ("deposit", "depositStartTime", "loan", "spinsTotal", "totalWon", "jackpotsWon"): cur_casino[ck] = int(cv) if isinstance(cv, (int, float)) else cv casino_json = json.dumps(cur_casino) # Zombie Lab merge cur_zombie_lab = json.loads(current["zombie_lab"] or "{}") if current and "zombie_lab" in current.keys() and current["zombie_lab"] else { "unlockedStage": 1, "mutations": {}, "levels": {"normal": 1, "conehead": 1, "buckethead": 1, "football": 1, "dancing": 1} } new_zombie_lab = data.get("zombieLab") or data.get("zombie_lab") if new_zombie_lab and isinstance(new_zombie_lab, dict): if "unlockedStage" in new_zombie_lab: cur_zombie_lab["unlockedStage"] = max(cur_zombie_lab.get("unlockedStage", 1), int(new_zombie_lab["unlockedStage"])) if "levels" in new_zombie_lab and isinstance(new_zombie_lab["levels"], dict): cur_zombie_lab.setdefault("levels", {}).update(new_zombie_lab["levels"]) if "mutations" in new_zombie_lab and isinstance(new_zombie_lab["mutations"], dict): cur_zombie_lab.setdefault("mutations", {}).update(new_zombie_lab["mutations"]) zombie_lab_json = json.dumps(cur_zombie_lab) if current: cursor.execute(""" UPDATE profiles SET coins = ?, level = ?, max_unlocked_level = ?, high_score = ?, theme = ?, upgrades = ?, materials = ?, plant_levels = ?, perks = ?, zen_garden = ?, tree_of_wisdom = ?, achievements = ?, stats = ?, casino = ?, zombie_lab = ?, admin_updated_at = ?, updated_at = ? WHERE user_id = ? """, ( coins, level, max_unlocked, high_score, theme, upgrades_json, materials_json, plant_levels_json, perks_json, zen_garden_json, tree_json, achievements_json, stats_json, casino_json, zombie_lab_json, admin_updated_val, now_str, user_id )) else: cursor.execute(""" INSERT INTO profiles ( user_id, coins, level, max_unlocked_level, high_score, theme, upgrades, materials, plant_levels, perks, zen_garden, tree_of_wisdom, achievements, stats, casino, zombie_lab, admin_updated_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( user_id, coins, level, max_unlocked, high_score, theme, upgrades_json, materials_json, plant_levels_json, perks_json, zen_garden_json, tree_json, achievements_json, stats_json, casino_json, zombie_lab_json, admin_updated_val, now_str )) conn.commit() conn.close() return {"success": True, "data": get_user_profile(username)} # ========================================== # Leaderboards & Bug Tracking # ========================================== def submit_leaderboard_score(user_id, player_name, mode, score, detail=""): """Submits or updates a player's high score on the server leaderboard.""" if not player_name or not mode or score is None: return {"success": False, "message": "Некорректные параметры счета"} clean_name = str(player_name).strip()[:30] clean_mode = str(mode).strip()[:20] clean_detail = str(detail).strip()[:100] score_val = int(score) conn = get_connection() cursor = conn.cursor() if user_id: cursor.execute("SELECT id, score FROM leaderboards WHERE user_id = ? AND mode = ?", (user_id, clean_mode)) else: cursor.execute("SELECT id, score FROM leaderboards WHERE player_name = ? AND mode = ?", (clean_name, clean_mode)) row = cursor.fetchone() now_str = datetime.now().strftime("%Y-%m-%d") if row: if score_val > row["score"]: cursor.execute( "UPDATE leaderboards SET score = ?, detail = ?, player_name = ?, created_at = ? WHERE id = ?", (score_val, clean_detail, clean_name, now_str, row["id"]) ) else: cursor.execute( "INSERT INTO leaderboards (user_id, player_name, mode, score, detail, created_at) VALUES (?, ?, ?, ?, ?, ?)", (user_id, clean_name, clean_mode, score_val, clean_detail, now_str) ) conn.commit() conn.close() return {"success": True} def get_leaderboard_scores(mode=None, limit=10): """Retrieves top leaderboard entries for a specific mode or all modes.""" conn = get_connection() cursor = conn.cursor() if mode: cursor.execute(""" SELECT id, user_id, player_name, mode, score, detail, created_at FROM leaderboards WHERE mode = ? ORDER BY score DESC LIMIT ? """, (mode, limit)) rows = cursor.fetchall() result = [{ "id": r["id"], "name": r["player_name"], "mode": r["mode"], "score": r["score"], "detail": r["detail"] or f"{r['score']} очков", "date": str(r["created_at"])[:10] } for r in rows] else: modes = ["adventure", "survival", "roguelike", "izombie", "bowling", "boss", "bossrush"] result = {} for m in modes: cursor.execute(""" SELECT id, user_id, player_name, mode, score, detail, created_at FROM leaderboards WHERE mode = ? ORDER BY score DESC LIMIT ? """, (m, limit)) rows = cursor.fetchall() result[m] = [{ "id": r["id"], "name": r["player_name"], "mode": r["mode"], "score": r["score"], "detail": r["detail"] or f"{r['score']} очков", "date": str(r["created_at"])[:10] } for r in rows] conn.close() return result def list_all_users(): conn = get_connection() cursor = conn.cursor() cursor.execute(""" SELECT u.id, u.username, u.created_at, u.last_login, u.is_banned, u.ban_reason, u.is_admin, p.coins, p.level, p.max_unlocked_level, p.high_score, p.theme, p.materials, p.plant_levels, p.perks, p.upgrades, p.zen_garden, p.tree_of_wisdom, p.achievements, p.stats FROM users u LEFT JOIN profiles p ON u.id = p.user_id ORDER BY u.id ASC """) rows = cursor.fetchall() conn.close() result = [] for r in rows: result.append({ "id": r["id"], "username": r["username"], "createdAt": r["created_at"], "lastLogin": r["last_login"], "isBanned": bool(r["is_banned"]), "banReason": r["ban_reason"] or "", "isAdmin": bool(r["is_admin"]), "coins": r["coins"] if r["coins"] is not None else 150, "level": r["level"] if r["level"] is not None else 1, "maxUnlockedLevel": r["max_unlocked_level"] if r["max_unlocked_level"] is not None else 1, "highScore": r["high_score"] if r["high_score"] is not None else 0, "theme": r["theme"] or "classic", "materials": json.loads(r["materials"] or "{}"), "plantLevels": json.loads(r["plant_levels"] or "{}"), "perks": json.loads(r["perks"] or "{}"), "upgrades": json.loads(r["upgrades"] or "{}"), "zenGarden": json.loads(r["zen_garden"] or "{}"), "treeOfWisdom": json.loads(r["tree_of_wisdom"] or "{}"), "achievements": json.loads(r["achievements"] or "{}"), "stats": json.loads(r["stats"] or "{}") if "stats" in r.keys() and r["stats"] else { "zombiesKilled": 0, "sunProduced": 0, "plantsPlanted": 0, "levelsCompleted": 0, "coinsEarned": 0 } }) return result # ========================================== # Game Configurator & Balance System # ========================================== def get_game_config(): """Retrieves active game balance configuration and presets.""" conn = get_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM game_config ORDER BY id DESC LIMIT 1") row = cursor.fetchone() conn.close() if not row: return DEFAULT_GAME_CONFIG.copy() try: data = json.loads(row["config_data"]) data["preset"] = row["preset_name"] or data.get("preset", "Classic 2009") data["updatedAt"] = row["updated_at"] return data except Exception: return DEFAULT_GAME_CONFIG.copy() def save_game_config(config_dict, preset_name=None): """Persists updated balance configuration to database.""" if not isinstance(config_dict, dict): raise ValueError("Config must be a valid dictionary") preset = preset_name or config_dict.get("preset", "Custom") config_dict["preset"] = preset now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") conn = get_connection() cursor = conn.cursor() cursor.execute( "INSERT INTO game_config (preset_name, config_data, updated_at) VALUES (?, ?, ?)", (preset, json.dumps(config_dict, ensure_ascii=False), now_str) ) conn.commit() conn.close() config_dict["updatedAt"] = now_str return config_dict # ========================================== # Admin Management Functions # ========================================== def admin_list_users(): """Returns list of all players with full account and profile details.""" return list_all_users() def admin_get_user_detail(user_id): """Fetches full player details, profile, active sessions, and leaderboard scores.""" conn = get_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,)) user = cursor.fetchone() if not user: conn.close() return None cursor.execute("SELECT * FROM profiles WHERE user_id = ?", (user_id,)) p = cursor.fetchone() cursor.execute("SELECT token, created_at, expires_at, ip_address, device_name, last_active FROM sessions WHERE user_id = ?", (user_id,)) sessions = [{ "token": s["token"], "createdAt": s["created_at"], "expiresAt": s["expires_at"], "ipAddress": s["ip_address"] or "", "deviceName": s["device_name"] or "Unknown Device", "lastActive": s["last_active"] or s["created_at"] } for s in cursor.fetchall()] cursor.execute("SELECT id, mode, score, detail, created_at FROM leaderboards WHERE user_id = ? ORDER BY score DESC LIMIT 10", (user_id,)) scores = [{"id": l["id"], "mode": l["mode"], "score": l["score"], "detail": l["detail"], "createdAt": l["created_at"]} for l in cursor.fetchall()] conn.close() if p: profile_dict = { "coins": p["coins"], "level": p["level"], "maxUnlockedLevel": p["max_unlocked_level"], "highScore": p["high_score"], "theme": p["theme"], "upgrades": json.loads(p["upgrades"] or "{}"), "materials": json.loads(p["materials"] or "{}"), "plantLevels": json.loads(p["plant_levels"] or "{}"), "perks": json.loads(p["perks"] or "{}"), "zenGarden": json.loads(p["zen_garden"] or "{}"), "treeOfWisdom": json.loads(p["tree_of_wisdom"] or "{}"), "achievements": json.loads(p["achievements"] or "{}"), "stats": json.loads(p["stats"] or "{}") if "stats" in p.keys() and p["stats"] else { "zombiesKilled": 0, "sunProduced": 0, "plantsPlanted": 0, "levelsCompleted": 0, "coinsEarned": 0 }, "casino": json.loads(p["casino"] or "{}") if "casino" in p.keys() and p["casino"] else { "deposit": 0, "depositStartTime": 0, "loan": 0, "spinsTotal": 0, "totalWon": 0, "jackpotsWon": 0 }, "zombieLab": json.loads(p["zombie_lab"] or "{}") if "zombie_lab" in p.keys() and p["zombie_lab"] else { "unlockedStage": 1, "mutations": {}, "levels": {"normal": 1, "conehead": 1, "buckethead": 1, "football": 1, "dancing": 1} }, "updatedAt": p["updated_at"] } else: profile_dict = get_default_profile_dict() return { "id": user["id"], "username": user["username"], "createdAt": user["created_at"], "lastLogin": user["last_login"], "isBanned": bool(user["is_banned"]), "banReason": user["ban_reason"] or "", "isAdmin": bool(user["is_admin"]), "coins": profile_dict["coins"], "level": profile_dict["level"], "maxUnlockedLevel": profile_dict["maxUnlockedLevel"], "highScore": profile_dict["highScore"], "theme": profile_dict["theme"], "materials": profile_dict["materials"], "stats": profile_dict.get("stats", {}), "zenGarden": profile_dict["zenGarden"], "profile": profile_dict, "sessions": sessions, "scores": scores } def admin_update_user_profile(user_id, updates): """Updates user profile and account details (coins, level, materials, perks, etc.).""" conn = get_connection() cursor = conn.cursor() cursor.execute("SELECT username FROM users WHERE id = ?", (user_id,)) u = cursor.fetchone() if not u: conn.close() return {"success": False, "message": "Пользователь не найден"} username = u["username"] conn.close() conn = get_connection() cursor = conn.cursor() if "isBanned" in updates or "is_banned" in updates: is_banned = 1 if updates.get("isBanned", updates.get("is_banned")) else 0 ban_reason = updates.get("banReason", updates.get("ban_reason", "")) cursor.execute("UPDATE users SET is_banned = ?, ban_reason = ? WHERE id = ?", (is_banned, ban_reason, user_id)) if is_banned: cursor.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,)) if "isAdmin" in updates or "is_admin" in updates: is_admin = 1 if updates.get("isAdmin", updates.get("is_admin")) else 0 cursor.execute("UPDATE users SET is_admin = ? WHERE id = ?", (is_admin, user_id)) conn.commit() conn.close() res = save_user_profile(username, updates, is_admin_override=True) return res def admin_set_user_ban(user_id, is_banned, reason=""): """Bans or unbans a player and clears sessions if banned.""" conn = get_connection() cursor = conn.cursor() cursor.execute("SELECT id FROM users WHERE id = ?", (user_id,)) if not cursor.fetchone(): conn.close() return {"success": False, "message": "Пользователь не найден"} ban_val = 1 if is_banned else 0 cursor.execute("UPDATE users SET is_banned = ?, ban_reason = ? WHERE id = ?", (ban_val, str(reason or "").strip(), user_id)) if ban_val == 1: cursor.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,)) conn.commit() conn.close() return {"success": True, "isBanned": bool(ban_val), "banReason": reason} def admin_delete_user(user_id): """Deletes user and cascades to profile, sessions, and leaderboards.""" conn = get_connection() cursor = conn.cursor() cursor.execute("SELECT id FROM users WHERE id = ?", (user_id,)) if not cursor.fetchone(): conn.close() return {"success": False, "message": "Пользователь не найден"} cursor.execute("DELETE FROM profiles WHERE user_id = ?", (user_id,)) cursor.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,)) cursor.execute("DELETE FROM leaderboards WHERE user_id = ?", (user_id,)) cursor.execute("DELETE FROM users WHERE id = ?", (user_id,)) conn.commit() conn.close() return {"success": True, "message": "Пользователь успешно удален"} def admin_create_user(username, password, initial_profile=None, is_admin=0): """Creates a new user with administrative parameters.""" res = register_user(username, password, initial_profile) if not res.get("success"): return res if is_admin: conn = get_connection() cursor = conn.cursor() cursor.execute("UPDATE users SET is_admin = 1 WHERE id = ?", (res["user_id"],)) conn.commit() conn.close() res["isAdmin"] = True return res def admin_reset_user_password(user_id, new_password): """Resets user's password and revokes all current sessions.""" if not new_password: return {"success": False, "message": "Пароль не может быть пустым"} conn = get_connection() cursor = conn.cursor() cursor.execute("SELECT id FROM users WHERE id = ?", (user_id,)) if not cursor.fetchone(): conn.close() return {"success": False, "message": "Пользователь не найден"} secure_hash = hash_password(new_password) cursor.execute("UPDATE users SET password_hash = ? WHERE id = ?", (secure_hash, user_id)) cursor.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,)) conn.commit() conn.close() return {"success": True, "message": "Пароль успешно сброшен"} def admin_get_stats(): """Calculates comprehensive dashboard statistics.""" conn = get_connection() cursor = conn.cursor() # Total users cursor.execute("SELECT COUNT(*) as cnt FROM users") total_users = cursor.fetchone()["cnt"] # Active in last 24h now = datetime.now() yesterday_str = (now - timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S") cursor.execute("SELECT COUNT(*) as cnt FROM users WHERE last_login >= ?", (yesterday_str,)) active_24h = cursor.fetchone()["cnt"] # Banned users cursor.execute("SELECT COUNT(*) as cnt FROM users WHERE is_banned = 1") banned_users = cursor.fetchone()["cnt"] # Total coins in economy cursor.execute("SELECT SUM(coins) as total_coins FROM profiles") row_coins = cursor.fetchone() total_coins = row_coins["total_coins"] or 0 # Max level reached across players cursor.execute("SELECT MAX(max_unlocked_level) as max_lvl, MAX(high_score) as max_score FROM profiles") row_max = cursor.fetchone() max_level = row_max["max_lvl"] or 1 max_high_score = row_max["max_score"] or 0 # Active sessions now_str = now.strftime("%Y-%m-%d %H:%M:%S") cursor.execute("SELECT COUNT(*) as cnt FROM sessions WHERE expires_at > ?", (now_str,)) active_sessions = cursor.fetchone()["cnt"] # Total login log events cursor.execute("SELECT COUNT(*) as cnt FROM login_logs") total_logins = cursor.fetchone()["cnt"] # Bug report counts cursor.execute("SELECT status, COUNT(*) as cnt FROM bugs GROUP BY status") bug_counts = {r["status"]: r["cnt"] for r in cursor.fetchall()} total_bugs = sum(bug_counts.values()) # Campaign levels count cursor.execute("SELECT COUNT(*) as cnt FROM campaign_levels") campaign_count = cursor.fetchone()["cnt"] # Aggregate gameplay stats across players cursor.execute("SELECT stats FROM profiles WHERE stats IS NOT NULL") tot_zombies = 0 tot_sun = 0 tot_plants = 0 tot_levels_cleared = 0 tot_coins_earned = 0 for r in cursor.fetchall(): try: st = json.loads(r["stats"] or "{}") if "stats" in r.keys() and r["stats"] else {} tot_zombies += int(st.get("zombiesKilled", 0)) tot_sun += int(st.get("sunProduced", 0)) tot_plants += int(st.get("plantsPlanted", 0)) tot_levels_cleared += int(st.get("levelsCompleted", 0)) tot_coins_earned += int(st.get("coinsEarned", 0)) except Exception: pass conn.close() active_config = get_game_config() return { "totalUsers": total_users, "active24h": active_24h, "bannedUsers": banned_users, "totalCoins": total_coins, "maxLevelReached": max_level, "maxHighScore": max_high_score, "activeSessions": active_sessions, "totalLoginLogs": total_logins, "campaignLevelsCount": campaign_count, "totalZombiesKilled": tot_zombies, "totalSunProduced": tot_sun, "totalPlantsPlanted": tot_plants, "totalLevelsCompleted": tot_levels_cleared, "totalCoinsEarned": tot_coins_earned, "bugReports": { "total": total_bugs, "new": bug_counts.get("новый", 0), "inProgress": bug_counts.get("в работе", 0), "fixed": bug_counts.get("исправлено", 0), "rejected": bug_counts.get("отклонено", 0) }, "gameConfig": { "preset": active_config.get("preset", "Classic 2009"), "events": active_config.get("events", {}), "updatedAt": active_config.get("updatedAt") } } def admin_export_db(): """Dumps all database tables into a JSON structure.""" conn = get_connection() cursor = conn.cursor() # Users cursor.execute("SELECT id, username, password_hash, created_at, last_login, is_banned, ban_reason, is_admin FROM users") users = [dict(r) for r in cursor.fetchall()] # Profiles cursor.execute("SELECT * FROM profiles") profiles = [dict(r) for r in cursor.fetchall()] # Leaderboards cursor.execute("SELECT * FROM leaderboards") leaderboards = [dict(r) for r in cursor.fetchall()] # Bugs cursor.execute("SELECT * FROM bugs") bugs = [dict(r) for r in cursor.fetchall()] # Game Config cursor.execute("SELECT * FROM game_config") game_configs = [dict(r) for r in cursor.fetchall()] # Campaign Levels cursor.execute("SELECT * FROM campaign_levels") campaign_levels = [dict(r) for r in cursor.fetchall()] # Admin Settings cursor.execute("SELECT * FROM admin_settings") admin_settings = [dict(r) for r in cursor.fetchall()] # Login Logs (latest 500) cursor.execute("SELECT * FROM login_logs ORDER BY id DESC LIMIT 500") login_logs = [dict(r) for r in cursor.fetchall()] conn.close() return { "exportedAt": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "version": "3.0", "users": users, "profiles": profiles, "leaderboards": leaderboards, "bugs": bugs, "gameConfigs": game_configs, "campaignLevels": campaign_levels, "adminSettings": admin_settings, "loginLogs": login_logs } def admin_import_db(data): """Restores database tables from a JSON export dictionary.""" if not isinstance(data, dict): return {"success": False, "message": "Неверный формат данных бэкапа"} conn = get_connection() cursor = conn.cursor() try: cursor.execute("BEGIN TRANSACTION") # Import Users if "users" in data and isinstance(data["users"], list): cursor.execute("DELETE FROM users") cursor.execute("DELETE FROM sessions") for u in data["users"]: cursor.execute(""" INSERT INTO users (id, username, password_hash, created_at, last_login, is_banned, ban_reason, is_admin) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( u.get("id"), u.get("username"), u.get("password_hash"), u.get("created_at"), u.get("last_login"), u.get("is_banned", 0), u.get("ban_reason", ""), u.get("is_admin", 0) )) # Import Profiles if "profiles" in data and isinstance(data["profiles"], list): cursor.execute("DELETE FROM profiles") for p in data["profiles"]: cursor.execute(""" INSERT INTO profiles ( user_id, coins, level, max_unlocked_level, high_score, theme, upgrades, materials, plant_levels, perks, zen_garden, tree_of_wisdom, achievements, stats, casino, zombie_lab, admin_updated_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( p.get("user_id"), p.get("coins", 150), p.get("level", 1), p.get("max_unlocked_level", 1), p.get("high_score", 0), p.get("theme", "classic"), p.get("upgrades", "{}"), p.get("materials", "{}"), p.get("plant_levels", "{}"), p.get("perks", "{}"), p.get("zen_garden", "{}"), p.get("tree_of_wisdom", "{}"), p.get("achievements", "{}"), p.get("stats", "{}"), p.get("casino", "{}"), p.get("zombie_lab", "{}"), p.get("admin_updated_at"), p.get("updated_at") )) # Import Leaderboards if "leaderboards" in data and isinstance(data["leaderboards"], list): cursor.execute("DELETE FROM leaderboards") for l in data["leaderboards"]: cursor.execute(""" INSERT INTO leaderboards (id, user_id, player_name, mode, score, detail, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( l.get("id"), l.get("user_id"), l.get("player_name"), l.get("mode"), l.get("score", 0), l.get("detail", ""), l.get("created_at") )) # Import Bugs if "bugs" in data and isinstance(data["bugs"], list): cursor.execute("DELETE FROM bugs") for b in data["bugs"]: cursor.execute(""" INSERT INTO bugs (id, author, category, text, status, timestamp, admin_reply) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( b.get("id"), b.get("author"), b.get("category", "bug"), b.get("text", ""), b.get("status", "новый"), b.get("timestamp"), b.get("admin_reply") or b.get("adminReply") )) # Import Game Config if "gameConfigs" in data and isinstance(data["gameConfigs"], list): cursor.execute("DELETE FROM game_config") for gc in data["gameConfigs"]: cursor.execute(""" INSERT INTO game_config (id, preset_name, config_data, updated_at) VALUES (?, ?, ?, ?) """, ( gc.get("id"), gc.get("preset_name", "Classic 2009"), gc.get("config_data"), gc.get("updated_at") )) # Import Campaign Levels if "campaignLevels" in data and isinstance(data["campaignLevels"], list): cursor.execute("DELETE FROM campaign_levels") for cl in data["campaignLevels"]: cursor.execute(""" INSERT INTO campaign_levels ( level, world, stage, code, title, desc, reward, biome, is_boss, waves_json, modifiers_json, dialogue_json, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( cl.get("level"), cl.get("world"), cl.get("stage"), cl.get("code"), cl.get("title"), cl.get("desc"), cl.get("reward", 100), cl.get("biome", "day"), cl.get("is_boss", 0), cl.get("waves_json", "{}"), cl.get("modifiers_json", "{}"), cl.get("dialogue_json", "{}"), cl.get("updated_at") )) # Import Admin Settings if "adminSettings" in data and isinstance(data["adminSettings"], list): cursor.execute("DELETE FROM admin_settings") for ast in data["adminSettings"]: cursor.execute(""" INSERT INTO admin_settings (key, value, updated_at) VALUES (?, ?, ?) """, ( ast.get("key"), ast.get("value"), ast.get("updated_at") )) conn.commit() conn.close() return {"success": True, "message": "База данных успешно восстановлена из бэкапа"} except Exception as e: conn.rollback() conn.close() return {"success": False, "message": f"Ошибка восстановления: {str(e)}"} if __name__ == "__main__": init_db() users = list_all_users() config = get_game_config() levels = get_campaign_levels() print(f"Database initialized. Total users: {len(users)}, Levels: {len(levels)}, Active config preset: {config.get('preset')}")