"""The way back from Dealer OS to this Mac. Dealer OS is the only screen people use for live auctions (Steven 2026-09-02); anything a person does there (star, hide, unhide, save a view, change a display setting, ask for a run) is written to Dealer OS's own tables AND to an append only action log. This Mac polls that log (GET /api/bidbrain/actions?since=) and applies each action through the SAME functions the cockpit's own buttons use (db.set_bid plus the watchlist queue, db.hide_car, db.save_view, db.set_settings, serve._start_run), so hide note learning, watchlist sync and suppression keep working unchanged. It also posts a heartbeat (POST /api/bidbrain/ status) carrying the run light and progress, whose reply says how many actions are waiting, so a poll never has to guess. Dealer OS is the source of record for anything a person does. Every star, hide, view and setting on either side carries an updated_at (UTC); last write wins per key. When an action is applied here it is stamped with the action's own `at`, never now(), so the echo in the next push can never look newer than the Dealer OS row and bounce back. Standard library only. Nothing here opens a browser; the hooks dict lets serve.py hand in the watchlist kick and run control so the same code runs from serve.py's own thread and from the command line. """ import datetime import json import urllib.error import urllib.request from . import db API_ACTIONS = "/api/bidbrain/actions" API_STATUS = "/api/bidbrain/status" API_DIAGNOSTICS = "/api/bidbrain/diagnostics" # The settings row that remembers how far through Dealer OS's action log # this Mac has read. SETTINGS_KEY = "dealer_os_sync" SYNCABLE_SETTINGS = ("details_style", "default_price_ceiling") class DealerOsSyncError(RuntimeError): """A poll or heartbeat that did not land, in plain words. missing_route is True when Dealer OS answered 404: an older Dealer OS without the sync routes yet, which is not a fault to flag, just something to wait for patiently.""" missing_route = False def utcnow(): return datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def _request(base_url, api_key, path, body=None, timeout=20, opener=None): base_url = (base_url or "").strip().rstrip("/") api_key = (api_key or "").strip() if not base_url or not api_key: raise DealerOsSyncError( "Dealer OS is not set up on this Mac: DEALER_OS_BASE_URL and " "DEALER_OS_API_KEY both need a value in dealer_config.py.") data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request( base_url + path, data=data, method="POST" if data is not None else "GET", headers={"Content-Type": "application/json", "Authorization": f"Bearer {api_key}", "User-Agent": "BidBrain"}) do_open = opener or urllib.request.urlopen try: with do_open(req, timeout=timeout) as resp: raw = resp.read() except urllib.error.HTTPError as e: if e.code in (401, 403): raise DealerOsSyncError("Dealer OS refused this Mac's token.") from e if e.code == 404: err = DealerOsSyncError(f"Dealer OS has no {path} route yet (an older Dealer OS).") err.missing_route = True raise err from e detail = "" try: detail = e.read().decode("utf-8", "replace")[:200].strip() except Exception: pass raise DealerOsSyncError(f"Dealer OS answered {e.code} to {path}." + (f" {detail}" if detail else "")) from e except urllib.error.URLError as e: raise DealerOsSyncError(f"Could not reach Dealer OS at {base_url}: {e.reason}") from e if not raw: return {} try: return json.loads(raw.decode("utf-8")) except ValueError: raise DealerOsSyncError(f"Dealer OS answered {path} with something that is not JSON.") def fetch_actions(base_url, api_key, since, limit=500, timeout=20, opener=None): """Every action Dealer OS has logged after cursor `since`, oldest first: {cursor, server_time, actions: [{seq, id, kind, at, by, payload}]}.""" return _request(base_url, api_key, f"{API_ACTIONS}?since={int(since)}&limit={int(limit)}", timeout=timeout, opener=opener) def post_status(base_url, api_key, snapshot, timeout=10, opener=None): """The heartbeat. Reply carries {"ok": true, "pending": n}.""" return _request(base_url, api_key, API_STATUS, body=snapshot, timeout=timeout, opener=opener) def post_diagnostics(base_url, api_key, doc, timeout=30, opener=None): """The diagnostics document (bidbrain/diagnostics.py): log tails, the full run records and site health. Reply carries {"ok": true}.""" return _request(base_url, api_key, API_DIAGNOSTICS, body=doc, timeout=timeout, opener=opener) # --------------------------------------------------------------------------- # The cursor def cursor(path=None): s = db.get_settings(path) if path else db.get_settings() return int((s.get(SETTINGS_KEY) or {}).get("cursor") or 0) def set_cursor(seq, path=None, **extra): s = db.get_settings(path) if path else db.get_settings() cur = dict(s.get(SETTINGS_KEY) or {}) cur["cursor"] = int(seq) cur.update(extra) if path: db.set_settings({SETTINGS_KEY: cur}, path) else: db.set_settings({SETTINGS_KEY: cur}) # --------------------------------------------------------------------------- # What this Mac echoes back in the push def local_state(sale_date, path=None): """The Mac's own stars for this sale date, every active hide and every saved view, each with its updated_at, plus the cursor, so Dealer OS can reconcile with last write wins and the Mac's own actions reach the page.""" kw = {"path": path} if path else {} stars = [{"reg": r["reg"], "sale_date": r["sale_date"], "on": bool(r["active"]), "platform": r.get("platform") or "", "updated_at": r.get("updated_at") or r.get("created_at")} for r in db.bids_with_updated(sale_date, **kw)] hidden = [{"reg": h["reg"], "reason": h.get("reason") or "", "name": h.get("name"), "make": h.get("make"), "model": h.get("model"), "active": bool(h.get("active", 1)), "hidden_at": h.get("hidden_at"), "updated_at": h.get("updated_at") or h.get("hidden_at")} for h in db.hidden_list(**kw)] views = [{"id": v["remote_id"], "name": v["name"], "criteria": v["criteria"], "starred": bool(v["starred"]), "deleted": bool(v.get("deleted")), "updated_at": v.get("updated_at") or v.get("created_at")} for v in db.list_saved_views(include_deleted=True, **kw)] return {"stars": stars, "hidden": hidden, "views": views, "cursor": cursor(path) if path else cursor()} # --------------------------------------------------------------------------- # Applying what Dealer OS logged def _newer(local_at, action_at): """True when the local row is already newer than the action (skip it).""" if not local_at or not action_at: return False return str(local_at) > str(action_at) def _who(action): """Who made a Dealer OS action, for the purchase timeline: the person's own name when Dealer OS sends one (v3.15.23, the team list's name), else their email, else plainly "Dealer OS".""" by = action.get("by") or {} if not isinstance(by, dict): return "Dealer OS" return (by.get("name") or "").strip() or by.get("email") or "Dealer OS" def apply_action(action, hooks=None, path=None): """Apply one logged action through the existing functions. Returns (applied, note). Never raises on a single bad action; the caller logs the note and moves on so one odd row cannot stall the whole log.""" hooks = hooks or {} kw = {"path": path} if path else {} kind = action.get("kind") or "" p = action.get("payload") or {} at = action.get("at") or utcnow() try: if kind == "star.set": reg = p.get("reg") or "" sale_date = p.get("sale_date") or "" if not reg or not sale_date: return False, "star.set without reg or sale_date" details = dict(p.get("details") or {}) # One star per listing (2026-09-14): the platform names which. details["platform"] = p.get("platform") or details.get("platform") or "" existing = db.bid_updated_at(reg, sale_date, platform=details["platform"], **kw) if _newer(existing, at): return False, f"star.set {reg} skipped, local row is newer" db.set_bid(reg, sale_date, bool(p.get("on", True)), p.get("make", "") or "", p.get("model", "") or "", p.get("name", "") or "", p.get("max_bid"), p.get("reserve"), details=details, updated_at=at, **kw) after = hooks.get("after_star") if after: after({"reg": reg, "on": bool(p.get("on", True)), "platform": details.get("platform") or p.get("platform"), "listing_url": p.get("listing_url")}) return True, f"star.set {reg} {'on' if p.get('on', True) else 'off'}" if kind == "hide.set": reg = p.get("reg") or "" if not reg: return False, "hide.set without reg" if _newer(db.hidden_updated_at(reg, **kw), at): return False, f"hide.set {reg} skipped, local row is newer" db.hide_car(reg, p.get("reason", "") or "", p.get("make", "") or "", p.get("model", "") or "", p.get("name", "") or "", updated_at=at, **kw) return True, f"hide.set {reg}" if kind == "hide.clear": reg = p.get("reg") or "" if not reg: return False, "hide.clear without reg" if _newer(db.hidden_updated_at(reg, **kw), at): return False, f"hide.clear {reg} skipped, local row is newer" db.unhide_car(reg, updated_at=at, **kw) return True, f"hide.clear {reg}" if kind == "hold.release": # Released from a hold on Dealer OS (2026-09-10): the reason is # kept with the rules the car broke, the car passes those rules # from now on, and the server moves it onto the list at once. reg = db._norm_reg(p.get("reg") or "") if not reg: return False, "hold.release without reg" if _newer(db.released_updated_at(reg, **kw), at): return False, f"hold.release {reg} skipped, local row is newer" db.release_car(reg, p.get("reason", "") or "", p.get("rules") or [], p.get("make", "") or "", p.get("model", "") or "", p.get("name", "") or "", _who(action), updated_at=at, **kw) door = hooks.get("hold_release") note = door(dict(p, reg=reg)) if door else None return True, f"hold.release {reg}" + (f", {note}" if note else "") if kind == "view.save": rid = p.get("id") or "" if not rid or not p.get("name"): return False, "view.save without id or name" if _newer(db.view_updated_at(rid, **kw), at): return False, f"view.save {rid} skipped, local row is newer" db.save_view(p.get("name"), p.get("criteria") or {}, remote_id=rid, starred=bool(p.get("starred", False)), updated_at=at, **kw) return True, f"view.save {p.get('name')}" if kind == "view.star": rid = p.get("id") or "" if _newer(db.view_updated_at(rid, **kw), at): return False, f"view.star {rid} skipped, local row is newer" db.star_view_remote(rid, bool(p.get("starred", True)), updated_at=at, **kw) return True, f"view.star {rid}" if kind == "view.delete": rid = p.get("id") or "" db.delete_view_remote(rid, updated_at=at, **kw) return True, f"view.delete {rid}" if kind == "settings.set" and p.get("section"): # Step three (2026-09-03): a whole section of the Mac's own # Settings, through the same validators its page uses. Newest # wins: skipped when the Mac saved that section more recently # than this action was made. door = hooks.get("settings_section") if not door: return False, "settings.set section needs the server to apply it" section = p.get("section") keys = [section] if section != "buying_rules" else list((p.get("value") or {}).keys()) for k in keys: if _newer(db.setting_updated_at(k, **kw) if hasattr(db, "setting_updated_at") else None, at): return False, f"settings.set {section} skipped, the Mac saved {k} more recently" try: saved = door(section, p.get("value"), _who(action), at) return True, f"settings.set {section} saved {saved} by {_who(action)}" except ValueError as e: return False, f"settings.set {section} refused: {e}" if kind == "settings.set": key = p.get("key") if key not in SYNCABLE_SETTINGS: return False, f"settings.set {key!r} is not a setting Dealer OS may change" if _newer(db.setting_updated_at(key, **kw), at): return False, f"settings.set {key} skipped, local row is newer" db.set_settings({key: p.get("value")}, updated_at=at, **kw) return True, f"settings.set {key}" if kind == "run.start": start = hooks.get("start_run") if not start: return False, "run.start needs the server to apply it" body = dict(p.get("body") or {}) body["_seq"] = action.get("seq") result = start(p.get("key") or "run", p.get("platforms"), body) return bool(result.get("started")), f"run.start {p.get('key')}: {result}" if kind == "purchase.set": # The everyday purchase fields (notes, the dropdowns, paid for, # collection arranged, and on a hand added car its own facts), # through the same door the Mac's own page uses, so the same # whitelist and dropdown checks apply. Applied in log order; # the purchases table has no per field stamp to compare, so # the later action simply wins. setter = hooks.get("set_purchase_field") who = _who(action) if setter: setter({"id": p.get("id"), "field": p.get("field"), "value": p.get("value")}, who) else: db.set_purchase_field(p.get("id"), p.get("field"), p.get("value"), who=who, **kw) return True, f"purchase.set {p.get('id')} {p.get('field')}" if kind == "purchase.override": # The two per car overrides a person may set or clear: the # retail estimate and the due in (collection) date. which = p.get("field") who = _who(action) if which == "retail_estimate": db.set_retail_estimate_override(p.get("id"), p.get("value"), who=who, **kw) # On to DealerKit at once rather than at the next automatic # run (2026-09-07). Best effort, exactly like every other # DealerKit half: a busy Mac writes it on the timeline and # the automatic run still has it to do. retail_hook = hooks.get("retail_changed") if retail_hook and p.get("value") not in (None, ""): try: retail_hook(p.get("id"), who) except Exception: pass elif which == "collection_date": db.set_collection_date_override(p.get("id"), p.get("value"), who=who, **kw) else: return False, f"purchase.override with unknown field {which!r}" return True, f"purchase.override {p.get('id')} {which}" if kind == "purchase.comment": db.add_purchase_comment(p.get("id"), p.get("text") or "", who=_who(action), **kw) return True, f"purchase.comment {p.get('id')}" if kind in ("purchase.chip", "purchase.unchip", "purchase.cancel", "purchase.checkin", "purchase.add", "purchase.dk", "purchase.dk_send", "purchase.delete", "views.save", "views.delete"): # Step two (2026-09-03): through the server's own doors, each # with the Mac page's own checks. A refusal comes back as a # ValueError with its reason, logged and skipped, never retried. door = hooks.get("purchase_action") if not door: return False, f"{kind} needs the server to apply it" try: return True, f"{kind}: {door(kind, p, _who(action))}" except ValueError as e: return False, f"{kind} refused: {e}" if kind == "purchase.document": # Dealer OS holds a document for a bought car (2026-09-16, the # MotorCheck history check PDF): fetched with this Mac's own # token, kept with the car's other documents, and sent on to # DealerKit. The server does it, it has the token and the door. door = hooks.get("document") if not door: return False, "purchase.document needs the server to apply it" try: return True, f"purchase.document: {door(p, _who(action))}" except ValueError as e: return False, f"purchase.document refused: {e}" if kind == "sale.price": # Dealer OS v3.15.42: DealerKit has just marked a car Sold; read # its own record for what it went for and post it back. door = hooks.get("sale_price") if not door: return False, "sale.price needs the server to apply it" return True, f"sale.price: {door(p)}" if kind == "purchase.dk_stage": # DealerKit's stage, live (2026-09-09, option one): queued by # Dealer OS itself off DealerKit's webhook, applied here through # the same door the timed check uses, then pushed at once. door = hooks.get("dk_stage") if not door: return False, "purchase.dk_stage needs the server to apply it" try: return True, f"purchase.dk_stage: {door(p)}" except ValueError as e: return False, f"purchase.dk_stage refused: {e}" if kind == "run.stop": stop = hooks.get("stop_run") if not stop: return False, "run.stop needs the server to apply it" stop() return True, "run.stop" if kind == "diagnostics.pull": # Dealer OS's Diagnostics card asked for the latest logs now # (2026-09-05): the server sends the document on its next # tick, whatever the fingerprint says. send = hooks.get("send_diagnostics") if not send: return False, "diagnostics.pull needs the server to apply it" send(action.get("seq")) return True, "diagnostics.pull" if kind == "shortlist.read": # The shortlist deep read (Dealer OS v4.1.0, 2026-09-17): only # the starred cars named, queued in the order Dealer OS sent # them (soonest closing first), read by the deep_read pass the # server starts now or on its next quiet moment. Never a bid, # never any action on the page (golden rule 1). cars = p.get("cars") or [] sale_date = p.get("sale_date") or "" n = db.queue_deep_reads(cars, sale_date, asked_at=p.get("asked_at") or at, **kw) if not n: return False, "shortlist.read named no readable car" start = hooks.get("deep_read") note = start(action.get("seq")) if start else "queued, the server reads it on its next quiet moment" return True, f"shortlist.read {n} car(s) queued for {sale_date}" + (f", {note}" if note else "") if kind == "update.start": # Dealer OS's Software update card (2026-09-05): the same one # click update the Mac's own settings page has, so nobody has # to open the Mac's screen to move to a new release. start = hooks.get("start_update") if not start: return False, "update.start needs the server to apply it" result = start(action.get("seq")) return bool(result.get("started")), f"update.start: {result}" return False, f"unknown action kind {kind!r}" except Exception as e: # one bad row never stalls the log return False, f"{kind} failed: {e}" def apply_actions(actions, hooks=None, path=None): applied = skipped = 0 notes = [] last_seq = None for a in actions: ok, note = apply_action(a, hooks, path) notes.append(note) if ok: applied += 1 else: skipped += 1 if a.get("seq") is not None: last_seq = a["seq"] return {"applied": applied, "skipped": skipped, "notes": notes, "last_seq": last_seq} def pull_and_apply(base_url, api_key, hooks=None, path=None, opener=None, limit=500): """Fetch everything after the saved cursor, apply it, move the cursor. Returns the apply summary plus the new cursor; raises DealerOsSyncError only for a transport failure (a bad action is a note, not an error).""" since = cursor(path) if path else cursor() reply = fetch_actions(base_url, api_key, since, limit=limit, opener=opener) actions = reply.get("actions") or [] summary = apply_actions(actions, hooks, path) new_cursor = reply.get("cursor") if summary["last_seq"] is not None: new_cursor = max(int(new_cursor or 0), int(summary["last_seq"])) if new_cursor is not None and int(new_cursor) != since: set_cursor(new_cursor, path, last_pull_at=utcnow()) summary["cursor"] = int(new_cursor or since) summary["fetched"] = len(actions) return summary