| from __future__ import annotations |
|
|
| import json |
| import os |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
| from pathlib import Path |
| from urllib.parse import urlparse |
|
|
| from .api import BackendApi |
| from .context_store import ContextStore |
| from .hermes_adapter import HermesAdapter |
| from .profiles import ProfileDiscovery |
| from .service import BackendService |
|
|
|
|
| def default_profiles_dir() -> Path: |
| return Path(os.environ.get("CLFGUI_HERMES_PROFILES_DIR", "Hermes/profiles")) |
|
|
|
|
| def default_state_dir() -> Path: |
| return Path(os.environ.get("CLFGUI_BACKEND_STATE_DIR", ".clfgui")) |
|
|
|
|
| def build_api() -> BackendApi: |
| service = BackendService( |
| profiles=ProfileDiscovery(default_profiles_dir()), |
| store=ContextStore(default_state_dir()), |
| hermes=HermesAdapter( |
| executable=os.environ.get("CLFGUI_HERMES_BIN", "hermes"), |
| timeout_s=int(os.environ.get("CLFGUI_HERMES_TIMEOUT", "180")), |
| cwd=Path(os.environ.get("CLFGUI_HERMES_CWD", os.getcwd())), |
| ), |
| ) |
| return BackendApi(service) |
|
|
|
|
| def make_handler(api: BackendApi): |
| class Handler(BaseHTTPRequestHandler): |
| def do_GET(self) -> None: |
| self._respond(api.handle("GET", urlparse(self.path).path)) |
|
|
| def do_POST(self) -> None: |
| length = int(self.headers.get("Content-Length", "0")) |
| raw_body = self.rfile.read(length) if length else b"{}" |
| try: |
| payload = json.loads(raw_body.decode("utf-8")) |
| except json.JSONDecodeError: |
| self._send_json(400, {"error": "Invalid JSON."}) |
| return |
| if not isinstance(payload, dict): |
| self._send_json(400, {"error": "Request body must be a JSON object."}) |
| return |
| self._respond(api.handle("POST", urlparse(self.path).path, payload)) |
|
|
| def log_message(self, format: str, *args) -> None: |
| return |
|
|
| def _respond(self, response) -> None: |
| self._send_json(response.status, response.body) |
|
|
| def _send_json(self, status: int, body) -> None: |
| encoded = json.dumps(body, ensure_ascii=False).encode("utf-8") |
| self.send_response(status) |
| self.send_header("Content-Type", "application/json; charset=utf-8") |
| self.send_header("Content-Length", str(len(encoded))) |
| self.end_headers() |
| self.wfile.write(encoded) |
|
|
| return Handler |
|
|
|
|
| def main() -> int: |
| host = os.environ.get("CLFGUI_API_HOST", "127.0.0.1") |
| port = int(os.environ.get("CLFGUI_API_PORT", "8765")) |
| server = ThreadingHTTPServer((host, port), make_handler(build_api())) |
| print(f"CLFGUI API listening on http://{host}:{port}") |
| try: |
| server.serve_forever() |
| except KeyboardInterrupt: |
| return 0 |
| finally: |
| server.server_close() |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
