File size: 2,927 Bytes
a4514b4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | 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())
|