from __future__ import annotations """ P3 投放周报 Markdown 格式化。 将 L1-L4 分析结果和诊断结论输出为钉钉/API 可用的 Markdown。 """ from typing import Any def _icon(color: str) -> str: return {"red": "🔴", "yellow": "🟡", "green": "🟢"}.get(color, "⚪") def _chg_arrow(val: float | None) -> str: if val is None: return "" return f" ↑{val:.1f}%" if val > 0 else f" ↓{abs(val):.1f}%" def _fmt_money(val: float | None) -> str: if val is None: return "—" if abs(val) >= 1e6: return f"${val / 1e6:.2f}M" if abs(val) >= 1e3: return f"${val / 1e3:.1f}K" return f"${val:,.2f}" def _fmt_pct(val: float | None) -> str: if val is None: return "—" return f"{val:.1f}%" def _fmt_roas(val: float | None) -> str: if val is None: return "—" return f"{val:.2f}" def format_p3_title(p3_result: dict) -> str: period = p3_result.get("period_label", "") return f"📊 P3 投放分析 — {period}" def format_p3_report(p3_result: dict, diagnostics: list[dict]) -> str: """完整 P3 周报 Markdown。""" parts: list[str] = [] period = p3_result.get("period_label", "") prev = p3_result.get("prev_label", "") parts.append(f"## 📊 P3 投放分析 — {period}\n") if prev: parts.append(f"> 环比 vs {prev}\n") # ── L1 市场总览 ── l1 = p3_result.get("L1", []) if l1: parts.append("### L1 市场总览\n") for row in l1: mkt = row.get("market", "?") ratings = row.get("ratings", {}) bp_icon = _icon(ratings.get("budget_pacing", "")) rvt_icon = _icon(ratings.get("revenue_vs_target", "")) parts.append(f"**{mkt}**") parts.append(f"- 花费: {_fmt_money(row.get('spend'))}{_chg_arrow(row.get('spend_chg'))}") parts.append(f"- 营收: {_fmt_money(row.get('revenue'))}{_chg_arrow(row.get('revenue_chg'))}") parts.append(f"- ROAS: {_fmt_roas(row.get('roas'))}{_chg_arrow(row.get('roas_chg'))}") bp = row.get("budget_pacing") if bp is not None: parts.append(f"- {bp_icon} Budget Pacing: {_fmt_pct(bp)}") rvt = row.get("revenue_vs_target") if rvt is not None: parts.append(f"- {rvt_icon} Revenue vs Target: {_fmt_pct(rvt)}") parts.append("") # ── L2 funnel×优化目标 ── l2 = p3_result.get("L2", []) if l2: parts.append("### L2 Funnel × 优化目标\n") parts.append("| 优化目标 | 受众 | 花费 | 占比 | 核心指标 | 状态 |") parts.append("|---------|------|------|------|---------|------|") for row in l2: obj = row.get("objective_group", "") aud = row.get("audience_type", "") spend = _fmt_money(row.get("spend")) share = _fmt_pct(row.get("spend_share")) ratings = row.get("ratings", {}) core_parts = [] from .p3_metrics import get_core_metrics_for_objective for m in get_core_metrics_for_objective(row.get("ad_objective", "")): val = row.get(m) if val is not None: icon = _icon(ratings.get(m, "")) if m in ("ROAS",): core_parts.append(f"{icon}{m}: {_fmt_roas(val)}") elif m in ("CPA", "CPL", "CPC", "CPM", "CostPerATC", "CostPerAtoB", "CostPerLPV", "CPE", "CostPerClick"): core_parts.append(f"{icon}{m}: {_fmt_money(val)}") elif m in ("CTR", "LeadRate", "ATCRate", "ClickRate"): core_parts.append(f"{icon}{m}: {val * 100:.2f}%") else: core_parts.append(f"{m}: {val:,.0f}") worst = "green" for c in ratings.values(): if c == "red": worst = "red" break if c == "yellow": worst = "yellow" parts.append(f"| {obj} | {aud} | {spend} | {share} | {', '.join(core_parts) or '—'} | {_icon(worst)} |") parts.append("") # ── L3 平台 ── l3 = p3_result.get("L3", []) if l3: parts.append("### L3 平台分析\n") parts.append("| 平台 | 花费 | 花费占比 | 营收 | 营收占比 | ROAS | CPC | CTR |") parts.append("|------|------|---------|------|---------|------|-----|-----|") for row in l3: plat = row.get("platform", "") roas_chg = row.get("roas_chg") roas_str = _fmt_roas(row.get("roas")) if roas_chg is not None: roas_str += _chg_arrow(roas_chg) ctr = row.get("ctr") ctr_str = f"{ctr * 100:.2f}%" if ctr else "—" parts.append( f"| {plat} | {_fmt_money(row.get('spend'))} | {_fmt_pct(row.get('spend_share'))} " f"| {_fmt_money(row.get('revenue'))} | {_fmt_pct(row.get('revenue_share'))} " f"| {roas_str} | {_fmt_money(row.get('cpc'))} | {ctr_str} |" ) parts.append("") # ── 诊断结论 ── if diagnostics: parts.append("### 诊断与建议\n") for d in diagnostics: icon = _icon(d.get("level", "")) layer = d.get("layer", "") parts.append(f"{icon} **[{layer}] {d['title']}**") parts.append(f" - {d['detail']}") parts.append(f" - 👉 {d['action']}") parts.append("") # ── L4 Top Campaign(只展示前 10)── l4 = p3_result.get("L4", []) if l4: parts.append("### L4 Top Campaign\n") parts.append("| Campaign | 平台 | 目的 | 花费 | ROAS | 环比 |") parts.append("|----------|------|------|------|------|------|") for row in l4[:10]: name = str(row.get("campaign_name", "")) if len(name) > 35: name = name[:32] + "..." plat = row.get("parsed_platform", row.get("platform", "")) obj = row.get("ad_objective", "") roas_chg = row.get("roas_chg") chg_str = _chg_arrow(roas_chg) if roas_chg is not None else "—" parts.append( f"| {name} | {plat} | {obj} " f"| {_fmt_money(row.get('spend'))} | {_fmt_roas(row.get('roas'))} | {chg_str} |" ) if len(l4) > 10: parts.append(f"\n> 共 {len(l4)} 个 campaign,仅展示 Top 10") parts.append("") return "\n".join(parts)