"""Slack notifier β€” Web API (Bot Token) con botones interactivos.""" from datetime import datetime import unicodedata import requests import config def _table_name(name: str, width: int) -> str: """Strip diacritics so accented chars don't break Slack's monospace alignment.""" normalized = unicodedata.normalize("NFKD", name) ascii_name = "".join(c for c in normalized if not unicodedata.combining(c)) return ascii_name[:width] _SLACK_API = "https://slack.com/api" _STRATEGY_LABELS = { "LOWEST_COST_WITHOUT_CAP": "Menor coste", "LOWEST_COST_WITH_BID_CAP": "Cap. puja", "COST_CAP": "Cap. coste", "MINIMUM_ROAS": "ROAS mΓ­n.", } _ACTION_DISPLAY = { "INCREASE_BUDGET": ("🟒", "AUMENTAR PRESUPUESTO"), "REDUCE_BUDGET": ("πŸ”΄", "REDUCIR PRESUPUESTO"), "PAUSE": ("β›”", "PAUSAR CAMPAΓ‘A"), "REVIEW_CREATIVES": ("πŸ”", "REVISAR CREATIVIDADES"), "MAINTAIN": ("βœ…", "MANTENER"), } # Solo estas acciones ejecutan algo real en Meta API β†’ botones _ACTIONABLE = {"INCREASE_BUDGET", "REDUCE_BUDGET", "PAUSE"} def _effect_text(action: dict, budget: float | None) -> str: """Texto que describe exactamente quΓ© ocurrirΓ‘ si se aprueba la acciΓ³n.""" atype = action.get("action_type", "") param = float(action.get("parameter") or 1.0) if atype == "PAUSE": return "⚠️ *La campaΓ±a serΓ‘ pausada en Meta Ads*" if atype in ("INCREASE_BUDGET", "REDUCE_BUDGET"): pct = round((param - 1) * 100) if pct == 0: return "" # parameter not available, skip misleading text sign = "+" if pct >= 0 else "" if budget: new_b = round(budget * param, 2) return f"πŸ“Š Presupuesto diario: *{budget:.0f}€ β†’ {new_b:.0f}€* ({sign}{pct}%)" return f"πŸ“Š Ajuste de presupuesto: *{sign}{pct}%*" return "" def _post(method: str, **payload) -> dict: resp = requests.post( f"{_SLACK_API}/{method}", headers={"Authorization": f"Bearer {config.SLACK_BOT_TOKEN}"}, json=payload, timeout=10, ) data = resp.json() if not data.get("ok"): raise RuntimeError(f"Slack {method} error: {data.get('error')} β€” {data.get('response_metadata', {}).get('messages', '')}") return data def update_message(channel: str, ts: str, text: str): """Reemplaza un mensaje con texto plano tras aprobaciΓ³n/rechazo.""" _post("chat.update", channel=channel, ts=ts, text=text, blocks=[]) def _ad_action_blocks(ads: list) -> list: """Genera bloques Slack con botΓ³n de pausa para anuncios que Claude recomienda pausar.""" blocks = [] for ad in ads: if not ad.get("row_id"): continue name = ad["name"] ev = ad.get("evaluacion", "") rec = ad.get("recomendacion", "") text = f"β›” *{name[:55]}*" if ev: text += f"\n_{ev}_" if rec: text += f"\nβ†’ {rec}" text += "\n⚠️ *El anuncio serΓ‘ pausado en Meta Ads si se aprueba*" blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": text}}) blocks.append({ "type": "actions", "elements": [ { "type": "button", "text": {"type": "plain_text", "text": "β›” Pausar anuncio"}, "style": "danger", "value": f"approve:{ad['row_id']}", "action_id": f"approve_{ad['row_id']}", }, { "type": "button", "text": {"type": "plain_text", "text": "❌ Rechazar"}, "value": f"reject:{ad['row_id']}", "action_id": f"reject_{ad['row_id']}", }, ], }) return blocks def _adset_ad_table(items: list, label: str, show_bid: bool = False, show_7d: bool = False) -> str: """Genera tabla monoespaciada de adsets o anuncios para Slack.""" if not items: return "" lines = [f"*{label}*"] lines.append("```") if show_7d: lines.append(f"{'Nombre':<45} {'Gasto':>6} {'Leads':>5} {'CPL(3d)':>8} {'CPL(7d)':>8} {'CTR':>5}") lines.append("─" * 82) elif show_bid: lines.append(f"{'Nombre':<45} {'Gasto':>6} {'Leads':>5} {'CPL':>6} {'CTR':>5} {'Cap':>7}") lines.append("─" * 79) else: lines.append(f"{'Nombre':<45} {'Gasto':>6} {'Leads':>5} {'CPL':>6} {'CTR':>5}") lines.append("─" * 71) for it in items: name = _table_name(it["name"], 45) leads_str = f"{it['leads']:>5}" if it["leads"] > 0 else " β€”" if show_7d: cpl_3d = it.get("cpl_3d", 0.0) cpl_7d = it.get("cpl_7d", it.get("cpl", 0.0)) cpl_3d_str = f"{cpl_3d:>6.2f}€" if cpl_3d > 0 else " β€”" cpl_7d_str = f"{cpl_7d:>6.2f}€" if cpl_7d > 0 else " β€”" lines.append( f"{name:<45} {it['spend']:>5.0f}€ {leads_str} {cpl_3d_str:>8} {cpl_7d_str:>8} {it['ctr']:>4.1f}%" ) elif show_bid: cpl_str = f"{it['cpl']:>5.2f}€" if it["leads"] > 0 else " β€”" cost_cap = it.get("cost_cap_eur") cap_str = f" {cost_cap:>5.2f}€" if cost_cap else " Auto" lines.append( f"{name:<45} {it['spend']:>5.0f}€ {leads_str} {cpl_str} {it['ctr']:>4.1f}%{cap_str}" ) else: cpl_str = f"{it['cpl']:>5.2f}€" if it["leads"] > 0 else " β€”" lines.append( f"{name:<45} {it['spend']:>5.0f}€ {leads_str} {cpl_str} {it['ctr']:>4.1f}%" ) lines.append("```") # Evaluations below the table for it in items: ev = it.get("evaluacion", "") rec = it.get("recomendacion", "") if ev or rec: lines.append(f"β€’ *{it['name'][:55]}*") if ev: lines.append(f" _{ev}_") if rec: lines.append(f" β†’ {rec}") return "\n".join(lines) def _vertical_status_emoji(v_cpl: float, v_obj: float, has_action: bool) -> str: if v_cpl == 0 or v_obj == 0: return "βšͺ" if v_cpl <= v_obj: return "⚠️" if has_action else "βœ…" if v_cpl <= v_obj * 1.3: return "⚠️" return "❌" def send_daily_report( daily_totals: list, best_campaigns: list, worst_campaigns: list, actions: list, target_cpl: float, campaigns_analyzed: int, mode: str = "DRY_RUN", verticals: dict = None, campaign_details: dict = None, monthly_verticals: dict = None, ) -> str | None: """EnvΓ­a el informe diario consolidado. Devuelve el ts del mensaje.""" now = datetime.now() date_label = now.strftime("%d/%m/%Y") month_name = now.strftime("%B %Y").capitalize() prefix = config.META_CAMPAIGN_PREFIX mode_label = "DRY RUN" if mode == "DRY_RUN" else "PRODUCCIΓ“N" action_map = {a["campaign_name"]: a for a in actions} details_map = campaign_details or {} name_to_cid = {d["name"]: cid for cid, d in details_map.items()} # ── Classify campaigns: issues vs OK ───────────────────────────────────── camps_issues: dict = {} # name β†’ (cid, detail, action_or_None) camps_ok: dict = {} # name β†’ (cid, detail) for cid, detail in details_map.items(): name = detail["name"] act = action_map.get(name) has_action = bool(act and act["action_type"] != "MAINTAIN") has_ad_pause = any( ad.get("accion") == "PAUSE" and ad.get("row_id") for ad in detail.get("ads", []) ) if has_action or has_ad_pause: camps_issues[name] = (cid, detail, act) else: camps_ok[name] = (cid, detail) # Group issues by vertical by_vertical_issues: dict = {} for name, (cid, detail, act) in camps_issues.items(): by_vertical_issues.setdefault(detail["vertical"], []).append((cid, detail, act)) # ── Message 1: Executive dashboard ─────────────────────────────────────── blocks: list = [ { "type": "header", "text": {"type": "plain_text", "text": f"Meta Optimizer β€” {prefix} β€” {date_label} ({mode_label})"}, }, ] # Monthly profitability table if daily_totals: v_order = ( sorted(monthly_verticals.keys(), key=lambda v: -monthly_verticals[v]["margin"]) if monthly_verticals else [] ) cw = 7 header = f"{'DΓ­a':<5} {'Gasto':>6} {'Leads':>5} {'CPL':>7}" for v in v_order: header += f" {v[:6]:>{cw}}" header += " Est" sep = "─" * len(header) lines = [header, sep] total_spend = total_leads = total_margin = 0.0 total_v = {v: 0.0 for v in v_order} for d in daily_totals: day = d["date"][8:10] + "/" + d["date"][5:7] margin = d.get("margin", 0.0) total_spend += d["spend"] total_leads += d["leads"] total_margin += margin v_day = d.get("v_margins", {}) icon = "βœ…" if d["leads"] > 0 else ("❌" if d["spend"] > 0 else "β€”") row = f"{day:<5} {d['spend']:>5.0f}€ {d['leads']:>5} {d['cpl']:>6.2f}€" for v in v_order: vm = v_day.get(v, 0.0) total_v[v] += vm vm_s = (f"+{vm:.0f}€" if vm >= 0 else f"{vm:.0f}€") if round(vm) != 0 else " β€”" row += f" {vm_s:>{cw}}" row += f" {icon}" lines.append(row) lines.append(sep) total_row = f"{'TOTAL':<5} {total_spend:>5.0f}€ {int(total_leads):>5} {'':>7}" for v in v_order: tv = total_v[v] tv_s = f"+{tv:.0f}€" if tv >= 0 else f"{tv:.0f}€" total_row += f" {tv_s:>{cw}}" lines.append(total_row) blocks.append({ "type": "section", "text": { "type": "mrkdwn", "text": f"*Rentabilidad {month_name}*\n```" + "\n".join(lines) + "```", }, }) else: blocks.append({ "type": "section", "text": {"type": "mrkdwn", "text": "_Sin datos del mes en curso aΓΊn._"}, }) blocks.append({"type": "divider"}) # Vertical scorecard if verticals: lines = [f"{'Vertical':<16} {'Gasto':>6} {'Leads':>5} {'CPL':>7} {'Obj':>7} {'Margen':>9}"] lines.append("─" * 58) for v, data in sorted(verticals.items(), key=lambda x: -x[1]["margin"]): v_leads = data["leads"] v_spend = data["spend"] v_cpl = round(v_spend / v_leads, 2) if v_leads > 0 else 0.0 v_m = data["margin"] v_obj = data.get("target_cpl", 0) m_sign = f"+{v_m:.0f}€" if v_m >= 0 else f"{v_m:.0f}€" obj_str = f"{v_obj:.2f}€" if v_obj else " β€”" has_act = v in by_vertical_issues st = _vertical_status_emoji(v_cpl, v_obj, has_act) lines.append( f"{st} {v:<14} {v_spend:>5.0f}€ {v_leads:>5} {v_cpl:>6.2f}€ {obj_str:>7} {m_sign:>9}" ) blocks.append({ "type": "section", "text": {"type": "mrkdwn", "text": "*Verticales Β· ayer*\n```" + "\n".join(lines) + "```"}, }) # Actions section if actions: blocks.append({"type": "divider"}) blocks.append({ "type": "section", "text": {"type": "mrkdwn", "text": "*🚨 Acciones recomendadas*"}, }) for act in actions: atype = act["action_type"] emoji, alabel = _ACTION_DISPLAY.get(atype, ("βšͺ", atype)) cid = name_to_cid.get(act["campaign_name"]) bid_cfg = details_map.get(cid, {}).get("bid_config", {}) if cid else {} budget = bid_cfg.get("daily_budget_eur") text = f"{emoji} *{act['campaign_name'][:60]}* β†’ *{alabel}*" if act.get("justification"): text += f"\n_{act['justification'][:200]}_" if act.get("alert"): text += f"\n:warning: {act['alert'][:150]}" effect = _effect_text(act, budget) if effect: text += f"\n{effect}" blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": text}}) if atype in _ACTIONABLE: blocks.append({ "type": "actions", "elements": [ { "type": "button", "text": {"type": "plain_text", "text": "βœ… Aprobar"}, "style": "primary", "value": f"approve:{act['row_id']}", "action_id": f"approve_{act['row_id']}", }, { "type": "button", "text": {"type": "plain_text", "text": "❌ Rechazar"}, "style": "danger", "value": f"reject:{act['row_id']}", "action_id": f"reject_{act['row_id']}", }, ], }) # Ad pause count notice total_ad_pauses = sum( 1 for _, detail, _ in camps_issues.values() for ad in detail.get("ads", []) if ad.get("accion") == "PAUSE" and ad.get("row_id") ) if total_ad_pauses > 0: noun = "anuncio" if total_ad_pauses == 1 else "anuncios" blocks.append({ "type": "context", "elements": [{"type": "mrkdwn", "text": f"β›” {total_ad_pauses} {noun} recomendados para pausar β€” ver detalle por vertical"}], }) # "Todo en orden" summary β€” grouped by vertical if camps_ok: blocks.append({"type": "divider"}) ok_by_v: dict = {} for name, (_, detail) in camps_ok.items(): ok_by_v.setdefault(detail["vertical"], []).append((name, detail)) ok_lines = ["*Todo en orden*"] for vert in sorted(ok_by_v.keys()): v_data = (verticals or {}).get(vert, {}) v_spend = v_data.get("spend", 0) v_leads = v_data.get("leads", 0) v_cpl = round(v_spend / v_leads, 2) if v_leads > 0 else 0.0 v_obj = v_data.get("target_cpl", 0) ok_lines.append(f"\nβœ… *{vert.upper()}* Β· CPL {v_cpl:.2f}€ obj {v_obj:.2f}€") for name, detail in sorted(ok_by_v[vert], key=lambda x: -x[1].get("spend_1d", 0)): ok_lines.append( f" β€’ {name} Β· " f"{detail.get('spend_1d', 0):.0f}€ / {detail.get('leads_1d', 0)} leads ayer" ) blocks.append({ "type": "section", "text": {"type": "mrkdwn", "text": "\n".join(ok_lines)}, }) blocks.append({ "type": "context", "elements": [{"type": "mrkdwn", "text": f"{campaigns_analyzed} campaΓ±as analizadas"}], }) result = _post( "chat.postMessage", channel=config.SLACK_CHANNEL_ID, blocks=blocks, text=f"Meta Optimizer β€” {prefix} β€” {date_label}", ) ts = result.get("ts") # ── Messages 2-N: one per vertical with issues ──────────────────────────── for v, camp_list in sorted(by_vertical_issues.items()): v_data = (verticals or {}).get(v, {}) v_spend = v_data.get("spend", 0) v_leads = v_data.get("leads", 0) v_cpl = round(v_spend / v_leads, 2) if v_leads > 0 else 0.0 v_obj = v_data.get("target_cpl", 0) v_margin = v_data.get("margin", 0) m_str = f"+{v_margin:.0f}€" if v_margin >= 0 else f"{v_margin:.0f}€" obj_str = f" Β· obj {v_obj:.2f}€" if v_obj else "" v_blocks: list = [ { "type": "header", "text": {"type": "plain_text", "text": f"πŸ“Š {v.upper()}"}, }, { "type": "section", "text": { "type": "mrkdwn", "text": ( f"{v_spend:.0f}€ Β· {v_leads} leads Β· " f"CPL {v_cpl:.2f}€{obj_str} Β· Margen {m_str}" ), }, }, ] for i, (cid, detail, act) in enumerate( sorted(camp_list, key=lambda x: -x[1].get("spend_1d", 0)) ): if i > 0: v_blocks.append({"type": "divider"}) name = detail["name"] spend_1d = detail.get("spend_1d", 0.0) leads_1d = detail.get("leads_1d", 0) margin = detail["margin"] m_str2 = f"+{margin:.2f}€" if margin >= 0 else f"{margin:.2f}€" adsets = detail.get("adsets", []) ads = detail.get("ads", []) bid_cfg = detail.get("bid_config", {}) budget = bid_cfg.get("daily_budget_eur") strategy = bid_cfg.get("bid_strategy", "") strat_label = _STRATEGY_LABELS.get(strategy, strategy or "β€”") atype = act["action_type"] if act else "MAINTAIN" emoji, alabel = _ACTION_DISPLAY.get(atype, ("βšͺ", atype)) camp_text = ( f"{emoji} *{name}*\n" f"Ayer: {spend_1d:.0f}€ / {leads_1d} leads Β· Margen: {m_str2} Β· " f"`{strat_label}`" + (f" Β· {budget:.0f}€/dΓ­a" if budget else "") + f"\n*{alabel}*" ) if act and act.get("justification"): camp_text += f" β€” _{act['justification'][:160]}_" if act and act.get("alert"): camp_text += f"\n:warning: {act['alert'][:130]}" v_blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": camp_text}}) # Approve/Reject buttons for campaign action if act and atype in _ACTIONABLE: effect = _effect_text(act, budget) if effect: v_blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": effect}}) v_blocks.append({ "type": "actions", "elements": [ { "type": "button", "text": {"type": "plain_text", "text": "βœ… Aprobar"}, "style": "primary", "value": f"approve:{act['row_id']}", "action_id": f"approve_{act['row_id']}", }, { "type": "button", "text": {"type": "plain_text", "text": "❌ Rechazar"}, "style": "danger", "value": f"reject:{act['row_id']}", "action_id": f"reject_{act['row_id']}", }, ], }) # Adset table β€” top 3, only for non-MAINTAIN campaigns if atype != "MAINTAIN" and adsets: adset_table = _adset_ad_table(adsets[:3], "Conjuntos de anuncios (3 dΓ­as)", show_bid=True) if adset_table: for chunk in [adset_table[i:i+2900] for i in range(0, len(adset_table), 2900)]: v_blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": chunk}}) # Ad pause buttons v_blocks.extend(_ad_action_blocks(ads)) _post( "chat.postMessage", channel=config.SLACK_CHANNEL_ID, blocks=v_blocks, text=f"Detalle vertical: {v}", ) return ts def send_execution_summary(log: dict): """Resumen plano de ejecuciΓ³n (fallback).""" mode_label = "DRY RUN" if log.get("mode") == "DRY_RUN" else "PRODUCCIΓ“N" text = ( f":bar_chart: *Meta Optimizer β€” Resumen diario* ({mode_label})\n" f"β€’ CampaΓ±as analizadas: {log.get('campaigns_analyzed', 0)}\n" f"β€’ Acciones propuestas: {log.get('actions_proposed', 0)}\n" f"β€’ Acciones ejecutadas: {log.get('actions_executed', 0)}\n" f"β€’ DuraciΓ³n: {log.get('duration_seconds', 0):.1f}s" ) _post("chat.postMessage", channel=config.SLACK_CHANNEL_ID, text=text)