#!/usr/bin/env python3 """Extract text and source metadata from the AI organization methodology folder.""" from __future__ import annotations import csv import hashlib import json import re import shutil import subprocess import tempfile import zipfile from pathlib import Path from xml.etree import ElementTree as ET SOURCE_DIR = Path("/Users/jack/Downloads/04_AI_组织_方法论") ROOT = Path(__file__).resolve().parents[1] EXTRACTED = ROOT / "extracted" TEXT_NS = "{http://schemas.openxmlformats.org/drawingml/2006/main}" WORD_NS = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}" SS_NS = "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}" SLIDE_RE = re.compile(r"ppt/slides/slide(\d+)\.xml$") def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def safe_stem(path: Path) -> str: stem = path.stem stem = re.sub(r"[^A-Za-z0-9\u4e00-\u9fff._-]+", "-", stem).strip("-") return stem or "source" def clean_text(value: str) -> str: value = value.replace("\u000b", "\n") lines = [re.sub(r"\s+", " ", line).strip() for line in value.splitlines()] return "\n".join(line for line in lines if line) def docx_text(path: Path) -> str: with zipfile.ZipFile(path) as zf: xml = zf.read("word/document.xml") root = ET.fromstring(xml) paragraphs: list[str] = [] for paragraph in root.iter(f"{WORD_NS}p"): parts = [node.text or "" for node in paragraph.iter(f"{WORD_NS}t")] text = clean_text("".join(parts)) if text: paragraphs.append(text) return "\n\n".join(paragraphs) def pptx_text(path: Path) -> str: slides = [] with zipfile.ZipFile(path) as zf: names = [] for name in zf.namelist(): match = SLIDE_RE.match(name) if match: names.append((int(match.group(1)), name)) for slide_no, name in sorted(names): root = ET.fromstring(zf.read(name)) paragraphs: list[str] = [] for paragraph in root.iter(f"{TEXT_NS}p"): parts = [node.text or "" for node in paragraph.iter(f"{TEXT_NS}t")] text = clean_text("".join(parts)) if text: paragraphs.append(text) slides.append(f"## Slide {slide_no}\n\n" + "\n".join(f"- {p}" for p in paragraphs)) return "\n\n".join(slides) def read_shared_strings(zf: zipfile.ZipFile) -> list[str]: try: root = ET.fromstring(zf.read("xl/sharedStrings.xml")) except KeyError: return [] strings = [] for si in root.iter(f"{SS_NS}si"): parts = [node.text or "" for node in si.iter(f"{SS_NS}t")] strings.append("".join(parts)) return strings def xlsx_text(path: Path) -> str: with zipfile.ZipFile(path) as zf: shared = read_shared_strings(zf) sheet_names = sorted( name for name in zf.namelist() if re.match(r"xl/worksheets/sheet\d+\.xml$", name) ) sections: list[str] = [] for sheet_idx, name in enumerate(sheet_names, start=1): root = ET.fromstring(zf.read(name)) rows = [] for row in root.iter(f"{SS_NS}row"): values = [] for cell in row.iter(f"{SS_NS}c"): value_node = cell.find(f"{SS_NS}v") if value_node is None or value_node.text is None: values.append("") continue value = value_node.text if cell.attrib.get("t") == "s": try: value = shared[int(value)] except (ValueError, IndexError): pass values.append(value) if any(v.strip() for v in values): rows.append(",".join(v.replace("\n", " ").strip() for v in values)) sections.append(f"## Sheet {sheet_idx}\n\n" + "\n".join(rows)) return "\n\n".join(sections) def pdf_text(path: Path) -> str: if not shutil.which("pdftotext"): return "[blocked] pdftotext not found" result = subprocess.run( ["pdftotext", "-layout", str(path), "-"], check=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) return result.stdout def xls_text(path: Path) -> str: soffice = shutil.which("soffice") or "/opt/homebrew/bin/soffice" if not Path(soffice).exists(): return "[blocked] soffice not found for .xls conversion" with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) result = subprocess.run( [soffice, "--headless", "--convert-to", "csv", "--outdir", str(tmp_path), str(path)], check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) csv_files = sorted(tmp_path.glob("*.csv")) if not csv_files: return f"[blocked] soffice conversion produced no csv: {result.stderr or result.stdout}" return "\n\n".join(file.read_text(encoding="utf-8", errors="replace") for file in csv_files) def extract_text(path: Path) -> tuple[str, str]: ext = path.suffix.lower() try: if ext == ".docx": return docx_text(path), "extracted" if ext == ".pptx": return pptx_text(path), "extracted" if ext == ".xlsx": return xlsx_text(path), "extracted" if ext == ".pdf": return pdf_text(path), "extracted" if ext == ".xls": return xls_text(path), "extracted" if ext in {".md", ".txt"}: return path.read_text(encoding="utf-8", errors="replace"), "copied-text" except Exception as exc: # noqa: BLE001 - evidence capture should continue. return f"[blocked] {type(exc).__name__}: {exc}", "blocked" return "[blocked] unsupported file type", "blocked" def main() -> int: EXTRACTED.mkdir(parents=True, exist_ok=True) files = sorted(path for path in SOURCE_DIR.iterdir() if path.is_file()) source_rows = [] asset_rows = [] combined_lines = ["# AI Organization Methodology Source Extraction", ""] for index, path in enumerate(files, start=1): source_id = f"AI-ORG-METHOD-{index:03d}" text, status = extract_text(path) out_name = f"{source_id}-{safe_stem(path)}.md" out_path = EXTRACTED / out_name out_path.write_text( f"# {path.name}\n\n- source_id: `{source_id}`\n- source_path: `{path}`\n- extraction_status: `{status}`\n\n{text}\n", encoding="utf-8", ) source_rows.append( { "source_id": source_id, "title": path.name, "source_type": path.suffix.lower().lstrip(".") or "unknown", "source_path": str(path), "size_bytes": path.stat().st_size, "sha256": sha256(path), "captured_at": "2026-05-29", "status": status, "derived_text_path": str(out_path.relative_to(ROOT)), "notes": "raw binary not copied into Git", } ) asset_rows.append( { "asset_id": source_id, "title": path.name, "canonical_source_type": "local_downloads_file", "canonical_root_key": "downloads_ai_org_method_jack", "canonical_rel_path": path.name, "original_root_key": "downloads_ai_org_method_jack", "original_rel_path": path.name, "owner_or_sender": "姜阳", "version_hint": path.stat().st_mtime_ns, "size_bytes": path.stat().st_size, "sha256": source_rows[-1]["sha256"], "access_policy": "jack-local-downloads", "raw_binary_in_git": "no", "derived_index_path": "work/2026-05-29-ai-organization-methodology-intake/source-index.csv", "derived_review_path": "work/2026-05-29-ai-organization-methodology-intake/ai-organization-methodology-review.html", "status": status, "next_action": "Use derived text and compiled wiki; do not copy raw binary.", } ) combined_lines.append(f"## {source_id} {path.name}") combined_lines.append("") combined_lines.append(text[:12000]) combined_lines.append("") with (ROOT / "source-index.csv").open("w", encoding="utf-8", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=list(source_rows[0].keys())) writer.writeheader() writer.writerows(source_rows) with (ROOT / "large-source-assets.csv").open("w", encoding="utf-8", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=list(asset_rows[0].keys())) writer.writeheader() writer.writerows(asset_rows) (EXTRACTED / "combined-extract.md").write_text("\n".join(combined_lines), encoding="utf-8") print(json.dumps({"files": len(files), "extracted_dir": str(EXTRACTED)}, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())