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.
509 lines
18 KiB
Python
509 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""Extract FFC/flat-cable PFMEA signals from indexed email attachments.
|
||
|
||
This intentionally uses only the Python standard library. The current host
|
||
does not have openpyxl/xlrd/libreoffice installed, so .xlsx files are parsed
|
||
from their OOXML parts and legacy .xls files are indexed with a conservative
|
||
string scan.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import hashlib
|
||
import os
|
||
import re
|
||
import sys
|
||
import zipfile
|
||
from collections import Counter, defaultdict
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from xml.etree import ElementTree as ET
|
||
|
||
|
||
VAULT = Path("/home/nac/repo/obsidian-wiki-linux/kb/bjd-vault")
|
||
META = VAULT / "_meta"
|
||
OUT_DIR = VAULT / "projects/bjd/references"
|
||
INVENTORY = META / "bjd-email-attachment-inventory-20260611.jsonl"
|
||
|
||
TARGET_RE = re.compile(
|
||
r"(PFMEA|P-FMEA|Process[_ ]?FMEA|过程失效|失效模式|AIAG.*VDA)",
|
||
re.I,
|
||
)
|
||
FFC_RE = re.compile(r"(FFC|排线|柔性扁平|flat cable|VPMASF|VPNYGF|VPRASF|Backlight FFC)", re.I)
|
||
|
||
RISK_TERMS = [
|
||
"失效",
|
||
"不良",
|
||
"断",
|
||
"短路",
|
||
"开路",
|
||
"错",
|
||
"反",
|
||
"偏",
|
||
"翘",
|
||
"破",
|
||
"裂",
|
||
"压伤",
|
||
"划伤",
|
||
"污染",
|
||
"脏污",
|
||
"毛刺",
|
||
"尺寸",
|
||
"导通",
|
||
"绝缘",
|
||
"阻抗",
|
||
"耐压",
|
||
"接地",
|
||
"GND",
|
||
"pad",
|
||
"焊盘",
|
||
"端子",
|
||
"线材",
|
||
"线缆",
|
||
"贴合",
|
||
"冲切",
|
||
"折弯",
|
||
"包装",
|
||
"标签",
|
||
"混料",
|
||
"Failure",
|
||
"Defect",
|
||
"Open",
|
||
"Short",
|
||
"Damage",
|
||
"Dimension",
|
||
"Continuity",
|
||
"Insulation",
|
||
]
|
||
|
||
CONTROL_TERMS = [
|
||
"全检",
|
||
"巡检",
|
||
"首检",
|
||
"自检",
|
||
"抽检",
|
||
"防错",
|
||
"治具",
|
||
"CCD",
|
||
"AOI",
|
||
"测试",
|
||
"检验",
|
||
"确认",
|
||
"控制",
|
||
"Control",
|
||
"Inspection",
|
||
"Test",
|
||
"Poka",
|
||
"Fixture",
|
||
]
|
||
|
||
NS = {
|
||
"s": "http://schemas.openxmlformats.org/spreadsheetml/2006/main",
|
||
"r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships",
|
||
}
|
||
|
||
|
||
def now_iso() -> str:
|
||
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
||
|
||
|
||
def read_jsonl(path: Path):
|
||
with path.open("r", encoding="utf-8") as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line:
|
||
yield json.loads(line)
|
||
|
||
|
||
def xml_text(elem: ET.Element) -> str:
|
||
return "".join(elem.itertext())
|
||
|
||
|
||
def col_to_idx(cell_ref: str) -> int:
|
||
m = re.match(r"([A-Z]+)", cell_ref or "")
|
||
if not m:
|
||
return 0
|
||
n = 0
|
||
for ch in m.group(1):
|
||
n = n * 26 + (ord(ch) - 64)
|
||
return n - 1
|
||
|
||
|
||
def load_shared_strings(zf: zipfile.ZipFile) -> list[str]:
|
||
try:
|
||
data = zf.read("xl/sharedStrings.xml")
|
||
except KeyError:
|
||
return []
|
||
root = ET.fromstring(data)
|
||
out = []
|
||
for si in root.findall("s:si", NS):
|
||
out.append(xml_text(si))
|
||
return out
|
||
|
||
|
||
def workbook_sheets(zf: zipfile.ZipFile) -> list[tuple[str, str]]:
|
||
wb = ET.fromstring(zf.read("xl/workbook.xml"))
|
||
rels = ET.fromstring(zf.read("xl/_rels/workbook.xml.rels"))
|
||
rid_to_target = {}
|
||
for rel in rels:
|
||
rid = rel.attrib.get("Id")
|
||
target = rel.attrib.get("Target")
|
||
if rid and target:
|
||
rid_to_target[rid] = "xl/" + target.lstrip("/")
|
||
sheets = []
|
||
for sheet in wb.findall("s:sheets/s:sheet", NS):
|
||
name = sheet.attrib.get("name", "Sheet")
|
||
rid = sheet.attrib.get(f"{{{NS['r']}}}id")
|
||
target = rid_to_target.get(rid)
|
||
if target:
|
||
sheets.append((name, target))
|
||
return sheets
|
||
|
||
|
||
def parse_xlsx(path: Path) -> dict[str, Any]:
|
||
result: dict[str, Any] = {
|
||
"parser": "xlsx-ooxml-stdlib",
|
||
"sheets": [],
|
||
"candidate_rows": [],
|
||
"term_counts": Counter(),
|
||
"control_counts": Counter(),
|
||
"error": None,
|
||
}
|
||
try:
|
||
with zipfile.ZipFile(path) as zf:
|
||
shared = load_shared_strings(zf)
|
||
for sheet_name, sheet_path in workbook_sheets(zf):
|
||
try:
|
||
root = ET.fromstring(zf.read(sheet_path))
|
||
except KeyError:
|
||
continue
|
||
rows_seen = 0
|
||
nonempty_rows = 0
|
||
candidate_count = 0
|
||
for row in root.findall(".//s:sheetData/s:row", NS):
|
||
rows_seen += 1
|
||
cells: dict[int, str] = {}
|
||
for c in row.findall("s:c", NS):
|
||
idx = col_to_idx(c.attrib.get("r", ""))
|
||
typ = c.attrib.get("t")
|
||
v = c.find("s:v", NS)
|
||
text = ""
|
||
if typ == "inlineStr":
|
||
is_elem = c.find("s:is", NS)
|
||
text = xml_text(is_elem) if is_elem is not None else ""
|
||
elif v is not None:
|
||
raw = v.text or ""
|
||
if typ == "s":
|
||
try:
|
||
text = shared[int(raw)]
|
||
except Exception:
|
||
text = raw
|
||
else:
|
||
text = raw
|
||
text = re.sub(r"\s+", " ", text).strip()
|
||
if text:
|
||
cells[idx] = text
|
||
if not cells:
|
||
continue
|
||
nonempty_rows += 1
|
||
row_text = " | ".join(cells[i] for i in sorted(cells))
|
||
for term in RISK_TERMS:
|
||
if re.search(re.escape(term), row_text, re.I):
|
||
result["term_counts"][term] += 1
|
||
for term in CONTROL_TERMS:
|
||
if re.search(re.escape(term), row_text, re.I):
|
||
result["control_counts"][term] += 1
|
||
is_candidate = (
|
||
any(re.search(re.escape(term), row_text, re.I) for term in RISK_TERMS)
|
||
or any(re.search(re.escape(term), row_text, re.I) for term in CONTROL_TERMS)
|
||
or TARGET_RE.search(row_text)
|
||
or FFC_RE.search(row_text)
|
||
)
|
||
if is_candidate:
|
||
candidate_count += 1
|
||
if len(result["candidate_rows"]) < 260:
|
||
result["candidate_rows"].append(
|
||
{
|
||
"sheet": sheet_name,
|
||
"row": int(row.attrib.get("r", rows_seen)),
|
||
"text": row_text[:1200],
|
||
}
|
||
)
|
||
result["sheets"].append(
|
||
{
|
||
"name": sheet_name,
|
||
"rows_seen": rows_seen,
|
||
"nonempty_rows": nonempty_rows,
|
||
"candidate_rows": candidate_count,
|
||
}
|
||
)
|
||
except Exception as exc: # noqa: BLE001 - written to ingest report
|
||
result["error"] = repr(exc)
|
||
result["term_counts"] = dict(result["term_counts"].most_common())
|
||
result["control_counts"] = dict(result["control_counts"].most_common())
|
||
return result
|
||
|
||
|
||
def scan_legacy_xls(path: Path) -> dict[str, Any]:
|
||
data = path.read_bytes()
|
||
latin = re.findall(rb"[\x20-\x7e]{4,}", data)
|
||
utf16 = re.findall((rb"(?:[\x20-\x7e]\x00){4,}"), data)
|
||
texts = []
|
||
for b in latin[:3000]:
|
||
texts.append(b.decode("latin1", "ignore"))
|
||
for b in utf16[:3000]:
|
||
texts.append(b.decode("utf-16le", "ignore"))
|
||
candidates = []
|
||
for text in texts:
|
||
if TARGET_RE.search(text) or FFC_RE.search(text) or any(t.lower() in text.lower() for t in ["failure", "control", "inspection"]):
|
||
candidates.append(text[:300])
|
||
return {
|
||
"parser": "xls-string-scan",
|
||
"sheets": [],
|
||
"candidate_rows": [{"sheet": "binary-string-scan", "row": i + 1, "text": t} for i, t in enumerate(candidates[:80])],
|
||
"term_counts": {},
|
||
"control_counts": {},
|
||
"error": "legacy .xls not structurally parsed; string scan only",
|
||
}
|
||
|
||
|
||
def classify_file(rec: dict[str, Any]) -> bool:
|
||
fn = rec.get("filename", "")
|
||
subj = " ".join(rec.get("subjects") or [])
|
||
text = f"{fn} {subj}"
|
||
return bool(TARGET_RE.search(text) and FFC_RE.search(text))
|
||
|
||
|
||
def summarize_patterns(parsed: list[dict[str, Any]]) -> tuple[list[tuple[str, int]], list[tuple[str, int]]]:
|
||
risk = Counter()
|
||
control = Counter()
|
||
for item in parsed:
|
||
risk.update(item["parsed"].get("term_counts", {}))
|
||
control.update(item["parsed"].get("control_counts", {}))
|
||
return risk.most_common(24), control.most_common(18)
|
||
|
||
|
||
def md_escape(s: str) -> str:
|
||
return s.replace("|", "\\|").replace("\n", " ")
|
||
|
||
|
||
def write_outputs(records: list[dict[str, Any]], parsed: list[dict[str, Any]]) -> None:
|
||
ts = now_iso()
|
||
data_path = META / "bjd-ffc-pfmea-extraction-20260611.json"
|
||
page_path = OUT_DIR / "bjd-ffc-pfmea-knowledge-20260611.md"
|
||
|
||
risk, control = summarize_patterns(parsed)
|
||
|
||
with data_path.open("w", encoding="utf-8") as f:
|
||
json.dump(
|
||
{
|
||
"generated_at": ts,
|
||
"target_records": records,
|
||
"parsed_files": parsed,
|
||
"aggregate_risk_terms": risk,
|
||
"aggregate_control_terms": control,
|
||
},
|
||
f,
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
)
|
||
|
||
lines = [
|
||
"---",
|
||
"title: BJD FFC/排线 PFMEA 知识整理",
|
||
"category: reference",
|
||
"project: bjd",
|
||
"tags: [bjd, ffc, pfmea, fmea, quality, ppap, visibility/internal]",
|
||
"visibility: internal",
|
||
"business_domains: [quality, products]",
|
||
"data_classification: quality-confidential",
|
||
f"updated: {ts}",
|
||
"---",
|
||
"",
|
||
"# BJD FFC/排线 PFMEA 知识整理",
|
||
"",
|
||
"本页从邮件附件索引中的 FFC/排线 PFMEA、AIAG & VDA PFMEA、PPAP 资料中抽取。由于当前环境没有 `openpyxl/xlrd/LibreOffice`，`.xlsx` 使用 OOXML 原始 XML 解析，旧版 `.xls` 仅做文件级定位和字符串级弱提取；正式提交客户资料前仍应打开原始 Excel 复核。",
|
||
"",
|
||
"## 结论",
|
||
"",
|
||
"- 母库中已经存在多份排线/FFC PFMEA 源文件，不是只有口头线索。",
|
||
"- 2026 年审核资料指出过一个体系性问题：柔性扁平线缆 NTF 清单相关知识没有完整传递到新项目 PFMEA，新项目 PFMEA 编制没有强制对标 NTF 文件。",
|
||
"- 整改方向已经在资料中出现：用最新 NTF 清单逐项修正新项目 PFMEA，补全新技术对应的失效风险、预防控制和探测控制。",
|
||
"- FFC PFMEA 的客户使用场景主要集中在 PPAP、客户审核、规格变更、样品/封样、背光/车载显示相关线缆项目。",
|
||
"",
|
||
"## 已定位的 FFC/排线 PFMEA 源文件",
|
||
"",
|
||
"| 文件 | 日期 | 邮件主题 | 解析状态 |",
|
||
"|---|---:|---|---|",
|
||
]
|
||
for item in parsed:
|
||
rec = item["record"]
|
||
p = item["parsed"]
|
||
status = p["parser"]
|
||
if p.get("error"):
|
||
status += f"; {p['error']}"
|
||
lines.append(
|
||
f"| `{md_escape(rec.get('filename',''))}` | {rec.get('date_end','未提取')[:10]} | {md_escape('; '.join(rec.get('subjects') or []))} | {md_escape(status)} |"
|
||
)
|
||
|
||
lines += [
|
||
"",
|
||
"## 聚合风险信号",
|
||
"",
|
||
"| 风险/失效关键词 | 出现行数 |",
|
||
"|---|---:|",
|
||
]
|
||
for term, count in risk:
|
||
lines.append(f"| {md_escape(term)} | {count} |")
|
||
|
||
lines += [
|
||
"",
|
||
"## 聚合控制信号",
|
||
"",
|
||
"| 控制/检验关键词 | 出现行数 |",
|
||
"|---|---:|",
|
||
]
|
||
for term, count in control:
|
||
lines.append(f"| {md_escape(term)} | {count} |")
|
||
|
||
lines += [
|
||
"",
|
||
"## 审核发现和整改要点",
|
||
"",
|
||
"客户审核 Action list 中对 FFC/PFMEA 的核心发现是：",
|
||
"",
|
||
"- 客户项目输入没有在内部充分整合。",
|
||
"- 柔性扁平线缆 NTF 清单相关知识没有完整传递到新项目 PFMEA。",
|
||
"- 部分 FFC 原材料及供应商管理不够系统，尤其是接地焊盘/GND pad 类物料，涉及供应商类别定义、图纸电气性能参数、PPAP Level 3、安全库存、货架有效期等。",
|
||
"",
|
||
"对应的 PFMEA 改进动作：",
|
||
"",
|
||
"- 新项目 PFMEA 必须强制对标最新 FFC NTF 清单。",
|
||
"- 将 NTF 中的新技术、新材料、新工艺风险转成 PFMEA 的失效模式、失效影响、失效原因、预防控制和探测控制。",
|
||
"- 对 GND pad、线材、端子、贴合/冲切/折弯/包装/标签等关键对象建立专项管控清单。",
|
||
"- 图纸评审、供应商准入、PPAP 审核、库存和效期管理要形成闭环，避免 PFMEA 写了但控制计划、检验规范和供应商管理没有同步。",
|
||
"",
|
||
"## 可复用的问答口径",
|
||
"",
|
||
"- 如果客户问有没有 FFC/排线 PFMEA：可以说有，BJD 内部已有 AIAG & VDA 版本和项目版本的 FFC PFMEA 资料，可按客户项目/料号提交受控版本。",
|
||
"- 如果客户追问是否覆盖新技术清单：应谨慎回答，内部审核资料曾指出 NTF 到新项目 PFMEA 的联动需要加强；现在正确做法是按最新 NTF 清单逐项复核后再提交。",
|
||
"- 如果是 PPAP 提交：不要直接用通用版替代项目版，应确认客户项目、料号、工艺路线、特殊特性、控制计划和检验规范是否一致。",
|
||
"",
|
||
"## 抽取样例行",
|
||
"",
|
||
]
|
||
|
||
for item in parsed:
|
||
rec = item["record"]
|
||
rows = item["parsed"].get("candidate_rows", [])[:20]
|
||
if not rows:
|
||
continue
|
||
lines += [f"### {rec.get('filename','')}", ""]
|
||
for row in rows[:12]:
|
||
lines.append(f"- `{row.get('sheet')}` R{row.get('row')}: {md_escape(row.get('text','')[:420])}")
|
||
lines.append("")
|
||
|
||
lines += [
|
||
"## 数据文件",
|
||
"",
|
||
f"- 结构化抽取结果：`_meta/{data_path.name}`",
|
||
"- 原始附件路径保存在 `_meta/bjd-email-attachment-inventory-20260611.jsonl` 中。",
|
||
]
|
||
|
||
page_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
update_vault_metadata(ts, records, parsed, page_path, data_path)
|
||
|
||
|
||
def update_vault_metadata(
|
||
ts: str,
|
||
records: list[dict[str, Any]],
|
||
parsed: list[dict[str, Any]],
|
||
page_path: Path,
|
||
data_path: Path,
|
||
) -> None:
|
||
rel_page = page_path.relative_to(VAULT).with_suffix("")
|
||
index_path = VAULT / "index.md"
|
||
if index_path.exists():
|
||
text = index_path.read_text(encoding="utf-8")
|
||
text = re.sub(r"最后更新：[^*]+", f"最后更新：{ts}", text)
|
||
entry = (
|
||
f"- [[{rel_page.as_posix()}]] - FFC/排线 PFMEA 源文件、AIAG & VDA FMEA、"
|
||
"审核发现和整改口径整理。 ( #bjd #ffc #pfmea #quality )"
|
||
)
|
||
if str(rel_page) not in text:
|
||
marker = "## 综合分析\n"
|
||
if marker in text:
|
||
text = text.replace(marker, marker + entry + "\n", 1)
|
||
else:
|
||
text += "\n## 综合分析\n" + entry + "\n"
|
||
index_path.write_text(text, encoding="utf-8")
|
||
|
||
hot_path = VAULT / "hot.md"
|
||
if hot_path.exists():
|
||
hot = hot_path.read_text(encoding="utf-8")
|
||
hot_entry = f"- [[{rel_page.as_posix()}]] - FFC/排线 PFMEA 知识整理。"
|
||
if str(rel_page) not in hot:
|
||
hot += "\n" + hot_entry + "\n"
|
||
hot_path.write_text(hot, encoding="utf-8")
|
||
|
||
log_path = VAULT / "log.md"
|
||
with log_path.open("a", encoding="utf-8") as f:
|
||
f.write(
|
||
"\n"
|
||
f"- [{ts}] DATA_INGEST source=\"_meta/bjd-email-attachment-inventory-20260611.jsonl\" "
|
||
f"project=\"bjd\" mode=\"ffc-pfmea-deep-extraction\" records={len(records)} "
|
||
f"parsed={len(parsed)} page=\"{page_path.relative_to(VAULT).as_posix()}\" "
|
||
f"data=\"{data_path.relative_to(VAULT).as_posix()}\" bot_visible=false\n"
|
||
)
|
||
|
||
manifest_path = VAULT / ".manifest.json"
|
||
manifest: dict[str, Any] = {}
|
||
if manifest_path.exists():
|
||
try:
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError:
|
||
manifest = {}
|
||
source_key = "derived:ffc-pfmea-extraction-20260611"
|
||
inv_hash = hashlib.sha256(INVENTORY.read_bytes()).hexdigest() if INVENTORY.exists() else ""
|
||
manifest.setdefault("sources", {})[source_key] = {
|
||
"ingested_at": ts,
|
||
"original_path": str(INVENTORY),
|
||
"source_type": "derived_email_attachment_pfmea",
|
||
"doc_type": "quality_pfmea",
|
||
"project": "bjd",
|
||
"content_hash": "sha256:" + inv_hash if inv_hash else "",
|
||
"records": len(records),
|
||
"parsed": len(parsed),
|
||
"pages_created": [page_path.relative_to(VAULT).as_posix()],
|
||
"data_path": data_path.relative_to(VAULT).as_posix(),
|
||
"bot_visible": False,
|
||
}
|
||
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
|
||
def main() -> int:
|
||
if not INVENTORY.exists():
|
||
print(f"missing inventory: {INVENTORY}", file=sys.stderr)
|
||
return 2
|
||
records = [r for r in read_jsonl(INVENTORY) if classify_file(r)]
|
||
records.sort(key=lambda r: (r.get("date_end") or "", r.get("filename") or ""))
|
||
parsed = []
|
||
for rec in records:
|
||
path = Path(rec["absolute_path"])
|
||
if rec.get("ext") == ".xlsx":
|
||
info = parse_xlsx(path)
|
||
elif rec.get("ext") == ".xls":
|
||
info = scan_legacy_xls(path)
|
||
else:
|
||
continue
|
||
parsed.append({"record": rec, "parsed": info})
|
||
write_outputs(records, parsed)
|
||
print(json.dumps({"records": len(records), "parsed": len(parsed)}, ensure_ascii=False))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|