#!/usr/bin/env python3 """Build a searchable, credential-safe knowledge package from Office sources.""" from __future__ import annotations import argparse import csv import hashlib import html import re import zipfile from collections import Counter from pathlib import Path from xml.etree import ElementTree as ET OFFICE_SUFFIXES = {".pptx", ".docx", ".xlsx", ".html"} SENSITIVE_NAME = "系统演示账号及操作视频链接202609.xlsx" URL_RE = re.compile(r"https?://[^\s<>'\"]*", re.I) SECRET_RE = re.compile(r"(?i)(密码|口令|password|token|secret|账号|帐号)") def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as source: for chunk in iter(lambda: source.read(8 * 1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def clean_text(value: str) -> str: value = URL_RE.sub("[链接已脱敏]", value) value = re.sub(r"\s+", " ", value).strip() if SECRET_RE.search(value): return "[涉及账号或口令的内容已脱敏]" if re.search(r"(?i)(www\.|[a-z0-9-]+\.(?:com|cn|net|org)(?:/|$))", value): return "[链接已脱敏]" return value def xml_text(blob: bytes) -> list[str]: try: root = ET.fromstring(blob) except ET.ParseError: return [] values = [] for node in root.iter(): if node.tag.endswith("}t") and node.text: text = clean_text(node.text) if text: values.append(text) return values def redact_context(values: list[str], following: int = 2) -> list[str]: redacted = [] skip = 0 for value in values: if value == "[涉及账号或口令的内容已脱敏]": if not redacted or redacted[-1] != value: redacted.append(value) skip = following continue if skip: skip -= 1 continue redacted.append(value) return redacted def extract_pptx(path: Path) -> list[str]: with zipfile.ZipFile(path) as archive: slides = sorted( (name for name in archive.namelist() if re.fullmatch(r"ppt/slides/slide\d+\.xml", name)), key=lambda name: int(re.search(r"(\d+)", name).group(1)), ) lines = [] for position, slide in enumerate(slides, 1): text = redact_context(xml_text(archive.read(slide))) if text: lines.append(f"## 第 {position} 页") lines.extend(f"- {item}" for item in text) return lines def extract_docx(path: Path) -> list[str]: with zipfile.ZipFile(path) as archive: names = [name for name in archive.namelist() if name == "word/document.xml" or name.startswith("word/header") or name.startswith("word/footer")] lines = [] for name in sorted(names): lines.extend(redact_context(xml_text(archive.read(name)))) return [f"- {item}" for item in lines] def extract_xlsx(path: Path) -> list[str]: from openpyxl import load_workbook workbook = load_workbook(path, read_only=True, data_only=True) lines = [] for sheet in workbook: lines.append(f"## 工作表：{sheet.title}") for row in sheet.iter_rows(values_only=True): raw_cells = [str(value) for value in row if value is not None] if any(SECRET_RE.search(value) for value in raw_cells): lines.append("- [涉及账号或口令的整行内容已脱敏]") continue cells = [clean_text(value) for value in raw_cells] cells = [value for value in cells if value] if cells: lines.append("- " + " | ".join(cells)) return lines def extract_html(path: Path) -> list[str]: from html.parser import HTMLParser class TextParser(HTMLParser): def __init__(self) -> None: super().__init__() self.values: list[str] = [] def handle_data(self, data: str) -> None: text = clean_text(data) if text: self.values.append(text) parser = TextParser() parser.feed(path.read_text(encoding="utf-8", errors="replace")) return [f"- {item}" for item in parser.values] def classify(relative_path: str) -> tuple[str, str]: if "智慧食安" in relative_path or "食品安全" in relative_path: return "food-safety", "product-overview" if relative_path.endswith(".pptx") else "delivery-assets" if "智慧采购" in relative_path or "进销存" in relative_path: return "smart-canteen", "product-solutions" if relative_path.endswith(".pptx") else "delivery-assets" if "康养" in relative_path or "医疗行业" in relative_path: return "cross-product", "product-solutions" if "知识问答库" in relative_path: return "smart-canteen", "knowledge-base" if relative_path.endswith(".docx"): return "smart-canteen", "delivery-assets" return "smart-canteen", "product-overview" def slug(index: int, path: Path) -> str: stem = re.sub(r"[^a-z0-9]+", "-", path.stem.lower()).strip("-") if not stem: stem = hashlib.sha1(path.name.encode()).hexdigest()[:10] return f"{index:03d}-{stem[:64]}.md" def build(source_root: Path, output_root: Path, canonical_prefix: str) -> None: output_root.mkdir(parents=True, exist_ok=True) extracts = output_root / "text-extracts" extracts.mkdir(exist_ok=True) paths = sorted(path for path in source_root.rglob("*") if path.is_file() and path.name != ".DS_Store") records = [] counts = Counter() for index, path in enumerate(paths, 1): relative = path.relative_to(source_root).as_posix() product_line, material_type = classify(relative) sensitive = path.name == SENSITIVE_NAME copied = not sensitive extract_name = "" extract_status = "restricted-not-extracted" if sensitive else "unsupported" if not sensitive and path.suffix.lower() in OFFICE_SUFFIXES: extract_name = slug(index, path) try: if path.suffix.lower() == ".pptx": lines = extract_pptx(path) elif path.suffix.lower() == ".docx": lines = extract_docx(path) elif path.suffix.lower() == ".xlsx": lines = extract_xlsx(path) else: lines = extract_html(path) content = [f"# {path.name}", "", f"来源批次：`{source_root.name}`", "", "说明：仅保留可检索文字；外部链接已脱敏，图片与版式以微盘原件为准。", ""] content.extend(lines or ["- 未提取到可搜索文字。"]) (extracts / extract_name).write_text("\n".join(content) + "\n", encoding="utf-8") extract_status = "extracted" except Exception as error: extract_status = f"extract-error:{type(error).__name__}" extract_name = "" digest = sha256(path) records.append( { "source_rel_path": relative, "title": path.name, "product_line": product_line, "material_type": material_type, "extension": path.suffix.lower(), "size_bytes": path.stat().st_size, "sha256": digest, "sensitivity": "restricted-credentials" if sensitive else "team-internal", "raw_upload_status": "blocked-sensitive" if sensitive else "uploaded-to-wedrive", "canonical_rel_path": "" if sensitive else f"{canonical_prefix}/{relative}", "text_extract": f"text-extracts/{extract_name}" if extract_name else "", "extract_status": extract_status, } ) counts[material_type] += 1 fieldnames = list(records[0]) with (output_root / "source-index.csv").open("w", newline="", encoding="utf-8-sig") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames, lineterminator="\n") writer.writeheader() writer.writerows(records) uploaded = sum(record["raw_upload_status"] == "uploaded-to-wedrive" for record in records) total_size = sum(int(record["size_bytes"]) for record in records) summary = f"""# 2026 年 9 月产品介绍资料知识入库 ## 结果 - 本批次共登记 {len(records)} 个文件，合计 {total_size / 1024 / 1024:.1f} MB。 - {uploaded} 个非敏感原文件已进入企业微信微盘标准产品归档目录。 - 1 个文件包含演示系统账号和密码，未上传、未抽取正文，只保留受限资产登记。 - 所有可解析的 PPTX、DOCX、XLSX 均生成脱敏文字抽取，便于后续 Agent 检索；链接统一替换为“链接已脱敏”。 ## 范围 资料覆盖智慧餐厅、营养结算、智慧食安、智慧采购、康养、医疗便民服务、产品清单、软件功能说明和问答库。原始二进制文件以微盘归档为恢复真源，Git 仅保存索引、哈希、摘要和可搜索文字，避免 1.5GB 二进制资料包进入仓库。 ## 使用方法 1. 先查 `source-index.csv`，按 `product_line`、`material_type` 或标题筛选。 2. 需要快速检索正文时查 `text-extracts/`。 3. 需要图片、版式或完整内容时，按 `canonical_rel_path` 回到企业微信微盘原件，并用 SHA256 校验版本。 4. 含账号密码的工作簿不得进入 Git 或普通共享目录；如需共享，应由负责人确认受限目录和访问范围后单独处理。 ## 分类统计 """ for name, count in sorted(counts.items()): summary += f"- {name}: {count}\n" (output_root / "summary.md").write_text(summary, encoding="utf-8") rows = [] for record in records: status = "受限，未上传" if record["raw_upload_status"] == "blocked-sensitive" else "已上传微盘" rows.append( "" f"{html.escape(record['title'])}" f"{html.escape(record['product_line'])}" f"{html.escape(record['material_type'])}" f"{int(record['size_bytes']) / 1024 / 1024:.1f} MB" f"{status}" f"{html.escape(record['extract_status'])}" "" ) page = f"""产品介绍资料知识入库

产品介绍资料知识入库

2026-09-16 · 企业微信微盘原件 + Git 可检索知识索引

{len(records)}
登记文件
{uploaded}
已上传微盘
1
含账号密码，已隔离

文件清单

{''.join(rows)}
文件产品线资料类型大小原件状态检索状态
""" (output_root / "product-introduction-materials.html").write_text(page, encoding="utf-8") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("source_root", type=Path) parser.add_argument("output_root", type=Path) parser.add_argument("canonical_prefix") args = parser.parse_args() build(args.source_root, args.output_root, args.canonical_prefix.rstrip("/")) if __name__ == "__main__": main()