Spaces:
Runtime error
Runtime error
| 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) | |