fix: alinear KPIs BPS al milímetro con Ads Manager y Clientify Online

Meta solo cuenta registration leads; Google guarda fracciones y redondea al sumar; ventas CRM usan pipeline Online con importe > 0.

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

scripts/dokploy_apply_ad_daily_leads_numeric.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+"""Apply sql/ad_daily_leads_numeric.sql via Dokploy supabase-db terminal."""
+from __future__ import annotations
+
+import base64
+import os
+import re
+import sys
+from pathlib import Path
+
+from playwright.sync_api import sync_playwright
+
+ROOT = Path("/Users/4ventures/4ventures-client-portal")
+BPS = Path("/Users/4ventures/Projects/bps-facturacion")
+SQL = ROOT / "sql" / "ad_daily_leads_numeric.sql"
+BASE = "https://dokploy.4ventures.es"
+
+
+def env(key: str) -> str:
+    v = os.environ.get(key, "").strip()
+    if v:
+        return v
+    p = BPS / ".env"
+    if not p.is_file():
+        return ""
+    for line in p.read_text().splitlines():
+        if line.startswith(f"{key}="):
+            return line.split("=", 1)[1].strip().strip('"').strip("'")
+    return ""
+
+
+def main() -> int:
+    email = env("DOKPLOY_EMAIL") or "marketing@4ventures.es"
+    password = env("DOKPLOY_PASSWORD")
+    if not password:
+        print("NO_PASSWORD", flush=True)
+        return 2
+    b64 = base64.b64encode(SQL.read_text(encoding="utf-8").encode()).decode()
+    chunks = [b64[i : i + 180] for i in range(0, len(b64), 180)]
+
+    with sync_playwright() as p:
+        browser = p.chromium.launch(headless=True)
+        page = browser.new_page(viewport={"width": 1400, "height": 1000})
+        page.goto(BASE + "/", wait_until="domcontentloaded", timeout=90000)
+        page.wait_for_timeout(800)
+        if "/dashboard" not in page.url:
+            page.locator('input[type="email"], input[name="email"]').first.fill(email)
+            page.locator('input[type="password"]').first.fill(password)
+            page.locator('button[type="submit"]').first.click()
+            page.wait_for_url(re.compile(r"/dashboard"), timeout=90000)
+        print("login ok", flush=True)
+        page.goto(f"{BASE}/dashboard/docker", wait_until="domcontentloaded", timeout=120000)
+        page.wait_for_timeout(800)
+        page.locator("input[placeholder*='ilter' i], input[placeholder*='earch' i]").first.fill(
+            "supabase-db"
+        )
+        page.wait_for_timeout(1000)
+        page.keyboard.press("Escape")
+        page.wait_for_timeout(200)
+        page.get_by_role("button", name=re.compile(r"Open menu", re.I)).first.click()
+        page.wait_for_timeout(400)
+        page.get_by_text("Terminal", exact=True).click()
+        page.wait_for_timeout(1000)
+        if page.locator("button", has_text="Bash").count():
+            page.locator("button", has_text="Bash").first.click()
+        page.wait_for_timeout(2800)
+        if page.locator("textarea.xterm-helper-textarea").count():
+            page.locator("textarea.xterm-helper-textarea").first.click()
+
+        def run(cmd: str, wait: int = 800) -> None:
+            page.keyboard.type(cmd, delay=1)
+            page.keyboard.press("Enter")
+            page.wait_for_timeout(wait)
+
+        run("rm -f /tmp/adl.b64 /tmp/adl.sql; echo READY", 1200)
+        for i, ch in enumerate(chunks):
+            op = ">" if i == 0 else ">>"
+            run(f"printf '%s' '{ch}' {op} /tmp/adl.b64", 400)
+        run(
+            "base64 -d /tmp/adl.b64 > /tmp/adl.sql && "
+            "psql -U postgres -d postgres -v ON_ERROR_STOP=1 -f /tmp/adl.sql; echo EXIT:$?",
+            8000,
+        )
+        text = page.evaluate(
+            "() => (document.querySelector('.xterm-rows')||{}).innerText||''"
+        )
+        browser.close()
+        ok = "EXIT:0" in text
+        print("SQL_OK", ok, flush=True)
+        if not ok:
+            print(text[-1500:], flush=True)
+        return 0 if ok else 6
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/dokploy_apply_deals_pipeline.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+"""Apply sql/deals_pipeline.sql via Dokploy supabase-db terminal."""
+from __future__ import annotations
+
+import base64
+import os
+import re
+import sys
+from pathlib import Path
+
+from playwright.sync_api import sync_playwright
+
+ROOT = Path("/Users/4ventures/4ventures-client-portal")
+BPS = Path("/Users/4ventures/Projects/bps-facturacion")
+SQL = ROOT / "sql" / "deals_pipeline.sql"
+BASE = "https://dokploy.4ventures.es"
+
+
+def env(key: str) -> str:
+    v = os.environ.get(key, "").strip()
+    if v:
+        return v
+    p = BPS / ".env"
+    if not p.is_file():
+        return ""
+    for line in p.read_text().splitlines():
+        if line.startswith(f"{key}="):
+            return line.split("=", 1)[1].strip().strip('"').strip("'")
+    return ""
+
+
+def main() -> int:
+    email = env("DOKPLOY_EMAIL") or "marketing@4ventures.es"
+    password = env("DOKPLOY_PASSWORD")
+    if not password:
+        print("NO_PASSWORD", flush=True)
+        return 2
+    b64 = base64.b64encode(SQL.read_text(encoding="utf-8").encode()).decode()
+    chunks = [b64[i : i + 180] for i in range(0, len(b64), 180)]
+
+    with sync_playwright() as p:
+        browser = p.chromium.launch(headless=True)
+        page = browser.new_page(viewport={"width": 1400, "height": 1000})
+        page.goto(BASE + "/", wait_until="domcontentloaded", timeout=90000)
+        page.wait_for_timeout(800)
+        if "/dashboard" not in page.url:
+            page.locator('input[type="email"], input[name="email"]').first.fill(email)
+            page.locator('input[type="password"]').first.fill(password)
+            page.locator('button[type="submit"]').first.click()
+            page.wait_for_url(re.compile(r"/dashboard"), timeout=90000)
+        print("login ok", flush=True)
+        page.goto(f"{BASE}/dashboard/docker", wait_until="domcontentloaded", timeout=120000)
+        page.wait_for_timeout(800)
+        page.locator("input[placeholder*='ilter' i], input[placeholder*='earch' i]").first.fill(
+            "supabase-db"
+        )
+        page.wait_for_timeout(1000)
+        page.keyboard.press("Escape")
+        page.wait_for_timeout(200)
+        page.get_by_role("button", name=re.compile(r"Open menu", re.I)).first.click()
+        page.wait_for_timeout(400)
+        page.get_by_text("Terminal", exact=True).click()
+        page.wait_for_timeout(1000)
+        if page.locator("button", has_text="Bash").count():
+            page.locator("button", has_text="Bash").first.click()
+        page.wait_for_timeout(2800)
+        if page.locator("textarea.xterm-helper-textarea").count():
+            page.locator("textarea.xterm-helper-textarea").first.click()
+
+        def run(cmd: str, wait: int = 800) -> None:
+            page.keyboard.type(cmd, delay=1)
+            page.keyboard.press("Enter")
+            page.wait_for_timeout(wait)
+
+        run("rm -f /tmp/dp.b64 /tmp/dp.sql; echo READY", 1200)
+        for i, ch in enumerate(chunks):
+            op = ">" if i == 0 else ">>"
+            run(f"printf '%s' '{ch}' {op} /tmp/dp.b64", 400)
+        run(
+            "base64 -d /tmp/dp.b64 > /tmp/dp.sql && "
+            "psql -U postgres -d postgres -v ON_ERROR_STOP=1 -f /tmp/dp.sql; echo EXIT:$?",
+            8000,
+        )
+        text = page.evaluate(
+            "() => (document.querySelector('.xterm-rows')||{}).innerText||''"
+        )
+        browser.close()
+        ok = "EXIT:0" in text
+        print("SQL_OK", ok, flush=True)
+        if not ok:
+            print(text[-1200:], flush=True)
+        return 0 if ok else 6
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/sync_clientify_tenant.py

