obsidian-wiki/scripts/analyze_email_jsonl.py

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


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

KEYWORDS = [
    "报价",
    "询价",
    "样品",
    "打样",
    "图纸",
    "承认书",
    "规格书",
    "替代",
    "替换",
    "竞品",
    "Hirose",
    "Molex",
    "TE",
    "泰科",
    "JAE",
    "Kyocera",
    "IRISO",
    "订单",
    "交期",
    "交货",
    "品质",
    "客诉",
    "投诉",
    "不良",
    "8D",
    "退货",
    "索赔",
    "对账",
    "付款",
    "发票",
    "合同",
    "送样",
    "开模",
    "开发",
    "零跑",
    "领跑",
    "维胜",
    "恒润",
    "涵润",
    "安悦",
    "先锋",
]


def safe_person(person):
    if not isinstance(person, dict):
        return None, None
    email = (person.get("email") or "").lower()
    name = (person.get("name") or "").strip()
    domain = email.split("@")[-1] if "@" in email else None
    return domain, name if name and len(name) <= 30 else None


def main() -> int:
    if len(sys.argv) != 2:
        print("usage: analyze_email_jsonl.py /path/to/llm_input.jsonl", file=sys.stderr)
        return 2
    path = Path(sys.argv[1])
    threads = msgs = attach = trunc = 0
    kw = collections.Counter()
    parts = collections.Counter()
    domains = collections.Counter()
    names = collections.Counter()
    years = collections.Counter()
    months = collections.Counter()
    attach_ext = collections.Counter()
    interesting = []

    with path.open(encoding="utf-8", errors="replace") as fh:
        for line in fh:
            obj = json.loads(line)
            threads += 1
            trunc += bool(obj.get("body_truncated"))
            messages = obj.get("messages", [])
            subject = obj.get("suggested_title_from_subject", "")
            text = " ".join(
                [subject]
                + [
                    (m.get("subject") or "") + " " + (m.get("body_text") or "")
                    for m in messages
                ]
            )
            lowered = text.lower()
            for key in KEYWORDS:
                if key.lower() in lowered:
                    kw[key] += 1
            found_parts = [x.upper() for x in PART_RE.findall(text)]
            for part in found_parts:
                if len(part) >= 6:
                    parts[part] += 1
            for msg in messages:
                msgs += 1
                month = (msg.get("date") or "")[:7]
                if len(month) == 7:
                    months[month] += 1
                    years[month[:4]] += 1
                people = [msg.get("from", {})] + msg.get("to", []) + msg.get("cc", [])
                for person in people:
                    domain, name = safe_person(person)
                    if domain:
                        domains[domain] += 1
                    if name:
                        names[name] += 1
                for item in msg.get("attachments", []) or []:
                    attach += 1
                    if isinstance(item, dict):
                        filename = item.get("filename") or item.get("name") or ""
                    else:
                        filename = str(item)
                    ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else "[no_ext]"
                    attach_ext[ext] += 1
            score = (
                sum(lowered.count(key.lower()) for key in KEYWORDS)
                + len(found_parts) * 2
                + len(messages) // 3
                + len(obj.get("attachments", [])) * 2
            )
            if score > 8:
                interesting.append(
                    {
                        "score": score,
                        "date_end": obj.get("date_end", ""),
                        "thread_id": obj.get("thread_id", ""),
                        "subject": subject[:120],
                        "message_count": len(messages),
                        "attachment_count": len(obj.get("attachments", [])),
                        "parts": found_parts[:12],
                        "keywords": [
                            key for key in KEYWORDS if key.lower() in lowered
                        ][:12],
                    }
                )

    result = {
        "source": str(path),
        "threads": threads,
        "messages": msgs,
        "attachments": attach,
        "truncated_threads": trunc,
        "years": years.most_common(),
        "months_top": months.most_common(18),
        "keywords": kw.most_common(50),
        "parts_top": parts.most_common(80),
        "domains_top": domains.most_common(60),
        "names_top": names.most_common(60),
        "attachment_extensions": attach_ext.most_common(40),
        "interesting_threads": sorted(
            interesting, key=lambda x: (x["score"], x["date_end"]), reverse=True
        )[:120],
    }
    json.dump(result, sys.stdout, ensure_ascii=False, indent=2)
    print()
    return 0


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