#!/usr/bin/env python3 """ BidBrain daily purchases capture. Both platforms close their daily sale at 3:30pm and the day's buys populate in the purchases screens about 15 minutes later. This run, scheduled at 3:50pm, reads what Really Easy Car Credit actually bought on Motorway and Carwow, stores it, and emails the NEW buys to accounts@reallyeasycarcredit.co.uk with the achievable retail price from that day's BidBrain run, so accounts can add the cars to the other systems. The email is sent through the Mail app on this Mac (signed in as sales@), no passwords are stored. No buys, no email. The very first run baselines the purchase history without emailing it. python3 purchases_run.py read, store, email new buys python3 purchases_run.py --dry-run read and store, print the email instead python3 purchases_run.py --baseline seed history without emailing it Reads screens only. Never bids, never acts on the platforms. """ import json import os import subprocess import sys from playwright.sync_api import sync_playwright from bidbrain import db from bidbrain.pricing import pounds from bidbrain.readers import motorway, carwow HERE = os.path.dirname(os.path.abspath(__file__)) CACHE = os.path.join(HERE, "data", "last_run.json") ACCOUNTS_EMAIL = "accounts@reallyeasycarcredit.co.uk" def _retail_map(): """Map normalised reg to the most recent governing retail value BidBrain computed in the last few days, from the stored assessments (every run saves them), plus the latest run cache which also knows each car's selling plate.""" import sqlite3 out = {} conn = sqlite3.connect(os.path.join(HERE, "data", "bidbrain.db"), timeout=5) try: for reg, gv in conn.execute( """SELECT reg, governing_value FROM assessments WHERE governing_value IS NOT NULL AND run_at >= datetime('now', '-4 days') ORDER BY id"""): k = "".join(str(reg or "").upper().split()) if k: out[k] = gv # later rows overwrite, keeping the newest value finally: conn.close() try: d = json.load(open(CACHE, encoding="utf-8")) for a in d.get("shortlist", []) + d.get("held", []): c, p = a.get("car", {}), a.get("pricing") if not p: continue for key in (c.get("reg"), c.get("selling_vrm")): k = "".join(str(key or "").upper().split()) if k: out[k] = p.get("governing_value") except OSError: pass return out def _is_cancelled(r): """A purchase whose sale was cancelled or voided on the platform. Carwow marks these voided; Motorway shows a cancel or withdrawn status.""" s = (r.get("status") or "").strip().lower() return any(t in s for t in ("void", "cancel", "withdrawn")) def read_all(cancelled_pages=1): """Read both purchases screens, plus Motorway's dedicated Cancelled list. Each read fails independently and loudly, one being down never hides another. cancelled_pages controls how deep the Motorway cancelled history is read (1 for the daily top up, more for a one off backfill).""" rows = [] failures = [] with sync_playwright() as p: for name, fn in (("Motorway", motorway.read_purchases), ("Carwow", carwow.read_won)): try: got = fn(p) print(f"{name}: {len(got)} purchases on the screen") rows.extend(got) except Exception as e: print(f"{name} purchases read FAILED: {e}") failures.append(name) # Motorway lists cancelled sales on their own page, so read it directly # rather than sniffing the Complete screen. Carwow's cancelled (voided) # cars already come through read_won. try: cx = motorway.read_cancelled(p, max_pages=cancelled_pages) print(f"Motorway cancelled list: {len(cx)} rows read") rows.extend(cx) except Exception as e: print(f"Motorway cancelled read FAILED: {e}") failures.append("Motorway cancelled") return rows, failures def compose(new): """The email body for accounts: what was bought, what was paid, and the retail each car can go up at. Plain text, house style, no dashes.""" lines = ["Cars bought today, confirmed on the platforms.", ""] for plat in ("Motorway", "Carwow"): cars = [r for r in new if r["platform"] == plat] if not cars: continue lines.append(plat) for r in cars: reg = r.get("reg") or "no plate" paid = pounds(r.get("total_price") or r.get("price")) paid_note = " including fees and VAT" if r.get("total_price") else "" retail = r.get("retail_estimate") retail_note = (f"Achievable retail {pounds(retail)}." if retail else "Not valued by BidBrain, price it by hand.") lines.append(f" {reg} {r.get('name', '')}") lines.append(f" Paid {paid}{paid_note}. {retail_note}") lines.append("") lines.append("") lines.append("Sent automatically by BidBrain after the 3:30pm sales closed.") return "\n".join(lines) def send_mail(subject, body, to=ACCOUNTS_EMAIL): """Send through the Mail app using the account already signed in on this Mac. The message text is passed as arguments, never built into the script.""" script = ''' on run argv set theSubject to item 1 of argv set theBody to item 2 of argv set theTo to item 3 of argv tell application "Mail" set m to make new outgoing message with properties {subject:theSubject, content:theBody, visible:false} tell m to make new to recipient with properties {address:theTo} send m end tell end run ''' subprocess.run(["osascript", "-e", script, subject, body, to], check=True, capture_output=True, timeout=60) def main(): db.init_db() dry = "--dry-run" in sys.argv baseline = "--baseline" in sys.argv # One off backfill reads Motorway's whole cancelled history (about 450 cars # across 18 pages); the daily run only reads page one for new cancellations. backfill = "--backfill-cancelled" in sys.argv rows, failures = read_all(cancelled_pages=20 if backfill else 1) # Cancelled or voided sales: record them so the daily run suppresses the car # for good (Steven never wants a car back that a sale fell through on), and # keep them out of the bought list and the accounts email. cancelled = [r for r in rows if _is_cancelled(r)] rows = [r for r in rows if not _is_cancelled(r)] if cancelled: newly = db.record_walked_away(cancelled) print(f"{len(cancelled)} cancelled/voided sales seen, {len(newly)} newly suppressed " "so the car is never shown again") retail = _retail_map() for r in rows: k = "".join(str(r.get("reg") or "").upper().split()) r["retail_estimate"] = retail.get(k) new = db.record_purchases(rows, baseline=baseline) print(f"{len(rows)} purchases stored, {len(new)} new") if baseline: print("Baseline seeded, nothing emailed.") return pending = db.unemailed_purchases() if not pending: print("No new buys, no email today.") if failures: print(f"ATTENTION: {', '.join(failures)} could not be read, " "a buy there would not have been seen.") sys.exit(1) return import datetime subject = f"Cars bought {datetime.date.today().strftime('%d %B %Y')}" body = compose(pending) if dry: print("\n--- DRY RUN, email not sent ---") print("To:", ACCOUNTS_EMAIL) print("Subject:", subject) print(body) return send_mail(subject, body) db.mark_purchases_emailed([r["id"] for r in pending]) print(f"Emailed {len(pending)} buys to {ACCOUNTS_EMAIL}.") if failures: print(f"ATTENTION: {', '.join(failures)} could not be read.") sys.exit(1) if __name__ == "__main__": main()