nac
777607dc29
Track the source-of-truth code, skills, bootstrap rules, docs, and packaging files before adding new features. Exclude local vault config, generated skill mirrors, temporary exports, virtual environments, and business data files from version control. Validation: inspected staged file list and git ignored status before commit.
287 lines
13 KiB
Python
287 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import tempfile
|
||
import zipfile
|
||
from collections import Counter, defaultdict
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from xml.etree import ElementTree as ET
|
||
|
||
ROOT = Path('/home/nac/smb/品质部/品质管理资料夹2009.11.3/PPAP资料')
|
||
VAULT = Path('/home/nac/repo/obsidian-wiki-linux/kb/bjd-vault')
|
||
META = VAULT / '_meta'
|
||
OUT = VAULT / 'projects/bjd/references'
|
||
RUN_DATE = '20260611'
|
||
MAX_TEXT = 20000
|
||
MAX_FILES = int(os.environ.get('PPAP_DEEP_MAX_FILES', '0')) or None
|
||
|
||
QUALITY_TERMS = [
|
||
'PPAP','APQP','PSW','零件提交保证书','全尺寸','寸法','MSA','GR&R','GRR','CPK','PPK','CMK',
|
||
'可靠性','DV','PV','接触电阻','绝缘','耐压','保持力','可焊性','耐焊热','重复插拔','耐湿','盐雾',
|
||
'PFMEA','DFMEA','控制计划','过程流程','特殊特性','规格书','SGS','RoHS','Halogen','判定标准','测试结果','合格','OK',
|
||
]
|
||
SERIES_RE = re.compile(r'(?<!\d)(?:0\.3|0\.5|0\.8|1\.0|1\.25|1\.8|2\.0|2\.2|3\.5|5\.0)-\d+(?:B)?|28\+4P|10PIN|18B966|VPKSJF|VPHNMF|VPDNNMF', re.I)
|
||
|
||
|
||
def now_iso():
|
||
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace('+00:00','Z')
|
||
|
||
|
||
def clean(s: str) -> str:
|
||
s = (s or '').replace('\x00', ' ')
|
||
s = re.sub(r'[\t\r\f\v]+', ' ', s)
|
||
s = re.sub(r'\n{3,}', '\n\n', s)
|
||
s = re.sub(r' {2,}', ' ', s)
|
||
return s.strip()
|
||
|
||
|
||
def run(cmd: list[str], timeout=30) -> str:
|
||
try:
|
||
p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, check=False)
|
||
data = p.stdout or b''
|
||
if not data and p.stderr:
|
||
data = p.stderr[:500]
|
||
return data.decode('utf-8', errors='ignore')
|
||
except Exception:
|
||
return ''
|
||
|
||
|
||
def extract_pdf(path: Path) -> str:
|
||
return run(['pdftotext', '-layout', str(path), '-'], timeout=45)
|
||
|
||
|
||
def extract_doc(path: Path) -> str:
|
||
txt = run(['antiword', str(path)], timeout=30)
|
||
if len(txt.strip()) < 50:
|
||
txt = run(['catdoc', str(path)], timeout=30)
|
||
return txt
|
||
|
||
|
||
def extract_unrtf(path: Path) -> str:
|
||
return run(['unrtf', '--text', str(path)], timeout=30)
|
||
|
||
|
||
def extract_xlsx(path: Path) -> str:
|
||
# Dependency-free OOXML shared strings/cell extraction; handles xlsx without openpyxl.
|
||
try:
|
||
with zipfile.ZipFile(path) as z:
|
||
names=set(z.namelist())
|
||
shared=[]
|
||
if 'xl/sharedStrings.xml' in names:
|
||
root=ET.fromstring(z.read('xl/sharedStrings.xml'))
|
||
ns={'a':'http://schemas.openxmlformats.org/spreadsheetml/2006/main'}
|
||
for si in root.findall('.//a:si', ns):
|
||
parts=[t.text or '' for t in si.findall('.//a:t', ns)]
|
||
if parts: shared.append(''.join(parts))
|
||
lines=[]
|
||
for name in sorted(n for n in names if n.startswith('xl/worksheets/sheet') and n.endswith('.xml'))[:8]:
|
||
root=ET.fromstring(z.read(name))
|
||
ns={'a':'http://schemas.openxmlformats.org/spreadsheetml/2006/main'}
|
||
vals=[]
|
||
for c in root.findall('.//a:c', ns)[:1000]:
|
||
v=c.find('a:v', ns)
|
||
if v is None or v.text is None: continue
|
||
val=v.text
|
||
if c.get('t') == 's':
|
||
try: val=shared[int(val)]
|
||
except Exception: pass
|
||
vals.append(val)
|
||
if vals:
|
||
lines.append(name + '\n' + ' '.join(vals))
|
||
return '\n'.join(lines)
|
||
except Exception:
|
||
return ''
|
||
|
||
|
||
def extract_xls_with_ssconvert(path: Path) -> str:
|
||
# Convert each workbook to csv in a temp dir. ssconvert may output one csv or multiple files depending workbook.
|
||
with tempfile.TemporaryDirectory(prefix='ppap-xls-') as td:
|
||
out = Path(td) / 'out.csv'
|
||
txt = run(['ssconvert', '--export-type=Gnumeric_stf:stf_csv', str(path), str(out)], timeout=60)
|
||
chunks=[]
|
||
for f in Path(td).glob('*'):
|
||
if f.is_file():
|
||
try:
|
||
chunks.append(f.read_text(encoding='utf-8', errors='ignore'))
|
||
except Exception:
|
||
pass
|
||
if chunks:
|
||
return '\n'.join(chunks)
|
||
return txt
|
||
|
||
|
||
def extract_libreoffice_text(path: Path) -> str:
|
||
with tempfile.TemporaryDirectory(prefix='ppap-lo-') as td:
|
||
run(['libreoffice', '--headless', '--convert-to', 'txt:Text', '--outdir', td, str(path)], timeout=90)
|
||
chunks=[]
|
||
for f in Path(td).glob('*.txt'):
|
||
try: chunks.append(f.read_text(encoding='utf-8', errors='ignore'))
|
||
except Exception: pass
|
||
return '\n'.join(chunks)
|
||
|
||
|
||
def extract_text(path: Path) -> tuple[str, str]:
|
||
ext=path.suffix.lower()
|
||
text=''; method=''
|
||
if ext == '.pdf':
|
||
text=extract_pdf(path); method='pdftotext'
|
||
elif ext == '.doc':
|
||
text=extract_doc(path); method='antiword/catdoc'
|
||
if len(clean(text)) < 50:
|
||
lo=extract_libreoffice_text(path)
|
||
if len(clean(lo)) > len(clean(text)):
|
||
text=lo; method='libreoffice-txt'
|
||
elif ext == '.rtf':
|
||
text=extract_unrtf(path); method='unrtf'
|
||
elif ext == '.xlsx':
|
||
text=extract_xlsx(path); method='xlsx-xml'
|
||
elif ext == '.xls':
|
||
text=extract_xls_with_ssconvert(path); method='ssconvert-csv'
|
||
if len(clean(text)) < 50:
|
||
lo=extract_libreoffice_text(path)
|
||
if len(clean(lo)) > len(clean(text)):
|
||
text=lo; method='libreoffice-txt'
|
||
elif ext in {'.txt','.log','.csv'}:
|
||
try: text=path.read_text(encoding='utf-8', errors='ignore'); method='plain'
|
||
except Exception: text=''
|
||
elif ext == '.ppt':
|
||
text=extract_libreoffice_text(path); method='libreoffice-txt'
|
||
return clean(text)[:MAX_TEXT], method
|
||
|
||
|
||
def snippets(text: str, max_snips=8) -> list[str]:
|
||
out=[]
|
||
for term in QUALITY_TERMS:
|
||
for m in re.finditer(re.escape(term), text, re.I):
|
||
start=max(0, m.start()-160); end=min(len(text), m.end()+320)
|
||
sn=clean(text[start:end])[:600]
|
||
if sn and sn not in out:
|
||
out.append(sn)
|
||
if len(out) >= max_snips:
|
||
return out
|
||
return out
|
||
|
||
|
||
def classify(text: str, path: Path) -> list[str]:
|
||
hay=(path.as_posix()+'\n'+text)
|
||
cats=[]
|
||
patterns={
|
||
'ppap_submission': r'PPAP提交清单|PPAP文件清单',
|
||
'psw': r'零件提交保证书|PSW',
|
||
'dimensional': r'全尺寸|寸法|尺寸测量|样品寸法',
|
||
'msa': r'MSA|GR&R|GRR|测量系统|量测系统',
|
||
'capability': r'CPK|PPK|CMK|过程能力|设备能力',
|
||
'reliability': r'可靠性|DV|PV|接触电阻|绝缘|耐压|保持力|可焊性|耐焊热|重复插拔|耐湿|盐雾',
|
||
'pfmea_dfmea': r'PFMEA|DFMEA|失效模式',
|
||
'control_plan': r'控制计划|Control Plan',
|
||
'process_flow': r'过程流程|工艺流程|流程图',
|
||
'material': r'SGS|RoHS|Halogen|C5210|C5191|C2680|LCP|PA9T',
|
||
'special_characteristics': r'特殊特性|关键特性|重要特性|Special Characteristics',
|
||
}
|
||
for k,pat in patterns.items():
|
||
if re.search(pat, hay, re.I): cats.append(k)
|
||
return cats or ['other']
|
||
|
||
|
||
def main():
|
||
records=[]
|
||
files=[]
|
||
for p in ROOT.rglob('*'):
|
||
if p.is_file() and p.suffix.lower() in {'.pdf','.doc','.rtf','.xls','.xlsx','.txt','.log','.csv','.ppt'}:
|
||
files.append(p)
|
||
if MAX_FILES:
|
||
files=files[:MAX_FILES]
|
||
for i,p in enumerate(files,1):
|
||
text, method = extract_text(p)
|
||
rel=p.relative_to(ROOT).as_posix()
|
||
cats=classify(text, p)
|
||
rec={
|
||
'relative_path': rel,
|
||
'path': str(p),
|
||
'top_package': rel.split('/')[0],
|
||
'filename': p.name,
|
||
'extension': p.suffix.lower().lstrip('.') or 'no_ext',
|
||
'method': method,
|
||
'text_length': len(text),
|
||
'categories': cats,
|
||
'series': sorted(set(SERIES_RE.findall(rel+'\n'+text))),
|
||
'snippets': snippets(text),
|
||
'text_sample': text[:1500],
|
||
}
|
||
records.append(rec)
|
||
if i % 100 == 0:
|
||
print(f'processed {i}/{len(files)}', flush=True)
|
||
META.mkdir(parents=True, exist_ok=True)
|
||
OUT.mkdir(parents=True, exist_ok=True)
|
||
jsonl=META/f'bjd-ppap-apqp-deep-text-index-{RUN_DATE}.jsonl'
|
||
with jsonl.open('w',encoding='utf-8') as f:
|
||
for r in records:
|
||
f.write(json.dumps(r,ensure_ascii=False)+'\n')
|
||
# Summary CSV
|
||
csvp=META/f'bjd-ppap-apqp-deep-text-index-{RUN_DATE}.csv'
|
||
with csvp.open('w',encoding='utf-8-sig',newline='') as f:
|
||
w=csv.DictWriter(f,fieldnames=['relative_path','top_package','extension','method','text_length','categories','series','snippet_count'])
|
||
w.writeheader()
|
||
for r in records:
|
||
w.writerow({
|
||
'relative_path':r['relative_path'], 'top_package':r['top_package'], 'extension':r['extension'], 'method':r['method'],
|
||
'text_length':r['text_length'], 'categories':'; '.join(r['categories']), 'series':'; '.join(r['series']), 'snippet_count':len(r['snippets'])
|
||
})
|
||
write_page(records,jsonl,csvp)
|
||
print(json.dumps({'files':len(records),'jsonl':str(jsonl),'csv':str(csvp)},ensure_ascii=False))
|
||
|
||
|
||
def esc(s): return str(s).replace('|','\\|').replace('\n',' ')
|
||
|
||
|
||
def write_page(records, jsonl, csvp):
|
||
ts=now_iso()
|
||
by_ext=Counter(r['extension'] for r in records)
|
||
by_method=Counter(r['method'] or 'none' for r in records)
|
||
by_cat=Counter(c for r in records for c in r['categories'])
|
||
by_pkg=defaultdict(list)
|
||
for r in records:
|
||
by_pkg[r['top_package']].append(r)
|
||
pkg_rows=[]
|
||
for pkg,rs in by_pkg.items():
|
||
pkg_rows.append((pkg,len(rs),sum(1 for r in rs if r['text_length']>50),Counter(c for r in rs for c in r['categories']).most_common(6)))
|
||
pkg_rows.sort(key=lambda x:(-x[1],x[0]))
|
||
page=OUT/f'bjd-ppap-apqp-deep-text-index-{RUN_DATE}.md'
|
||
lines=['---','title: BJD PPAP/APQP 深度文本索引','category: reference','project: bjd','tags: [bjd, quality, ppap, apqp, deep-extraction, visibility/internal]','visibility: internal','data_classification: technical-confidential',f'updated: {ts}','---','', '# BJD PPAP/APQP 深度文本索引','', f'本页基于已安装工具 `pdftotext`、`antiword/catdoc`、`ssconvert`、`libreoffice` 对 PPAP 目录做二次文本抽取。共处理 `{len(records)}` 个可解析文件。', '', '## 抽取覆盖','', '- 文件类型：' + '；'.join(f'{k}={v}' for k,v in by_ext.most_common()), '- 抽取方法：' + '；'.join(f'{k}={v}' for k,v in by_method.most_common()), '- 质量类别：' + '；'.join(f'{k}={v}' for k,v in by_cat.most_common()), '', '## 资料包覆盖', '', '| 资料包 | 文件数 | 抽到文本文件数 | 主要质量类别 |', '|---|---:|---:|---|']
|
||
for pkg,n,txtn,cats in pkg_rows[:80]:
|
||
lines.append(f'| {esc(pkg)} | {n} | {txtn} | {esc("；".join(f"{k}={v}" for k,v in cats))} |')
|
||
# selected snippets
|
||
interesting=[r for r in records if r['snippets'] and any(c in r['categories'] for c in ['reliability','capability','msa','pfmea_dfmea','control_plan','special_characteristics'])]
|
||
lines += ['', '## 代表性证据片段', '', '| 来源 | 类别 | 片段 |', '|---|---|---|']
|
||
for r in interesting[:80]:
|
||
lines.append(f'| {esc(r["relative_path"])} | {esc("; ".join(r["categories"]))} | {esc(r["snippets"][0][:500])} |')
|
||
lines += ['', '## 数据文件', '', f'- JSONL：`_meta/{jsonl.name}`', f'- CSV：`_meta/{csvp.name}`', '', '## 使用注意', '', '- 这是文本抽取索引，不替代原始 PPAP 文件。正式对外提交仍需复核原文件签核、版本、日期和客户适用性。', '- 扫描版 PDF 或图片型图纸如果没有文本层，仍需要 OCR 才能抽出正文。']
|
||
page.write_text('\n'.join(lines)+'\n',encoding='utf-8')
|
||
idx=VAULT/'index.md'
|
||
if idx.exists():
|
||
text=idx.read_text(encoding='utf-8')
|
||
rel=page.relative_to(VAULT).with_suffix('').as_posix()
|
||
if rel not in text:
|
||
text=text.replace('## 综合分析\n', f'## 综合分析\n- [[{rel}]] - PPAP/APQP 全目录深度文本索引，包含 PDF/DOC/XLS 二次抽取。 ( #bjd #quality #ppap #apqp )\n',1)
|
||
text=re.sub(r'最后更新：[^*]+', f'最后更新：{ts}', text)
|
||
idx.write_text(text,encoding='utf-8')
|
||
with (VAULT/'log.md').open('a',encoding='utf-8') as f:
|
||
f.write(f'\n- [{ts}] DATA_INGEST source="{ROOT}" project="bjd" mode="ppap-apqp-deep-text-index" files={len(records)} page="{page.relative_to(VAULT).as_posix()}" jsonl="{jsonl.relative_to(VAULT).as_posix()}" csv="{csvp.relative_to(VAULT).as_posix()}" bot_visible=false\n')
|
||
mp=VAULT/'.manifest.json'
|
||
data={}
|
||
if mp.exists():
|
||
try: data=json.loads(mp.read_text(encoding='utf-8'))
|
||
except Exception: data={}
|
||
data.setdefault('sources',{})[f'filesystem:ppap-apqp-deep-text-index-{RUN_DATE}']={'ingested_at':ts,'source_path':str(ROOT),'source_type':'filesystem_directory','doc_type':'ppap_apqp_deep_text_index','project':'bjd','files':len(records),'pages_created':[page.relative_to(VAULT).as_posix()],'data_paths':[jsonl.relative_to(VAULT).as_posix(),csvp.relative_to(VAULT).as_posix()],'bot_visible':False,'tools':'pdftotext, antiword/catdoc, ssconvert, libreoffice'}
|
||
mp.write_text(json.dumps(data,ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
|
||
|
||
if __name__ == '__main__':
|
||
main()
|