#!/usr/bin/env python3 """Sync Clientify contacts → leads and deals → deals for one tenant. Requires in .env.local: CLIENTIFY_API_KEY (or CLIENTIFY_API_KEY_) NEXT_PUBLIC_SUPABASE_URL SUPABASE_SERVICE_ROLE_KEY Usage: python3 scripts/sync_clientify_tenant.py --tenant-id a7a6eb49-... --label agua-miia """ from __future__ import annotations import argparse import json import os import sys import urllib.error import urllib.parse import urllib.request import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] CFY = "https://api.clientify.com/v1" # Clientify deal status_desc → warehouse stage STAGE_MAP = { "won": "won", "lost": "lost", "open": "open", } def load_env() -> None: for line in (ROOT / ".env.local").read_text().splitlines(): if "=" in line and not line.startswith("#"): k, v = line.split("=", 1) k = k.strip() if k and k not in os.environ: os.environ[k] = v.strip().strip('"').strip("'") def env_key(label: str) -> str: slug = label.upper().replace("-", "_").replace(" ", "_") return ( os.environ.get(f"CLIENTIFY_API_KEY_{slug}") or os.environ.get("CLIENTIFY_API_KEY") or "" ) def sb_headers() -> dict[str, str]: key = os.environ["SUPABASE_SERVICE_ROLE_KEY"] return { "apikey": key, "Authorization": f"Bearer {key}", "Content-Type": "application/json", } def sb(method: str, path: str, body: Any = None, prefer: str = "return=minimal"): url = os.environ["NEXT_PUBLIC_SUPABASE_URL"].rstrip("/") + path headers = dict(sb_headers()) if prefer: headers["Prefer"] = prefer data = None if body is None else json.dumps(body).encode() req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=120) as r: raw = r.read().decode() return r.status, (json.loads(raw) if raw.strip() else None) except urllib.error.HTTPError as e: return e.code, e.read().decode()[:500] def cfy_get(token: str, path: str) -> dict: req = urllib.request.Request( CFY + path, headers={"Authorization": f"Token {token}", "Accept": "application/json"}, ) with urllib.request.urlopen(req, timeout=60) as r: return json.loads(r.read()) def cfy_paginate(token: str, path: str, page_size: int = 100) -> list[dict]: out: list[dict] = [] # path may already have query sep = "&" if "?" in path else "?" url_path = f"{path}{sep}page_size={page_size}" page = 1 while True: q = f"{url_path}&page={page}" if "page=" not in url_path else url_path # rebuild clean base, _, query = path.partition("?") params = urllib.parse.parse_qs(query) params["page_size"] = [str(page_size)] params["page"] = [str(page)] q = base + "?" + urllib.parse.urlencode({k: v[0] for k, v in params.items()}) j = cfy_get(token, q) batch = j.get("results") or [] out.extend(batch) print(f" {path} page={page} +{len(batch)} total={len(out)}/{j.get('count')}", flush=True) if not j.get("next") or not batch: break page += 1 return out def first_email(c: dict) -> str | None: emails = c.get("emails") or [] if isinstance(emails, list) and emails: e = emails[0] if isinstance(e, dict): return (e.get("email") or e.get("value") or "").strip() or None return str(e).strip() or None return None def first_phone(c: dict) -> str | None: phones = c.get("phones") or [] if isinstance(phones, list) and phones: p = phones[0] if isinstance(p, dict): return (p.get("phone") or p.get("value") or "").strip() or None return str(p).strip() or None return None def contact_id_from_url(url: str | None) -> str | None: if not url: return None # https://api.clientify.com/v1/contacts/12345/ parts = url.rstrip("/").split("/") return parts[-1] if parts else None def map_stage(deal: dict) -> str: desc = (deal.get("status_desc") or "").strip().lower() if desc in STAGE_MAP: return STAGE_MAP[desc] st = deal.get("status") # Observed: 1 open-ish, 3 won, 4 lost (API filter unreliable) if st == 3 or desc == "won": return "won" if st == 4 or desc == "lost": return "lost" return "open" def parse_date(s: str | None) -> str | None: if not s: return None # date-only or iso try: if "T" in s: return datetime.fromisoformat(s.replace("Z", "+00:00")).astimezone(timezone.utc).isoformat() return datetime.fromisoformat(s + "T12:00:00+00:00").astimezone(timezone.utc).isoformat() except Exception: return None def main() -> int: load_env() ap = argparse.ArgumentParser() ap.add_argument("--tenant-id", required=True) ap.add_argument("--label", default="tenant") ap.add_argument("--contacts-only", action="store_true") ap.add_argument("--deals-only", action="store_true") args = ap.parse_args() token = env_key(args.label) if not token: print("missing CLIENTIFY_API_KEY", file=sys.stderr) return 2 tid = args.tenant_id print(f"Sync Clientify → warehouse tenant={tid} label={args.label}", flush=True) do_contacts = not args.deals_only do_deals = not args.contacts_only contact_to_lead: dict[str, str] = {} if do_contacts: print("Fetching contacts…", flush=True) contacts = cfy_paginate(token, "/contacts/") # wipe previous clientify leads for tenant st, _ = sb( "DELETE", f"/rest/v1/leads?tenant_id=eq.{tid}&crm=eq.clientify", ) print(f"DELETE leads → {st}", flush=True) rows = [] for c in contacts: cid = str(c["id"]) lid = str(uuid.uuid5(uuid.NAMESPACE_URL, f"clientify-lead:{tid}:{cid}")) contact_to_lead[cid] = lid source = ( (c.get("contact_source") or c.get("channel") or c.get("medium") or "clientify") ) if isinstance(source, str): source = source.strip().lower()[:80] or "clientify" else: source = "clientify" status = (c.get("status") or "new") if isinstance(status, str): status = status.strip().lower()[:40] or "new" else: status = "new" rows.append( { "id": lid, "tenant_id": tid, "first_name": ( " ".join( x for x in [(c.get("first_name") or "").strip(), (c.get("last_name") or "").strip()] if x ) or None ), "email": first_email(c), "phone": first_phone(c), "source": source, "medium": (c.get("medium") or None), "crm": "clientify", "crm_contact_id": cid, "status": status, "created_at": parse_date(c.get("created")) or datetime.now(timezone.utc).isoformat(), } ) ok = 0 for i in range(0, len(rows), 200): chunk = rows[i : i + 200] st, err = sb("POST", "/rest/v1/leads", chunk) if st not in (200, 201): print("FAIL leads", st, err, flush=True) return 1 ok += len(chunk) print(f" INSERT leads {ok}/{len(rows)}", flush=True) print(f"OK contacts→leads {ok}", flush=True) if do_deals: if not contact_to_lead and do_contacts is False: # load existing map st, data = sb( "GET", f"/rest/v1/leads?tenant_id=eq.{tid}&crm=eq.clientify&select=id,crm_contact_id&limit=20000", prefer="", ) if st == 200 and isinstance(data, list): for r in data: if r.get("crm_contact_id"): contact_to_lead[str(r["crm_contact_id"])] = r["id"] print("Fetching deals…", flush=True) deals = cfy_paginate(token, "/deals/") st, _ = sb( "DELETE", f"/rest/v1/deals?tenant_id=eq.{tid}&crm_deal_id=like.clientify:*", ) print(f"DELETE deals → {st}", flush=True) rows = [] by_stage = {"won": 0, "lost": 0, "open": 0} for d in deals: did = str(d["id"]) stage = map_stage(d) by_stage[stage] = by_stage.get(stage, 0) + 1 cid = contact_id_from_url(d.get("contact")) lead_id = contact_to_lead.get(cid) if cid else None amount = float(d.get("amount") or d.get("amount_user") or 0) closed = parse_date(d.get("actual_closed_date")) pipe_url = str(d.get("pipeline") or "") pipe_id = "" if "/pipelines/" in pipe_url: pipe_id = pipe_url.rstrip("/").split("/")[-1] raw_src = ( d.get("deal_source") or d.get("contact_source") or "" ) if isinstance(raw_src, str): raw_src = raw_src.strip()[:80] else: raw_src = "" tags = d.get("tags") or [] tag_blob = " ".join(str(t) for t in tags).lower() if isinstance(tags, list) else "" name = (d.get("name") or "")[:200] if not raw_src: if any(x in tag_blob for x in ("facebook", "instagram", "meta", " fb ")): raw_src = "meta" elif "google" in tag_blob: raw_src = "google" elif "woocommerce" in tag_blob or name.lower().startswith("order "): raw_src = "ecommerce" row = { "id": str(uuid.uuid5(uuid.NAMESPACE_URL, f"clientify-deal:{tid}:{did}")), "tenant_id": tid, "lead_id": lead_id, "crm_deal_id": f"clientify:{did}", "stage": stage, "amount": amount, "won_at": closed if stage == "won" else None, "lost_at": closed if stage == "lost" else None, "pipeline_id": pipe_id or None, "pipeline_name": (d.get("pipeline_desc") or None), "source": raw_src.lower() if raw_src else None, "name": name or None, } rows.append(row) ok = 0 for i in range(0, len(rows), 200): chunk = rows[i : i + 200] st, err = sb("POST", "/rest/v1/deals", chunk) if st not in (200, 201): print("FAIL deals", st, err, flush=True) return 1 ok += len(chunk) print(f" INSERT deals {ok}/{len(rows)}", flush=True) won_amt = sum(r["amount"] for r in rows if r["stage"] == "won") print(f"OK deals {ok} by_stage={by_stage} won_amount={won_amt:.2f}", flush=True) # touch integration now = datetime.now(timezone.utc).isoformat() sb( "PATCH", f"/rest/v1/integrations?tenant_id=eq.{tid}&provider=eq.clientify", {"status": "connected", "last_sync_at": now, "error_message": None}, ) print("DONE", flush=True) return 0 if __name__ == "__main__": raise SystemExit(main())