José Manuel Gómez
94745cfda3
- run.py: skip metricas update when campaign has no prior-month GACampaignMes record instead of silently writing into the new month - slack_reporter.py: round before formatting so small negative margins don't render as "-0€"
379 lines
15 KiB
Python
379 lines
15 KiB
Python
import json
|
||
import subprocess
|
||
import requests
|
||
import config
|
||
from datetime import datetime, timedelta, date
|
||
|
||
|
||
def _git_version() -> str:
|
||
try:
|
||
return subprocess.check_output(
|
||
["git", "rev-parse", "--short", "HEAD"],
|
||
stderr=subprocess.DEVNULL,
|
||
text=True,
|
||
).strip()
|
||
except Exception:
|
||
return "unknown"
|
||
|
||
|
||
ALERT_LOSS_EUR = 200 # pérdida absoluta > 200€ → alerta
|
||
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 {}
|
||
except (json.JSONDecodeError, TypeError):
|
||
return {}
|
||
|
||
|
||
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:]
|
||
total_coste = round(sum(v.get("coste", 0) for _, v in last_n), 2)
|
||
total_ingreso = round(sum(v.get("ingreso", 0) for _, v in last_n), 2)
|
||
return {
|
||
"coste": total_coste,
|
||
"ingreso": total_ingreso,
|
||
"margen": round(total_ingreso - total_coste, 2),
|
||
"n_days": len(last_n),
|
||
}
|
||
|
||
|
||
def _trunc(text: str, limit: int = 2950) -> str:
|
||
return text if len(text) <= limit else text[:limit - 3] + "…"
|
||
|
||
|
||
def _fmt_eur(v: float) -> str:
|
||
v_int = round(v)
|
||
sign = "+" if v_int > 0 else ""
|
||
return f"{sign}{v_int:,.0f}€".replace(",", ".")
|
||
|
||
|
||
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, prev_month_metricas: dict = None,
|
||
portfolio_analysis_text: str = 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_")]
|
||
|
||
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 = 0.0
|
||
coste_sumatorio = 0.0
|
||
for item in fco:
|
||
for d in _parse_metricas(item["campaign"].get("metricas_diarias", "{}")).values():
|
||
ing_sumatorio += d.get("ingreso", 0)
|
||
coste_sumatorio += d.get("coste", 0)
|
||
ing_sumatorio = round(ing_sumatorio, 2)
|
||
coste_sumatorio = round(coste_sumatorio, 2)
|
||
margen_sumatorio = round(ing_sumatorio - coste_sumatorio, 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
|
||
|
||
# ── 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:
|
||
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({
|
||
"curso": item["campaign"]["curso"],
|
||
"margen": s["margen"],
|
||
"ingreso": s["ingreso"],
|
||
"coste": s["coste"],
|
||
})
|
||
last5_rows.sort(key=lambda x: x["margen"])
|
||
worst_last5 = last5_rows[:TOP_N]
|
||
best_last5 = list(reversed(last5_rows[-TOP_N:]))
|
||
|
||
# ── Mes en curso ─────────────────────────────────────────────────────────
|
||
month_rows = []
|
||
for item in fco:
|
||
cost = item["metrics"].get("cost", 0)
|
||
rev = item["analysis"]["revenue_estimado"]
|
||
month_rows.append({
|
||
"curso": item["campaign"]["curso"],
|
||
"margen": round(rev - cost, 2),
|
||
"margen_pct": round(item["analysis"]["margen"] * 100, 1),
|
||
"ingreso": round(rev, 2),
|
||
"coste": round(cost, 2),
|
||
})
|
||
month_rows.sort(key=lambda x: x["margen"])
|
||
worst_month = month_rows[:TOP_N]
|
||
best_month = list(reversed(month_rows[-TOP_N:]))
|
||
|
||
# ── Tabla de márgenes diarios ─────────────────────────────────────────────
|
||
daily_totals: dict[int, dict] = {}
|
||
for item in fco:
|
||
ppl = item["campaign"].get("ppl", 0)
|
||
md = _parse_metricas(item["campaign"].get("metricas_diarias", "{}"))
|
||
for day_str, vals in md.items():
|
||
try:
|
||
d = int(day_str)
|
||
except ValueError:
|
||
continue
|
||
if d not in daily_totals:
|
||
daily_totals[d] = {"coste": 0.0, "ing_gads": 0.0, "ing_at": 0.0, "leads_gads": 0, "leads_at": 0}
|
||
daily_totals[d]["coste"] += vals.get("coste", 0)
|
||
daily_totals[d]["ing_gads"] += vals.get("ingreso", 0) # conv_hoy × PPL
|
||
daily_totals[d]["ing_at"] += vals.get("leads_lake", 0) * ppl # leads_lake × PPL
|
||
daily_totals[d]["leads_gads"] += int(vals.get("leads", 0))
|
||
daily_totals[d]["leads_at"] += int(vals.get("leads_lake", 0))
|
||
|
||
def _eur(v: float) -> str:
|
||
return f"{v:,.0f}€".replace(",", ".")
|
||
|
||
def _marg(v: float) -> str:
|
||
v_int = round(v)
|
||
return ("+" if v_int >= 0 else "") + f"{v_int:,.0f}€".replace(",", ".")
|
||
|
||
def _pct(v: float) -> str:
|
||
return ("+" if v >= 0 else "") + f"{v:.1f}%"
|
||
|
||
margin_table_block = None
|
||
if daily_totals and not primer_dia_mes:
|
||
hdr = f"{'Día':>3} {'L.AT':>5} {'L.GA':>5} {'Coste':>7} {'€.AT':>7} {'€.GA':>7} {'Margen':>9} {'%':>7}"
|
||
sep = "─" * len(hdr)
|
||
rows = [hdr, sep]
|
||
tot = {"coste": 0.0, "ing_gads": 0.0, "ing_at": 0.0, "leads_gads": 0, "leads_at": 0}
|
||
for d in sorted(daily_totals):
|
||
v = daily_totals[d]
|
||
ing = v["ing_gads"] # margen: conv Google Ads × PPL
|
||
marg = ing - v["coste"]
|
||
pct_v = round(marg / ing * 100, 1) if ing > 0 else 0.0
|
||
for k in tot:
|
||
tot[k] += v[k]
|
||
rows.append(
|
||
f"{d:02d} {v['leads_at']:>5} {v['leads_gads']:>5}"
|
||
f" {_eur(v['coste']):>7}"
|
||
f" {_eur(v['ing_at']):>7} {_eur(v['ing_gads']):>7}"
|
||
f" {_marg(marg):>9} {_pct(pct_v):>7}"
|
||
)
|
||
tot_ing = tot["ing_gads"]
|
||
tot_marg = tot_ing - tot["coste"]
|
||
tot_pct = round(tot_marg / tot_ing * 100, 1) if tot_ing > 0 else 0.0
|
||
rows.append(sep)
|
||
rows.append(
|
||
f"{'TOT':>3} {tot['leads_at']:>5} {tot['leads_gads']:>5}"
|
||
f" {_eur(tot['coste']):>7}"
|
||
f" {_eur(tot['ing_at']):>7} {_eur(tot['ing_gads']):>7}"
|
||
f" {_marg(tot_marg):>9} {_pct(tot_pct):>7}"
|
||
)
|
||
margin_table_block = {
|
||
"type": "section",
|
||
"text": {
|
||
"type": "mrkdwn",
|
||
"text": _trunc(f"📅 *MÉTRICAS POR DÍA — {MESES_ES[now.month].upper()}* _(L.AT=leads Airtable · L.GA=conv Google Ads · €.AT=fact×PPL Airtable · €.GA=fact×PPL Google Ads)_\n```\n" + "\n".join(rows) + "\n```"),
|
||
},
|
||
}
|
||
|
||
# ── Alertas ──────────────────────────────────────────────────────────────
|
||
alerts = [
|
||
r for r in month_rows
|
||
if r["margen"] < -ALERT_LOSS_EUR or r["margen_pct"] < ALERT_MARGIN_PCT
|
||
]
|
||
alerts.sort(key=lambda x: x["margen"])
|
||
|
||
# ── Construir bloques ─────────────────────────────────────────────────────
|
||
mode = "🔵 DRY RUN" if dry_run else "⚡ PRODUCCIÓN"
|
||
version = _git_version()
|
||
|
||
blocks = [
|
||
{
|
||
"type": "header",
|
||
"text": {"type": "plain_text", "text": f"LEADS OPTIMIZER v{version} — {now.strftime('%d/%m/%Y %H:%M')} {mode}"},
|
||
},
|
||
{
|
||
"type": "section",
|
||
"text": {
|
||
"type": "mrkdwn",
|
||
"text": (
|
||
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']}%)"
|
||
for a in alerts
|
||
)
|
||
blocks.append({"type": "divider"})
|
||
blocks.append({
|
||
"type": "section",
|
||
"text": {
|
||
"type": "mrkdwn",
|
||
"text": f"🚨 *ALERTAS — CAMPAÑAS CON PÉRDIDAS IMPORTANTES*\n{alert_lines}",
|
||
},
|
||
})
|
||
|
||
def _ranking_last5_text(rows, label):
|
||
lines = [label]
|
||
for i, r in enumerate(rows, 1):
|
||
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)
|
||
|
||
if margin_table_block:
|
||
blocks.append({"type": "divider"})
|
||
blocks.append(margin_table_block)
|
||
|
||
blocks.append({"type": "divider"})
|
||
blocks.append({
|
||
"type": "section",
|
||
"text": {
|
||
"type": "mrkdwn",
|
||
"text": _trunc(_ranking_last5_text(worst_last5, "📉 *PEOR — ÚLTIMOS 5 DÍAS*")),
|
||
},
|
||
})
|
||
blocks.append({
|
||
"type": "section",
|
||
"text": {
|
||
"type": "mrkdwn",
|
||
"text": _trunc(_ranking_last5_text(best_last5, "📈 *MEJOR — ÚLTIMOS 5 DÍAS*")),
|
||
},
|
||
})
|
||
|
||
if not primer_dia_mes:
|
||
blocks.append({"type": "divider"})
|
||
blocks.append({
|
||
"type": "section",
|
||
"text": {
|
||
"type": "mrkdwn",
|
||
"text": _trunc(_ranking_month_text(worst_month, "📉 *PEOR — MES EN CURSO*")),
|
||
},
|
||
})
|
||
blocks.append({
|
||
"type": "section",
|
||
"text": {
|
||
"type": "mrkdwn",
|
||
"text": _trunc(_ranking_month_text(best_month, "📈 *MEJOR — MES EN CURSO*")),
|
||
},
|
||
})
|
||
|
||
if portfolio_analysis_text:
|
||
blocks.append({"type": "divider"})
|
||
blocks.append({
|
||
"type": "section",
|
||
"text": {
|
||
"type": "mrkdwn",
|
||
"text": _trunc(f"🤖 *DIAGNÓSTICO ESTRATÉGICO*\n{portfolio_analysis_text}"),
|
||
},
|
||
})
|
||
|
||
payload = {"blocks": blocks}
|
||
try:
|
||
resp = requests.post(config.SLACK_WEBHOOK_URL, json=payload, timeout=10)
|
||
if resp.status_code != 200:
|
||
print(f" ⚠️ Slack respondió {resp.status_code}: {resp.text[:200]}")
|
||
except Exception as e:
|
||
print(f" ⚠️ Error enviando a Slack: {e}")
|