feat: KPIs con totales absolutos, backfill ads y ventas en cruce canal

Comparativas WoW/MoM/YoY muestran prev/Δ además del %; sync ads admite
histórico desde 2025; tabla ads↔CRM añade cierres, ventas y ROAS.

Co-authored-by: Cursor <cursoragent@cursor.com>

scripts/sync_clientify_tenant.py

@@ -0,0 +1,316 @@
+#!/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_<SLUG>)
+  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"))
+            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,
+            }
+            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())