@@ -277,6 +277,10 @@ def main() -> int:
             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]
             row = {
                 "id": str(uuid.uuid5(uuid.NAMESPACE_URL, f"clientify-deal:{tid}:{did}")),
                 "tenant_id": tid,
@@ -286,6 +290,8 @@ def main() -> int:
                 "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),
             }
             rows.append(row)
 

sql/ad_daily_leads_numeric.sql

@@ -0,0 +1,5 @@
+-- Store fractional Google Ads conversions so period totals match Ads Manager
+-- (Math.round of the exact sum) for any date range.
+ALTER TABLE public.ad_daily
+  ALTER COLUMN platform_leads TYPE numeric(14, 6)
+  USING platform_leads::numeric(14, 6);

sql/deals_pipeline.sql

@@ -0,0 +1,4 @@
+-- Pipeline on deals so KPI ventas can match Clientify board filters (e.g. Online).
+ALTER TABLE public.deals
+  ADD COLUMN IF NOT EXISTS pipeline_id text,
+  ADD COLUMN IF NOT EXISTS pipeline_name text;

src/app/api/webhooks/clientify/route.ts

@@ -268,6 +268,11 @@ async function upsertDeal(tenantId: string, d: Json) {
     }
   }
   const closed = str(d.actual_closed_date, 40);
