"""Сервер пульта и виджетов KATSBY. Работает на удалённом сервере (VPS), чтобы на стрим-ПК не было Python: покерные румы сети GG (ПокерОК) скрывают столы от захвата, когда видят запущенный Python. Что делает: * раздаёт файлы пульта, виджетов и сцен; * хранит состояние сессии (пульт пишет, виджеты и сцены читают); * принимает нажатия Stream Deck: /api/do?key=...&a=moment * сохраняет выгрузки в папку exports и отдаёт их на скачивание. Настройки через переменные окружения: KATSBY_KEY секретный ключ, обязателен на сервере в интернете KATSBY_HOST адрес для прослушивания, по умолчанию 127.0.0.1 (за nginx) KATSBY_PORT порт, по умолчанию 8777 KATSBY_DATA папка для данных, по умолчанию ./data """ import hmac import json import os import re import threading import time from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, quote, urlparse ROOT = os.path.dirname(os.path.abspath(__file__)) KEY = os.environ.get("KATSBY_KEY", "") HOST = os.environ.get("KATSBY_HOST", "127.0.0.1") PORT = int(os.environ.get("KATSBY_PORT", "8777")) DATA = os.environ.get("KATSBY_DATA", os.path.join(ROOT, "data")) EXPORTS = os.path.join(DATA, "exports") STATE_FILE = os.path.join(DATA, "state.json") DATA_REAL = os.path.realpath(DATA) ACTIONS = {"moment", "reg", "rebuy", "addon", "ko", "cash", "tplus", "tminus", "undo"} PRIVATE_KEYS = {"pult_prefs"} # настройки пульта (в т.ч. пароль OBS) на сервер не отправляются lock = threading.Lock() queue = [] last_poll = 0.0 state = {"ver": 0, "data": {}, "changed": {}} def load_state(): try: with open(STATE_FILE, encoding="utf-8") as f: saved = json.load(f) state["data"] = saved.get("data", {}) state["ver"] = int(saved.get("ver", 0)) state["changed"] = {k: state["ver"] for k in state["data"]} except (OSError, ValueError): pass def save_state(): os.makedirs(DATA, exist_ok=True) tmp = STATE_FILE + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump({"ver": state["ver"], "data": state["data"]}, f, ensure_ascii=False) os.replace(tmp, STATE_FILE) class Handler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=ROOT, **kwargs) def end_headers(self): self.send_header("Cache-Control", "no-store") self.send_header("Access-Control-Allow-Origin", "*") super().end_headers() def log_message(self, *args): pass def send_head(self): real = os.path.realpath(self.translate_path(self.path)) if real == DATA_REAL or real.startswith(DATA_REAL + os.sep): self.send_error(404) # папку данных не отдаём и по /%64ata, /x/../data return None return super().send_head() def reply(self, code, obj): body = json.dumps(obj, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def query(self): return {k: v[0] for k, v in parse_qs(urlparse(self.path).query).items()} def authorized(self, q): if not KEY: return True return hmac.compare_digest(q.get("key", ""), KEY) def body_json(self): length = int(self.headers.get("Content-Length", 0)) if length > 2_000_000: raise ValueError("too large") return json.loads(self.rfile.read(length).decode("utf-8") or "{}") def do_OPTIONS(self): self.send_response(204) self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() def do_GET(self): global last_poll path = urlparse(self.path).path q = self.query() if path.startswith("/api/") or path.startswith("/exports/"): if not self.authorized(q): return self.reply(403, {"ok": False, "error": "bad key"}) if path in ("/data", "/data/") or path.startswith("/data/"): return self.reply(404, {"ok": False}) # папку данных наружу не отдаём if path == "/api/do": action = q.get("a", "") if action not in ACTIONS: return self.reply(400, {"ok": False, "error": "unknown action"}) if time.time() - last_poll > 3: return self.reply(503, {"ok": False, "error": "pult is not open"}) with lock: queue.append({"a": action, "v": q.get("v", ""), "cur": q.get("cur", ""), "note": q.get("note", ""), "t": time.time()}) return self.reply(200, {"ok": True}) if path == "/api/next": last_poll = time.time() with lock: items = queue[:] queue.clear() return self.reply(200, items) if path == "/api/state": since = int(q.get("since", "0") or 0) with lock: changed = {k: state["data"][k] for k, v in state["changed"].items() if v > since and k in state["data"]} return self.reply(200, {"ver": state["ver"], "data": changed}) if path == "/api/exports": os.makedirs(EXPORTS, exist_ok=True) files = sorted(os.listdir(EXPORTS), reverse=True) return self.reply(200, [{"name": n, "size": os.path.getsize(os.path.join(EXPORTS, n))} for n in files]) if path.startswith("/exports/"): name = re.sub(r"[^\w.\-]", "_", path[len("/exports/"):]) full = os.path.join(EXPORTS, name) if not os.path.isfile(full): return self.reply(404, {"ok": False}) with open(full, "rb") as f: data = f.read() self.send_response(200) self.send_header("Content-Type", "text/csv; charset=utf-8") self.send_header("Content-Disposition", "attachment; filename*=UTF-8''" + quote(name)) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) return return super().do_GET() def do_POST(self): path = urlparse(self.path).path q = self.query() if not self.authorized(q): return self.reply(403, {"ok": False, "error": "bad key"}) try: data = self.body_json() except Exception as e: # noqa: BLE001 return self.reply(400, {"ok": False, "error": str(e)}) if path == "/api/state": k = str(data.get("k", "")) if not k or k in PRIVATE_KEYS or len(k) > 64: return self.reply(400, {"ok": False}) with lock: state["ver"] += 1 state["data"][k] = data.get("v") state["changed"][k] = state["ver"] save_state() return self.reply(200, {"ok": True, "ver": state["ver"]}) if path == "/api/save": try: name = re.sub(r"[^\w.\-]", "_", str(data.get("name", "export.txt")))[:120] mode = "a" if data.get("append") else "w" os.makedirs(EXPORTS, exist_ok=True) full = os.path.join(EXPORTS, name) new_file = not os.path.exists(full) with open(full, mode, encoding="utf-8-sig" if (mode == "w" or new_file) else "utf-8", newline="") as f: if mode == "a" and new_file and data.get("header"): f.write(data["header"]) f.write(str(data.get("content", ""))) return self.reply(200, {"ok": True, "path": "exports/" + name}) except Exception as e: # noqa: BLE001 return self.reply(500, {"ok": False, "error": str(e)}) return self.reply(404, {"ok": False}) if __name__ == "__main__": load_state() if HOST not in ("127.0.0.1", "localhost") and not KEY: raise SystemExit("Задай KATSBY_KEY: сервер слушает не только локальный адрес.") print(f"KATSBY: http://{HOST}:{PORT}/control.html" + (f"?key={KEY}" if KEY else "")) ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()