#!/usr/bin/env python3 """ Lokaler Server für Lara Schachturnier Serviert statische Dateien und bietet Stockfish-Analyse. """ import http.server import socketserver import os import subprocess import json import re import sys import threading import socket PORT = int(os.environ.get("PORT", 8111)) BASE_DIR = os.path.dirname(os.path.abspath(__file__)) STOCKFISH_PATH = os.environ.get("STOCKFISH_PATH") if not STOCKFISH_PATH: win_path = os.path.join(BASE_DIR, "stockfish.exe") if os.path.exists(win_path): STOCKFISH_PATH = win_path else: STOCKFISH_PATH = "stockfish" STOCKFISH_DEPTH = int(os.environ.get("STOCKFISH_DEPTH", 15)) _stockfish_lock = threading.Lock() class StockfishEngine: def __init__(self, path): self.path = path self.proc = None def start(self): if self.proc: return flags = 0 if sys.platform == "win32": flags = subprocess.CREATE_NO_WINDOW self.proc = subprocess.Popen( [self.path], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, creationflags=flags, ) self._send("uci") self._read_until("uciok") self._send("isready") self._read_until("readyok") print(f"[STOCKFISH] Engine gestartet: {self.path}") def _send(self, cmd): self.proc.stdin.write(cmd + "\n") self.proc.stdin.flush() def _read_until(self, marker): while True: line = self.proc.stdout.readline().strip() if line == marker: return def evaluate(self, fen): self._send(f"position fen {fen}") self._send(f"go movetime 10000") score_cp = None score_mate = None bestmove = None pv = None last_depth = 0 while True: line = self.proc.stdout.readline().strip() if not line: continue if "score cp" in line: m = re.search(r"score cp (-?\d+)", line) if m: score_cp = int(m.group(1)) if "score mate" in line: m = re.search(r"score mate (-?\d+)", line) if m: score_mate = int(m.group(1)) if " pv " in line: pv = line.split(" pv ", 1)[1] depth_m = re.search(r"depth (\d+)", line) if depth_m and score_cp is not None: new_depth = int(depth_m.group(1)) if new_depth != last_depth: last_depth = new_depth yield { "scoreCp": score_cp, "scoreMate": score_mate, "bestMove": None, "pv": pv, "depth": new_depth, } if line.startswith("bestmove"): parts = line.split() bestmove = parts[1] if len(parts) > 1 else None break yield { "scoreCp": score_cp, "scoreMate": score_mate, "bestMove": bestmove, "pv": pv, } def stop(self): if self.proc: self.proc.terminate() self.proc = None _engine = StockfishEngine(STOCKFISH_PATH) class Handler(http.server.BaseHTTPRequestHandler): def do_GET(self): if self.path == "/": self.path = "/index.html" filepath = os.path.join(BASE_DIR, self.path.lstrip("/")) if os.path.isfile(filepath): content_types = { ".html": "text/html", ".css": "text/css", ".js": "application/javascript", ".json": "application/json", ".png": "image/png", ".jpg": "image/jpeg", ".svg": "image/svg+xml", } ext = os.path.splitext(filepath)[1] content_type = content_types.get(ext, "application/octet-stream") with open(filepath, "rb") as f: content = f.read() self.send_response(200) self.send_header("Content-Type", f"{content_type}; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Cache-Control", "no-cache") self.end_headers() self.wfile.write(content) else: self.send_response(404) self.end_headers() def do_POST(self): if self.path == "/evaluate": length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) if length else b"{}" try: data = json.loads(body) fen = data.get("fen", "") except (json.JSONDecodeError, TypeError): self._send_json({"error": "Invalid JSON"}, 400) return if not fen: self._send_json({"error": "Missing fen"}, 400) return self.send_response(200) self.send_header("Content-Type", "application/x-ndjson; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "close") self.end_headers() self.wfile.flush() # TCP_NODELAY für sofortiges Senden try: self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) except Exception: pass self.close_connection = True try: with _stockfish_lock: try: _engine.start() except FileNotFoundError: self.wfile.write( json.dumps({"error": "Stockfish nicht gefunden"}).encode("utf-8") + b"\n" ) self.wfile.flush() return for result in _engine.evaluate(fen): data = json.dumps(result).encode("utf-8") + b"\n" self.wfile.write(data) self.wfile.flush() except Exception as e: print(f"[STOCKFISH] Fehler: {e}") self.wfile.write( json.dumps({"error": str(e)}).encode("utf-8") + b"\n" ) self.wfile.flush() else: self._send_json({"error": "Not found"}, 404) def _send_json(self, obj, status=200): self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Cache-Control", "no-cache") self.end_headers() self.wfile.write(json.dumps(obj).encode("utf-8")) def log_message(self, format, *args): print(f"[{self.log_date_time_string()}] {args[0]}") def main(): print("=" * 50) print(" [TROPHY] Lara Kiesewetter - Live Schachturnier") print("=" * 50) print(f" Server laeuft auf: http://localhost:{PORT}") if os.path.exists(STOCKFISH_PATH) or STOCKFISH_PATH == "stockfish": print(f" Stockfish-Analyse aktiv (depth={STOCKFISH_DEPTH})") else: print(f" Stockfish nicht gefunden unter: {STOCKFISH_PATH}") print(f" Druecke Ctrl+C zum Beenden") print("=" * 50) socketserver.ThreadingTCPServer.allow_reuse_address = True with socketserver.ThreadingTCPServer(("", PORT), Handler) as httpd: print(f"\n[SERVER] Server gestartet: http://localhost:{PORT}\n") try: httpd.serve_forever() except KeyboardInterrupt: print("\n[BYE] Server gestoppt.") _engine.stop() if __name__ == "__main__": main()