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.
282 lines
12 KiB
Python
282 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import math
|
||
import re
|
||
from collections import defaultdict
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pandas as pd
|
||
|
||
ROOT = Path('/home/nac/smb/品质部/品质管理资料夹2009.11.3/PPAP资料/0.5-18 APQP资料 2022年8月修正')
|
||
VAULT = Path('/home/nac/repo/obsidian-wiki-linux/kb/bjd-vault')
|
||
META = VAULT / '_meta'
|
||
OUT = VAULT / 'projects/bjd/references'
|
||
RUN_DATE = '20260611'
|
||
|
||
KEY_TERMS = ['规格','norm','USL','LSL','SL','UCL','LCL','CL','判定','judgment','OK','NG','测试目的','测试条件','参照标准','判定标准','测试设备','测试方法','测试结果','报告No','测试日','品名','本厂料号','试验数量']
|
||
SPECIAL_HINTS = ['总间距','间距','直线度','平面度','插入深度','GAP','卡点','板厚','电镀','PIN孔','内腔长度','变形量','厚度','保持力','可焊性','接触电阻','绝缘','耐压','耐焊热','重复插拔']
|
||
|
||
|
||
def now_iso():
|
||
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace('+00:00','Z')
|
||
|
||
|
||
def clean(v: Any) -> str:
|
||
if v is None:
|
||
return ''
|
||
if isinstance(v, float) and math.isnan(v):
|
||
return ''
|
||
s=str(v).replace('\n',' ').replace('\r',' ')
|
||
return re.sub(r'\s+', ' ', s).strip()
|
||
|
||
|
||
def row_values(row) -> list[str]:
|
||
return [clean(x) for x in row.tolist() if clean(x)]
|
||
|
||
|
||
def read_excel(path: Path):
|
||
try:
|
||
return pd.ExcelFile(path)
|
||
except Exception as e:
|
||
return None
|
||
|
||
|
||
def find_value_after(vals: list[str], labels: list[str]) -> str:
|
||
def norm(x: str) -> str:
|
||
return re.sub(r'\s+', '', x).lower()
|
||
for i,v in enumerate(vals):
|
||
nv=norm(v)
|
||
for lab in labels:
|
||
nl=norm(lab)
|
||
exact = nv == nl or nv.startswith(nl + ':') or nv.startswith(nl + '：')
|
||
contains = nl not in {'cl','sl'} and nl in nv
|
||
if exact or contains:
|
||
if i+1 < len(vals):
|
||
return vals[i+1]
|
||
m=re.search(re.escape(lab)+r'[:：]?\s*(.+)', v, re.I)
|
||
if m:
|
||
return m.group(1).strip()
|
||
return ''
|
||
|
||
|
||
def extract_header_fields(df) -> dict[str,str]:
|
||
vals=[]
|
||
for _,row in df.head(20).iterrows():
|
||
vals.extend(row_values(row))
|
||
return {
|
||
'part_name': find_value_after(vals, ['产品名称','Part Name','品名']),
|
||
'item': find_value_after(vals, ['管制项目','Control the project']),
|
||
'unit': find_value_after(vals, ['测量单位','Unit']),
|
||
'spec': find_value_after(vals, ['规 格','规格','norm']),
|
||
'usl': find_value_after(vals, ['上限USL','USL']),
|
||
'sl': find_value_after(vals, ['中心限SL']),
|
||
'lsl': find_value_after(vals, ['下限LSL','LSL']) if not re.search(r'\b(?:Max|Min)\b', find_value_after(vals, ['规 格','规格','norm']), re.I) else '',
|
||
'ucl': find_value_after(vals, ['上限UCL','UCL']),
|
||
'cl': find_value_after(vals, ['中心限CL']),
|
||
'lcl': find_value_after(vals, ['下限LCL','LCL']),
|
||
'report_no': find_value_after(vals, ['报告No']),
|
||
'test_date': find_value_after(vals, ['测试日']),
|
||
'bjd_pn': find_value_after(vals, ['本厂料号']),
|
||
'sample_qty': find_value_after(vals, ['试验数量']),
|
||
}
|
||
|
||
|
||
def extract_judgement(df) -> str:
|
||
for _,row in df.iterrows():
|
||
vals=row_values(row)
|
||
if not vals:
|
||
continue
|
||
line=' | '.join(vals)
|
||
if re.search(r'判定|judgment', line, re.I):
|
||
# prefer OK/合格 near end
|
||
for v in reversed(vals):
|
||
if re.search(r'合格|OK|NG|不合格', v, re.I):
|
||
return v
|
||
return line[:200]
|
||
return ''
|
||
|
||
|
||
def extract_key_lines(df, max_lines=12) -> list[str]:
|
||
lines=[]
|
||
for _,row in df.iterrows():
|
||
vals=row_values(row)
|
||
if not vals:
|
||
continue
|
||
line=' | '.join(vals)
|
||
if any(k.lower() in line.lower() for k in KEY_TERMS) or any(k in line for k in SPECIAL_HINTS):
|
||
if line not in lines:
|
||
lines.append(line[:600])
|
||
if len(lines)>=max_lines:
|
||
break
|
||
return lines
|
||
|
||
|
||
def extract_numeric_result_rows(df, max_rows=8) -> list[str]:
|
||
out=[]
|
||
for _,row in df.iterrows():
|
||
vals=row_values(row)
|
||
if not vals:
|
||
continue
|
||
nums=[]
|
||
for v in vals:
|
||
if re.fullmatch(r'-?\d+(?:\.\d+)?', v):
|
||
nums.append(v)
|
||
if len(nums) >= 3 and re.search(r'S-\d|平均|AVG|Max|Min|Ave|OK|合格', ' '.join(vals), re.I):
|
||
out.append(' | '.join(vals)[:500])
|
||
if len(out)>=max_rows:
|
||
break
|
||
return out
|
||
|
||
|
||
def classify_file(path: Path) -> str:
|
||
s=path.as_posix()
|
||
if 'PPK' in s or '过程能力' in s or 'CMK' in s:
|
||
return 'process_capability'
|
||
if 'MSA' in s or 'GRR' in s or 'GR&R' in s or '测量系统' in s:
|
||
return 'msa'
|
||
if '可靠性' in s or '测试' in s or any(x in path.name for x in ['保持力','可焊性','接触电阻','绝缘','耐压','耐焊热','重复插拔']):
|
||
return 'reliability'
|
||
if '全尺寸' in s or '尺寸' in s:
|
||
return 'dimensional'
|
||
if '试验测试工作计划' in s or 'DV' in s or 'PV' in s:
|
||
return 'test_plan'
|
||
return 'other'
|
||
|
||
|
||
def extract_records():
|
||
records=[]
|
||
paths=[p for p in ROOT.rglob('*') if p.is_file() and p.suffix.lower() in {'.xls','.xlsx'}]
|
||
for path in sorted(paths):
|
||
kind=classify_file(path)
|
||
if kind == 'other':
|
||
continue
|
||
xls=read_excel(path)
|
||
if xls is None:
|
||
continue
|
||
for sheet in xls.sheet_names[:12]:
|
||
try:
|
||
df=pd.read_excel(path, sheet_name=sheet, header=None)
|
||
except Exception:
|
||
continue
|
||
if df.empty:
|
||
continue
|
||
fields=extract_header_fields(df)
|
||
key_lines=extract_key_lines(df)
|
||
result_rows=extract_numeric_result_rows(df)
|
||
judgement=extract_judgement(df)
|
||
text=' '.join([path.name, sheet] + list(fields.values()) + key_lines + result_rows + [judgement])
|
||
if not any(x in text for x in SPECIAL_HINTS) and kind not in {'process_capability','msa'}:
|
||
continue
|
||
rec={
|
||
'relative_path': path.relative_to(ROOT).as_posix(),
|
||
'filename': path.name,
|
||
'sheet': sheet,
|
||
'kind': kind,
|
||
'fields': {k:v for k,v in fields.items() if v},
|
||
'judgement': judgement,
|
||
'key_lines': key_lines,
|
||
'result_rows': result_rows,
|
||
}
|
||
records.append(rec)
|
||
return records
|
||
|
||
|
||
def summarize(records):
|
||
special=[]
|
||
seen=set()
|
||
for r in records:
|
||
item=r['fields'].get('item') or ''
|
||
if not item:
|
||
# from filename
|
||
name=r['filename']
|
||
for h in SPECIAL_HINTS:
|
||
if h in name:
|
||
item=name.rsplit('.',1)[0]
|
||
break
|
||
if item and item not in seen:
|
||
seen.add(item)
|
||
special.append(item)
|
||
return special
|
||
|
||
|
||
def esc(s):
|
||
return str(s).replace('|','\\|').replace('\n',' ')
|
||
|
||
|
||
def table(records, kind, limit=80):
|
||
rows=[r for r in records if r['kind']==kind]
|
||
lines=['| 项目/表 | 规格/标准 | 判定 | 关键数据/说明 | 来源 |','|---|---|---|---|---|']
|
||
for r in rows[:limit]:
|
||
f=r['fields']
|
||
item=f.get('item') or r['sheet']
|
||
spec=f.get('spec') or ''
|
||
if f.get('usl') or f.get('lsl'):
|
||
spec='; '.join(x for x in [spec, f"USL {f.get('usl','')}", f"SL {f.get('sl','')}", f"LSL {f.get('lsl','')}"] if x.strip())
|
||
key='; '.join((r['result_rows'][:2] or r['key_lines'][:2]))
|
||
lines.append(f"| {esc(item)} | {esc(spec)} | {esc(r.get('judgement') or '')} | {esc(key[:500])} | {esc(r['relative_path']+' :: '+r['sheet'])} |")
|
||
return '\n'.join(lines)
|
||
|
||
|
||
def write_outputs(records):
|
||
ts=now_iso()
|
||
META.mkdir(parents=True, exist_ok=True)
|
||
OUT.mkdir(parents=True, exist_ok=True)
|
||
jsonp=META/f'bjd-ppap-0.5-18-deep-extraction-{RUN_DATE}.json'
|
||
jsonp.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding='utf-8')
|
||
page=OUT/f'bjd-ppap-0.5-18-deep-extraction-{RUN_DATE}.md'
|
||
special=summarize(records)
|
||
counts=defaultdict(int)
|
||
for r in records:
|
||
counts[r['kind']]+=1
|
||
lines=[
|
||
'---','title: BJD 0.5-18 PPAP/APQP 深度解析','category: reference','project: bjd',
|
||
'tags: [bjd, quality, ppap, apqp, 0.5-18, msa, ppk, reliability, visibility/internal]',
|
||
'visibility: internal','business_domains: [quality, engineering, products]','data_classification: technical-confidential',f'updated: {ts}','---','',
|
||
'# BJD 0.5-18 PPAP/APQP 深度解析','',
|
||
f'来源目录：`{ROOT}`。本次用 pandas/openpyxl/xlrd 对 0.5-18 相关 `.xls/.xlsx` 做深度解析，抽取 `{len(records)}` 个工作表记录。记录分布：' + '；'.join(f'{k}={v}' for k,v in sorted(counts.items())) + '。','',
|
||
'## 可识别的特殊特性/重点管控项目','',
|
||
]
|
||
for s in special[:80]:
|
||
lines.append(f'- {s}')
|
||
lines += ['', '## 过程能力 / PPK / CMK', '', table(records,'process_capability', limit=80), '', '## MSA / GR&R / 测量系统', '', table(records,'msa', limit=80), '', '## 可靠性 / DV / PV / 样品阶段测试', '', table(records,'reliability', limit=120), '', '## 全尺寸 / 尺寸测量', '', table(records,'dimensional', limit=80), '', '## 使用注意', '', '- 旧 Excel 中有些能力指数位于图表/公式区域，若文件未保存缓存，解析器可能只能抽到规格、控制限和判定，不能保证每个 PPK/GR&R 数值都完整。', '- 对客户正式回复时，应以原文件截图/导出 PDF 复核版本、签核、日期和适用料号。', f'- 结构化 JSON：`_meta/{jsonp.name}`']
|
||
page.write_text('\n'.join(lines)+'\n', encoding='utf-8')
|
||
update_meta(ts, page, jsonp, len(records))
|
||
return page, jsonp
|
||
|
||
|
||
def update_meta(ts, page, jsonp, n):
|
||
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}]] - 0.5-18 PPAP/APQP 深度解析，包含特殊特性、PPK/MSA/可靠性/全尺寸抽取。 ( #bjd #quality #ppap #0.5-18 )\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-0.5-18-deep-extraction" records={n} page="{page.relative_to(VAULT).as_posix()}" data="{jsonp.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-0.5-18-deep-extraction-{RUN_DATE}']={
|
||
'ingested_at':ts,'source_path':str(ROOT),'source_type':'filesystem_directory','doc_type':'ppap_deep_extraction','project':'bjd','records':n,
|
||
'pages_created':[page.relative_to(VAULT).as_posix()],'data_paths':[jsonp.relative_to(VAULT).as_posix()],'bot_visible':False,
|
||
'parser':'pandas/openpyxl/xlrd in /tmp/ppap-parse-venv'
|
||
}
|
||
mp.write_text(json.dumps(data, ensure_ascii=False, indent=2)+'\n', encoding='utf-8')
|
||
|
||
|
||
def main():
|
||
records=extract_records()
|
||
page,jsonp=write_outputs(records)
|
||
print(json.dumps({'records':len(records),'page':str(page),'json':str(jsonp)}, ensure_ascii=False))
|
||
|
||
if __name__=='__main__':
|
||
main()
|