#!/usr/bin/env python3 """Local webserver with healthcheck endpoint.""" from __future__ import annotations import hashlib import json import os from datetime import datetime, timezone from http import HTTPStatus from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, urlparse from uuid import uuid4 DEFAULT_HOST = "0.0.0.0" DEFAULT_PORT = 8808 DEFAULT_SUGGESTIONS_FILE = "suggestions.jsonl" class RequestHandler(SimpleHTTPRequestHandler): def do_GET(self) -> None: # noqa: N802 parsed = urlparse(self.path) if parsed.path == "/health": payload = json.dumps({"status": "ok"}).encode("utf-8") self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) return if parsed.path == "/api/suggestions": include_done = parse_qs(parsed.query).get("includeDone", ["0"])[0] in {"1", "true", "yes"} self._handle_list_suggestions(include_done=include_done) return # Serve static files from the current working directory (repo root). super().do_GET() def do_POST(self) -> None: # noqa: N802 parsed = urlparse(self.path) if parsed.path == "/api/suggestions": self._handle_create_suggestion() return if parsed.path == "/api/suggestions/mark-done": self._handle_mark_suggestion_done() return self.send_error(HTTPStatus.NOT_FOUND) def log_message(self, format: str, *args: object) -> None: # noqa: A003 # Keep default HTTP logs quiet for local developer usage/tests. return def _send_json(self, status: HTTPStatus, payload: dict | list) -> None: response = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(response))) self.end_headers() self.wfile.write(response) def _read_json_body(self) -> dict | None: content_length = int(self.headers.get("Content-Length", "0")) if content_length <= 0: return None raw = self.rfile.read(content_length) try: payload = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): return None if not isinstance(payload, dict): return None return payload def _suggestions_file_path(self) -> str: return os.getenv("SUGGESTIONS_FILE", DEFAULT_SUGGESTIONS_FILE) def _load_suggestions(self) -> list[dict]: suggestions: list[dict] = [] suggestions_file = self._suggestions_file_path() if os.path.exists(suggestions_file): with open(suggestions_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: entry = json.loads(line) except json.JSONDecodeError: continue if not isinstance(entry, dict): continue suggestions.append(entry) return suggestions def _write_suggestions(self, suggestions: list[dict]) -> None: suggestions_file = self._suggestions_file_path() with open(suggestions_file, "w", encoding="utf-8") as f: for item in suggestions: f.write(json.dumps(item) + "\n") def _legacy_id_for(self, entry: dict) -> str: seed = "|".join([ str(entry.get("submittedAt", "")), str(entry.get("requester", "")), str(entry.get("title", "")), str(entry.get("details", "")), ]) digest = hashlib.sha1(seed.encode("utf-8"), usedforsecurity=False).hexdigest() # noqa: S324 return f"legacy-{digest[:12]}" def _normalize_entry(self, entry: dict) -> dict: normalized = dict(entry) normalized.setdefault("id", self._legacy_id_for(normalized)) normalized.setdefault("doneAt", None) return normalized def _handle_create_suggestion(self) -> None: payload = self._read_json_body() if payload is None: self._send_json(HTTPStatus.BAD_REQUEST, {"error": "Invalid JSON body."}) return requester = str(payload.get("requester", "")).strip() title = str(payload.get("title", "")).strip() details = str(payload.get("details", "")).strip() if not requester or not title or not details: self._send_json( HTTPStatus.BAD_REQUEST, {"error": "requester, title, and details are required."}, ) return entry = { "id": str(uuid4()), "submittedAt": datetime.now(timezone.utc).isoformat(), "requester": requester, "title": title, "details": details, "doneAt": None, } suggestions = [self._normalize_entry(item) for item in self._load_suggestions()] suggestions.append(entry) self._write_suggestions(suggestions) self._send_json( HTTPStatus.CREATED, {"message": "Suggestion submitted to the Founding Engineer.", "suggestion": entry}, ) def _handle_mark_suggestion_done(self) -> None: payload = self._read_json_body() if payload is None: self._send_json(HTTPStatus.BAD_REQUEST, {"error": "Invalid JSON body."}) return target_id = str(payload.get("id", "")).strip() target_submitted = str(payload.get("submittedAt", "")).strip() target_title = str(payload.get("title", "")).strip() if not target_id and not (target_submitted and target_title): self._send_json( HTTPStatus.BAD_REQUEST, {"error": "Provide id, or submittedAt + title to identify a suggestion."}, ) return suggestions = [self._normalize_entry(item) for item in self._load_suggestions()] done_at = datetime.now(timezone.utc).isoformat() updated: dict | None = None for item in suggestions: matches_id = target_id and item.get("id") == target_id matches_legacy = ( not target_id and item.get("submittedAt") == target_submitted and item.get("title") == target_title ) if matches_id or matches_legacy: item["doneAt"] = done_at updated = item break if updated is None: self._send_json(HTTPStatus.NOT_FOUND, {"error": "Suggestion not found."}) return self._write_suggestions(suggestions) self._send_json(HTTPStatus.OK, {"message": "Suggestion marked done.", "suggestion": updated}) def _handle_list_suggestions(self, include_done: bool = False) -> None: suggestions = [self._normalize_entry(item) for item in self._load_suggestions()] if not include_done: suggestions = [item for item in suggestions if not item.get("doneAt")] self._send_json(HTTPStatus.OK, {"count": len(suggestions), "suggestions": suggestions}) def create_server(host: str, port: int) -> ThreadingHTTPServer: return ThreadingHTTPServer((host, port), RequestHandler) def resolve_bind_address() -> tuple[str, int]: host = os.getenv("HOST", DEFAULT_HOST) port = int(os.getenv("PORT", str(DEFAULT_PORT))) return host, port def main() -> None: host, port = resolve_bind_address() server = create_server(host, port) print(f"Serving on http://{host}:{port}") server.serve_forever() if __name__ == "__main__": main()