#!/usr/bin/env python3 """Report demo proxy: one saved program with parameters, run on the device that holds the file. 127.0.0.1:18453 behind Caddy (handle /api/report/*). The platform token stays in the environment. There is no model client in this server: nothing here calls a model. GET /api/report/state devices, the allowed parameter values, the program's source and its sha256 POST /api/report/run {device: vps|second, group_by, value, month, top} -> the report, as the device returned it The program is the expert csv_report (csv_report.fython, next to this file). It takes a path to a CSV file, so a visitor is never allowed to send the path: this proxy always sends the path of the sample file and accepts only the parameter values listed in ALLOWED. On your own machine you pass any path you like (see first_program.py). """ import hashlib import json import os import sys import threading import time import urllib.error import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer HERE = os.path.dirname(os.path.abspath(__file__)) API = "https://api.extella.ai/api" TOKEN = os.environ["EXTELLA_TOKEN"] AGENT = os.environ.get("ROBOT_AGENT", "agent_Fgw-hPvLsg9vyn2QDg5GO") PORT = int(os.environ.get("REPORT_PORT", "18453")) DEVICES = {"vps": {"id": os.environ.get("ROBOT_DEVICE", "85800354-f7b7-449f-b526-9357cd91f780"), "name": "extella-team", "label": "Server 1 · Linux"}, "second": {"id": os.environ.get("ROBOT_DEVICE_SECOND", "9ed87a11-61f9-4985-b997-584de73d028f"), "name": "extella-second", "label": "Server 2 · Linux"}} EXPERT, SAMPLE, PATH = "csv_report", "csv_report_sample", "~/extella-demo-data/orders.csv" ALLOWED = {"group_by": ["region", "product"], "value": ["amount", "quantity"], "month": [""] + ["2026-%02d" % m for m in range(1, 10)], "top": [3, 5, 7]} DESCRIPTIONS = {EXPERT: "Report over a CSV file on the device: group rows by one column, sum a numeric column, return the top groups. Parameters: path, group_by, value, top, filter_column, filter_value.", SAMPLE: "Writes a file of synthetic orders (CSV) on the device it runs on; the data depends on the host name."} SOURCE = {name: open(os.path.join(HERE, name + ".fython"), encoding="utf-8").read() for name in (EXPERT, SAMPLE)} LOCK = threading.Lock() RATE, RUNS = {}, {} # requests and runs per address, last minute SLOTS = {key: threading.BoundedSemaphore(2) for key in DEVICES} # at most two runs on one device at a time SAVED = {} # name -> True or False: the platform holds the text of the file next to this server (checked at start) # what a visitor gets back from a run; anything else the platform returns stays in the server log PUBLIC = ("status", "reason", "columns", "table", "group_by", "value", "filter", "rows_in_file", "rows_used", "rows_skipped", "groups", "grand_total", "file", "file_sha256", "host", "os", "seconds_on_device") def platform(path, body, timeout=120): req = urllib.request.Request(API + path, data=json.dumps(body, ensure_ascii=False).encode(), headers={"X-Auth-Token": TOKEN, "X-Profile-Id": "default", "X-Agent-Id": AGENT, "Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read()) def unwrap(out): import ast res = out.get("result", out) if isinstance(out, dict) else out if isinstance(res, dict) and "result" in res: res = res["result"] if isinstance(res, str): for loader in (json.loads, ast.literal_eval): try: return loader(res) except Exception: pass return res def run_expert(expert, params, device): dev = DEVICES[device] t0 = time.time() try: out = platform("/expert/run", {"name": expert, "params": params, "targets": [dev["id"]]}, timeout=150) task = out.get("task_id") if isinstance(out, dict) else None while task: time.sleep(2) chk = platform("/tasks/check", {"task_id": task}, timeout=30) if str(chk.get("status")) in ("completed", "done", "success", "finished", "error", "failed"): out = chk break if time.time() - t0 > 140: return {"status": "error", "reason": "the device did not finish in 140 seconds", "platform_seconds": round(time.time() - t0, 1)} except urllib.error.HTTPError as e: body = e.read().decode("utf-8", "ignore")[:200] sys.stderr.write("platform HTTP %d: %s\n" % (e.code, body)) return {"status": "error", "reason": ("device %s is offline right now" % dev["name"]) if "unavailable" in body.lower() else "the platform answered HTTP %d" % e.code, "platform_seconds": round(time.time() - t0, 1)} except Exception as e: sys.stderr.write("platform call failed: %r\n" % (e,)) return {"status": "error", "reason": "the platform did not answer", "platform_seconds": round(time.time() - t0, 1)} res = unwrap(out) if not isinstance(res, dict): sys.stderr.write("unexpected answer: %s\n" % str(res)[:300]) res = {"status": "error", "reason": "unavailable" if "unavailable" in str(res).lower() else "the device returned an answer this server could not read"} if isinstance(res.get("reason"), str) and "unavailable" in res["reason"].lower(): res = {"status": "error", "reason": "device %s is offline right now" % dev["name"]} if res.get("status") != "success": sys.stderr.write("device answer: %s\n" % str(res)[:300]) res = {k: res[k] for k in PUBLIC if k in res} if "reason" in res: res["reason"] = res["reason"][:200] if isinstance(res["reason"], str) else "the device gave no readable reason" res["platform_seconds"] = round(time.time() - t0, 1) return res def prepare(): """Save the two programs as they are in the files next to this server, and make sure each device has the sample file.""" for name in (EXPERT, SAMPLE): try: r = platform("/expert/save", {"name": name, "description": DESCRIPTIONS[name], "code": SOURCE[name], "cspl": "fython", "kwargs": {}, "global": False}) held = platform("/expert/get", {"name": name}).get("expert_code") SAVED[name] = isinstance(held, str) and held.rstrip("\n") == SOURCE[name].rstrip("\n") # the platform stores the text without its final newline sys.stderr.write("save %s: %s, the platform holds this text: %s\n" % (name, r.get("status"), SAVED[name])) except Exception as e: SAVED[name] = False sys.stderr.write("save %s failed: %s\n" % (name, str(e)[:120])) for device in DEVICES: r = run_expert(EXPERT, {"path": PATH, "top": 1}, device) if r.get("status") != "success": r = run_expert(SAMPLE, {"path": PATH}, device) sys.stderr.write("sample file on %s: %s\n" % (device, r.get("status"))) def allowed(ip, bucket=RATE, limit=30): now = time.time() with LOCK: if len(bucket) > 5000: # forget addresses that have been quiet for a minute for key in [k for k, v in bucket.items() if not v or now - v[-1] > 60]: del bucket[key] hits = [t for t in bucket.get(ip, []) if now - t < 60] if len(hits) >= limit: bucket[ip] = hits return False hits.append(now) bucket[ip] = hits return True class Handler(BaseHTTPRequestHandler): server_version, sys_version = "extella-report", "" timeout = 15 # a client that stops sending is dropped after 15 seconds def _send(self, code, obj): data = json.dumps(obj, ensure_ascii=False).encode() self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(data) def _ip(self): return self.headers.get("X-Forwarded-For", self.client_address[0]).split(",")[0].strip() def log_message(self, fmt, *args): pass def do_GET(self): if not allowed(self._ip()): return self._send(429, {"status": "error", "reason": "too many requests, slow down"}) if self.path.split("?")[0] == "/api/report/state": return self._send(200, {"status": "success", "expert": EXPERT, "language": "fython", "source": SOURCE[EXPERT], "source_sha256": hashlib.sha256(SOURCE[EXPERT].encode()).hexdigest(), "devices": {k: {"name": v["name"], "label": v["label"]} for k, v in DEVICES.items()}, "allowed": ALLOWED, "file": PATH, "saved_matches_source": SAVED.get(EXPERT), "note": "the file is synthetic and differs between the two servers; no model is called by this server; source_sha256 is the hash of the file published on the site, the platform stores the same text without its final newline"}) return self._send(404, {"status": "error", "reason": "unknown path"}) def do_POST(self): if not allowed(self._ip()): return self._send(429, {"status": "error", "reason": "too many requests, slow down"}) if self.path.split("?")[0] != "/api/report/run": return self._send(404, {"status": "error", "reason": "unknown path"}) if (self.headers.get("Content-Type") or "").split(";")[0].strip().lower() != "application/json": return self._send(415, {"status": "error", "reason": "send the body as Content-Type: application/json"}) length = self.headers.get("Content-Length") or "" if not (length.isascii() and length.isdigit()): return self._send(411, {"status": "error", "reason": "a Content-Length header is required"}) n = int(length) try: body = json.loads(self.rfile.read(n)) if 0 < n <= 2000 else None except Exception: body = None if not isinstance(body, dict): return self._send(400, {"status": "error", "reason": "bad request"}) device = body.get("device") if not isinstance(device, str) or device not in DEVICES: return self._send(400, {"status": "error", "reason": "device must be vps or second"}) picked = {} for key, values in ALLOWED.items(): v = body.get(key, values[0]) if type(v) is not type(values[0]) or v not in values: return self._send(400, {"status": "error", "reason": "%s must be one of %s" % (key, values)}) picked[key] = v params = {"path": PATH, "group_by": picked["group_by"], "value": picked["value"], "top": picked["top"]} if picked["month"]: params.update({"filter_column": "date", "filter_value": picked["month"]}) if not allowed(self._ip(), RUNS, 10): return self._send(429, {"status": "error", "reason": "too many runs from your address, wait a minute"}) if not SLOTS[device].acquire(blocking=False): return self._send(429, {"status": "error", "reason": "this server is busy with other visitors' runs, try again in a few seconds"}) try: res = run_expert(EXPERT, params, device) finally: SLOTS[device].release() res.update({"expert": EXPERT, "params_sent": params, "device_key": device}) return self._send(200, res) if __name__ == "__main__": print("report proxy on 127.0.0.1:%d" % PORT) threading.Thread(target=prepare, daemon=True).start() ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever()