"""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"] text = f"β›” *{name[:80]}* _(0 leads Β· 7d)_" 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("```") return "\n".join(lines) def _vert_status(v_cpl: float, v_obj: float, has_issues: bool) -> str: if v_cpl == 0 or v_obj == 0: return "βšͺ" if has_issues: return "🚨" if v_cpl > v_obj * 1.3 else "⚠️" return "βœ…" if v_cpl <= v_obj else "⚠️" 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 {} # Group ALL campaigns by vertical by_vertical: dict = {} for cid, detail in details_map.items(): act = action_map.get(detail["name"]) by_vertical.setdefault(detail["vertical"], []).append((cid, detail, act)) def _has_issues(camp_list): return any( (act and act["action_type"] != "MAINTAIN") or any(ad.get("accion") == "PAUSE" and ad.get("row_id") for ad in detail.get("ads", [])) for _, detail, act in camp_list ) # Sort verticals: issues first (by margin asc), then OK (by margin desc) def _vert_sort_key(item): v, cl = item v_data = (verticals or {}).get(v, {}) margin = v_data.get("margin", 0) has_iss = _has_issues(cl) return (0 if has_iss else 1, margin if has_iss else -margin) sorted_verticals = sorted(by_vertical.items(), key=_vert_sort_key) # ── Message 1: 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 hdr = f"{'DΓ­a':<5} {'Gasto':>6} {'Leads':>5} {'CPL':>7}" for v in v_order: hdr += f" {v[:6]:>{cw}}" hdr += " Est" sep = "─" * len(hdr) lines = [hdr, 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"{'':>2} {'Vertical':<14} {'Gasto':>6} {'Leads':>5} {'CPL':>7} {'Obj':>7} {'Margen':>9}"] lines.append("─" * 60) for v, cl in sorted_verticals: data = (verticals or {}).get(v, {}) v_leads = data.get("leads", 0) v_spend = data.get("spend", 0) v_cpl = round(v_spend / v_leads, 2) if v_leads > 0 else 0.0 v_m = data.get("margin", 0) 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 " β€”" st = _vert_status(v_cpl, v_obj, _has_issues(cl)) 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": "*Resumen Β· ayer*\n```" + "\n".join(lines) + "```"}, }) blocks.append({ "type": "context", "elements": [{"type": "mrkdwn", "text": f"{campaigns_analyzed} campaΓ±as analizadas β€” detalle por vertical a continuaciΓ³n"}], }) result = _post( "chat.postMessage", channel=config.SLACK_CHANNEL_ID, blocks=blocks, text=f"Meta Optimizer β€” {prefix} β€” {date_label}", ) ts = result.get("ts") # ── One message per vertical ────────────────────────────────────────────── for v, camp_list in sorted_verticals: 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 "" has_iss = _has_issues(camp_list) st = _vert_status(v_cpl, v_obj, has_iss) v_blocks: list = [ { "type": "header", "text": {"type": "plain_text", "text": f"{st} {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}" ), }, }, {"type": "divider"}, ] 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" cemoji, alabel = _ACTION_DISPLAY.get(atype, ("βšͺ", atype)) if atype == "MAINTAIN" and not any( ad.get("accion") == "PAUSE" and ad.get("row_id") for ad in ads ): # Compact header for clean campaigns v_blocks.append({ "type": "section", "text": { "type": "mrkdwn", "text": ( f"{cemoji} *{name}*\n" f"Ayer: {spend_1d:.0f}€ / {leads_1d} leads Β· " f"Margen: {m_str2} Β· `{strat_label}`" + (f" Β· {budget:.0f}€/dΓ­a" if budget else "") ), }, }) # Still show adset breakdown for context if adsets: tbl = _adset_ad_table(adsets[:3], "Conjuntos (3 dΓ­as)", show_bid=True) if tbl: for chunk in [tbl[j:j+2900] for j in range(0, len(tbl), 2900)]: v_blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": chunk}}) else: # Full block for campaigns with action or ad pauses camp_text = ( f"{cemoji} *{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 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 if atype != "MAINTAIN" and adsets: tbl = _adset_ad_table(adsets[:3], "Conjuntos (3 dΓ­as)", show_bid=True) if tbl: for chunk in [tbl[j:j+2900] for j in range(0, len(tbl), 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"{v.upper()} Β· {v_spend:.0f}€ Β· {v_leads} leads", ) return ts def _score_emoji(score: float) -> str: if score >= 8: return "🟒" if score >= 6: return "🟑" if score >= 4: return "🟠" return "πŸ”΄" def _brief_action(score: float, fatigue: bool) -> str: if score == 0: return "sin imagen" if fatigue: return "renovar urgente" if score >= 8: return "mantener" if score >= 6: return "optimizar" if score >= 4: return "renovar" return "reemplazar" def send_creative_analysis_report(all_results: dict) -> None: """EnvΓ­a scorecard compacto de creatividades a Slack. Un mensaje por campaΓ±a.""" now = datetime.now() date_label = now.strftime("%d/%m/%Y %H:%M") total_ads = sum(len(as_d["ads"]) for c in all_results.values() for as_d in c["adsets"].values()) total_fatigue = sum( 1 for c in all_results.values() for as_d in c["adsets"].values() for ad in as_d["ads"] if ad.get("fatigue") ) scored = [ ad.get("score", 0) for c in all_results.values() for as_d in c["adsets"].values() for ad in as_d["ads"] if ad.get("score", 0) > 0 ] avg_score = round(sum(scored) / len(scored), 1) if scored else 0.0 summary = f"*{len(all_results)} campaΓ±as* Β· *{total_ads} anuncios* Β· score medio *{avg_score}/10*" if total_fatigue: summary += f"\n⚠️ *{total_fatigue} con fatiga creativa detectada*" _post( "chat.postMessage", channel=config.SLACK_CHANNEL_ID, blocks=[ {"type": "header", "text": {"type": "plain_text", "text": f"Creatividades β€” {date_label}"}}, {"type": "section", "text": {"type": "mrkdwn", "text": summary}}, ], text=f"Creatividades β€” {date_label}", ) # ── One message per campaign ────────────────────────────────────────────── for cid, camp_data in all_results.items(): camp_name = camp_data["name"] adsets = camp_data.get("adsets", {}) if not adsets: continue def _flush(buf: list) -> list: if len(buf) > 1: try: _post("chat.postMessage", channel=config.SLACK_CHANNEL_ID, blocks=buf, text=camp_name) except RuntimeError as e: print(f" [WARN] Slack: {e}") return [{"type": "header", "text": {"type": "plain_text", "text": f"{camp_name} (cont.)"}}] blocks: list = [{"type": "header", "text": {"type": "plain_text", "text": camp_name}}] for as_data in adsets.values(): adset_name = as_data["name"] ads = as_data["ads"] ads_sorted = sorted(ads, key=lambda x: -x.get("score", 0)) # Compact monospace table β€” one line per ad lines = [ f"*{adset_name}* _({len(ads)} anuncios)_", "```", f"{'Nombre':<33} {'Sc':>4} {'CTR':>5} {'CPL':>6} AcciΓ³n", "─" * 63, ] for ad in ads_sorted: name = _table_name(ad["ad_name"], 33) score = ad.get("score", 0) ctr = ad.get("ctr_7d", 0) cpl = ad.get("cpl_7d", 0) fat = "⚠" if ad.get("fatigue") else " " action = _brief_action(score, ad.get("fatigue", False)) cpl_s = f"{cpl:.2f}€" if cpl > 0 else " β€”" sc_s = f"{score:.1f}" if score > 0 else " β€”" lines.append(f"{name:<33}{fat} {sc_s:>4} {ctr:>4.1f}% {cpl_s:>6} {action}") lines.append("```") winner = as_data.get("comparison", {}) or {} if winner.get("winner"): lines.append(f"πŸ† _{winner['winner'][:70]}_") ab = [{"type": "section", "text": {"type": "mrkdwn", "text": "\n".join(lines)}}] if len(blocks) + len(ab) > 48: blocks = _flush(blocks) blocks.extend(ab) _flush(blocks) 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)