Spaces:
Runtime error
Runtime error
| from __future__ import annotations | |
| """ | |
| Campaign Name 12 字段解析器。 | |
| 按 '_' 分割 campaign_name，提取 12 个标准字段， | |
| 自动归类市场/funnel/平台/优化目标/产品线/受众类型。 | |
| 字段内部用 '-' 连接（如 Bone-Conduction），不与 '_' 分隔符混淆。 | |
| 位置: 1 2 3 4 5 6 7 8 9 10 11 12 | |
| 字段: 市场 funnel 平台 渠道 优化目标 产品线 产品型号 项目类型 受众分类 素材类型 日期 备注 | |
| 示例: US Conversion Meta All Purchase Bone-Conduction Others AO ASC Mix 20260123 coreNumber | |
| """ | |
| import logging | |
| from typing import Any | |
| import pandas as pd | |
| log = logging.getLogger("ad-agent") | |
| VALID_MARKETS = {"US", "CA"} | |
| VALID_FUNNELS = {"Awareness", "Consideration", "Conversion"} | |
| VALID_PLATFORMS = {"Meta", "Google", "Criteo", "Bing", "TikTok"} | |
| VALID_OBJECTIVES = { | |
| "Purchase", "Lead", "ATC", "AtoB", | |
| "Linkclick", "Buttonclick", "Pageview", "PE", | |
| } | |
| AUDIENCE_REMARKETING = "remarketing" | |
| AUDIENCE_ASC = "asc" | |
| AUDIENCE_COLD = "cold" | |
| _FIELD_KEYS = [ | |
| "market", "funnel", "parsed_platform", "channel", | |
| "ad_objective", "product_line", "product_model", "project_type", | |
| "audience_raw", "creative_type", "campaign_date", "note", | |
| ] | |
| _OUTPUT_COLUMNS = _FIELD_KEYS + ["audience_type", "parse_ok", "objective_group"] | |
| def _classify_audience(raw: str) -> str: | |
| upper = raw.upper() | |
| if upper == "RMKT": | |
| return AUDIENCE_REMARKETING | |
| if upper == "ASC": | |
| return AUDIENCE_ASC | |
| return AUDIENCE_COLD | |
| def _derive_objective_group(objective: str | None, audience: str) -> str: | |
| if objective == "Purchase": | |
| if audience == AUDIENCE_REMARKETING: | |
| return "Purchase_RMKT" | |
| if audience == AUDIENCE_ASC: | |
| return "Purchase_ASC" | |
| return "Purchase_Cold" | |
| if objective in VALID_OBJECTIVES: | |
| return objective | |
| return "Other" | |
| def parse_campaign_name(name: str) -> dict[str, Any]: | |
| """ | |
| 解析单个 campaign name，返回 12 个原始字段 + 派生字段。 | |
| 不足 12 段时缺失字段为 None；超出 12 段时多余部分合并到 note。 | |
| """ | |
| parts = str(name).split("_") | |
| result: dict[str, Any] = {} | |
| for i, key in enumerate(_FIELD_KEYS): | |
| if i < len(parts): | |
| if key == "note" and len(parts) > 12: | |
| result[key] = "_".join(parts[11:]) | |
| else: | |
| result[key] = parts[i] | |
| else: | |
| result[key] = None | |
| result["audience_type"] = ( | |
| _classify_audience(result["audience_raw"]) | |
| if result["audience_raw"] | |
| else AUDIENCE_COLD | |
| ) | |
| result["parse_ok"] = ( | |
| len(parts) >= 5 | |
| and result.get("market") in VALID_MARKETS | |
| and result.get("funnel") in VALID_FUNNELS | |
| ) | |
| result["objective_group"] = _derive_objective_group( | |
| result.get("ad_objective"), result["audience_type"] | |
| ) | |
| return result | |
| def enrich_dataframe(df: pd.DataFrame) -> pd.DataFrame: | |
| """ | |
| 批量解析 campaign_name 列，新增解析字段列。返回新 DataFrame。 | |
| """ | |
| if "campaign_name" not in df.columns: | |
| log.warning("DataFrame 中没有 campaign_name 列，跳过解析") | |
| return df | |
| df = df.copy() | |
| parsed = df["campaign_name"].apply(parse_campaign_name) | |
| parsed_df = pd.DataFrame(parsed.tolist(), index=df.index) | |
| for col in _OUTPUT_COLUMNS: | |
| if col in parsed_df.columns: | |
| df[col] = parsed_df[col] | |
| ok_count = df["parse_ok"].sum() if "parse_ok" in df.columns else 0 | |
| total = len(df) | |
| if total > 0: | |
| rate = ok_count / total * 100 | |
| log.info(f"Campaign name 解析: {ok_count}/{total} 成功 ({rate:.1f}%)") | |
| if rate < 90: | |
| log.warning("解析成功率低于 90%，可能存在命名不规范的 campaign") | |
| return df | |