Spaces:
Runtime error
Runtime error
File size: 3,965 Bytes
ebb46dc 8d0d365 ebb46dc 8d0d365 ebb46dc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | 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
|