← back to Rentv Sheet Enrich Refine
live_server.py
309 lines
#!/usr/bin/env python3
"""
live_server.py — LIVE two-way web console over the RENTV CRE contacts Google Sheet.
Reuses lib.py (OAuth token + read_tab + green-stamping batch_fill), so no new auth.
GET / -> the console UI (public_live/index.html)
GET /api/meta -> [{gid,title}] (all tabs)
GET /api/tab?gid=N -> {gid,title,headers,rows} (pull latest — Sheet->UI)
POST /api/write -> {gid,row0,col0,value} write one cell green (UI->Sheet)
POST /api/enhance -> {prompt} optional local-Ollama draft polish ($0, if up)
Two independent sync directions: writes are event-driven (on cell edit); the UI polls
/api/tab to pull external Sheet edits. Basic-auth admin/DW2024! per the viewer convention.
All contacts are Steve Bloom's (RENTV) — ADD only, never bulk-overwrite. $0 (local).
"""
import base64, json, os, socket, sys, threading, urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import lib
import warmth # derived 🔥 Warmth column — response-only, never written to the sheet
HERE = os.path.dirname(os.path.abspath(__file__))
USER, PW = "admin", "DW2024!"
OLLAMA = os.environ.get("OLLAMA_URL", "http://localhost:11434/api/generate")
OLLAMA_MODEL = os.environ.get("DRAFT_MODEL", "qwen3:14b")
# token is cached process-wide and refreshed lazily by lib.access_token()
_tok_lock = threading.Lock()
def tok():
with _tok_lock:
return lib.access_token()
class QuotaError(Exception):
"""Raised when the Sheets read/write quota (429) is hit after retries."""
def retry(fn, *a, tries=5):
"""Call a Sheets-reading fn, retrying on a 429/RESOURCE_EXHAUSTED. lib._req raises
SystemExit (a BaseException) on any API error, so we catch BaseException, inspect the
message, back off on quota errors, and re-raise everything else."""
import time as _t
for i in range(tries):
try:
return fn(*a)
except BaseException as e: # noqa: BLE001 - lib._req raises SystemExit
m = str(e)
if ("429" in m or "RESOURCE_EXHAUSTED" in m or "RATE_LIMIT" in m) and i < tries - 1:
_t.sleep(1.5 * (i + 1)); continue
if "429" in m or "RESOURCE_EXHAUSTED" in m or "RATE_LIMIT" in m:
raise QuotaError("Sheets read quota (60/min) hit — retrying shortly")
raise
import snapshot as snap_mod # our own local-copy builder
SNAP = None
_snap_lock = threading.Lock()
def load_snap():
"""Load our OWN local copy (data/snapshot.json) into memory — no live Sheets reads."""
global SNAP
with open(os.path.join(HERE, "data", "snapshot.json")) as f:
s = json.load(f)
s["_by_gid"] = {t["gid"]: t for t in s["tabs"]}
SNAP = s
def refresh_snap():
"""Rebuild the local copy from the live sheet (a few batch reads) and swap it in."""
global SNAP
s = snap_mod.build(tok())
s["_by_gid"] = {t["gid"]: t for t in s["tabs"]}
with _snap_lock:
SNAP = s
json.dump({k: v for k, v in s.items() if k != "_by_gid"},
open(os.path.join(HERE, "data", "snapshot.json"), "w"))
return {"ok": True, "generated": s["generated"], "tabs": len(s["tabs"]),
"rows": sum(len(t["rows"]) for t in s["tabs"])}
def meta():
return {"generated": SNAP["generated"],
"status_field": SNAP.get("status_field"),
"status_colors": SNAP.get("status_colors", {}),
"status_values": SNAP.get("status_values", []),
"tabs": [{"gid": t["gid"], "title": t["title"], "rows": len(t["rows"])} for t in SNAP["tabs"]]}
def tab(gid):
t = SNAP["_by_gid"][int(gid)]
return warmth.augment({"gid": t["gid"], "title": t["title"], "headers": t["headers"],
"rows": t["rows"], "green": t["green"], "colors": t.get("colors", {}),
"generated": SNAP["generated"]})
def green_map(gid):
"""Enriched (green) cells for a tab — precomputed in the local snapshot."""
t = SNAP["_by_gid"][int(gid)]
return {"gid": t["gid"], "green": t["green"]}
def combined(gids):
"""Merge several tabs into ONE view: columns = union of their headers (by name),
rows concatenated, each already carrying its Sheet Name. Green coords remapped."""
tabs = [SNAP["_by_gid"][int(g)] for g in gids if int(g) in SNAP["_by_gid"]]
order = []; seen = set()
for t in tabs:
for h in t["headers"]:
if h and h not in seen:
seen.add(h); order.append(h)
if "Sheet Name" in order:
order.remove("Sheet Name"); order.insert(0, "Sheet Name")
colidx = {h: i for i, h in enumerate(order)}
rows = []; green = []; colors = {}
for t in tabs:
hmap = [colidx.get(h) for h in t["headers"]] # tab column -> unified column
gset = set(t.get("green", []))
cmap = t.get("colors", {})
base = len(rows)
for ri, r in enumerate(t["rows"]):
newrow = [""] * len(order)
for ci, val in enumerate(r):
ui = hmap[ci] if ci < len(hmap) else None
if ui is not None:
newrow[ui] = val
rows.append(newrow)
for ci in range(len(t["headers"])):
ui = hmap[ci]
if ui is None:
continue
if f"{ri},{ci}" in gset:
green.append(f"{base+ri},{ui}")
hx = cmap.get(f"{ri},{ci}")
if hx:
colors[f"{base+ri},{ui}"] = hx
return warmth.augment({"gid": "combined", "title": f"{len(tabs)} sheets combined",
"headers": order, "rows": rows, "green": green, "colors": colors,
"generated": SNAP["generated"]})
def resolve_combined(gids, crow, ucol):
"""Reverse-map a combined-view (row, union-col) back to the SOURCE (gid, tab_row0, tab_col0).
Mirrors combined()'s row/col construction EXACTLY so a write lands on the right sheet cell.
Returns (gid, tab_row0, tab_col0); tab_col0 == -1 means that union column doesn't exist on
the row's source tab (not editable in combined view)."""
tabs = [SNAP["_by_gid"][int(g)] for g in gids if str(g).strip() and int(g) in SNAP["_by_gid"]]
order = []; seen = set()
for t in tabs:
for h in t["headers"]:
if h and h not in seen:
seen.add(h); order.append(h)
if "Sheet Name" in order:
order.remove("Sheet Name"); order.insert(0, "Sheet Name")
header = order[ucol] if 0 <= ucol < len(order) else None
base = 0
for t in tabs:
n = len(t["rows"])
if crow < base + n:
try:
tcol = t["headers"].index(header) if header is not None else -1
except ValueError:
tcol = -1
return (t["gid"], crow - base, tcol)
base += n
return None
def write_cell(gid, row0, col0, value, formula=False):
# row0 is 0-based over DATA (body) rows; the real sheet grid row = hr + 1 + row0
# (hr = the tab's header row; tabs with a title row ABOVE the header have hr>0)
t = SNAP["_by_gid"].get(int(gid))
# 🔥 Warmth is a derived, response-only column (index == real header count).
# It has no sheet cell behind it — reject the write instead of corrupting the grid.
if t and int(col0) >= len(t["headers"]):
return {"ok": False, "error": "derived column (🔥 Warmth) is read-only"}
hr = t.get("hr", 0) if t else 0
cell = {"row0": int(row0) + hr + 1, "col0": int(col0), "value": value}
if formula:
cell["formula"] = True
retry(lambda: lib.batch_fill(tok(), int(gid), [cell]))
# 2) patch the in-memory local copy (body-indexed) so the UI reflects it instantly + marks green
if t:
r, c = int(row0), int(col0)
while len(t["rows"]) <= r:
t["rows"].append([])
while len(t["rows"][r]) <= c:
t["rows"][r].append("")
t["rows"][r][c] = value
if f"{r},{c}" not in t["green"]:
t["green"].append(f"{r},{c}")
return {"ok": True}
def enhance(prompt):
try:
body = json.dumps({"model": OLLAMA_MODEL, "prompt": prompt, "stream": False,
"options": {"temperature": 0.5}}).encode()
r = urllib.request.urlopen(urllib.request.Request(
OLLAMA, data=body, headers={"Content-Type": "application/json"}), timeout=60)
out = json.load(r).get("response", "").strip()
# strip any <think> blocks qwen emits
if "</think>" in out:
out = out.split("</think>", 1)[1].strip()
return {"ok": True, "text": out}
except Exception as e:
return {"ok": False, "error": f"local model unavailable ({e})"}
class H(BaseHTTPRequestHandler):
def _auth(self):
h = self.headers.get("Authorization", "")
if h.startswith("Basic "):
try:
u, p = base64.b64decode(h[6:]).decode().split(":", 1)
if u == USER and p == PW:
return True
except Exception:
pass
self.send_response(401)
self.send_header("WWW-Authenticate", 'Basic realm="RENTV Live Console"')
self.end_headers()
return False
def _send(self, code, ctype, data):
if isinstance(data, str):
data = data.encode()
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(data)))
# never cache — the console updates constantly; this kills the stale-copy "doesn't work" bug
self.send_header("Cache-Control", "no-store, no-cache, must-revalidate, max-age=0")
self.send_header("Pragma", "no-cache")
self.send_header("Expires", "0")
self.end_headers()
self.wfile.write(data)
def _json(self, obj, code=200):
self._send(code, "application/json", json.dumps(obj))
def log_message(self, *a): # quiet
pass
def do_GET(self):
if not self._auth():
return
path = self.path.split("?", 1)[0]
q = {}
if "?" in self.path:
for kv in self.path.split("?", 1)[1].split("&"):
if "=" in kv:
k, v = kv.split("=", 1)
q[k] = urllib.parse.unquote(v)
try:
if path == "/" or path == "/index.html":
with open(os.path.join(HERE, "public_live", "index.html"), "rb") as f:
self._send(200, "text/html; charset=utf-8", f.read())
elif path == "/api/meta":
self._json(meta())
elif path == "/api/tab":
self._json(tab(q.get("gid", "0")))
elif path == "/api/format":
self._json(green_map(q.get("gid", "0")))
elif path == "/api/combined":
self._json(combined([g for g in q.get("gids", "").split(",") if g.strip()]))
elif "/../" not in path and (path.startswith("/nav-agent/") or path.endswith((".css", ".js"))):
fp = os.path.join(HERE, "public_live", path.lstrip("/"))
if os.path.isfile(fp):
ct = "text/css" if fp.endswith(".css") else "application/javascript"
with open(fp, "rb") as f:
self._send(200, ct + "; charset=utf-8", f.read())
else:
self._json({"error": "not found"}, 404)
else:
self._json({"error": "not found"}, 404)
except QuotaError as e:
self._json({"error": str(e), "retry": True}, 503)
except BaseException as e: # incl. SystemExit from lib._req on API errors
self._json({"error": str(e)}, 500)
def do_POST(self):
if not self._auth():
return
try:
n = int(self.headers.get("Content-Length", 0))
body = json.loads(self.rfile.read(n) or b"{}")
if self.path == "/api/write":
gid, row0, col0 = body["gid"], body["row0"], body["col0"]
if gid == "combined": # reverse-map the combined (row, union-col) to the real source cell
res = resolve_combined(body.get("gids", []), int(row0), int(col0))
if not res or res[2] < 0:
self._json({"ok": False, "error": "That column isn't editable in the combined view for this row's sheet — open the single sheet to edit."})
return
gid, row0, col0 = res
self._json(write_cell(gid, row0, col0,
body.get("value", ""), body.get("formula", False)))
elif self.path == "/api/enhance":
self._json(enhance(body.get("prompt", "")))
elif self.path == "/api/refresh":
self._json(refresh_snap())
else:
self._json({"error": "not found"}, 404)
except QuotaError as e:
self._json({"error": str(e), "retry": True}, 503)
except BaseException as e: # incl. SystemExit from lib._req on API errors
self._json({"error": str(e)}, 500)
import urllib.parse # noqa: E402 (used in handlers)
def main():
load_snap() # serve from OUR local copy, not the live sheet
port = int(sys.argv[1]) if len(sys.argv) > 1 else 0
srv = ThreadingHTTPServer(("127.0.0.1", port), H)
real = srv.server_address[1]
open(os.path.join(HERE, ".liveviewer.port"), "w").write(str(real))
print(f"RENTV Live Console -> http://127.0.0.1:{real} (admin / DW2024!) $0 local", flush=True)
srv.serve_forever()
if __name__ == "__main__":
main()