#!/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/save, /api/users) - Community Bug Reports & Feedback (/api/bugs) """ import http.server import json import os import sys 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 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() 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 = 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 = b.get("text", "").replace("|", "/").replace("\n", " ") reply = (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 _set_cors(self): self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") 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)) 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.end_headers() self.wfile.write(json.dumps({"error": f"Invalid JSON: {e}"}).encode("utf-8")) return # 1. User Authentication (Login & Auto-Registration) if path in ["/api/auth/login", "/api/bugs/auth/login", "/api/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.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Имя пользователя и пароль обязательны"}).encode("utf-8")) return # Check if user exists user = db.get_user_by_username(username) if user: res = db.authenticate_user(username, password_hash) if not res or not res.get("success"): self.send_response(401) self._set_cors() self.end_headers() self.wfile.write(json.dumps(res or {"success": False, "message": "Неверный пароль"}).encode("utf-8")) return resp_json = res else: # Auto-register new user res = db.register_user(username, password_hash, initial_data) resp_json = res data = json.dumps(resp_json, 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 # 2. User Progress Profile Save & Sync if path in ["/api/auth/save", "/api/bugs/auth/save", "/api/auth/sync"]: username = str(payload.get("username", "")).strip() user_data = payload.get("data", {}) if not username or not user_data: self.send_response(400) self._set_cors() self.end_headers() self.wfile.write(json.dumps({"success": False, "message": "Username and data are required"}).encode("utf-8")) return res = db.save_user_profile(username, user_data) 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 # 3. Bug Report Submission if path.startswith("/api/bugs") or path.startswith("/bugs"): text = str(payload.get("text", "")).strip() if not text: self.send_response(400) self._set_cors() self.end_headers() self.wfile.write(json.dumps({"error": "Empty report text"}).encode("utf-8")) return author = str(payload.get("author", "Гость")).strip() or "Гость" category = str(payload.get("category", "bug")).strip() if category not in ["bug", "feature", "visual", "balance"]: category = "bug" bugs = load_bugs() new_id = max([b.get("id", 0) for b in bugs] + [0]) + 1 new_bug = { "id": new_id, "author": author[:30], "category": category, "text": text[:1000], "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()