#!/usr/bin/env python3 """ Backend Server for Plants vs. Zombies (ai.qpid.pro) Zero external dependencies. Manages: - SQLite User Accounts & Game Progression Profiles (/api/auth/login, /api/auth/register, /api/auth/save, /api/auth/sync, /api/users) - Community Bug Reports & Feedback (/api/bugs) """ import http.server import json import os import sys import html import urllib.parse from datetime import datetime BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, os.path.join(BASE_DIR, "scripts")) import db PORT = 8001 MAX_BODY_SIZE = 1 * 1024 * 1024 # 1MB REPORTS_DIR = os.path.join(BASE_DIR, "reports") BUGS_JSON = os.path.join(REPORTS_DIR, "bugs.json") BUGS_MD = os.path.join(BASE_DIR, "BUG_REPORTS.md") os.makedirs(REPORTS_DIR, exist_ok=True) db.init_db() # In-memory rate limiter for failed login attempts FAILED_LOGIN_ATTEMPTS = {} # ip -> list of timestamps def is_login_rate_limited(ip, max_attempts=5, window_seconds=60): now = datetime.now().timestamp() attempts = [t for t in FAILED_LOGIN_ATTEMPTS.get(ip, []) if now - t < window_seconds] FAILED_LOGIN_ATTEMPTS[ip] = attempts return len(attempts) >= max_attempts def record_failed_login(ip): now = datetime.now().timestamp() attempts = [t for t in FAILED_LOGIN_ATTEMPTS.get(ip, []) if now - t < 60] attempts.append(now) FAILED_LOGIN_ATTEMPTS[ip] = attempts def clear_failed_logins(ip): FAILED_LOGIN_ATTEMPTS.pop(ip, None) DEFAULT_BUGS = [ { "id": 1, "author": "Player_Kopri", "category": "bug", "text": "Зомби после поедания растения не шел дальше, а оставался стоять на месте.", "status": "исправлено", "timestamp": "2026-09-16 09:30", "adminReply": "Исправлено в v1.6: сброшен флаг isEating и таймер поедания." }, { "id": 2, "author": "DaveFan", "category": "bug", "text": "Режим 'Я, Зомби' не работал: сразу после старта выводилось сообщение 'победа'.", "status": "исправлено", "timestamp": "2026-09-16 09:45", "adminReply": "Исправлено в v1.6: устранено состояние гонки при инициализации газона." }, { "id": 3, "author": "ZombossHater", "category": "visual", "text": "Элементы главного меню и кнопки вываливаются за границы экрана на небольших дисплеях.", "status": "исправлено", "timestamp": "2026-09-16 10:50", "adminReply": "Исправлено в v2.2: адаптивная панель навигации, компактные карточки и вертикальный скролл." } ] def load_bugs(): if not os.path.exists(BUGS_JSON): save_bugs(DEFAULT_BUGS) return DEFAULT_BUGS try: with open(BUGS_JSON, "r", encoding="utf-8") as f: return json.load(f) except Exception: return DEFAULT_BUGS def save_bugs(bugs): with open(BUGS_JSON, "w", encoding="utf-8") as f: json.dump(bugs, f, ensure_ascii=False, indent=2) sync_markdown(bugs) def sync_markdown(bugs): lines = [ "# Журнал баг-репортов и обратной связи от игроков", "", "> Данный файл обновляется автоматически при поступлении новых баг-репортов из игры.", "> При каждом обновлении и новой итерации разработки ассистент анализирует этот список.", "", f"**Всего репортов:** {len(bugs)} | **Последнее обновление:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", "", "| ID | Дата | Автор | Категория | Статус | Описание | Ответ / Фикс |", "|:---:|:---:|:---|:---:|:---:|:---|:---|" ] for b in reversed(bugs): bid = b.get("id", 0) ts = b.get("timestamp", "") author = str(b.get("author", "Аноним")).replace("|", "/") cat = b.get("category", "bug") cat_icon = "🔴 Баг" if cat == "bug" else ("💡 Идея" if cat == "feature" else ("🎨 UI" if cat == "visual" else "⚖️ Баланс")) st = b.get("status", "новый") st_badge = "✅ Исправлено" if st == "исправлено" else ("🟡 В работе" if st == "в работе" else "🔴 Новый") text = str(b.get("text", "")).replace("|", "/").replace("\n", " ") reply = str(b.get("adminReply") or "—").replace("|", "/").replace("\n", " ") lines.append(f"| {bid} | {ts} | {author} | {cat_icon} | {st_badge} | {text} | {reply} |") lines.append("") with open(BUGS_MD, "w", encoding="utf-8") as f: f.write("\n".join(lines)) class FeedbackHandler(http.server.BaseHTTPRequestHandler): def get_client_ip(self): xff = self.headers.get("X-Forwarded-For") if xff: return xff.split(",")[0].strip() x_real = self.headers.get("X-Real-IP") if x_real: return x_real.strip() return self.client_address[0] def _set_cors(self): origin = self.headers.get("Origin", "") if not origin: self.send_header("Access-Control-Allow-Origin", "*") else: try: parsed = urllib.parse.urlparse(origin) host = parsed.hostname or "" if host in ["localhost", "127.0.0.1", "ai.qpid.pro", "qpid.pro", "ai", "racks"] or host.endswith(".qpid.pro") or host.endswith(".ai") or host.endswith(".racks"): self.send_header("Access-Control-Allow-Origin", origin) else: self.send_header("Access-Control-Allow-Origin", "https://ai.qpid.pro") except Exception: self.send_header("Access-Control-Allow-Origin", "https://ai.qpid.pro") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization") self.send_header("Access-Control-Allow-Credentials", "true") def do_OPTIONS(self): self.send_response(204) self._set_cors() self.end_headers() def do_GET(self): path = self.path.split("?")[0] # 1. Health check if path == "/health": self.send_response(200) self._set_cors() self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"OK") return # 2. List Registered Users in Database if path in ["/api/users", "/api/bugs/users", "/api/auth/users"]: users = db.list_all_users() data = json.dumps({"success": True, "users": users}, ensure_ascii=False).encode("utf-8") self.send_response(200) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) return # 3. Bug Reports if path.startswith("/api/bugs") or path.startswith("/bugs"): bugs = load_bugs() data = json.dumps(bugs, ensure_ascii=False).encode("utf-8") self.send_response(200) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) return self.send_response(404) self._set_cors() self.end_headers() def do_POST(self): path = self.path.split("?")[0] length = int(self.headers.get("Content-Length", 0)) # Check payload size limit (Max 1MB) if length > MAX_BODY_SIZE: self.send_response(413) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Connection", "close") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Payload Too Large (Max 1MB)"}).encode("utf-8")) self.close_connection = True return body = self.rfile.read(length).decode("utf-8") if length > 0 else "{}" try: payload = json.loads(body) if body else {} except Exception as e: self.send_response(400) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"error": f"Invalid JSON: {e}"}).encode("utf-8")) return # 1a. User Login with Rate Limiting if path in ["/api/auth/login", "/api/bugs/auth/login"]: client_ip = self.get_client_ip() if is_login_rate_limited(client_ip): self.send_response(429) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({ "success": False, "message": "Слишком много неудачных попыток входа. Пожалуйста, подождите 1 минуту." }).encode("utf-8")) return username = str(payload.get("username", "")).strip() password_hash = str(payload.get("passwordHash", "")).strip() if not username or not password_hash: record_failed_login(client_ip) self.send_response(400) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Введите логин и пароль"}).encode("utf-8")) return import re if not re.match(r"^[a-zA-Zа-яА-ЯёЁ0-9_-]{3,20}$", username): record_failed_login(client_ip) self.send_response(400) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Логин может содержать только буквы, цифры, '-' и '_' (3-20 симв.)"}).encode("utf-8")) return user = db.get_user_by_username(username) if not user: record_failed_login(client_ip) self.send_response(404) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Пользователь с таким логином не найден. Зарегистрируйтесь."}).encode("utf-8")) return res = db.authenticate_user(username, password_hash) if not res or not res.get("success"): record_failed_login(client_ip) self.send_response(401) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps(res or {"success": False, "message": "Неверный логин или пароль"}).encode("utf-8")) return clear_failed_logins(client_ip) data = json.dumps(res, ensure_ascii=False).encode("utf-8") self.send_response(200) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) return # 1b. User Registration if path in ["/api/auth/register", "/api/bugs/auth/register"]: username = str(payload.get("username", "")).strip() password_hash = str(payload.get("passwordHash", "")).strip() initial_data = payload.get("initialData") if not username or not password_hash: self.send_response(400) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Введите логин и пароль для регистрации"}).encode("utf-8")) return import re if not re.match(r"^[a-zA-Zа-яА-ЯёЁ0-9_-]{3,20}$", username): self.send_response(400) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Логин может содержать только буквы, цифры, '-' и '_' (3-20 симв.)"}).encode("utf-8")) return if len(password_hash) < 1: self.send_response(400) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Пароль не может быть пустым"}).encode("utf-8")) return if username.lower() in ["гость", "guest", "admin", "null", "undefined", "system", "anonymous"]: self.send_response(400) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Это имя пользователя зарезервировано"}).encode("utf-8")) return user = db.get_user_by_username(username) if user: self.send_response(409) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Пользователь с таким логином уже существует"}).encode("utf-8")) return res = db.register_user(username, password_hash, initial_data) data = json.dumps(res, ensure_ascii=False).encode("utf-8") self.send_response(200 if res.get("success") else 400) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) return # 1c. User Logout if path in ["/api/auth/logout", "/api/bugs/auth/logout"]: auth_header = self.headers.get("Authorization", "") token = None if auth_header.startswith("Bearer "): token = auth_header[7:].strip() elif "token" in payload and payload.get("token"): token = str(payload.get("token")).strip() if token: db.delete_session(token) self.send_response(200) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": True, "message": "Сессия успешно завершена"}).encode("utf-8")) return # 2. User Progress Profile Save & Sync (Token Authenticated) if path in ["/api/auth/save", "/api/bugs/auth/save", "/api/auth/sync"]: auth_header = self.headers.get("Authorization", "") token = None if auth_header.startswith("Bearer "): token = auth_header[7:].strip() elif "token" in payload and payload.get("token"): token = str(payload.get("token")).strip() if not token: self.send_response(401) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Требуется авторизация (отсутствует токен сессии)"}).encode("utf-8")) return session_user = db.validate_session(token) if not session_user: self.send_response(401) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Недействительный или истекший токен сессии"}).encode("utf-8")) return username = str(payload.get("username", "")).strip() or session_user["username"] if session_user["username"].lower() != username.lower(): self.send_response(403) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Доступ запрещен (несоответствие пользователя)"}).encode("utf-8")) return user_data = payload.get("data", {}) if not user_data: self.send_response(400) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Отсутствуют данные профиля для сохранения"}).encode("utf-8")) return res = db.save_user_profile(username, user_data, token=token) data = json.dumps(res, ensure_ascii=False).encode("utf-8") self.send_response(200 if res.get("success") else 400) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) return # 3. Bug Report Submission (Sanitized) if path.startswith("/api/bugs") or path.startswith("/bugs"): raw_text = str(payload.get("text", "")).strip() if not raw_text: self.send_response(400) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps({"error": "Empty report text"}).encode("utf-8")) return raw_author = str(payload.get("author", "Гость")).strip() or "Гость" category = str(payload.get("category", "bug")).strip() if category not in ["bug", "feature", "visual", "balance"]: category = "bug" clean_author = html.escape(raw_author)[:30] clean_text = html.escape(raw_text)[:1000] bugs = load_bugs() new_id = max([b.get("id", 0) for b in bugs] + [0]) + 1 new_bug = { "id": new_id, "author": clean_author, "category": category, "text": clean_text, "status": "новый", "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M"), "adminReply": None } bugs.append(new_bug) save_bugs(bugs) resp_data = json.dumps({"success": True, "report": new_bug, "total": len(bugs)}, ensure_ascii=False).encode("utf-8") self.send_response(201) self._set_cors() self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(resp_data))) self.end_headers() self.wfile.write(resp_data) return self.send_response(404) self._set_cors() self.end_headers() def log_message(self, format, *args): sys.stderr.write(f"[{datetime.now().strftime('%H:%M:%S')}] {self.address_string()} - {format%args}\n") if __name__ == "__main__": load_bugs() server = http.server.HTTPServer(("0.0.0.0", PORT), FeedbackHandler) print(f"PvZ Backend Server running on port {PORT} with SQLite database...") try: server.serve_forever() except KeyboardInterrupt: print("\nShutting down server.") server.server_close()