obsidian-wiki/scripts/extract_email_project_signals.py

#!/usr/bin/env python3
import collections
import json
import re
import sys
from pathlib import Path


OEM_TERMS = [
    "零跑",
    "Leapmotor",
    "BYD",
    "比亚迪",
    "Toyota",
    "SVW",
    "上汽大众",
    "MZD",
    "Mazda",
    "VW",
    "STLA",
    "Stellantis",
    "Ford",
    "Honda",
    "Hyundai",
    "Kia",
    "CATARC",
    "Yazaki",
    "Marelli",
    "Valeo",
    "Visteon",
]

APP_TERMS = [
    "吸顶屏",
    "显示",
    "显示屏",
    "CID",
    "FID",
    "Cluster",
    "仪表",
    "Full Digital Cluster",
    "氛围灯",
    "Ambient Lighting",
    "储能",
    "车灯",
    "lighting",
    "霍尔小板",
    "排线",
    "FPC",
    "FFC",
    "Fakra",
    "FAKRA",
    "Socket Header",
    "Pin Header",
    "SMB",
    "BTB",
    "WTB",
    "Wafer",
]

PRODUCT_RE = re.compile(
    r"\b(?:\d{6}-\d{4,6}(?:-HF)?|\d{3,}-\d{3,}(?:-HF)?|FH\d+[A-Z0-9-]*|C-\d{7})\b",
    re.I,
)

PROJECT_RE = re.compile(
    r"(?:项目名称[:：]\s*([^\n\r；;，,]{2,60})|项目车型[:：]\s*([^\n\r；;，,]{2,60})|"
    r"\b(?:eP|EP)\s*\d{4,6}\b|CU\d+|BR\d+|MZD[-\w]+|Scout|RUV|PICKUP)",
    re.I,
)


def domain_of(person):
    if not isinstance(person, dict):
        return None
    email = (person.get("email") or "").lower()
    return email.split("@")[-1] if "@" in email else None


def context_snippet(text, term, limit=130):
    idx = text.lower().find(term.lower())
    if idx < 0:
        return ""
    start = max(0, idx - 45)
    end = min(len(text), idx + len(term) + 85)
    s = re.sub(r"\s+", " ", text[start:end]).strip()
    return s[:limit]


def main():
    if len(sys.argv) != 2:
        print("usage: extract_email_project_signals.py /path/to/llm_input.jsonl", file=sys.stderr)
        return 2
    path = Path(sys.argv[1])
    pairs = collections.Counter()
    oems = collections.Counter()
    apps = collections.Counter()
    products = collections.Counter()
    domains = collections.Counter()
    rows = []

    with path.open(encoding="utf-8", errors="replace") as fh:
        for line in fh:
            obj = json.loads(line)
            messages = obj.get("messages", [])
            text = " ".join(
                [obj.get("suggested_title_from_subject", "")]
                + [(m.get("subject") or "") + " " + (m.get("body_text") or "") for m in messages]
            )
            found_oems = [x for x in OEM_TERMS if x.lower() in text.lower()]
            found_apps = [x for x in APP_TERMS if x.lower() in text.lower()]
            found_products = [x.upper() for x in PRODUCT_RE.findall(text)]
            project_hits = []
            for m in PROJECT_RE.finditer(text):
                project_hits.append(next((g for g in m.groups() if g), m.group(0)))
            if not (found_oems or found_apps or project_hits):
                continue
            thread_domains = collections.Counter()
            for msg in messages:
                for person in [msg.get("from", {})] + msg.get("to", []) + msg.get("cc", []):
                    d = domain_of(person)
                    if d:
                        thread_domains[d] += 1
            for d, n in thread_domains.items():
                domains[d] += n
            for x in found_oems:
                oems[x] += 1
            for x in found_apps:
                apps[x] += 1
            for x in found_products[:25]:
                products[x] += 1
            for oem in found_oems:
                for app in found_apps:
                    pairs[(oem, app)] += 1
            score = len(found_oems) * 4 + len(found_apps) * 2 + len(project_hits) * 3 + len(found_products)
            if score >= 8:
                anchor = (found_oems + found_apps + project_hits + found_products)[:1]
                rows.append(
                    {
                        "thread_id": obj.get("thread_id"),
                        "subject": obj.get("suggested_title_from_subject", "")[:160],
                        "date_end": obj.get("date_end"),
                        "customer_domains": [d for d, _ in thread_domains.most_common(5)],
                        "oems": found_oems[:8],
                        "applications": found_apps[:10],
                        "projects": project_hits[:8],
                        "products": found_products[:12],
                        "snippet": context_snippet(text, anchor[0]) if anchor else "",
                    }
                )

    out = {
        "oems_top": oems.most_common(50),
        "applications_top": apps.most_common(50),
        "products_top": products.most_common(80),
        "oem_application_pairs_top": [
            {"oem": k[0], "application": k[1], "threads": v}
            for k, v in pairs.most_common(80)
        ],
        "customer_domains_top": domains.most_common(50),
        "project_signal_threads": sorted(rows, key=lambda x: x["date_end"] or "", reverse=True)[:120],
    }
    json.dump(out, sys.stdout, ensure_ascii=False, indent=2)
    print()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())