#!/usr/bin/env python3 """ SQLite Database Manager for Plants vs. Zombies Game (ai.qpid.pro) Zero external dependencies. Manages users, game progress profiles, leaderboards, and bug reports. """ import sqlite3 import json import os from datetime import datetime 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) def get_connection(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): conn = get_connection() cursor = conn.cursor() 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 ) """) 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 '{}', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) """) 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 ) """) 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 ) """) conn.commit() conn.close() 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": {} } 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): username = username.strip() conn = get_connection() cursor = conn.cursor() try: now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") cursor.execute( "INSERT INTO users (username, password_hash, created_at, last_login) VALUES (?, ?, ?, ?)", (username, password_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, 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", {})), now_str )) conn.commit() conn.close() return {"success": True, "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): 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() return None if user["password_hash"] != password_hash: conn.close() return {"success": False, "message": "Неверный пароль"} # Update last login now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") 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() 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 "{}") } return {"success": True, "user_id": user["id"], "username": user["username"], "data": profile_dict} def save_user_profile(username, data): username = username.strip() conn = get_connection() cursor = conn.cursor() cursor.execute("SELECT id FROM users WHERE username = ? COLLATE NOCASE", (username,)) user = cursor.fetchone() if not user: conn.close() return {"success": False, "message": "Пользователь не найден"} 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") coins = data.get("coins", current["coins"] if current else 150) level = data.get("level", current["level"] if current else 1) max_unlocked = data.get("maxUnlockedLevel", current["max_unlocked_level"] if current else 1) high_score = data.get("highScore", current["high_score"] if current else 0) theme = data.get("theme", current["theme"] if current else "classic") # Merge JSON dictionaries def merge_json(key, default): cur_obj = json.loads(current[key] or "{}") if current and current[key] else default new_val = data.get(key) if new_val is not None and isinstance(new_val, dict): cur_obj.update(new_val) return json.dumps(cur_obj) 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): upg_obj["unlockedPlants"] = data["unlockedPlants"] if "selectedDeck" in data and isinstance(data["selectedDeck"], list): upg_obj["selectedDeck"] = data["selectedDeck"] upgrades_json = json.dumps(upg_obj) materials_json = merge_json("materials", {}) plant_levels_json = merge_json("plant_levels", {}) if "plant_levels" in data else (json.dumps(data["plantLevels"]) if "plantLevels" in data else (current["plant_levels"] if current else "{}")) perks_json = merge_json("perks", {}) zen_garden_json = json.dumps(data["zenGarden"]) if "zenGarden" in data else (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 "{}") 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 = ?, 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, 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, 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, now_str )) conn.commit() conn.close() return {"success": True} def list_all_users(): conn = get_connection() cursor = conn.cursor() cursor.execute(""" SELECT u.id, u.username, u.created_at, u.last_login, 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 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"], "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 "{}") }) return result if __name__ == "__main__": init_db() users = list_all_users() print(f"Database initialized. Total registered users: {len(users)}")