"""Local backend process skeleton (U-007): pure stdlib `http.server`, no third-party deps (05 §5.7) -- this process is also where the Graph Engine scheduler will live later (05 §5.4.1), but that's a different Unit. MVP scope here: process start/stop, `c_pick_folder` (03-C §3.-1) as the one real API route -- macOS only for now (`osascript`); Windows/Linux native dialogs are PH3 (U-022) -- and serving the U-008 frontend shell as static files (no build step, 05 §5.7: whitelisted by exact path, not a generic directory walker, so there's no path-traversal surface to reason about). """ import json import os import platform import subprocess from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer _FRONTEND_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend") # U-008: exact-path whitelist, not a generic static-file server -- the # frontend is 3 fixed files until PH3 adds real pages. _STATIC_FILES = { "/": ("index.html", "text/html; charset=utf-8"), "/index.html": ("index.html", "text/html; charset=utf-8"), "/app.js": ("app.js", "text/javascript; charset=utf-8"), "/style.css": ("style.css", "text/css; charset=utf-8"), } class DialogUnavailableError(Exception): """DIALOG_UNAVAILABLE -- 03-C §3.-1: no native-dialog capability here.""" def pick_folder(): """macOS implementation of c_pick_folder. Returns {"selected_path": str} on success, {"selected_path": None} on user cancel (03-C: not an error). Raises DialogUnavailableError off macOS or if osascript itself fails for a reason other than the user cancelling.""" if platform.system() != "Darwin": raise DialogUnavailableError("native folder picker only implemented for macOS so far (PH3 adds Windows/Linux)") script = 'POSIX path of (choose folder with prompt "Select a graphlemon project folder")' result = subprocess.run(["osascript", "-e", script], capture_output=True, text=True) if result.returncode != 0: if "User canceled" in result.stderr or "(-128)" in result.stderr: return {"selected_path": None} raise DialogUnavailableError(result.stderr.strip()) return {"selected_path": result.stdout.strip()} class _Handler(BaseHTTPRequestHandler): def do_GET(self): entry = _STATIC_FILES.get(self.path) if entry is None: self._send_json(404, {"error": "NOT_FOUND"}) return filename, content_type = entry with open(os.path.join(_FRONTEND_DIR, filename), "rb") as f: body = f.read() self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_POST(self): if self.path == "/api/system/pick-folder": self._handle_pick_folder() else: self._send_json(404, {"error": "NOT_FOUND"}) def _handle_pick_folder(self): try: self._send_json(200, pick_folder()) except DialogUnavailableError as e: self._send_json(422, {"error": "DIALOG_UNAVAILABLE", "detail": str(e)}) def _send_json(self, status, obj): body = json.dumps(obj).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, format, *args): pass # quiets test output; real request logging is a later Unit's concern def create_server(host="127.0.0.1", port=0): """port=0 lets the OS assign a free port -- avoids collisions in tests and lets multiple graphlemon instances run side by side if ever needed.""" return ThreadingHTTPServer((host, port), _Handler)