Spaces:
Runtime error
Runtime error
| """ | |
| 固定消息模板：钉钉 Markdown 周报。 | |
| 结构不变，数据填充，专家预警分级展示。 | |
| """ | |
| from typing import Any | |
| ALERT_ICON = { | |
| "red": "🔴", | |
| "yellow": "🟡", | |
| "green": "🟢", | |
| # 兼容旧格式 | |
| "critical": "🔴", | |
| "warning": "🟡", | |
| "positive": "🟢", | |
| } | |
| # ─── 数值格式化 ─── | |
| def _n(val: float | None, prefix: str = "", suffix: str = "", d: int = 2) -> str: | |
| if val is None or val != val: | |
| return "—" | |
| if abs(val) >= 1_000_000: | |
| return f"{prefix}{val / 1_000_000:,.{d}f}M{suffix}" | |
| if abs(val) >= 1_000: | |
| return f"{prefix}{val / 1_000:,.{d}f}K{suffix}" | |
| return f"{prefix}{val:,.{d}f}{suffix}" | |
| def _pct(val: float | None) -> str: | |
| if val is None or val != val: | |
| return "—" | |
| return f"{val * 100:.2f}%" if abs(val) < 1 else f"{val:.2f}%" | |
| def _arrow(change: float | None) -> str: | |
| """根据变化幅度返回方向箭头。""" | |
| if change is None or change != change: | |
| return "" | |
| if change >= 10: | |
| return "↑↑" | |
| if change > 0: | |
| return "↑" | |
| if change <= -10: | |
| return "↓↓" | |
| if change < 0: | |
| return "↓" | |
| return "→" | |
| def _deltas_zh(wow_val: float | None, yoy_val: float | None) -> str: | |
| """环比/同比（中文标签），适合钉钉 Markdown；大幅波动加粗。""" | |
| parts: list[str] = [] | |
| if wow_val is not None and wow_val == wow_val: | |
| sign = "+" if wow_val >= 0 else "" | |
| ar = _arrow(wow_val) | |
| s = f"环比 {ar}{sign}{wow_val:.1f}%" | |
| if abs(wow_val) >= 10: | |
| s = f"**{s}**" | |
| parts.append(s) | |
| if yoy_val is not None and yoy_val == yoy_val: | |
| sign = "+" if yoy_val >= 0 else "" | |
| ar = _arrow(yoy_val) | |
| s = f"同比 {ar}{sign}{yoy_val:.1f}%" | |
| if abs(yoy_val) >= 10: | |
| s = f"**{s}**" | |
| parts.append(s) | |
| if not parts: | |
| return "" | |
| return " ｜ " + " ｜ ".join(parts) | |
| def _append_funnel_dim_lines(L: list[str], m: dict[str, Any], cw: dict, cy: dict) -> None: | |
| """漏斗维度：每指标一行，钉钉端更易扫读。""" | |
| L.append(f"- **花费** {_n(m.get('Cost'), prefix='$')}{_deltas_zh(cw.get('Cost'), cy.get('Cost'))}") | |
| L.append(f"- **销售额** {_n(m.get('Ads.Sales'), prefix='$')}{_deltas_zh(cw.get('Ads.Sales'), cy.get('Ads.Sales'))}") | |
| L.append(f"- **ROAS** {m.get('ROAS', 0):.2f}{_deltas_zh(cw.get('ROAS'), cy.get('ROAS'))}") | |
| L.append(f"- **展示量** {_n(m.get('Impressions'), d=0)}{_deltas_zh(cw.get('Impressions'), cy.get('Impressions'))}") | |
| L.append(f"- **CTR** {_pct(m.get('CTR'))}{_deltas_zh(cw.get('CTR'), cy.get('CTR'))}") | |
| L.append(f"- **CPC** {_n(m.get('CPC'), prefix='$')}{_deltas_zh(cw.get('CPC'), cy.get('CPC'))}") | |
| if m.get("ATC"): | |
| L.append(f"- **加购 ATC** {_n(m.get('ATC'), d=0)}{_deltas_zh(cw.get('ATC'), cy.get('ATC'))}") | |
| if m.get("GA4.Sessions"): | |
| L.append(f"- **会话** {_n(m.get('GA4.Sessions'), d=0)}{_deltas_zh(cw.get('GA4.Sessions'), cy.get('GA4.Sessions'))}") | |
| def _append_country_dim_lines(L: list[str], m: dict[str, Any], cw: dict, cy: dict, *, full: bool) -> None: | |
| """市场维度：full=True 与周报一致；False 仅花费+ROAS（日报/简版）。""" | |
| L.append(f"- **花费** {_n(m.get('Cost'), prefix='$')}{_deltas_zh(cw.get('Cost'), cy.get('Cost'))}") | |
| if full: | |
| L.append(f"- **销售额** {_n(m.get('Ads.Sales'), prefix='$')}{_deltas_zh(cw.get('Ads.Sales'), cy.get('Ads.Sales'))}") | |
| L.append(f"- **ROAS** {m.get('ROAS', 0):.2f}{_deltas_zh(cw.get('ROAS'), cy.get('ROAS'))}") | |
| if full: | |
| L.append(f"- **会话** {_n(m.get('GA4.Sessions'), d=0)}{_deltas_zh(cw.get('GA4.Sessions'), cy.get('GA4.Sessions'))}") | |
| L.append(f"- **加购** {_n(m.get('ATC'), d=0)}{_deltas_zh(cw.get('ATC'), cy.get('ATC'))}") | |
| L.append(f"- **转化率** {_pct(m.get('CVR'))}{_deltas_zh(cw.get('CVR'), cy.get('CVR'))}") | |
| # ─── 固定模板 ─── | |
| def format_report(data: dict[str, Any]) -> str: | |
| cp = data.get("current_period", {}) | |
| pp = data.get("previous_period") | |
| year, week = cp.get("year"), cp.get("week") | |
| L: list[str] = [] | |
| L.append(f"## DM 周报 · {year} W{week}") | |
| cmp_bits: list[str] = [] | |
| if pp: | |
| cmp_bits.append(f"环比：{pp['year']} 年第 {pp['week']} 周") | |
| if data.get("yoy_period"): | |
| yp = data["yoy_period"] | |
| cmp_bits.append(f"同比：{yp['year']} 年第 {yp['week']} 周") | |
| if cmp_bits: | |
| L.append("> " + " ｜ ".join(cmp_bits)) | |
| L.append("") | |
| curr = data.get("current", {}) | |
| wow_t = data.get("wow_total") or {} | |
| yoy_t = data.get("yoy_total") or {} | |
| total = curr.get("total", {}) | |
| L.append("### 一、整体表现") | |
| L.append("") | |
| _block_metrics(L, total, wow_t, yoy_t if yoy_t else None) | |
| L.append("") | |
| by_plat = curr.get("by_platform", {}) | |
| wow_plat = data.get("wow_by_platform") or {} | |
| yoy_plat = data.get("yoy_by_platform") or {} | |
| L.append("### 二、平台对比") | |
| L.append("") | |
| for plat in ["Google", "Meta"]: | |
| m = by_plat.get(plat) | |
| if not m: | |
| continue | |
| pw = wow_plat.get(plat, {}) | |
| py = yoy_plat.get(plat, {}) | |
| L.append(f"**{plat}**") | |
| _block_metrics(L, m, pw, py if py else None) | |
| L.append("") | |
| by_ct = curr.get("by_campaign_type", {}) | |
| wow_ct = data.get("wow_by_campaign") or {} | |
| yoy_ct = data.get("yoy_by_campaign") or {} | |
| L.append("### 三、广告漏斗") | |
| L.append("") | |
| for ct in ["Awareness", "Consideration", "Conversion"]: | |
| m = by_ct.get(ct) | |
| if not m: | |
| continue | |
| cw = wow_ct.get(ct, {}) | |
| cy = yoy_ct.get(ct, {}) | |
| L.append(f"**{ct}**") | |
| _append_funnel_dim_lines(L, m, cw, cy) | |
| L.append("") | |
| by_country = curr.get("by_country", {}) | |
| wow_country = data.get("wow_by_country") or {} | |
| yoy_country = data.get("yoy_by_country") or {} | |
| L.append("### 四、市场表现") | |
| L.append("") | |
| for c in sorted(by_country.keys()): | |
| m = by_country[c] | |
| cw = wow_country.get(c, {}) | |
| cy = yoy_country.get(c, {}) | |
| L.append(f"**{c}**") | |
| _append_country_dim_lines(L, m, cw, cy, full=True) | |
| L.append("") | |
| _append_expert_section(L, data.get("expert", {})) | |
| return "\n".join(L).strip() | |
| def _append_expert_section(L: list[str], expert: dict[str, Any]) -> None: | |
| """将专家分析追加到报告末尾。兼容新格式（verdict）和旧格式（summary）。""" | |
| if not expert: | |
| return | |
| L.append("### 陈策点评") | |
| L.append("") | |
| verdict = expert.get("verdict") or expert.get("summary", "") | |
| if verdict: | |
| L.append(f"> 💡 {verdict}") | |
| L.append("") | |
| findings = expert.get("key_findings") or [] | |
| if not findings: | |
| for key in ["efficiency", "platform", "funnel", "market"]: | |
| findings.extend(expert.get(key, [])) | |
| if findings: | |
| for i, f in enumerate(findings, 1): | |
| L.append(f"**{i}.** {f}") | |
| L.append("") | |
| alerts = expert.get("alerts", []) | |
| if alerts: | |
| for a in alerts: | |
| level = a.get("level", "yellow") | |
| icon = ALERT_ICON.get(level, "🟡") | |
| msg = a.get("msg") or f"{a.get('title', '')}：{a.get('detail', '')}" | |
| L.append(f"{icon} **{msg}**" if level == "red" else f"{icon} {msg}") | |
| L.append("") | |
| action = expert.get("next_action", "") | |
| if not action: | |
| recs = expert.get("recommendations", []) | |
| if recs: | |
| action = recs[0] | |
| if action: | |
| L.append(f"> 👉 **下一步：{action}**") | |
| L.append("") | |
| def _flag(wow_val: float | None, yoy_val: float | None) -> str: | |
| """大幅异动时加前缀标记，让读者快速定位重点。""" | |
| for v in [wow_val, yoy_val]: | |
| if v is not None and v == v: | |
| if v <= -20: | |
| return "⚠️ " | |
| if v >= 20: | |
| return "🔥 " | |
| return "" | |
| def _block_metrics( | |
| L: list[str], | |
| m: dict[str, float], | |
| wow: dict[str, float | None], | |
| yoy: dict[str, float | None] | None = None, | |
| ) -> None: | |
| """每个核心指标独立一行，大幅异动自动标记。""" | |
| y = yoy or {} | |
| rows = [ | |
| ("花费", _n(m.get("Cost", 0), prefix="$"), wow.get("Cost"), y.get("Cost")), | |
| ("销售额", _n(m.get("Ads.Sales", 0), prefix="$"), wow.get("Ads.Sales"), y.get("Ads.Sales")), | |
| ("ROAS", f"{m.get('ROAS', 0):.2f}", wow.get("ROAS"), y.get("ROAS")), | |
| ("展示量", _n(m.get("Impressions", 0), d=0), wow.get("Impressions"), y.get("Impressions")), | |
| ("CTR", _pct(m.get("CTR", 0)), wow.get("CTR"), y.get("CTR")), | |
| ("转化率", _pct(m.get("CVR", 0)), wow.get("CVR"), y.get("CVR")), | |
| ("CPC", _n(m.get("CPC", 0), prefix="$"), wow.get("CPC"), y.get("CPC")), | |
| ("会话", _n(m.get("GA4.Sessions", 0), d=0), wow.get("GA4.Sessions"), y.get("GA4.Sessions")), | |
| ] | |
| if m.get("ATC", 0): | |
| rows.append(("加购 ATC", _n(m["ATC"], d=0), wow.get("ATC"), y.get("ATC"))) | |
| for name, val, w, yr in rows: | |
| prefix = _flag(w, yr) | |
| L.append(f"- {prefix}**{name}** {val}{_deltas_zh(w, yr)}") | |
| def format_title(data: dict[str, Any]) -> str: | |
| cp = data.get("current_period", {}) | |
| return f"DM Weekly Report {cp.get('year')} W{cp.get('week')}" | |
| # ───────────────────────────────────────────── | |
| # 月度报告模板（与周报结构一致） | |
| # ───────────────────────────────────────────── | |
| def format_monthly_report(data: dict[str, Any]) -> str: | |
| """月度汇总报告，与周报结构一致，含环比+同比。""" | |
| period = data.get("period_label", "") | |
| prev = data.get("prev_label") | |
| yoy_label = data.get("yoy_label") | |
| error = data.get("error") | |
| L: list[str] = [] | |
| L.append(f"## DM 月报 · {period}") | |
| cmp_parts = [] | |
| if prev: | |
| cmp_parts.append(f"环比：{prev}") | |
| if yoy_label: | |
| cmp_parts.append(f"同比：{yoy_label}") | |
| if cmp_parts: | |
| L.append("> " + " ｜ ".join(cmp_parts)) | |
| L.append("") | |
| if error: | |
| L.append(f"> {error}") | |
| return "\n".join(L).strip() | |
| curr = data.get("current", {}) | |
| wow_t = data.get("wow_total") or {} | |
| yoy_t = data.get("yoy_total") or {} | |
| total = curr.get("total", {}) | |
| L.append("### 一、整体表现") | |
| L.append("") | |
| _block_metrics(L, total, wow_t, yoy_t if yoy_t else None) | |
| L.append("") | |
| by_plat = curr.get("by_platform", {}) | |
| wow_plat = data.get("wow_by_platform") or {} | |
| yoy_plat = data.get("yoy_by_platform") or {} | |
| L.append("### 二、平台对比") | |
| L.append("") | |
| for plat in ["Google", "Meta"]: | |
| m = by_plat.get(plat) | |
| if not m: | |
| continue | |
| pw = wow_plat.get(plat, {}) | |
| py = yoy_plat.get(plat, {}) | |
| L.append(f"**{plat}**") | |
| _block_metrics(L, m, pw, py if py else None) | |
| L.append("") | |
| by_ct = curr.get("by_campaign_type", {}) | |
| wow_ct = data.get("wow_by_campaign") or {} | |
| yoy_ct = data.get("yoy_by_campaign") or {} | |
| L.append("### 三、广告漏斗") | |
| L.append("") | |
| for ct in ["Awareness", "Consideration", "Conversion"]: | |
| m = by_ct.get(ct) | |
| if not m: | |
| continue | |
| cw = wow_ct.get(ct, {}) | |
| cy = yoy_ct.get(ct, {}) | |
| L.append(f"**{ct}**") | |
| _append_funnel_dim_lines(L, m, cw, cy) | |
| L.append("") | |
| by_country = curr.get("by_country", {}) | |
| wow_country = data.get("wow_by_country") or {} | |
| yoy_country = data.get("yoy_by_country") or {} | |
| L.append("### 四、市场表现") | |
| L.append("") | |
| for c in sorted(by_country.keys()): | |
| m = by_country[c] | |
| cw = wow_country.get(c, {}) | |
| cy = yoy_country.get(c, {}) | |
| L.append(f"**{c}**") | |
| _append_country_dim_lines(L, m, cw, cy, full=True) | |
| L.append("") | |
| _append_expert_section(L, data.get("expert", {})) | |
| return "\n".join(L).strip() | |
| def format_monthly_title(data: dict[str, Any]) -> str: | |
| return f"DM Monthly Report {data.get('period_label', '')}" | |
| # ───────────────────────────────────────────── | |
| # 日报模板（复用月报结构） | |
| # ───────────────────────────────────────────── | |
| def format_daily_report(data: dict[str, Any]) -> str: | |
| """日度报告，结构与月报一致，标签改为 DoD/YoY。""" | |
| period = data.get("period_label", "") | |
| prev = data.get("prev_label") | |
| yoy_label = data.get("yoy_label") | |
| error = data.get("error") | |
| L: list[str] = [] | |
| L.append(f"## DM 日报 · {period}") | |
| cmp_parts = [] | |
| if prev: | |
| cmp_parts.append(f"环比：{prev}") | |
| if yoy_label: | |
| cmp_parts.append(f"同比：{yoy_label}") | |
| if cmp_parts: | |
| L.append("> " + " ｜ ".join(cmp_parts)) | |
| L.append("") | |
| if error: | |
| L.append(f"> {error}") | |
| return "\n".join(L).strip() | |
| curr = data.get("current", {}) | |
| wow_t = data.get("wow_total") or {} | |
| yoy_t = data.get("yoy_total") or {} | |
| total = curr.get("total", {}) | |
| L.append("### 整体表现") | |
| L.append("") | |
| _block_metrics(L, total, wow_t, yoy_t if yoy_t else None) | |
| L.append("") | |
| by_plat = curr.get("by_platform", {}) | |
| wow_plat = data.get("wow_by_platform") or {} | |
| yoy_plat = data.get("yoy_by_platform") or {} | |
| L.append("### 平台对比") | |
| L.append("") | |
| for plat in ["Google", "Meta"]: | |
| m = by_plat.get(plat) | |
| if not m: | |
| continue | |
| pw = wow_plat.get(plat, {}) | |
| py = yoy_plat.get(plat, {}) | |
| L.append(f"**{plat}**") | |
| _block_metrics(L, m, pw, py if py else None) | |
| L.append("") | |
| by_country = curr.get("by_country", {}) | |
| wow_country = data.get("wow_by_country") or {} | |
| yoy_country = data.get("yoy_by_country") or {} | |
| if by_country: | |
| L.append("### 市场表现") | |
| L.append("") | |
| for c in sorted(by_country.keys()): | |
| m = by_country[c] | |
| cw = wow_country.get(c, {}) | |
| cy = yoy_country.get(c, {}) | |
| L.append(f"**{c}**") | |
| _append_country_dim_lines(L, m, cw, cy, full=False) | |
| L.append("") | |
| _append_expert_section(L, data.get("expert", {})) | |
| return "\n".join(L).strip() | |
| def format_daily_title(data: dict[str, Any]) -> str: | |
| return f"DM Daily Report {data.get('period_label', '')}" | |
| # ───────────────────────────────────────────── | |
| # 自定义时间段报告 | |
| # ───────────────────────────────────────────── | |
| def format_range_report(data: dict[str, Any]) -> str: | |
| """自定义时间段报告，结构与月报一致。""" | |
| period = data.get("period_label", "") | |
| prev = data.get("prev_label") | |
| yoy_label = data.get("yoy_label") | |
| days = data.get("days_count", "") | |
| error = data.get("error") | |
| L: list[str] = [] | |
| L.append(f"## DM 自定义报告 · {period}") | |
| sub_parts = [] | |
| if days: | |
| sub_parts.append(f"统计 {days} 天") | |
| if prev: | |
| sub_parts.append(f"环比：{prev}") | |
| if yoy_label: | |
| sub_parts.append(f"同比：{yoy_label}") | |
| if sub_parts: | |
| L.append("> " + " ｜ ".join(sub_parts)) | |
| L.append("") | |
| if error: | |
| L.append(f"> {error}") | |
| return "\n".join(L).strip() | |
| curr = data.get("current", {}) | |
| wow_t = data.get("wow_total") or {} | |
| yoy_t = data.get("yoy_total") or {} | |
| total = curr.get("total", {}) | |
| L.append("### 整体表现") | |
| L.append("") | |
| _block_metrics(L, total, wow_t, yoy_t if yoy_t else None) | |
| L.append("") | |
| by_plat = curr.get("by_platform", {}) | |
| wow_plat = data.get("wow_by_platform") or {} | |
| yoy_plat = data.get("yoy_by_platform") or {} | |
| L.append("### 平台对比") | |
| L.append("") | |
| for plat in ["Google", "Meta"]: | |
| m = by_plat.get(plat) | |
| if not m: | |
| continue | |
| pw = wow_plat.get(plat, {}) | |
| py = yoy_plat.get(plat, {}) | |
| L.append(f"**{plat}**") | |
| _block_metrics(L, m, pw, py if py else None) | |
| L.append("") | |
| by_ct = curr.get("by_campaign_type", {}) | |
| wow_ct = data.get("wow_by_campaign") or {} | |
| yoy_ct = data.get("yoy_by_campaign") or {} | |
| if by_ct: | |
| L.append("### 广告漏斗") | |
| L.append("") | |
| for ct in ["Awareness", "Consideration", "Conversion"]: | |
| m = by_ct.get(ct) | |
| if not m: | |
| continue | |
| cw = wow_ct.get(ct, {}) | |
| cy = yoy_ct.get(ct, {}) | |
| L.append(f"**{ct}**") | |
| _append_funnel_dim_lines(L, m, cw, cy) | |
| L.append("") | |
| by_country = curr.get("by_country", {}) | |
| wow_country = data.get("wow_by_country") or {} | |
| yoy_country = data.get("yoy_by_country") or {} | |
| if by_country: | |
| L.append("### 市场表现") | |
| L.append("") | |
| for c in sorted(by_country.keys()): | |
| m = by_country[c] | |
| cw = wow_country.get(c, {}) | |
| cy = yoy_country.get(c, {}) | |
| L.append(f"**{c}**") | |
| _append_country_dim_lines(L, m, cw, cy, full=False) | |
| L.append("") | |
| _append_expert_section(L, data.get("expert", {})) | |
| return "\n".join(L).strip() | |
| def format_range_title(data: dict[str, Any]) -> str: | |
| return f"DM Custom Report {data.get('period_label', '')}" | |