+  const pipeUrl = str(d.pipeline, 200);
+  let pipelineId: string | null = null;
+  if (pipeUrl.includes('/pipelines/')) {
+    pipelineId = pipeUrl.replace(/\/$/, '').split('/').pop() || null;
+  }
   const row = {
     id,
     tenant_id: tenantId,
@@ -277,6 +282,8 @@ async function upsertDeal(tenantId: string, d: Json) {
     amount: n(d.amount ?? d.amount_user),
     won_at: stage === 'won' && closed ? closed : null,
     lost_at: stage === 'lost' && closed ? closed : null,
+    pipeline_id: pipelineId,
+    pipeline_name: str(d.pipeline_desc, 120) || null,
   };
   const base = supabaseUrl();
   await fetch(

src/app/dashboard/kpis/page.tsx

@@ -138,7 +138,7 @@ async function fetchWonDeals(
   tenantId: string | null,
   useServiceRole: boolean
 ): Promise<DealWonRow[]> {
-  const select = 'amount,won_at,stage,tenant_id';
+  const select = 'amount,won_at,stage,tenant_id,pipeline_id,pipeline_name';
   const pageSize = 1000;
   const out: DealWonRow[] = [];
 

src/lib/ads-sync.ts

@@ -71,21 +71,13 @@ function leadFromMetaActions(
   actions?: { action_type?: string; value?: string }[]
 ): number {
   if (!actions?.length) return 0;
-  const prefer = [
+  // Exact Ads Manager «Clientes potenciales» for Meta Leads CRM campaigns.
-    'lead',
+  // Do not fall back to lead / lead_grouped / pixel — those inflate past the
-    'onsite_conversion.lead_grouped',
+  // Ads Manager period total when mixed with registration actions.
-    'offsite_conversion.fb_pixel_lead',
+  const primary = actions.find(
-    'offsite_complete_registration_add_meta_leads',
+    (a) => a.action_type === 'offsite_complete_registration_add_meta_leads'
-    'complete_registration',
-  ];
-  for (const key of prefer) {
-    const hit = actions.find((a) => a.action_type === key);
-    if (hit) return n(hit.value);
-  }
-  const anyLead = actions.find((a) =>
-    (a.action_type || '').toLowerCase().includes('lead')
   );
-  return anyLead ? n(anyLead.value) : 0;
+  return primary ? n(primary.value) : 0;
 }
 
 async function googleAccessToken(): Promise<string> {
@@ -167,8 +159,9 @@ async function fetchMetaInsightsWindow(
   const out: AdDailyInsert[] = [];
   let url: string | null =
     `${META_GRAPH}/${act}/insights` +
-    `?fields=campaign_id,campaign_name,ad_id,ad_name,impressions,clicks,spend,actions,date_start` +
+    // Campaign level: ad-level daily lead sums inflate vs Ads Manager period totals.
-    `&level=ad&time_increment=1&time_range=${encodeURIComponent(timeRange)}` +
+    `?fields=campaign_id,campaign_name,impressions,clicks,spend,actions,date_start` +
+    `&level=campaign&time_increment=1&time_range=${encodeURIComponent(timeRange)}` +
     `&limit=500&access_token=${encodeURIComponent(token)}`;
 
   while (url) {
@@ -180,8 +173,6 @@ async function fetchMetaInsightsWindow(
       data?: {
         campaign_id?: string;
         campaign_name?: string;
-        ad_id?: string;
-        ad_name?: string;
         impressions?: string;
         clicks?: string;
         spend?: string;
@@ -192,14 +183,16 @@ async function fetchMetaInsightsWindow(
     };
     for (const row of j.data || []) {
       if (!row.date_start) continue;
+      const campaignId = row.campaign_id || null;
       out.push({
         tenant_id: tenantId,
         date: row.date_start,
         platform: 'meta',
-        campaign_id: row.campaign_id || null,
+        campaign_id: campaignId,
         campaign_name: row.campaign_name || null,
-        ad_id: row.ad_id || null,
+        // campaign-level rows: reuse campaign id as ad_id (NOT NULL column)
-        ad_name: row.ad_name || null,
+        ad_id: campaignId || '',
+        ad_name: '',
         impressions: n(row.impressions),
         clicks: n(row.clicks),
         spend: Math.round(n(row.spend) * 100) / 100,
@@ -327,28 +320,46 @@ async function fetchGoogleWindow(
     throw new Error(`Google Ads ${cid}: ${lastErr.slice(0, 220)}`);
   }
 
-  const out: AdDailyInsert[] = [];
+  const rawRows: {
+    date: string;
+    campaignId: string | null;
+    campaignName: string | null;
+    impressions: number;
+    clicks: number;
+    spend: number;
+    conversions: number;
+  }[] = [];
   for (const row of j.results || []) {
     const date = row.segments?.date;
     if (!date) continue;
     const spend = Math.round((n(row.metrics?.costMicros) / 1_000_000) * 100) / 100;
     const campaignId = row.campaign?.id ? String(row.campaign.id) : null;
-    out.push({
+    rawRows.push({
-      tenant_id: tenantId,
       date,
-      platform: 'google',
+      campaignId,
-      campaign_id: campaignId,
+      campaignName: row.campaign?.name || null,
-      campaign_name: row.campaign?.name || null,
-      // ad_daily.ad_id is NOT NULL — campaign-level rows reuse campaign id
-      ad_id: campaignId || '',
-      ad_name: '',
       impressions: n(row.metrics?.impressions),
       clicks: n(row.metrics?.clicks),
       spend,
-      platform_leads: Math.round(n(row.metrics?.conversions)),
+      conversions: n(row.metrics?.conversions),
     });
   }
-  return out;
+
+  // Keep fractional Google conversions (numeric column). UI totals use
+  // Math.round(sum) once so any date range matches Ads Manager.
+  return rawRows.map((r) => ({
+    tenant_id: tenantId,
+    date: r.date,
+    platform: 'google',
+    campaign_id: r.campaignId,
+    campaign_name: r.campaignName,
+    ad_id: r.campaignId || '',
+    ad_name: '',
+    impressions: r.impressions,
+    clicks: r.clicks,
+    spend: r.spend,
+    platform_leads: Math.round(r.conversions * 1e6) / 1e6,
+  }));
 }
 
 async function fetchGoogleCampaignDaily(

src/lib/metrics.ts

@@ -76,7 +76,8 @@ export function sumDaily(rows: AdDailyRow[]): {
 } {
   return {
     spend: rows.reduce((a, r) => a + n(r.spend), 0),
-    leads: rows.reduce((a, r) => a + n(r.platform_leads), 0),
+    // Round once after summing fractional Google conversions (Ads Manager style).
+    leads: Math.round(rows.reduce((a, r) => a + n(r.platform_leads), 0)),
     impressions: rows.reduce((a, r) => a + n(r.impressions), 0),
     clicks: rows.reduce((a, r) => a + n(r.clicks), 0),
   };
@@ -488,23 +489,25 @@ export function channelTotalsFromDaily(rows: AdDailyRow[]): {
     map.set(channel, cur);
   }
   return [...map.entries()]
-    .map(([channel, v]) => ({
+    .map(([channel, v]) => {
+      const leads = Math.round(v.leads);
+      return {
         channel,
         spend: v.spend,
-      leads: v.leads,
+        leads,
         deals: 0,
         revenue: 0,
-      cpl: v.leads > 0 ? v.spend / v.leads : 0,
+        cpl: leads > 0 ? v.spend / leads : 0,
         roas: 0,
-    }))
+      };
+    })
     .sort((a, b) => b.spend - a.spend);
 }
 
 export function metricTotalsFromDaily(rows: AdDailyRow[]): MetricTotals {
   const spend = rows.reduce((a, r) => a + n(r.spend), 0);
-  const leads = rows.reduce(
+  const leads = Math.round(
-    (a, r) => a + n(r.platform_leads ?? r.conversions),
+    rows.reduce((a, r) => a + n(r.platform_leads ?? r.conversions), 0)
-    0
   );
   return {
     spend,
@@ -525,14 +528,32 @@ export type DealWonRow = {
   tenant_id?: string | null;
   /** Optional source/channel when joined from lead or Clientify contact. */
   source?: string | null;
+  pipeline_id?: string | null;
+  pipeline_name?: string | null;
 };
 
+/**
+ * Match Clientify board «Online» (paid won). If the tenant has an Online
+ * pipeline, prefer it so totals match the board; otherwise keep all paid won.
+ */
+export function preferOnlinePaidDeals(deals: DealWonRow[]): DealWonRow[] {
+  const paid = deals.filter((d) => n(d.amount) > 0);
+  const online = paid.filter((d) => {
+    const id = String(d.pipeline_id || '');
+    const name = d.pipeline_name || '';
+    return id === '77547' || /online/i.test(name);
+  });
+  return online.length ? online : paid;
+}
+
 export function filterDealsForRange(
   deals: DealWonRow[],
   since: string | null,
   until: string | null
 ): DealWonRow[] {
-  const won = deals.filter((d) => !d.stage || d.stage === 'won');
+  const won = preferOnlinePaidDeals(
+    deals.filter((d) => !d.stage || d.stage === 'won')
+  );
   if (!since || !until) return won;
   return won.filter((d) => {
     const day = (d.won_at || '').slice(0, 10);
@@ -566,7 +587,7 @@ export function mergeDealsIntoWeekly(
   deals: DealWonRow[]
 ): KpiWeeklyRow[] {
   const byWeek = new Map<string, { deals: number; revenue: number }>();
-  for (const d of deals) {
+  for (const d of preferOnlinePaidDeals(deals)) {
     if (d.stage && d.stage !== 'won') continue;
     const day = (d.won_at || '').slice(0, 10);
     if (!day) continue;
@@ -901,7 +922,7 @@ export function attributionCross(
     cur.crmLeads += 1;
   }
 
-  for (const d of deals) {
+  for (const d of preferOnlinePaidDeals(deals)) {
     if (d.stage && d.stage !== 'won') continue;
     const raw = (d.source || '').trim();
     const channel = raw ? normalizeLeadChannel(raw) : 'sin-atribuir';
@@ -911,19 +932,22 @@ export function attributionCross(
   }
 
   return [...map.entries()]
-    .map(([channel, v]) => ({
+    .map(([channel, v]) => {
+      const platformLeads = Math.round(v.platformLeads);
+      return {
         channel,
         spend: v.spend,
-      platformLeads: v.platformLeads,
+        platformLeads,
         crmLeads: v.crmLeads,
-      gap: v.crmLeads - v.platformLeads,
+        gap: v.crmLeads - platformLeads,
-      cplPlatform: v.platformLeads > 0 ? v.spend / v.platformLeads : 0,
+        cplPlatform: platformLeads > 0 ? v.spend / platformLeads : 0,
         cplCrm: v.crmLeads > 0 ? v.spend / v.crmLeads : 0,
         ventas: v.ventas,
         importe: v.importe,
         roas: v.spend > 0 ? v.importe / v.spend : 0,
         cac: v.ventas > 0 ? v.spend / v.ventas : 0,
-    }))
+      };
+    })
     .sort(
       (a, b) =>
         b.spend - a.spend ||