obsidian-wiki/scripts/ingest_email_attachments_to_bjd_wiki.py

#!/usr/bin/env python3
from __future__ import annotations
import collections, hashlib, html, json, re, zipfile
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

VAULT=Path('/home/nac/repo/obsidian-wiki-linux/kb/bjd-vault')
SOURCE=Path('/home/nac/repo/emails_export/parsed/llm_input.jsonl')
ATTACH_BASE=Path('/home/nac/repo/emails_export')
RUN_DATE='20260611'
NOW=datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace('+00:00','Z')
TODAY=NOW[:10]
META=VAULT/'_meta'
INV=META/f'bjd-email-attachment-inventory-{RUN_DATE}.jsonl'
SNIP=META/f'bjd-email-attachment-extracted-snippets-{RUN_DATE}.jsonl'
SUMMARY=META/f'bjd-email-attachment-ingest-{RUN_DATE}.json'
KW_CLASSES={
 'drawing':['图纸','drawing','2d','3d','dwg','step','stp','结构图','线规','工程图'],
 'spec':['规格书','spec','datasheet','承认书','acknowledge','approval','承認書','产品资料'],
 'quote':['报价','quote','rfq','询价','quotation','price','价格','单价'],
 'order':['订单','po','purchase order','cgdd','合同','采购单','订单合同','order'],
 'forecast':['forecast','fcst','预测','mrp','排期','交期','delivery schedule','需求计划','提货计划'],
 'quality':['8d','不良','客诉','quality','complaint','退货','改善','异常','索赔','报废','返工'],
 'compliance':['rohs','reach','halogen','无卤','ppap','iatf','audit','审核','camds','msds','sgs','vda','环保','gp'],
}
BUSINESS_TERMS=['BJD','北极大','Beijida','RFQ','报价','询价','订单','交期','样品','图纸','规格书','承认书','PPAP','RoHS','REACH','8D','客诉','不良','forecast','MRP','SOP','VAVE','替代','Hirose','Molex','TE','JAE','Kyocera','IRISO','零跑','理想','巨一','Marelli','Valeo','Visteon','BYD','比亚迪']
PARSE_EXT={'.xlsx','.xlsm','.csv','.txt','.htm','.html','.xml','.docx','.pptx','.eml'}
BINARY_TEXT_EXT={'.pdf','.xls','.doc'}

def clean(s:str, limit:int=1200)->str:
    s=html.unescape(str(s or '')).replace('\x00',' ')
    s=re.sub(r'\s+',' ',s).strip()
    return s[:limit]

def cls_for(fn:str, subject:str='')->str:
    low=(fn+' '+subject).lower()
    for k,vs in KW_CLASSES.items():
        if any(v.lower() in low for v in vs): return k
    return 'other'

def domain_of(person:Any)->str|None:
    if not isinstance(person,dict): return None
    e=(person.get('email') or '').lower()
    return e.split('@')[-1] if '@' in e else None

def safe_decode(data:bytes, limit:int=2_000_000)->str:
    data=data[:limit]
    for enc in ('utf-8','gb18030','big5','latin1'):
        try: return data.decode(enc, errors='ignore')
        except Exception: pass
    return data.decode('utf-8', errors='ignore')

def extract_xml_text_from_zip(path:Path, suffixes:tuple[str,...], limit:int=12000)->str:
    vals=[]
    try:
        with zipfile.ZipFile(path) as z:
            names=[n for n in z.namelist() if n.endswith(suffixes)]
            for n in names[:80]:
                try:
                    root=ET.fromstring(z.read(n))
                    for elem in root.iter():
                        if elem.text and elem.tag.rsplit('}',1)[-1] in {'t','v'}:
                            vals.append(elem.text)
                            if sum(map(len,vals))>limit: return clean(' '.join(vals), limit)
                except Exception:
                    continue
    except Exception:
        return ''
    return clean(' '.join(vals), limit)

