浅念
218 lines
9.9 KiB
Python
218 lines
9.9 KiB
Python
"""来源归因规则引擎：裸域名 → 渠道/平台/搜索词/证据可信度分层识别。
|
||
|
||
纯函数、无 IO、无 DB——classify 是唯一入口，ingest 增量计算与
|
||
recompute_attribution 历史重算共用同一套规则（改规则必须 bump
|
||
constants.ATTRIBUTION_RULE_VERSION 并重跑重算）。
|
||
|
||
诚实状态语义（resolution）：
|
||
resolved 观察到 referrer/点击参数/UA 等硬证据
|
||
combined referrer 观察与 UTM 标签互相印证
|
||
conflict referrer 结论与 UTM 标签矛盾——观察证据优先，标签并列存 detail
|
||
label_only 仅 UTM 自述，无观察证据
|
||
unknown 全空（直接访问）——不猜测手输/书签
|
||
"""
|
||
import re
|
||
from urllib.parse import parse_qsl, urlsplit
|
||
|
||
from tracker import constants as C
|
||
|
||
_CTRL_CHARS = re.compile(r"[\x00-\x1f\x7f]")
|
||
_GENERIC_TLDS = ("com",) # 引擎 host 落在这些 TLD 时不作为国家 variant 显示
|
||
|
||
# 广告平台家族归一表（冲突判定前）：同一家族的各种写法（缩写/全称/带 ads 后缀/
|
||
# 中文俄文别称）→ 家族名。fbclid 推出的 "Meta Ads" 与 utm_source=fb/ig 同属 meta
|
||
# 家族 → 印证而非矛盾。键为去空格后的紧凑形态（"meta ads" → "metaads"）。
|
||
_AD_FAMILIES = {
|
||
"fb": "meta", "facebook": "meta", "fbads": "meta",
|
||
"ig": "meta", "instagram": "meta", "meta": "meta", "metaads": "meta",
|
||
"google": "google", "googleads": "google", "adwords": "google",
|
||
"bing": "bing", "bingads": "bing", "microsoftads": "bing",
|
||
"yandex": "yandex", "yandexads": "yandex", "директ": "yandex",
|
||
}
|
||
# UTM 标签 ↔ 平台名别名（家族表之外的零散别名）；baidu 恒等仅为可读
|
||
_UTM_ALIASES = {"tt": "tiktok",
|
||
"wa": "whatsapp", "g": "google", "baidu": "baidu", "百度": "baidu"}
|
||
|
||
|
||
def _host(url: str) -> str:
|
||
"""提取 referrer host；异常/空 → ""。hostname 自动去端口并小写。"""
|
||
try:
|
||
return (urlsplit(url).hostname or "").lower()
|
||
except ValueError:
|
||
return ""
|
||
|
||
|
||
def _canonical(host: str) -> str:
|
||
"""去掉单个 www. 前缀后做域名匹配（www.bing.com → bing.com）。"""
|
||
return host[4:] if host.startswith("www.") else host
|
||
|
||
|
||
def _domain_match(host: str, key: str) -> bool:
|
||
"""host 是否命中规则键。key 以 "." 结尾（google./yandex.）→ 前缀匹配任意国家站
|
||
TLD；否则精确或子域（m.facebook.com 命中 facebook.com）。"""
|
||
h = _canonical(host)
|
||
k = key.lower()
|
||
if k.endswith("."):
|
||
return h == k[:-1] or h.startswith(k)
|
||
return h == k or h.endswith("." + k)
|
||
|
||
|
||
def _tld_variant(host: str) -> str:
|
||
"""引擎 host 的 TLD 国家（www.google.co.ru → "ru"）；通用 TLD 不显示。"""
|
||
tld = _canonical(host).rsplit(".", 1)[-1]
|
||
return "" if tld in _GENERIC_TLDS else tld
|
||
|
||
|
||
def _clean_term(raw: str) -> str:
|
||
"""搜索词清洗：strip → 剔除控制字符 → 截 100。"""
|
||
return _CTRL_CHARS.sub("", (raw or "").strip())[:100]
|
||
|
||
|
||
def _search_term(engine: str, referrer_url: str) -> str:
|
||
"""按引擎参数名白名单从 referrer URL 提取搜索词。Google 不在表内 → 恒空（绝不虚构）。"""
|
||
for param in C.SEARCH_TERM_PARAMS.get(engine, []):
|
||
for k, v in parse_qsl(urlsplit(referrer_url).query, keep_blank_values=True):
|
||
if k == param and v.strip():
|
||
return _clean_term(v)
|
||
return ""
|
||
|
||
|
||
def _paid_param(landing_path: str, referrer_url: str) -> tuple[str, str] | None:
|
||
"""付费优先于自然：落地页（或 referrer）query 命中点击参数 → (参数名, 平台名)。"""
|
||
for source in (landing_path, referrer_url):
|
||
if not source:
|
||
continue
|
||
try:
|
||
params = [k for k, _ in parse_qsl(urlsplit(source).query,
|
||
keep_blank_values=True)]
|
||
except ValueError:
|
||
continue
|
||
for k in params:
|
||
if k in C.PAID_CLICK_PARAMS:
|
||
return k, C.PAID_CLICK_PARAMS[k]
|
||
return None
|
||
|
||
|
||
def _in_app_name(ua: str) -> str:
|
||
"""UA 命中应用内浏览器标记（大小写不敏感）→ 应用显示名；未命中 → ""。"""
|
||
low = (ua or "").lower()
|
||
for marker, name in C.INAPP_UA_MARKERS.items():
|
||
if marker.lower() in low:
|
||
return name
|
||
return ""
|
||
|
||
|
||
def _canon(name: str) -> str:
|
||
"""平台名/UTM 标签宽松归一（仅用于冲突判定）：小写 → 去协议与路径 → 去 www. 前缀
|
||
→ 去最后一段域名后缀（chatgpt.com→chatgpt、google.ru→google）→ 去空格（"meta
|
||
ads"→"metaads"）→ 家族表（fb/ig↔Meta Ads→meta；adwords→google；директ→yandex）
|
||
→ 别名表 → 仅留字母数字（含汉字）。同广告家族（meta/google/bing/yandex）归一后
|
||
同源，不再误报 conflict。"""
|
||
s = (name or "").lower().strip()
|
||
if "://" in s:
|
||
s = s.split("://", 1)[1]
|
||
s = s.split("/", 1)[0]
|
||
if s.startswith("www."):
|
||
s = s[4:]
|
||
if "." in s:
|
||
s = s.rsplit(".", 1)[0]
|
||
s = re.sub(r"\s+", "", s)
|
||
if s in _AD_FAMILIES: # 家族命中即返回家族名（已是规范 token）
|
||
return _AD_FAMILIES[s]
|
||
s = _UTM_ALIASES.get(s, s)
|
||
return re.sub(r"[^0-9a-z一-鿿]", "", s)
|
||
|
||
|
||
def _utm_channel(label: str) -> str:
|
||
"""UTM 标签尽力归类：已知 AI/引擎/社交标签（裸标签 "baidu"/显示名 "Google" 皆可）
|
||
映射到对应渠道，未知 → referral（有名来源，可信度由 label_only 单独表达）。"""
|
||
h = (label or "").lower().strip()
|
||
if not h:
|
||
return "referral"
|
||
for table, channel in ((C.AI_REFERRAL_DOMAINS, "ai_referral"),
|
||
(C.SEARCH_ENGINE_DOMAINS, "natural_search"),
|
||
(C.SOCIAL_DOMAINS, "social")):
|
||
for key, name in table.items():
|
||
if _domain_match(h, key) or key.startswith(h + ".") or h == name.lower():
|
||
return channel
|
||
return "referral"
|
||
|
||
|
||
def classify(*, referrer_url: str = "", utm: dict | None = None,
|
||
landing_path: str = "", ua: str = "", internal_domains: list | None = None) -> dict:
|
||
"""把一次到访的原始信号分类为 {channel, platform, variant, search_term,
|
||
evidence_kind, resolution, detail}。各字段默认空串/空 dict，纯函数无副作用。
|
||
|
||
规则序：本站域名去信号 → 付费点击参数 → AI → 搜索引擎 → 社交 → 泛 referral
|
||
→ UTM 证据叠加（combined/conflict/label_only）→ 应用内 UA 叠加/兜底 → direct。
|
||
"""
|
||
out = {"channel": "", "platform": "", "variant": "", "search_term": "",
|
||
"evidence_kind": "", "resolution": "", "detail": {}}
|
||
utm = utm or {}
|
||
utm_source = str(utm.get("utm_source") or "").strip()
|
||
|
||
host = _host(referrer_url)
|
||
if host and any(_domain_match(host, d) for d in (internal_domains or []) if d):
|
||
host = "" # 本站跳转不是外部来源信号
|
||
|
||
paid = _paid_param(landing_path, referrer_url)
|
||
if paid:
|
||
out.update(channel="paid_search", platform=paid[1],
|
||
evidence_kind="observed_referrer", resolution="resolved",
|
||
detail={"click_param": paid[0]})
|
||
elif host:
|
||
for key, name in C.AI_REFERRAL_DOMAINS.items():
|
||
if _domain_match(host, key):
|
||
out.update(channel="ai_referral", platform=name)
|
||
break
|
||
else:
|
||
for key, name in C.SEARCH_ENGINE_DOMAINS.items():
|
||
if _domain_match(host, key):
|
||
out.update(channel="natural_search", platform=name,
|
||
variant=_tld_variant(host),
|
||
search_term=_search_term(name, referrer_url))
|
||
break
|
||
else:
|
||
for key, name in C.SOCIAL_DOMAINS.items():
|
||
if _domain_match(host, key):
|
||
out.update(channel="social", platform=name)
|
||
break
|
||
else:
|
||
out.update(channel="referral", platform=host)
|
||
if not out["resolution"]:
|
||
out.update(evidence_kind="observed_referrer", resolution="resolved")
|
||
|
||
# UTM 证据叠加：观察结论之上只做印证/并列，不覆盖
|
||
if utm_source:
|
||
if out["resolution"] == "resolved":
|
||
if _canon(utm_source) == _canon(out["platform"]):
|
||
# 归一后同源 → 互相印证；UTM 原文并记进 detail（顶层展示观察结论）
|
||
out["evidence_kind"] = "combined"
|
||
out["detail"] = {**out["detail"], "utm_source": utm_source}
|
||
else: # 矛盾：观察证据保留在顶层，两路证据各自完整并列进 detail
|
||
out["resolution"] = "conflict"
|
||
out["detail"] = {**out["detail"],
|
||
"referrer_evidence": {
|
||
"channel": out["channel"], "platform": out["platform"],
|
||
"variant": out["variant"], "search_term": out["search_term"],
|
||
"evidence_kind": "observed_referrer", "referrer_host": host},
|
||
"utm_evidence": {
|
||
"channel": _utm_channel(utm_source), "platform": utm_source,
|
||
"evidence_kind": "utm_label"}}
|
||
else: # 无 referrer 观察 → 仅标签：按 utm_source 归类显示，状态诚实标注
|
||
out.update(channel=_utm_channel(utm_source), platform=utm_source,
|
||
evidence_kind="utm_label", resolution="label_only")
|
||
|
||
# 应用内浏览器 UA：有来源 → 作为 surface 叠加；无任何来源 → 本身就是来源
|
||
app = _in_app_name(ua)
|
||
if app:
|
||
if out["resolution"]:
|
||
out["detail"] = {**out["detail"], "in_app": app}
|
||
else:
|
||
out.update(channel="in_app", platform=app,
|
||
evidence_kind="observed_ua", resolution="resolved")
|
||
|
||
if not out["channel"]:
|
||
out.update(channel="direct", resolution="unknown") # 不猜测手输/书签
|
||
return out
|