#!/usr/bin/env python3 """ Experiment: quote conversion % for Bunbury Electrics via Simpro REST API. Live tenant lists company ID 2 as "ELECTRICS BUNBURY" (see GET /companies/). App config defaults still label company 2 as "Residential" — names may have drifted. Usage (repo root): python3 scripts/experiment_simpro_quotes_electrics_bunbury.py Requires .env: SIMPRO_BRANCH2_API_KEY, SIMPRO_BRANCH2_API_URL. Cohort: quotes with DateIssued in the period (Simpro does not expose DateCreated on quotes). Converted: Status.Name contains "converted to job" (matches JobNo presence in checks). """ from __future__ import annotations import json import os import sys import urllib.error import urllib.parse import urllib.request from calendar import monthrange from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] ENV_PATH = ROOT / ".env" # Verified against GET {api}/companies/ — Electrics Bunbury COMPANY_ID = 2 def load_dotenv_simple(path: Path) -> dict[str, str]: out: dict[str, str] = {} if not path.exists(): return out for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue k, _, v = line.partition("=") k, v = k.strip(), v.strip().strip('"').strip("'") out[k] = v return out def parse_ts(s: str | None) -> datetime | None: if not s or not str(s).strip(): return None s = str(s).replace("Z", "+00:00") try: return datetime.fromisoformat(s) except ValueError: return None def naive_utc(dt: datetime | None) -> datetime | None: if dt is None: return None if dt.tzinfo is not None: return dt.astimezone(timezone.utc).replace(tzinfo=None) return dt def in_calendar_month(dt: datetime | None, year: int, month: int) -> bool: dt = naive_utc(dt) if dt is None: return False return dt.year == year and dt.month == month def in_datetime_range(dt: datetime | None, start: datetime, end: datetime) -> bool: dt = naive_utc(dt) if dt is None: return False return start <= dt <= end def status_text(status: object | None) -> str: if status is None: return "" if isinstance(status, dict): return str(status.get("Name") or status.get("value") or "").strip() return str(status).strip() def is_converted_status(status: object | None) -> bool: s = status_text(status).lower() if not s: return False return "converted to job" in s def fetch_json(url: str, api_key: str) -> object: req = urllib.request.Request( url, headers={ "Authorization": f"Bearer {api_key}", "Accept": "application/json", }, method="GET", ) with urllib.request.urlopen(req, timeout=60) as resp: return json.loads(resp.read().decode("utf-8")) def fetch_company_label(base: str, api_key: str, company_id: int) -> str: try: data = fetch_json(f"{base}/companies/", api_key) if isinstance(data, list): for c in data: if c.get("ID") == company_id: return str(c.get("Name") or f"Company {company_id}") except Exception: pass return f"Company {company_id}" def fetch_quotes_until_before_month( base: str, api_key: str, company_id: int, year: int, month: int, ) -> list[dict]: """Newest-first pagination until a page’s oldest DateIssued is before month_start.""" month_start = datetime(year, month, 1) page_size = 250 all_rows: list[dict] = [] page = 1 while True: q = urllib.parse.urlencode( { "columns": "ID,Status,DateIssued,DateModified,Total", "pageSize": page_size, "page": page, "orderby": "-DateIssued", } ) url = f"{base}/companies/{company_id}/quotes/?{q}" try: data = fetch_json(url, api_key) except urllib.error.HTTPError as e: print(f"HTTP {e.code}: {e.reason}", file=sys.stderr) try: print(e.read().decode("utf-8", errors="replace")[:2000], file=sys.stderr) except Exception: pass raise batch = data if isinstance(data, list) else data.get("items") or data.get("values") or [] if not batch: break all_rows.extend(batch) oldest_di = None for row in batch: di = naive_utc(parse_ts(row.get("DateIssued"))) if di is not None and (oldest_di is None or di < oldest_di): oldest_di = di if oldest_di is not None and oldest_di < month_start: break if len(batch) < page_size: break page += 1 if page > 500: print("Stopped at page 500 (safety)", file=sys.stderr) break return all_rows def summarize_month(year: int, month: int, rows: list[dict]) -> tuple[int, int, float]: issued = [q for q in rows if in_calendar_month(parse_ts(q.get("DateIssued")), year, month)] created = len(issued) converted = sum(1 for q in issued if is_converted_status(q.get("Status"))) pct = (converted / created * 100.0) if created else 0.0 return created, converted, pct def main() -> int: env = {**load_dotenv_simple(ENV_PATH), **os.environ} base = (env.get("SIMPRO_BRANCH2_API_URL") or "").rstrip("/") key = env.get("SIMPRO_BRANCH2_API_KEY") or "" if not base or not key: print("Missing SIMPRO_BRANCH2_API_URL or SIMPRO_BRANCH2_API_KEY in .env", file=sys.stderr) return 1 label = fetch_company_label(base, key, COMPANY_ID) print("Simpro API — Bunbury Electrics") print(f" Company ID {COMPANY_ID} = {label}") print() # --- March 2026 (full month) — overlaps quote-data-bunbury.csv --- y, m = 2026, 3 rows = fetch_quotes_until_before_month(base, key, COMPANY_ID, y, m) created, converted, pct = summarize_month(y, m, rows) status_breakdown: dict[str, int] = {} for q in rows: if not in_calendar_month(parse_ts(q.get("DateIssued")), y, m): continue s = status_text(q.get("Status")) or "(empty)" status_breakdown[s] = status_breakdown.get(s, 0) + 1 print(f"March {y} (DateIssued in month, full month)") print(f" Quotes in cohort: {created}") print(f" Converted (status): {converted}") print(f" Conversion %: {pct:.1f}%") print() # --- April 2026 MTD (through 20th; matches user session date Apr 20, 2026) --- y2, m2 = 2026, 4 last = 20 rows2 = fetch_quotes_until_before_month(base, key, COMPANY_ID, y2, m2) start = datetime(y2, m2, 1) end = datetime(y2, m2, last, 23, 59, 59) apr = [q for q in rows2 if in_datetime_range(parse_ts(q.get("DateIssued")), start, end)] c2 = sum(1 for q in apr if is_converted_status(q.get("Status"))) p2 = (c2 / len(apr) * 100.0) if apr else 0.0 print(f"April {y2} MTD (DateIssued Apr 1–{last})") print(f" Quotes in cohort: {len(apr)}") print(f" Converted (status): {c2}") print(f" Conversion %: {p2:.1f}%") print() print("Status breakdown (March, issued in month, top 8):") for s, n in sorted(status_breakdown.items(), key=lambda x: -x[1])[:8]: print(f" {n:4d} {s[:90]}") print() print("Compare to quote-data-bunbury.csv (Mar 2026 snapshot): ELECTRICS BUNBURY shows 17 created / 16 converted") print(" → 94.1% in that export. Raw API DateIssued counts are much larger — BI is using different filters/definitions.") print(" Align your Simpro report filters (quote types, stages, etc.) to the API query or expect a gap.") return 0 if __name__ == "__main__": raise SystemExit(main())