"""
规则版本: name_normalization.v1
目标/Goal:
- 将 crm_contacts.full_name 统一为可比的规范形态，支持中英文团队协作。
- Normalize crm_contacts.full_name into a comparable canonical form for bilingual (Chinese-English) collaboration.
规则/Rules:
- 预清洗 Pre-cleaning:
- 去除首尾空格，合并连续空格为1个。
- 清除 HTML 标签、控制字符(Cc)、Emoji。
- 记录原始数字比例；若数字比例>20%，标记为 invalid:numeric_ratio 并返回 NULL。
- 移除阿拉伯数字与尾随标点(.,;:_-)；保留合法中间分隔符（-、·、’）。
- 大小写统一 Case normalization:
- 对拉丁字符采用 Title Case；保留合法撇号与连字符（如 McDonald、O’Connor、Anna-Maria）。
- 非拉丁/中文保持原文，仅统一冗余空格与符号。
- 符号标准 Symbol standardization:
- 统一全角/半角、引号和连字符：将各类撇号标准化为 ’ (U+2019)，连字符统一为 - (U+002D)，保留中间分隔符（-、·、’）。
- 文化姓名处理 Cultural handling:
- 中文名不拆分、不重排（例：王小明）。
- 西文名按 Firstname Lastname，保留中间名缩写与敬称位置。
- 去重 Deduplication:
- 以规范化后字符串的 SHA-256 为键分组。
- 保留最新记录（按 updated_at 最大），汇总来源ID与出现频次。
- 缺失值 Missing values:
- 空字符串、只含空白或清洗后为空的记为 NULL。
- 在 audit 列写入原因（empty/invalid），可附加 notes。
- 白名单与例外 Whitelist & exceptions:
- 保留合法前/后缀（如 Dr., Ms., PhD），在 lookup_names_titles 字典表维护与标准化。
- 输出 Output:
- 生成中英文规则说明与 Python 示例代码；代码函数化、幂等、可复用，可在 Airflow/Prefect 任务中调用。
- 校验 Validation:
- 规范化后长度需在 1–120 字符；超长截断为 120，并记录 audit 原因 truncated 与 notes。
- 版本化 Versioning:
- 规则版本固定为 name_normalization.v1，后续追加不破坏兼容。
Examples:
- ' ms. anna o’connor ' -> 'Ms. Anna O’Connor'
- '张三' -> '张三'
"""
from future import annotations
import re
import hashlib
import unicodedata
from datetime import datetime
from typing import Dict, List, Optional, Tuple, Any
RULE_VERSION = "name_normalization.v1"
Title whitelist/lookup dictionary table (可扩展/Extensible)
lookup_names_titles: Dict[str, Dict[str, str]] = {
"prefix": {
# English
"mr": "Mr.",
"mrs": "Mrs.",
"ms": "Ms.",
"miss": "Ms.",
"dr": "Dr.",
"prof": "Prof.",
"sir": "Sir",
"madam": "Ms.",
"mx": "Mx.",
# Chinese to English mappings (optional normalization)
"先生": "Mr.",
"女士": "Ms.",
"小姐": "Ms.",
"博士": "Dr.",
"教授": "Prof.",
},
"suffix": {
# English
"phd": "PhD",
"md": "MD",
"dds": "DDS",
"dvm": "DVM",
"jd": "JD",
"mba": "MBA",
"jr": "Jr.",
"sr": "Sr.",
"iii": "III",
"iv": "IV",
# Chinese degrees could be mapped if desired; left empty to avoid lossy conversion
},
}
Emoji ranges for broad removal (BMP + supplementary planes)
_EMOJI_PATTERN = re.compile(
"[" +
"\U0001F600-\U0001F64F" + # Emoticons
"\U0001F300-\U0001F5FF" + # Misc Symbols and Pictographs
"\U0001F680-\U0001F6FF" + # Transport & Map
"\U0001F1E6-\U0001F1FF" + # Regional Indicators
"\U0001F900-\U0001F9FF" + # Supplemental Symbols and Pictographs
"\U0001FA70-\U0001FAFF" + # Symbols & Pictographs Extended-A
"\U00002700-\U000027BF" + # Dingbats
"\U00002600-\U000026FF" + # Misc symbols
"]",
flags=re.UNICODE
)
_HTML_PATTERN = re.compile(r"<[^>]*>")
_MULTI_SPACE_PATTERN = re.compile(r"\s+")
Allowed middle separators inside names
_ALLOWED_SEPARATORS = {"-", "·", "’"} # hyphen, middle dot, right single quotation mark
Trailing punctuation to strip from end of string
TRAILING_PUNCT_CHARS = ".,;:-"
def _strip_html(s: str) -> str:
return _HTML_PATTERN.sub("", s)
def _remove_emoji(s: str) -> str:
return _EMOJI_PATTERN.sub("", s)
def _remove_control_chars(s: str) -> str:
# Remove Unicode category Cc (control chars)
return "".join(ch for ch in s if unicodedata.category(ch) != "Cc")
def _normalize_width_and_symbols(s: str) -> str:
# Normalize width (fullwidth -> halfwidth) and standardize symbols
s = unicodedata.normalize("NFKC", s)
# Standardize apostrophes to U+2019
s = s.replace("\u2018", "’").replace("\u2019", "’").replace("'", "’").replace("`", "’")
# Standardize hyphens to ASCII hyphen-minus
s = s.replace("\u2013", "-").replace("\u2014", "-").replace("\u2212", "-").replace("－", "-")
# Leave middle dot as is
return s
def _strip_trailing_punct(s: str) -> str:
return s.rstrip(_TRAILING_PUNCT_CHARS)
def _collapse_spaces(s: str) -> str:
s = s.strip()
s = _MULTI_SPACE_PATTERN.sub(" ", s)
return s
def _digit_ratio(s: str) -> float:
if not s:
return 0.0
total = sum(1 for ch in s if not ch.isspace())
digits = sum(1 for ch in s if ch.isdigit())
return digits / total if total > 0 else 0.0
def _is_latin_token(token: str) -> bool:
# Consider a token Latin if all letters are from Latin scripts or separators/punct
for ch in token:
if ch.isalpha():
try:
name = unicodedata.name(ch)
except ValueError:
return False
if "LATIN" not in name:
return False
elif ch in _ALLOWED_SEPARATORS or ch in {".", " "}:
continue
else:
# other characters like Chinese/Cyrillic
return False
return True
def _titlecase_latin_token(token: str) -> str:
# Break on allowed separators and apply casing per segment
def tc_segment(seg: str) -> str:
if not seg:
return seg
lower = seg.lower()
# Preserve Mc/Mac style (basic heuristics)
if lower.startswith("mc") and len(lower) > 2:
return "Mc" + lower[2:3].upper() + lower[3:]
# Keep single-letter initials uppercase (e.g., "j." -> "J.")
if len(seg) == 1:
return seg.upper()
return lower[0].upper() + lower[1:]
# Tokens can have periods (e.g., "dr."), leave them for title mapping step
# Split by apostrophe and hyphen to title-case each segment
parts_apostrophe = token.split("’")
parts_apostrophe = [subpart for subpart in parts_apostrophe]
for i, ap in enumerate(parts_apostrophe):
subparts = ap.split("-")
subparts = [tc_segment(sp) for sp in subparts]
parts_apostrophe[i] = "-".join(subparts)
return "’".join(parts_apostrophe)
def _standardize_prefix(token: str, title_lookup: Dict[str, Dict[str, str]]) -> Optional[str]:
# Normalize token for matching (strip trailing dot)
candidate = token.replace(".", "")
canonical = title_lookup.get("prefix", {}).get(candidate.lower())
if canonical:
return canonical
return None
def _standardize_suffix(token: str, title_lookup: Dict[str, Dict[str, str]]) -> Optional[str]:
candidate = token.replace(".", "")
canonical = title_lookup.get("suffix", {}).get(candidate.lower())
if canonical:
return canonical
return None
def _process_tokens(tokens: List[str], title_lookup: Dict[str, Dict[str, str]]) -> List[str]:
if not tokens:
return tokens
# Handle prefix at first token
first = tokens[0]
pref = _standardize_prefix(first, title_lookup)
if pref:
tokens[0] = pref
start_idx = 1
else:
start_idx = 0
# Handle suffix at last token
last = tokens[-1]
suff = _standardize_suffix(last, title_lookup)
if suff:
tokens[-1] = suff
end_idx = len(tokens) - 1
else:
end_idx = len(tokens)
# Title-case middle tokens that are Latin
for i in range(start_idx, end_idx):
tok = tokens[i]
if _is_latin_token(tok):
tokens[i] = _titlecase_latin_token(tok)
else:
# Non-Latin: leave as-is (e.g., Chinese), only width/symbol normalization already applied
tokens[i] = tok
return tokens
def _remove_digits(s: str) -> str:
return re.sub(r"\d+", "", s)
def _normalize_name_string(raw: str, title_lookup: Dict[str, Dict[str, str]]) -> Tuple[Optional[str], Dict[str, Any]]:
"""
Returns (normalized_name or None, audit_info)
audit_info contains: reason, notes
"""
audit = {"reason": None, "notes": {}}
if raw is None:
audit["reason"] = "empty"
return None, audit
original = raw
# Numeric ratio check on original text (excluding spaces)
ratio = _digit_ratio(original)
if ratio > 0.20:
audit["reason"] = "invalid:numeric_ratio"
audit["notes"]["digit_ratio"] = ratio
return None, audit
# Pre-cleaning
s = _strip_html(original)
s = _remove_control_chars(s)
s = _remove_emoji(s)
s = _normalize_width_and_symbols(s)
s = _collapse_spaces(s)
# Remove digits (Arabic numerals) per rule
s = _remove_digits(s)
# Strip trailing punctuation
s = _strip_trailing_punct(s)
# Space collapse again if removal created double spaces
s = _collapse_spaces(s)
# Missing value check
if not s or s.strip() == "":
audit["reason"] = "empty"
return None, audit
# Tokenization by spaces for prefix/suffix handling and Latin title-case
tokens = s.split(" ")
# Standardize titles and apply Latin title-case
tokens = _process_tokens(tokens, title_lookup)
normalized = " ".join(tokens)
# Length validation
nlen = len(normalized)
if nlen == 0:
audit["reason"] = "empty"
return None, audit
if nlen > 120:
normalized = normalized[:120]
audit["reason"] = "truncated"
audit["notes"]["original_length"] = nlen
audit["notes"]["truncated_to"] = 120
else:
audit["reason"] = "ok"
return normalized, audit
def _sha256_hex(s: str) -> str:
return hashlib.sha256(s.encode("utf-8")).hexdigest()
def normalize_full_name(
name: Optional[str],
*,
title_lookup: Dict[str, Dict[str, str]] = lookup_names_titles,
) -> Dict[str, Any]:
"""
Normalize a full name string according to name_normalization.v1.
Returns a dict:
{
"normalized_full_name": Optional[str],
"full_name_hash": Optional[str], # SHA-256 of normalized string
"audit_reason": str, # "ok", "empty", "invalid:numeric_ratio", "truncated"
"audit_notes": Dict[str, Any],
"version": RULE_VERSION,
}
Idempotent: applying the function multiple times yields the same result.
"""
normalized, audit = _normalize_name_string(name, title_lookup)
result = {
"normalized_full_name": normalized,
"full_name_hash": _sha256_hex(normalized) if normalized else None,
"audit_reason": audit["reason"],
"audit_notes": audit.get("notes", {}),
"version": RULE_VERSION,
}
return result
def _parse_datetime(dt: Any) -> Optional[datetime]:
if dt is None:
return None
if isinstance(dt, datetime):
return dt
# Try ISO 8601
try:
# Support 'Z' suffix
if isinstance(dt, str):
if dt.endswith("Z"):
dt = dt[:-1]
return datetime.fromisoformat(dt)
except Exception:
return None
return None
def deduplicate_contacts(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Deduplicate contacts by normalized_full_name SHA-256 hash.
Input records: list of dicts, each containing at least:
{
"id": Any,
"full_name": str,
"source_id": Any,
"updated_at": datetime or ISO8601 str,
... (other fields)
}
Process:
- Normalize full_name (apply rules).
- Group by normalized_full_name hash.
- Keep the latest record (max updated_at) per group.
- Summarize source_ids (unique) and occurrences (count).
- Records with NULL normalized_full_name are excluded from grouping and returned individually with audit.
Returns list of dicts:
{
"id": Any, # id of retained (latest) record
"normalized_full_name": Optional[str],
"full_name_hash": Optional[str],
"audit_reason": str,
"audit_notes": Dict[str, Any],
"version": RULE_VERSION,
"source_ids": List[Any], # aggregated for grouped records
"occurrences": int, # count in group
"representative_record": Dict[str, Any], # latest record snapshot
}
"""
groups: Dict[str, Dict[str, Any]] = {}
invalids: List[Dict[str, Any]] = []
for rec in records:
norm = normalize_full_name(rec.get("full_name"))
updated = _parse_datetime(rec.get("updated_at"))
enriched = {
"id": rec.get("id"),
"normalized_full_name": norm["normalized_full_name"],
"full_name_hash": norm["full_name_hash"],
"audit_reason": norm["audit_reason"],
"audit_notes": norm["audit_notes"],
"version": RULE_VERSION,
"source_ids": [rec.get("source_id")],
"occurrences": 1,
"representative_record": rec,
"_updated_at": updated or datetime.min,
}
if norm["normalized_full_name"] is None or norm["full_name_hash"] is None:
# keep as standalone invalid/empty
invalids.append(enriched)
continue
key = norm["full_name_hash"]
if key not in groups:
groups[key] = enriched
else:
grp = groups[key]
# Aggregate source_ids
grp["source_ids"].append(rec.get("source_id"))
grp["occurrences"] += 1
# Choose latest record
cur_dt = grp["_updated_at"]
new_dt = enriched["_updated_at"]
if new_dt and new_dt > cur_dt:
grp["id"] = enriched["id"]
grp["representative_record"] = rec
grp["_updated_at"] = new_dt
# Audit reason: if any member truncated, keep "ok"/"truncated" precedence
# Prefer "ok" over "truncated" only if representative not truncated; else keep truncated
# If any invalid slipped here (shouldn't), ignore since norm exists
# Dedup group keeps normalized_full_name from representative
groups[key] = grp
# Finalize output: dedup source_ids unique, remove internal fields
output: List[Dict[str, Any]] = []
for g in groups.values():
g["source_ids"] = [sid for sid in set(g["source_ids"]) if sid is not None]
g.pop("_updated_at", None)
output.append(g)
# Include invalids as independent (no grouping)
for inv in invalids:
inv["source_ids"] = [sid for sid in set(inv["source_ids"]) if sid is not None]
inv.pop("_updated_at", None)
output.append(inv)
return output
Example reusable task wrappers
def process_batch(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Batch processor suitable for Airflow/Prefect tasks:
- Applies normalization and deduplication.
- Returns normalized, deduplicated list with audit info.
"""
return deduplicate_contacts(records)
def airflow_task(**context) -> List[Dict[str, Any]]:
"""
Airflow-compatible callable.
Expects 'records' in context['params'] or context['ti'].xcom_pull().
"""
params = context.get("params", {}) if context else {}
records = params.get("records")
if records is None and context:
try:
records = context["ti"].xcom_pull(task_ids=params.get("upstream_task_id"))
except Exception:
records = []
records = records or []
return process_batch(records)
def prefect_task(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Prefect-compatible callable.
"""
return process_batch(records)
---- Demonstration ----
def _demo():
samples = [
{"id": 1, "full_name": " ms. anna o’connor ", "source_id": "crm_a", "updated_at": "2024-01-02T12:00:00Z"},
{"id": 2, "full_name": "Ms anna O'CONNOR", "source_id": "crm_b", "updated_at": "2024-03-05T09:30:00Z"},
{"id": 3, "full_name": "张三", "source_id": "crm_c", "updated_at": "2024-05-01T08:00:00Z"},
{"id": 4, "full_name": "Dr john mcDonald phd", "source_id": "crm_d", "updated_at": "2024-02-02T10:00:00Z"},
{"id": 5, "full_name": " ", "source_id": "crm_e", "updated_at": "2024-02-03T10:00:00Z"},
{"id": 6, "full_name": "Anna 1234 O’Connor 5678", "source_id": "crm_f", "updated_at": "2024-02-04T10:00:00Z"},
]
result = process_batch(samples)
for r in result:
print(r)
if name == "main":
_demo()
