"""Ramco Intelligence — the PO brain (answer-first UI). White, minimal, SF system font, soft pink/violet. The answer renders directly below the question as sectioned paragraphs; the screen->service->API->SP->tables chain is a compact left-to-right pipeline below it; citations are cards that reveal their exact passage. stdlib only. GET / UI GET /ask?q=... grounded answer JSON (sections, chain, citations, handoff) GET /cite?source=&loc= the cited passage text GET /questions the 100-question catalog (grouped) """ from __future__ import annotations import json import urllib.parse from functools import lru_cache from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import config from query.ask import answer @lru_cache(maxsize=1) def _chunk_index(): idx = {} try: for c in json.loads(config.DOC_CHUNKS_JSON.read_text()): idx[(c["source"], c["loc"])] = c["text"] except Exception: pass return idx PAGE = r""" Ramco Intelligence · PO Brain
Ramco Intelligence

Ask the Purchase Order brain

Plain-English answers about how the PO system works — grounded in the source truth.

Ask about a process, the tables it populates, an API, a stored procedure, a validation, or what something means. Every answer links back to its source.

""" class Handler(BaseHTTPRequestHandler): def _send(self, code, body, ctype="application/json"): data = body.encode("utf-8") if isinstance(body, str) else body self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def log_message(self, *a): pass def do_GET(self): u = urllib.parse.urlparse(self.path) qs = urllib.parse.parse_qs(u.query) if u.path in ("/", "/index.html"): return self._send(200, PAGE, "text/html; charset=utf-8") if u.path == "/ask": q = (qs.get("q", [""])[0]).strip() res = answer(q) if q else {"answer": "", "chain": [], "citations": []} return self._send(200, json.dumps(res)) if u.path == "/cite": src = qs.get("source", [""])[0] loc = qs.get("loc", [""])[0] text = _chunk_index().get((src, loc), "") return self._send(200, json.dumps({"source": src, "loc": loc, "text": text})) if u.path == "/questions": try: return self._send(200, config.QUESTIONS_JSON.read_text()) except Exception: return self._send(200, "[]") return self._send(404, json.dumps({"error": "not found"})) def main(port: int = 8800): import os host = os.environ.get("HOST", "127.0.0.1") port = int(os.environ.get("PORT", port)) answer("warmup") srv = ThreadingHTTPServer((host, port), Handler) print(f"Ramco Intelligence · PO Brain at http://{host}:{port} (Ctrl-C to stop)") try: srv.serve_forever() except KeyboardInterrupt: srv.shutdown() if __name__ == "__main__": import sys main(int(sys.argv[1]) if len(sys.argv) > 1 else 8800)