def extract_xlsx(path:Path, limit:int=12000)->str:
    vals=[]; shared=[]
    try:
        with zipfile.ZipFile(path) as z:
            if 'xl/sharedStrings.xml' in z.namelist():
                try:
                    root=ET.fromstring(z.read('xl/sharedStrings.xml'))
                    for si in root.iter():
                        if si.tag.rsplit('}',1)[-1]=='si':
                            shared.append(''.join(t.text or '' for t in si.iter() if t.tag.rsplit('}',1)[-1]=='t'))
                except Exception: pass
            sheets=[n for n in z.namelist() if n.startswith('xl/worksheets/sheet') and n.endswith('.xml')]
            for n in sheets[:20]:
                try:
                    root=ET.fromstring(z.read(n))
                    for c in root.iter():
                        if c.tag.rsplit('}',1)[-1] != 'c': continue
                        t=c.attrib.get('t')
                        v=None
                        for ch in c:
                            if ch.tag.rsplit('}',1)[-1]=='v': v=ch.text; break
                        if v is None: continue
                        if t=='s':
                            try: vals.append(shared[int(v)])
                            except Exception: vals.append(v)
                        else: vals.append(v)
                        if sum(map(len,vals))>limit: return clean(' '.join(vals), limit)
                except Exception: continue
    except Exception:
        return ''
    return clean(' '.join(vals), limit)

def extract_text_file(path:Path, limit:int=12000)->str:
    try: text=safe_decode(path.read_bytes(), 1_500_000)
    except Exception: return ''
    return clean(text, limit)

def extract_html(path:Path, limit:int=12000)->str:
    try: text=safe_decode(path.read_bytes(), 1_500_000)
    except Exception: return ''
    text=re.sub(r'<(script|style)[^>]*>.*?</\1>',' ',text,flags=re.I|re.S)
    text=re.sub(r'<[^>]+>',' ',text)
    return clean(text, limit)

def extract_binary_strings(path:Path, limit:int=8000)->str:
    try: data=path.read_bytes()[:3_000_000]
    except Exception: return ''
    txt=safe_decode(data, 3_000_000)
    runs=re.findall(r'[\u4e00-\u9fffA-Za-z0-9_@./:：;；,，()（）\-\s]{8,}', txt)
    return clean(' '.join(runs[:300]), limit)

def extract(path:Path, ext:str)->tuple[str,str]:
    try:
        if ext in {'.xlsx','.xlsm'}: return extract_xlsx(path), 'xlsx-xml'
        if ext == '.docx': return extract_xml_text_from_zip(path, ('document.xml',), 12000), 'docx-xml'
        if ext == '.pptx': return extract_xml_text_from_zip(path, ('.xml',), 12000), 'pptx-xml'
        if ext in {'.htm','.html'}: return extract_html(path), 'html'
        if ext in {'.txt','.csv','.xml','.eml'}: return extract_text_file(path), 'text'
        if ext in BINARY_TEXT_EXT: return extract_binary_strings(path), 'binary-strings'
    except Exception as e:
        return '', 'error:'+type(e).__name__
    return '', 'metadata-only'

def should_extract(ext:str, klass:str, size:int, filename:str, subject:str)->bool:
    low=(filename+' '+subject).lower()
    important = klass != 'other' or any(t.lower() in low for t in BUSINESS_TERMS)
    if ext in PARSE_EXT and size <= 4_000_000 and important: return True
    if ext in BINARY_TEXT_EXT and size <= 3_000_000 and klass in {'quote','order','forecast','quality','compliance','spec','drawing'}: return True
    return False

def frontmatter(title,tags,summary):
    return """---
title: {title}
category: references
project: bjd
tags: {tags}
visibility: internal
data_classification: commercial-confidential
allowed_bots: []
excluded_bots: [bjd-general-bot]
sources:
  - _meta/bjd-email-attachment-ingest-{run}.json
  - _meta/bjd-email-attachment-inventory-{run}.jsonl
  - _meta/bjd-email-attachment-extracted-snippets-{run}.jsonl
summary: >-
  {summary}
provenance:
  extracted: 0.76
  inferred: 0.16
  ambiguous: 0.08
base_confidence: 0.70
lifecycle: draft
lifecycle_changed: {today}
created: {now}
updated: {now}
---

""".format(title=title,tags=tags,summary=summary,run=RUN_DATE,today=TODAY,now=NOW)

