jm/leads-optimizer
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Fix Slack monthly summary: correct investment, conversions and margin calculations

Browse Source 
- Use CosteMes and ConvMes from all previous month fco_ campaigns (not just
  those active in current month) to avoid missing paused campaigns
- Show day-1 cierre block with Google Ads data labeled explicitly
- Show current month as 0 on day 1 (no data yet)
- Fetch Campaign Name in get_metricas_diarias_prev_month for fco_ filtering
- Fetch prev month data for days 1-5 (last 5 days can span two months)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Jose Manuel 2026-05-01 10:55:58 +02:00
parent 64fb151de9
commit a85bc90b80
3 changed files with 220 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
airtable_client.py
View File

@ -379,6 +379,59 @@ class AirtableClient:
for i in range(0, len(batch), 10):
self.gacampaignmes.batch_update(batch[i:i+10])
def get_metricas_diarias_prev_month(self) -> dict:
"""
Devuelve {google_campaign_id: {metricas, coste_mes, conv_mes}} del mes anterior.
Usado en el reporter de Slack cuando hay cambio de mes.
"""
now = datetime.now()
prev_month = now.month - 1 if now.month > 1 else 12
prev_year = now.year if now.month > 1 else now.year - 1
formula = f"AND({{Mes}}='{prev_month}',{{Año}}='{prev_year}')"
records = self.gacampaignmes.all(
formula=formula,
fields=["CampaignID", "MetricasDiarias", "CosteMes", "ConvMes",
"Campaign Name (from CampaignID)"],
)
campaigns_records = self.campaigns.all(fields=["CampaignID"])
at_id_to_gid = {r["id"]: str(r["fields"].get("CampaignID", "")).strip() for r in campaigns_records}
result = {}
for r in records:
at_cids = r["fields"].get("CampaignID", [])
metricas = r["fields"].get("MetricasDiarias") or "{}"
coste = float(r["fields"].get("CosteMes") or 0)
conv = float(r["fields"].get("ConvMes") or 0)
name_list = r["fields"].get("Campaign Name (from CampaignID)", [])
name = name_list[0] if name_list else ""
if at_cids:
gid = at_id_to_gid.get(at_cids[0], "")
if gid:
result[gid] = {
"metricas": metricas,
"coste_mes": coste,
"conv_mes": conv,
"nombre": name,
}
return result
def get_gcm_id_map_for_month(self, mes: int, anio: int) -> dict:
"""
Devuelve {google_campaign_id: gcm_record_id} para el mes/año indicado.
Útil para escribir MetricasDiarias en el mes correcto cuando hay cambio de mes.
"""
formula = f"AND({{Mes}}='{mes}',{{Año}}='{anio}')"
records = self.gacampaignmes.all(formula=formula, fields=["CampaignID"])
campaigns_records = self.campaigns.all(fields=["CampaignID"])
at_id_to_gid = {r["id"]: str(r["fields"].get("CampaignID", "")).strip() for r in campaigns_records}
result = {}
for r in records:
at_cids = r["fields"].get("CampaignID", [])
if at_cids:
gid = at_id_to_gid.get(at_cids[0], "")
if gid:
result[gid] = r["id"]
return result
def batch_update_metricas_diarias(self, updates: list[dict]) -> None:
"""
Actualiza MetricasDiarias en GACampaignMes.

18
run.py
View File

@ -218,7 +218,9 @@ def run():
advice_updates = [] # (gcm_record_id, consejo, criticidad) para batch update final
metricas_updates = [] # {airtable_id, metricas_json} para MetricasDiarias
from datetime import timedelta
dia_hoy = (datetime.now() - timedelta(days=1)).strftime("%d")
ayer = datetime.now() - timedelta(days=1)
dia_hoy = ayer.strftime("%d")
cambio_mes = ayer.month != datetime.now().month
last_priority = -1
for item in collected:
@ -337,6 +339,17 @@ def run():
# Guardar métricas diarias en MetricasDiarias
if metricas_updates:
print(f"→ Actualizando MetricasDiarias ({len(metricas_updates)} registros)...")
if cambio_mes:
# Ayer pertenece al mes anterior: redirigir escritura al GACampaignMes correcto
prev_map = at.get_gcm_id_map_for_month(ayer.month, ayer.year)
for u in metricas_updates:
gid = next(
(item["campaign"]["google_campaign_id"]
for item in collected if item["campaign"]["airtable_id"] == u["airtable_id"]),
None,
)
if gid and gid in prev_map:
u["airtable_id"] = prev_map[gid]
at.batch_update_metricas_diarias(metricas_updates)
print(" ✓ MetricasDiarias actualizado.")
@ -372,7 +385,8 @@ def run():
# Enviar resumen a Slack
print("→ Enviando resumen a Slack...")
build_and_send(collected, config.DRY_RUN)
prev_month_metricas = at.get_metricas_diarias_prev_month() if (cambio_mes or datetime.now().day <= 5) else {}
build_and_send(collected, config.DRY_RUN, prev_month_metricas)
print(" ✓ Resumen enviado a Slack.")

165
slack_reporter.py
View File

@ -1,7 +1,7 @@
import json
import requests
import config
from datetime import datetime
from datetime import datetime, timedelta, date
ALERT_LOSS_EUR = 200 # pérdida absoluta > 200€ → alerta
@ -9,6 +9,13 @@ ALERT_MARGIN_PCT = -50 # margen % < -50% → alerta
TOP_N = 5 # campañas a mostrar en rankings
MESES_ES = {
1: "Enero", 2: "Febrero", 3: "Marzo", 4: "Abril",
5: "Mayo", 6: "Junio", 7: "Julio", 8: "Agosto",
9: "Septiembre", 10: "Octubre", 11: "Noviembre", 12: "Diciembre",
}
def _parse_metricas(metricas_json: str) -> dict:
try:
return json.loads(metricas_json) if metricas_json else {}
@ -16,13 +23,34 @@ def _parse_metricas(metricas_json: str) -> dict:
return {}
def _last_n_days_sum(md: dict, n: int) -> dict:
days = sorted(md.keys())[-n:]
def _last_n_days_combined(current_md: dict, prev_md: dict, now: datetime, n: int) -> dict:
"""
Combina MetricasDiarias del mes actual y del anterior para calcular los últimos n días.
Las claves son strings de día ("01"…"31"); para ordenar por fecha real se añade el mes.
"""
entries = []
prev_month = now.month - 1 if now.month > 1 else 12
prev_year = now.year if now.month > 1 else now.year - 1
for day_str, vals in prev_md.items():
try:
entries.append((date(prev_year, prev_month, int(day_str)), vals))
except ValueError:
pass
for day_str, vals in current_md.items():
try:
entries.append((date(now.year, now.month, int(day_str)), vals))
except ValueError:
pass
entries.sort(key=lambda x: x[0])
last_n = entries[-n:]
return {
"coste": round(sum(md[d].get("coste", 0) for d in days), 2),
"ingreso": round(sum(md[d].get("ingreso", 0) for d in days), 2),
"margen": round(sum(md[d].get("margen", 0) for d in days), 2),
"n_days": len(days),
"coste": round(sum(v.get("coste", 0) for _, v in last_n), 2),
"ingreso": round(sum(v.get("ingreso", 0) for _, v in last_n), 2),
"margen": round(sum(v.get("margen", 0) for _, v in last_n), 2),
"n_days": len(last_n),
}
@ -31,38 +59,70 @@ def _fmt_eur(v: float) -> str:
return f"{sign}{v:,.0f}€".replace(",", ".")
def _curso(name: str, max_len: int = 42) -> str:
def _curso(name: str, max_len: int = 40) -> str:
return name[:max_len] + ("…" if len(name) > max_len else "")
def build_and_send(collected: list, dry_run: bool) -> None:
def build_and_send(collected: list, dry_run: bool, prev_month_metricas: dict = None) -> None:
if not config.SLACK_WEBHOOK_URL:
print(" ⚠️ SLACK_WEBHOOK_URL no configurada, omitiendo envío.")
return
now = datetime.now()
ayer = now - timedelta(days=1)
cambio_mes = ayer.month != now.month
mes_sumatorio = MESES_ES[ayer.month] if cambio_mes else None
prev_md_map = prev_month_metricas or {}
fco = [item for item in collected if item["campaign"]["curso"].lower().startswith("fco_")]
# ── Totales del mes ──────────────────────────────────────────────────────
inv_total = round(sum(item["metrics"].get("cost", 0) for item in fco), 2)
primer_dia_mes = now.day == 1
# ── Totales del mes en curso ─────────────────────────────────────────────
inv_total = round(sum(item["metrics"].get("cost", 0) for item in fco), 2)
conv_total = int(sum(item["analysis"]["conversiones_google"] for item in fco))
ing_leads_ppl = round(sum(item["analysis"]["revenue_estimado"] for item in fco), 2)
if primer_dia_mes:
# Mes recién empezado: no hay datos de sumatorio para el mes en curso
ing_sumatorio = 0.0
margen_sumatorio = 0.0
pct_sumatorio = 0.0
margen_leads_ppl = 0.0
pct_leads_ppl = 0.0
else:
ing_sumatorio = round(sum(
sum(d.get("ingreso", 0) for d in _parse_metricas(item["campaign"].get("metricas_diarias", "{}")).values())
for item in fco
), 2)
ing_leads_ppl = round(sum(item["analysis"]["revenue_estimado"] for item in fco), 2)
margen_sumatorio = round(ing_sumatorio - inv_total, 2)
margen_leads_ppl = round(ing_leads_ppl - inv_total, 2)
pct_sumatorio = round(margen_sumatorio / ing_sumatorio * 100, 1) if ing_sumatorio > 0 else 0.0
pct_leads_ppl = round(margen_leads_ppl / ing_leads_ppl * 100, 1) if ing_leads_ppl > 0 else 0.0
# ── Últimos 5 días (desde MetricasDiarias) ───────────────────────────────
# ── Totales del mes anterior (solo día 1) ────────────────────────────────
if primer_dia_mes and prev_md_map:
# Iterar sobre TODAS las campañas del mes anterior filtrando fco_
# (no solo las activas en el mes en curso, para no perder campañas pausadas)
prev_fco = [v for v in prev_md_map.values() if v.get("nombre", "").lower().startswith("fco_")]
prev_inv = round(sum(v.get("coste_mes", 0) for v in prev_fco), 2)
prev_conv = int(sum(v.get("conv_mes", 0) for v in prev_fco))
prev_ing = round(sum(
sum(d.get("ingreso", 0) for d in _parse_metricas(v.get("metricas", "{}")).values())
for v in prev_fco
), 2)
prev_margen = round(prev_ing - prev_inv, 2)
prev_pct = round(prev_margen / prev_ing * 100, 1) if prev_ing > 0 else 0.0
else:
prev_inv = prev_conv = prev_ing = prev_margen = prev_pct = None
# ── Últimos 5 días (combinando meses si hay cambio de mes) ───────────────
last5_rows = []
for item in fco:
md = _parse_metricas(item["campaign"].get("metricas_diarias", "{}"))
s = _last_n_days_sum(md, 5)
gid = item["campaign"]["google_campaign_id"]
current_md = _parse_metricas(item["campaign"].get("metricas_diarias", "{}"))
prev_md = _parse_metricas(prev_md_map.get(gid, {}).get("metricas", "{}"))
s = _last_n_days_combined(current_md, prev_md, now, 5)
if s["n_days"] == 0:
continue
last5_rows.append({
@ -80,12 +140,10 @@ def build_and_send(collected: list, dry_run: bool) -> None:
for item in fco:
cost = item["metrics"].get("cost", 0)
rev = item["analysis"]["revenue_estimado"]
loss = round(rev - cost, 2)
pct = round(item["analysis"]["margen"] * 100, 1)
month_rows.append({
"curso": item["campaign"]["curso"],
"margen": loss,
"margen_pct": pct,
"margen": round(rev - cost, 2),
"margen_pct": round(item["analysis"]["margen"] * 100, 1),
"ingreso": round(rev, 2),
"coste": round(cost, 2),
})
@ -113,15 +171,35 @@ def build_and_send(collected: list, dry_run: bool) -> None:
"text": {
"type": "mrkdwn",
"text": (
f"📊 *RESUMEN DEL MES*\n"
f"Inversión: *{inv_total:,.0f}€*\n"
f"📊 *RESUMEN DEL MES EN CURSO ({MESES_ES[now.month]})*\n"
f"Inversión: *{inv_total:,.0f}€* | Conversiones: *{conv_total}*\n"
+ (
f"Ingreso por sumatorio: *0€* | Margen por sumatorio: *0€*\n"
f"Ingreso LeadsxPPL: *0€* | Margen por LeadsxPPL: *0€*"
if primer_dia_mes else
f"Ingreso por sumatorio: *{ing_sumatorio:,.0f}€* | Margen por sumatorio: *{_fmt_eur(margen_sumatorio)}* ({pct_sumatorio}%)\n"
f"Ingreso LeadsxPPL: *{ing_leads_ppl:,.0f}€* | Margen por LeadsxPPL: *{_fmt_eur(margen_leads_ppl)}* ({pct_leads_ppl}%)"
)
).replace(",", "."),
},
},
]
if primer_dia_mes and prev_inv is not None:
blocks.append({"type": "divider"})
blocks.append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
f"📅 *CIERRE {MESES_ES[ayer.month].upper()} (mes anterior)*\n"
f"_Datos de Google Ads_\n"
f"Inversión: *{prev_inv:,.0f}€* | Conversiones: *{prev_conv}*\n"
f"Ingreso por sumatorio: *{prev_ing:,.0f}€* | Margen: *{_fmt_eur(prev_margen)}* ({prev_pct}%)"
).replace(",", "."),
},
})
if alerts:
alert_lines = "\n".join(
f" 🔴 `{_curso(a['curso'])}` → Pérdida *{_fmt_eur(a['margen'])}* ({a['margen_pct']}%)"
@ -136,41 +214,52 @@ def build_and_send(collected: list, dry_run: bool) -> None:
},
})
blocks.append({"type": "divider"})
def _ranking_text(rows, label, show_pct=False):
lines = []
def _ranking_last5_text(rows, label):
lines = [label]
for i, r in enumerate(rows, 1):
pct_str = f" ({r['margen_pct']}%)" if show_pct else ""
lines.append(f" {i}. `{_curso(r['curso'])}` → *{_fmt_eur(r['margen'])}*{pct_str}")
return f"{label}\n" + "\n".join(lines)
lines.append(f" {i}. `{_curso(r['curso'])}`")
lines.append(
f" Coste: {r['coste']:,.0f}€ | Ingreso: {r['ingreso']:,.0f}€ | Margen: *{_fmt_eur(r['margen'])}*"
.replace(",", ".")
)
return "\n".join(lines)
def _ranking_month_text(rows, label):
lines = [label]
for i, r in enumerate(rows, 1):
lines.append(
f" {i}. `{_curso(r['curso'])}` → *{_fmt_eur(r['margen'])}* ({r['margen_pct']}%)"
)
return "\n".join(lines)
blocks.append({"type": "divider"})
blocks.append({
"type": "section",
"fields": [
{
"text": {
"type": "mrkdwn",
"text": _ranking_text(worst_last5, "📉 *PEOR — ÚLTIMOS 5 DÍAS*"),
"text": _ranking_last5_text(worst_last5, "📉 *PEOR — ÚLTIMOS 5 DÍAS*"),
},
{
})
blocks.append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": _ranking_text(best_last5, "📈 *MEJOR — ÚLTIMOS 5 DÍAS*"),
"text": _ranking_last5_text(best_last5, "📈 *MEJOR — ÚLTIMOS 5 DÍAS*"),
},
],
})
if not primer_dia_mes:
blocks.append({"type": "divider"})
blocks.append({
"type": "section",
"fields": [
{
"type": "mrkdwn",
"text": _ranking_text(worst_month, "📉 *PEOR — MES EN CURSO*", show_pct=True),
"text": _ranking_month_text(worst_month, "📉 *PEOR — MES EN CURSO*"),
},
{
"type": "mrkdwn",
"text": _ranking_text(best_month, "📈 *MEJOR — MES EN CURSO*", show_pct=True),
"text": _ranking_month_text(best_month, "📈 *MEJOR — MES EN CURSO*"),
},
],
})