def esc(x): return str(x or '').replace('|',' / ').replace('\n',' ')
def md(items, empty='未提取'): return '、'.join(items) if items else empty

def main():
    META.mkdir(parents=True, exist_ok=True)
    by_sha={}; refs=0
    with SOURCE.open(encoding='utf-8', errors='replace') as f:
        for line in f:
            o=json.loads(line); refs+=len(o.get('attachments') or [])
            tid=o.get('thread_id'); subject=clean(o.get('suggested_title_from_subject',''),220); end=o.get('date_end')
            domains=collections.Counter()
            for m in o.get('messages') or []:
                if not isinstance(m,dict): continue
                for p in [m.get('from',{})]+(m.get('to') or [])+(m.get('cc') or []):
                    d=domain_of(p)
                    if d and d not in {'beijida.com','bjdffc.com'}: domains[d]+=1
            for a in o.get('attachments') or []:
                sha=a.get('content_sha1') or a.get('attachment_id')
                fn=a.get('filename') or ''
                ext=Path(fn).suffix.lower() or '[none]'
                path=ATTACH_BASE/(a.get('local_path') or '')
                rec=by_sha.get(sha)
                if not rec:
                    klass=cls_for(fn, subject)
                    rec={'content_sha1':sha,'attachment_id':a.get('attachment_id'),'filename':fn,'ext':ext,'content_type':a.get('content_type'),'size_bytes':a.get('size_bytes') or (path.stat().st_size if path.exists() else None),'local_path':a.get('local_path'),'absolute_path':str(path),'exists':path.exists(),'class':klass,'thread_count':0,'ref_count':0,'threads':[],'subjects':[],'domains':collections.Counter()}
                    by_sha[sha]=rec
                rec['ref_count']+=1
                if tid not in rec['threads']:
                    rec['thread_count']+=1
                    if len(rec['threads'])<8: rec['threads'].append(tid)
                if subject and subject not in rec['subjects'] and len(rec['subjects'])<5: rec['subjects'].append(subject)
                rec['domains'].update(domains)
                if end and not rec.get('date_end'): rec['date_end']=end
    stats={'refs':refs,'unique':len(by_sha),'exists':0,'missing':0,'bytes':0}
    ext_counts=collections.Counter(); class_counts=collections.Counter(); parser_counts=collections.Counter(); extracted_counts=collections.Counter()
    rows=[]; snippets=[]
    with INV.open('w', encoding='utf-8') as inv, SNIP.open('w', encoding='utf-8') as snip:
        for rec in by_sha.values():
            p=Path(rec['absolute_path']); size=rec.get('size_bytes') or 0
            if rec['exists']: stats['exists']+=1; stats['bytes']+=int(size or 0)
            else: stats['missing']+=1
            ext_counts[rec['ext']]+=1; class_counts[rec['class']]+=1
            rec['top_domains']=[d for d,_ in rec['domains'].most_common(8)]
            rec.pop('domains', None)
            inv.write(json.dumps(rec, ensure_ascii=False)+'\n')
            if should_extract(rec['ext'], rec['class'], int(size or 0), rec['filename'], ' '.join(rec['subjects'])):
                text, parser=extract(p, rec['ext']) if rec['exists'] else ('','missing')
                parser_counts[parser]+=1
                if text and len(text)>=20:
                    key_terms=[term for term in BUSINESS_TERMS if term.lower() in text.lower()]
                    srec={k:rec.get(k) for k in ['content_sha1','filename','ext','size_bytes','absolute_path','class','threads','subjects','top_domains','date_end']}
                    srec.update({'parser':parser,'text_excerpt':text[:2500],'key_terms':key_terms[:20]})
                    snip.write(json.dumps(srec, ensure_ascii=False)+'\n')
                    snippets.append(srec); extracted_counts[rec['class']]+=1
            rows.append(rec)
    def score(r):
        return {'order':50,'quote':48,'forecast':44,'quality':45,'compliance':40,'drawing':38,'spec':38,'other':0}.get(r['class'],0)+{'.xlsx':12,'.xlsm':12,'.xls':9,'.pdf':8,'.docx':7,'.doc':6,'.csv':6,'.htm':5,'.html':5,'.txt':4,'.stp':3,'.step':3,'.dwg':3}.get(r['ext'],0)+min(r.get('thread_count',0),20)+min(r.get('ref_count',0)//2,20)
    rows.sort(key=score, reverse=True)
    snippets.sort(key=lambda r: ({'order':50,'quote':48,'forecast':44,'quality':45,'compliance':40,'drawing':38,'spec':38,'other':0}.get(r['class'],0), r.get('size_bytes') or 0), reverse=True)
    summary={'generated_at':NOW,'source':str(SOURCE),'attachment_base':str(ATTACH_BASE/'attachments'),'stats':stats,'ext_counts':ext_counts.most_common(50),'class_counts':class_counts.most_common(),'parser_counts':parser_counts.most_common(),'extracted_counts':extracted_counts.most_common(),'inventory':str(INV.relative_to(VAULT)),'snippets':str(SNIP.relative_to(VAULT))}
    SUMMARY.write_text(json.dumps(summary, ensure_ascii=False, indent=2, default=list)+'\n', encoding='utf-8')
    pages={}
    overview=frontmatter(f'BJD 邮件附件 ingest {RUN_DATE}','[bjd, emails, attachments, internal, evidence]','邮件附件全量 inventory 与高价值附件内容摘录索引，覆盖图纸、规格书、订单、报价、预测、质量和合规资料。')
    overview += f"""# BJD 邮件附件 ingest {RUN_DATE}

本次针对 `/home/nac/repo/emails_export/parsed/llm_input.jsonl` 中引用的附件做单独 ingest。附件真实根目录为 `/home/nac/repo/emails_export/attachments`。^[extracted]

## 规模

- 附件引用数：{stats['refs']:,}
- 去重后附件数：{stats['unique']:,}
- 本地存在附件：{stats['exists']:,}
- 本地附件总大小：{stats['bytes']/1024**3:.2f} GB
- 全量 inventory：`{summary['inventory']}`
- 内容摘录索引：`{summary['snippets']}`

## 内容类型

| 类别 | 去重附件数 |
|---|---:|
"""
    for k,v in class_counts.most_common(): overview+=f'| {esc(k)} | {v:,} |\n'
    overview+='\n## 文件类型 Top 25\n\n| 扩展名 | 去重附件数 |\n|---|---:|\n'
    for k,v in ext_counts.most_common(25): overview+=f'| `{esc(k)}` | {v:,} |\n'
    overview+='\n## 解析覆盖\n\n| 解析器 | 附件数 |\n|---|---:|\n'
    for k,v in parser_counts.most_common(): overview+=f'| {esc(k)} | {v:,} |\n'
    overview+='\n## 内部页面\n\n- [[projects/bjd/references/bjd-email-attachment-high-value-index-20260611]]\n- [[projects/bjd/references/bjd-email-attachment-extracted-content-index-20260611]]\n- [[projects/bjd/references/bjd-email-attachment-parser-coverage-20260611]]\n\n这些页面均为母库 internal-only，不进入 `bjd-general-bot`。^[extracted]\n'
    pages[f'projects/bjd/references/bjd-email-attachment-ingest-{RUN_DATE}.md']=overview
    high=frontmatter(f'BJD 邮件高价值附件索引 {RUN_DATE}','[bjd, emails, attachments, orders, rfq, quality, internal]','按业务类别、文件名、客户域名、thread 和本地路径索引高价值邮件附件。')
    high += f'# BJD 邮件高价值附件索引 {RUN_DATE}\n\n| 类别 | 扩展名 | 文件名 | 大小 | 引用/线程 | 客户域名 | 代表主题 | 本地路径 |\n|---|---|---|---:|---:|---|---|---|\n'
    for r in rows[:350]: high += f"| {esc(r['class'])} | `{esc(r['ext'])}` | {esc(r['filename'])} | {int(r.get('size_bytes') or 0):,} | {r.get('ref_count',0)}/{r.get('thread_count',0)} | {esc(md(r.get('top_domains',[])[:5]))} | {esc('；'.join(r.get('subjects',[])[:2]))} | `{esc(r.get('absolute_path'))}` |\n"
    pages[f'projects/bjd/references/bjd-email-attachment-high-value-index-{RUN_DATE}.md']=high
    cont=frontmatter(f'BJD 邮件附件内容摘录索引 {RUN_DATE}','[bjd, emails, attachments, extracted-content, internal]','可解析附件的文本摘录索引，用于检索报价、订单、预测、质量和合规附件内容。')
    cont += f'# BJD 邮件附件内容摘录索引 {RUN_DATE}\n\n| 类别 | 文件名 | 解析器 | 客户域名 | 代表主题 | 关键词 | 内容摘录 |\n|---|---|---|---|---|---|---|\n'
    for r in snippets[:260]: cont += f"| {esc(r['class'])} | {esc(r['filename'])} | {esc(r['parser'])} | {esc(md(r.get('top_domains',[])[:5]))} | {esc('；'.join(r.get('subjects',[])[:2]))} | {esc(md(r.get('key_terms',[])[:8]))} | {esc(r.get('text_excerpt','')[:550])} |\n"
    pages[f'projects/bjd/references/bjd-email-attachment-extracted-content-index-{RUN_DATE}.md']=cont
    cov=frontmatter(f'BJD 邮件附件解析覆盖 {RUN_DATE}','[bjd, emails, attachments, parser-coverage, internal]','说明附件 ingest 的覆盖范围、未解析类型和后续 OCR/PDF/CAD 处理建议。')
    cov += f"""# BJD 邮件附件解析覆盖 {RUN_DATE}

## 已完成

- 全量附件 inventory：所有 {stats['unique']:,} 个去重附件均已登记路径、sha1、文件名、大小、类型、关联 thread、主题和客户域名。^[extracted]
- 可用标准库解析的高价值附件已写入内容摘录索引，包括 `.xlsx/.xlsm/.csv/.txt/.html/.docx/.pptx/.eml`，以及部分 `.pdf/.xls/.doc` 的二进制字符串摘录。^[extracted]

## 需要后续增强

- PDF：当前没有 `pdftotext` / PyPDF / OCR，部分 PDF 只能建索引或抽少量未压缩字符串；规格书、图纸 PDF 要高质量 ingest 需要安装 PDF 解析/OCR。^[inferred]
- 图片：`.png/.jpg/.tif` 当前只建索引，没有 OCR；质量照片、截图、标签图片要另跑 OCR/视觉模型。^[inferred]
- CAD/3D：`.stp/.step/.dwg` 当前只建索引；几何内容和图纸参数需要 CAD 专用解析或人工/视觉流程。^[inferred]
- 压缩包：`.zip/.rar/.7z` 当前只建外层索引；需要后续安全解包再逐项 ingest。^[inferred]

## 解析器统计

| 解析器 | 附件数 |
|---|---:|
"""
    for k,v in parser_counts.most_common(): cov+=f'| {esc(k)} | {v:,} |\n'
    cov+='\n## 按类别的内容摘录数\n\n| 类别 | 已摘录附件数 |\n|---|---:|\n'
    for k,v in extracted_counts.most_common(): cov+=f'| {esc(k)} | {v:,} |\n'
    pages[f'projects/bjd/references/bjd-email-attachment-parser-coverage-{RUN_DATE}.md']=cov
    for rel,text in pages.items():
        path=VAULT/rel; path.parent.mkdir(parents=True, exist_ok=True); path.write_text(text, encoding='utf-8')
    idx=VAULT/'index.md'; text=idx.read_text(encoding='utf-8'); text=re.sub(r'最后更新：[^*]+', f'最后更新：{NOW}', text)
    marker='## 综合分析\n'
    if marker not in text: text+='\n## 综合分析\n'
    ins=text.index(marker)+len(marker)
    entries=[f'- [[projects/bjd/references/bjd-email-attachment-ingest-{RUN_DATE}]] - 邮件附件全量 inventory 和高价值附件内容摘录总览。 ( #bjd #emails #attachments )',f'- [[projects/bjd/references/bjd-email-attachment-high-value-index-{RUN_DATE}]] - 高价值附件索引，关联客户、主题、thread 和本地路径。 ( #bjd #attachments #internal )',f'- [[projects/bjd/references/bjd-email-attachment-extracted-content-index-{RUN_DATE}]] - 可解析附件内容摘录索引。 ( #bjd #attachments #extracted-content )',f'- [[projects/bjd/references/bjd-email-attachment-parser-coverage-{RUN_DATE}]] - 附件解析覆盖和后续 OCR/PDF/CAD 处理建议。 ( #bjd #attachments #parser )']
    block=''.join(e+'\n' for e in entries if e not in text); idx.write_text(text[:ins]+block+text[ins:], encoding='utf-8')
    hot=VAULT/'hot.md'; h=hot.read_text(encoding='utf-8'); h=re.sub(r'updated: [^\n]+', f'updated: {NOW}', h, count=1)
    item=f"- [{NOW}] DATA_INGEST - 新增邮件附件 ingest：{stats['refs']:,} 个引用、{stats['unique']:,} 个去重附件、{stats['bytes']/1024**3:.2f}GB；已生成全量 inventory 和高价值附件内容摘录，internal-only。\n"
    if item not in h: h=h.replace('## 近期活动\n\n','## 近期活动\n\n'+item,1)
    hot.write_text(h, encoding='utf-8')
    with (VAULT/'log.md').open('a', encoding='utf-8') as f: f.write(f'\n- [{NOW}] DATA_INGEST source="{SOURCE}" format=email_attachments refs={stats["refs"]} unique={stats["unique"]} pages_created={len(pages)} inventory="_meta/{INV.name}" snippets="_meta/{SNIP.name}" bot_visible=false\n')
    man=VAULT/'.manifest.json'; data=json.loads(man.read_text(encoding='utf-8'))
    key=f'{SOURCE}::email-attachments::{RUN_DATE}'
    data.setdefault('sources',{})[key]={'ingested_at':NOW,'original_path':str(SOURCE),'source_type':'email_attachment_corpus','doc_type':'followup_email_attachments','project':'bjd','content_hash':'sha256:'+hashlib.sha256(SOURCE.read_bytes()).hexdigest(),'attachment_refs':stats['refs'],'unique_attachments':stats['unique'],'existing_attachments':stats['exists'],'attachment_total_bytes':stats['bytes'],'attachment_base':str(ATTACH_BASE/'attachments'),'inventory_path':str(INV.relative_to(VAULT)),'snippet_path':str(SNIP.relative_to(VAULT)),'pages_created':sorted(pages),'pages_updated':['index.md','hot.md','log.md','.manifest.json'],'bot_visible':False}
    data['updated']=NOW; data.setdefault('stats',{})['total_sources_ingested']=len(data.get('sources',{})); man.write_text(json.dumps(data, ensure_ascii=False, indent=2)+'\n', encoding='utf-8')
    print(json.dumps({'stats':stats,'pages':sorted(pages),'inventory':str(INV),'snippets':str(SNIP),'parser_counts':parser_counts.most_common(10),'generated_at':NOW}, ensure_ascii=False, indent=2))
if __name__=='__main__': main()