| from __future__ import annotations |
| |
| |
|
|
| import argparse |
| import csv |
| import hashlib |
| import json |
| import os |
| import re |
| import tarfile |
| 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(__file__).resolve().parents[1] |
| MANIFEST = ROOT / "manifests" / "raw_sources.csv" |
| LOCK_FILE = ROOT / "manifests" / "raw_index.json" |
| REPORT_FILE = ROOT / "manifests" / "intake_report.md" |
| DEFAULT_RAW_ROOT = (ROOT.parent / "__RAW_ROOT_NAME__").resolve() |
| EXPECTED_COLUMNS = [ |
| "source_id", |
| "company", |
| "vendor", |
| "kind", |
| "filename", |
| "raw_rel_path", |
| "status", |
| "compiled_into", |
| "notes", |
| ] |
| TRACKED_EXTENSIONS = { |
| ".pdf", ".md", ".txt", |
| ".xlsx", ".xls", ".xlsm", ".csv", ".tsv", |
| ".doc", ".docx", ".ppt", ".pptx", |
| ".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".svg", |
| ".zip", ".rar", ".7z", ".tar", ".gz", |
| } |
| SKIP_DIRS = { |
| ".git", ".svn", "__pycache__", ".DS_Store", |
| "node_modules", ".venv", "venv", |
| } |
| KEY_COLUMN_HINTS = ("sku", "part", "item", "model", "spec", "code", "pn", "material", "id", "series") |
| VALUE_COLUMN_HINTS = ("price", "cost", "amount", "qty", "quantity", "lead", "currency", "discount", "rate") |
| MAX_HEADERS = 8 |
| MAX_ROW_WIDTH = 12 |
| MAX_TRACKED_ROWS = 2000 |
| MAX_SHEETS = 6 |
|
|
|
|
| def utc_now() -> str: |
| return datetime.now(timezone.utc).replace(microsecond=0).isoformat() |
|
|
|
|
| def sha256_prefix(path: Path) -> str: |
| h = hashlib.sha256() |
| with path.open("rb") as handle: |
| for chunk in iter(lambda: handle.read(8192), b""): |
| h.update(chunk) |
| return h.hexdigest()[:16] |
|
|
|
|
| def safe_read_text(path: Path, limit: int = 2000) -> str: |
| try: |
| return path.read_text(encoding="utf-8", errors="ignore")[:limit] |
| except Exception: |
| return "" |
|
|
|
|
| def first_nonempty_line(text: str) -> str: |
| for line in text.splitlines(): |
| stripped = line.strip(" # -") |
| if stripped: |
| return stripped[:120] |
| return "" |
|
|
|
|
| def detect_kind(path: Path) -> str: |
| ext = path.suffix.lower() |
| if ext in {".xlsx", ".xls", ".xlsm", ".csv", ".tsv"}: |
| return "spreadsheet" |
| if ext in {".pdf", ".doc", ".docx", ".ppt", ".pptx", ".md", ".txt"}: |
| return "document" |
| if ext in {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".svg"}: |
| return "image" |
| if ext in {".zip", ".rar", ".7z", ".tar", ".gz"}: |
| return "archive" |
| return "raw" |
|
|
|
|
| def clean_cells(values: list[str], *, limit: int = MAX_ROW_WIDTH) -> list[str]: |
| cleaned = [ |
| str(value or "").replace("\n", " ").replace("\r", " ").strip()[:80] |
| for value in values[:limit] |
| ] |
| while cleaned and not cleaned[-1]: |
| cleaned.pop() |
| return cleaned |
|
|
|
|
| def guess_key_column(headers: list[str]) -> str: |
| lowered = [(header, header.lower()) for header in headers if header] |
| for hint in KEY_COLUMN_HINTS: |
| for original, value in lowered: |
| if hint in value: |
| return original |
| return headers[0] if headers else "" |
|
|
|
|
| def suspicious_columns(headers: list[str]) -> list[str]: |
| flagged: list[str] = [] |
| for header in headers: |
| lowered = header.lower() |
| if any(hint in lowered for hint in KEY_COLUMN_HINTS + VALUE_COLUMN_HINTS): |
| flagged.append(header) |
| return flagged[:6] |
|
|
|
|
| def row_signature(cells: list[str]) -> str: |
| payload = "|".join(clean_cells(cells, limit=MAX_ROW_WIDTH)) |
| return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] |
|
|
|
|
| def compare_named_lists(before: list[str], after: list[str]) -> tuple[list[str], list[str]]: |
| old = set(before) |
| new = set(after) |
| return sorted(new - old), sorted(old - new) |
|
|
|
|
| def compare_row_signatures(before: dict[str, str], after: dict[str, str]) -> str: |
| if not before or not after: |
| return "" |
| old_keys = set(before) |
| new_keys = set(after) |
| added = len(new_keys - old_keys) |
| removed = len(old_keys - new_keys) |
| changed = len([key for key in old_keys & new_keys if before[key] != after[key]]) |
| parts: list[str] = [] |
| if added: |
| parts.append(f"{added} added") |
| if removed: |
| parts.append(f"{removed} removed") |
| if changed: |
| parts.append(f"{changed} changed") |
| return ", ".join(parts) |
|
|
|
|
| def summarize_delimited(path: Path, delimiter: str, parser: str) -> dict[str, object]: |
| headers: list[str] = [] |
| key_column = "" |
| key_index = -1 |
| row_count = 0 |
| column_count = 0 |
| tracked_rows: dict[str, str] = {} |
| sample_rows: list[str] = [] |
| truncated = False |
| with path.open("r", encoding="utf-8-sig", errors="ignore", newline="") as handle: |
| reader = csv.reader(handle, delimiter=delimiter) |
| for raw_row in reader: |
| cleaned = clean_cells(raw_row) |
| if not cleaned: |
| continue |
| column_count = max(column_count, len(cleaned)) |
| if not headers: |
| headers = [cell[:60] for cell in cleaned[:MAX_HEADERS]] |
| key_column = guess_key_column(headers) |
| key_index = headers.index(key_column) if key_column in headers else -1 |
| continue |
| row_count += 1 |
| if len(sample_rows) < 3: |
| sample_rows.append(" | ".join(cleaned[:MAX_HEADERS])) |
| if row_count > MAX_TRACKED_ROWS: |
| truncated = True |
| continue |
| if key_index >= 0 and key_index < len(cleaned): |
| key = cleaned[key_index] |
| if key: |
| tracked_rows[key] = row_signature(cleaned) |
| return { |
| "parser": parser, |
| "summary": ( |
| f"{row_count} data row(s); {column_count} column(s); " |
| f"headers: {', '.join(headers) if headers else 'none'}" |
| ), |
| "metadata": { |
| "row_count": row_count, |
| "column_count": column_count, |
| "headers": headers, |
| "delimiter": delimiter, |
| "key_column": key_column, |
| "suspicious_columns": suspicious_columns(headers), |
| "sample_rows": sample_rows, |
| "tracked_rows": tracked_rows, |
| "truncated": truncated, |
| }, |
| } |
|
|
|
|
| def summarize_csv(path: Path) -> dict[str, object]: |
| delimiter = "," if path.suffix.lower() == ".csv" else "\t" |
| return summarize_delimited(path, delimiter, "csv-local") |
|
|
|
|
| def normalize_zip_path(base: str, target: str) -> str: |
| parts = [part for part in (Path(base).parent / target).as_posix().split("/") if part not in {"", "."}] |
| normalized: list[str] = [] |
| for part in parts: |
| if part == "..": |
| if normalized: |
| normalized.pop() |
| continue |
| normalized.append(part) |
| return "/".join(normalized) |
|
|
|
|
| def spreadsheet_column_index(label: str) -> int: |
| total = 0 |
| for char in label.upper(): |
| if not char.isalpha(): |
| break |
| total = total * 26 + (ord(char) - 64) |
| return total |
|
|
|
|
| def parse_sheet_dimension(ref: str) -> tuple[int, int]: |
| if not ref: |
| return 0, 0 |
| tail = ref.split(":")[-1] |
| letters = "".join(char for char in tail if char.isalpha()) |
| digits = "".join(char for char in tail if char.isdigit()) |
| return int(digits or 0), spreadsheet_column_index(letters) |
|
|
|
|
| def read_shared_strings(zf: zipfile.ZipFile) -> list[str]: |
| try: |
| with zf.open("xl/sharedStrings.xml") as handle: |
| root = ET.fromstring(handle.read()) |
| return ["".join(node.itertext()).strip() for node in root.findall(".//{*}si")] |
| except Exception: |
| return [] |
|
|
|
|
| def read_workbook_sheets(zf: zipfile.ZipFile) -> list[tuple[str, str]]: |
| with zf.open("xl/workbook.xml") as handle: |
| workbook_root = ET.fromstring(handle.read()) |
| rel_map: dict[str, str] = {} |
| try: |
| with zf.open("xl/_rels/workbook.xml.rels") as handle: |
| rel_root = ET.fromstring(handle.read()) |
| for node in rel_root.findall(".//{*}Relationship"): |
| rel_id = node.attrib.get("Id", "") |
| target = node.attrib.get("Target", "") |
| if rel_id and target: |
| rel_map[rel_id] = normalize_zip_path("xl/workbook.xml", target) |
| except Exception: |
| rel_map = {} |
| sheets: list[tuple[str, str]] = [] |
| for node in workbook_root.findall(".//{*}sheet"): |
| name = node.attrib.get("name", "") |
| rel_id = "" |
| for key, value in node.attrib.items(): |
| if key.endswith("}id") or key == "id": |
| rel_id = value |
| break |
| target = rel_map.get(rel_id, "") |
| if name and target: |
| sheets.append((name, target)) |
| return sheets |
|
|
|
|
| def resolve_xlsx_cell(cell: ET.Element, shared_strings: list[str]) -> str: |
| cell_type = cell.attrib.get("t", "") |
| if cell_type == "inlineStr": |
| return "".join(cell.itertext()).strip() |
| value = cell.findtext("{*}v", default="").strip() |
| if cell_type == "s": |
| try: |
| return shared_strings[int(value)] |
| except Exception: |
| return value |
| if cell_type == "b": |
| return "TRUE" if value == "1" else "FALSE" |
| if value: |
| return value |
| return "".join(cell.itertext()).strip() |
|
|
|
|
| def summarize_xlsx_sheet(zf: zipfile.ZipFile, sheet_name: str, sheet_path: str, shared_strings: list[str]) -> dict[str, object]: |
| with zf.open(sheet_path) as handle: |
| root = ET.fromstring(handle.read()) |
| dimension_node = root.find("{*}dimension") |
| dim_rows, dim_cols = parse_sheet_dimension(dimension_node.attrib.get("ref", "") if dimension_node is not None else "") |
| headers: list[str] = [] |
| key_column = "" |
| key_index = -1 |
| row_count = 0 |
| column_count = 0 |
| tracked_rows: dict[str, str] = {} |
| sample_rows: list[str] = [] |
| truncated = False |
| for row_node in root.findall(".//{*}sheetData/{*}row"): |
| cells_by_index: dict[int, str] = {} |
| for cell in row_node.findall("{*}c"): |
| ref = cell.attrib.get("r", "") |
| letters = "".join(char for char in ref if char.isalpha()) |
| index = spreadsheet_column_index(letters) or (len(cells_by_index) + 1) |
| cells_by_index[index] = resolve_xlsx_cell(cell, shared_strings) |
| if not cells_by_index: |
| continue |
| max_index = min(max(cells_by_index), MAX_ROW_WIDTH) |
| cleaned = clean_cells([cells_by_index.get(idx, "") for idx in range(1, max_index + 1)]) |
| if not cleaned: |
| continue |
| column_count = max(column_count, len(cleaned)) |
| if not headers: |
| headers = [cell[:60] for cell in cleaned[:MAX_HEADERS]] |
| key_column = guess_key_column(headers) |
| key_index = headers.index(key_column) if key_column in headers else -1 |
| continue |
| row_count += 1 |
| if len(sample_rows) < 3: |
| sample_rows.append(" | ".join(cleaned[:MAX_HEADERS])) |
| if row_count > MAX_TRACKED_ROWS: |
| truncated = True |
| continue |
| if key_index >= 0 and key_index < len(cleaned): |
| key = cleaned[key_index] |
| if key: |
| tracked_rows[key] = row_signature(cleaned) |
| return { |
| "name": sheet_name, |
| "row_count": row_count or dim_rows, |
| "column_count": column_count or dim_cols, |
| "headers": headers, |
| "key_column": key_column, |
| "suspicious_columns": suspicious_columns(headers), |
| "sample_rows": sample_rows, |
| "tracked_rows": tracked_rows, |
| "truncated": truncated, |
| } |
|
|
|
|
| def summarize_xlsx(path: Path) -> dict[str, object]: |
| sheets: list[dict[str, object]] = [] |
| try: |
| with zipfile.ZipFile(path) as zf: |
| shared_strings = read_shared_strings(zf) |
| for sheet_name, sheet_path in read_workbook_sheets(zf)[:MAX_SHEETS]: |
| try: |
| sheets.append(summarize_xlsx_sheet(zf, sheet_name, sheet_path, shared_strings)) |
| except Exception: |
| sheets.append({ |
| "name": sheet_name, |
| "row_count": 0, |
| "column_count": 0, |
| "headers": [], |
| "key_column": "", |
| "suspicious_columns": [], |
| "sample_rows": [], |
| "tracked_rows": {}, |
| "truncated": False, |
| }) |
| except Exception: |
| sheets = [] |
| sheet_names = [sheet["name"] for sheet in sheets] |
| summary_bits = [ |
| f"{sheet['name']}[{sheet.get('row_count', 0)}x{sheet.get('column_count', 0)}]" |
| for sheet in sheets[:4] |
| ] |
| return { |
| "parser": "xlsx-local", |
| "summary": f"{len(sheet_names)} sheet(s): {', '.join(summary_bits) if summary_bits else 'unknown'}", |
| "metadata": {"sheet_names": sheet_names, "sheets": sheets}, |
| } |
|
|
|
|
| def summarize_docx(path: Path) -> dict[str, object]: |
| paragraph_count = 0 |
| try: |
| with zipfile.ZipFile(path) as zf: |
| with zf.open("word/document.xml") as handle: |
| root = ET.fromstring(handle.read()) |
| paragraph_count = len(root.findall(".//{*}p")) |
| except Exception: |
| pass |
| return { |
| "parser": "docx-local", |
| "summary": f"{paragraph_count} paragraph block(s)", |
| "metadata": {"paragraph_blocks": paragraph_count}, |
| } |
|
|
|
|
| def summarize_pptx(path: Path) -> dict[str, object]: |
| slide_count = 0 |
| try: |
| with zipfile.ZipFile(path) as zf: |
| slide_count = len([name for name in zf.namelist() if name.startswith("ppt/slides/slide") and name.endswith(".xml")]) |
| except Exception: |
| pass |
| return { |
| "parser": "pptx-local", |
| "summary": f"{slide_count} slide(s)", |
| "metadata": {"slide_count": slide_count}, |
| } |
|
|
|
|
| def summarize_pdf(path: Path) -> dict[str, object]: |
| raw = path.read_bytes() |
| page_count = raw.count(b"/Type /Page") |
| return { |
| "parser": "pdf-local", |
| "summary": f"{page_count or 'unknown'} page(s)", |
| "metadata": {"page_count": int(page_count)}, |
| } |
|
|
|
|
| def image_size(path: Path) -> tuple[int | None, int | None]: |
| raw = path.read_bytes()[:64] |
| if raw.startswith(b"\x89PNG\r\n\x1a\n") and len(raw) >= 24: |
| return int.from_bytes(raw[16:20], "big"), int.from_bytes(raw[20:24], "big") |
| if raw[:6] in {b"GIF87a", b"GIF89a"} and len(raw) >= 10: |
| return int.from_bytes(raw[6:8], "little"), int.from_bytes(raw[8:10], "little") |
| if raw.startswith(b"BM") and len(raw) >= 26: |
| return int.from_bytes(raw[18:22], "little"), int.from_bytes(raw[22:26], "little") |
| if raw.startswith(b"\xff\xd8"): |
| with path.open("rb") as handle: |
| handle.read(2) |
| while True: |
| marker_prefix = handle.read(1) |
| if marker_prefix != b"\xff": |
| break |
| marker = handle.read(1) |
| while marker == b"\xff": |
| marker = handle.read(1) |
| if marker in {b"\xc0", b"\xc1", b"\xc2", b"\xc3", b"\xc5", b"\xc6", b"\xc7", b"\xc9", b"\xca", b"\xcb", b"\xcd", b"\xce", b"\xcf"}: |
| _length = int.from_bytes(handle.read(2), "big") |
| handle.read(1) |
| height = int.from_bytes(handle.read(2), "big") |
| width = int.from_bytes(handle.read(2), "big") |
| return width, height |
| if marker in {b"\xd8", b"\xd9"}: |
| continue |
| seg_length = int.from_bytes(handle.read(2), "big") |
| handle.seek(seg_length - 2, 1) |
| return None, None |
|
|
|
|
| def summarize_image(path: Path) -> dict[str, object]: |
| width, height = image_size(path) |
| dims = f"{width}x{height}" if width and height else "unknown" |
| return { |
| "parser": "image-local", |
| "summary": f"dimensions: {dims}", |
| "metadata": {"width": width, "height": height}, |
| } |
|
|
|
|
| def summarize_archive(path: Path) -> dict[str, object]: |
| ext = path.suffix.lower() |
| entry_count = None |
| parser = "archive-local" |
| try: |
| if ext == ".zip": |
| with zipfile.ZipFile(path) as zf: |
| entry_count = len(zf.namelist()) |
| elif ext in {".tar", ".gz"} or path.name.endswith(".tar.gz"): |
| with tarfile.open(path) as tf: |
| entry_count = len(tf.getmembers()) |
| except Exception: |
| entry_count = None |
| return { |
| "parser": parser, |
| "summary": f"archive entries: {entry_count if entry_count is not None else 'unknown'}", |
| "metadata": {"entry_count": entry_count, "format": ext.lstrip('.')}, |
| } |
|
|
|
|
| def summarize_plaintext(path: Path) -> dict[str, object]: |
| text = safe_read_text(path) |
| summary = first_nonempty_line(text) or "no preview" |
| return { |
| "parser": "text-local", |
| "summary": summary, |
| "metadata": {"preview": summary}, |
| } |
|
|
|
|
| def summarize_xls_legacy(path: Path) -> dict[str, object]: |
| return { |
| "parser": "xls-legacy", |
| "summary": "legacy .xls workbook (sheet metadata unavailable without extra parser)", |
| "metadata": {"format": "xls", "size_bytes": path.stat().st_size}, |
| } |
|
|
|
|
| def summarize_csv_change(previous: dict[str, object], current: dict[str, object]) -> list[str]: |
| notes: list[str] = [] |
| prev_headers = [str(item) for item in previous.get("headers", [])] |
| curr_headers = [str(item) for item in current.get("headers", [])] |
| added_headers, removed_headers = compare_named_lists(prev_headers, curr_headers) |
| if added_headers: |
| notes.append(f"headers added: {', '.join(added_headers[:4])}") |
| if removed_headers: |
| notes.append(f"headers removed: {', '.join(removed_headers[:4])}") |
| prev_rows = int(previous.get("row_count", 0) or 0) |
| curr_rows = int(current.get("row_count", 0) or 0) |
| if prev_rows != curr_rows: |
| notes.append(f"rows {prev_rows} -> {curr_rows}") |
| prev_key = str(previous.get("key_column", "") or "") |
| curr_key = str(current.get("key_column", "") or "") |
| if prev_key != curr_key and curr_key: |
| notes.append(f"key column {prev_key or 'none'} -> {curr_key}") |
| row_changes = compare_row_signatures( |
| {str(k): str(v) for k, v in dict(previous.get("tracked_rows", {})).items()}, |
| {str(k): str(v) for k, v in dict(current.get("tracked_rows", {})).items()}, |
| ) |
| if row_changes: |
| notes.append(f"tracked rows: {row_changes}") |
| if not notes: |
| notes.append("content changed; structural summary unchanged") |
| return notes |
|
|
|
|
| def summarize_xlsx_change(previous: dict[str, object], current: dict[str, object]) -> list[str]: |
| notes: list[str] = [] |
| prev_sheets = {str(sheet.get("name", "")): sheet for sheet in previous.get("sheets", []) if sheet.get("name")} |
| curr_sheets = {str(sheet.get("name", "")): sheet for sheet in current.get("sheets", []) if sheet.get("name")} |
| added_sheets, removed_sheets = compare_named_lists(list(prev_sheets), list(curr_sheets)) |
| if added_sheets: |
| notes.append(f"sheets added: {', '.join(added_sheets[:4])}") |
| if removed_sheets: |
| notes.append(f"sheets removed: {', '.join(removed_sheets[:4])}") |
| for sheet_name in sorted(prev_sheets.keys() & curr_sheets.keys())[:4]: |
| before = prev_sheets[sheet_name] |
| after = curr_sheets[sheet_name] |
| sheet_notes: list[str] = [] |
| if int(before.get("row_count", 0) or 0) != int(after.get("row_count", 0) or 0): |
| sheet_notes.append(f"rows {before.get('row_count', 0)} -> {after.get('row_count', 0)}") |
| if int(before.get("column_count", 0) or 0) != int(after.get("column_count", 0) or 0): |
| sheet_notes.append(f"cols {before.get('column_count', 0)} -> {after.get('column_count', 0)}") |
| added_headers, removed_headers = compare_named_lists( |
| [str(item) for item in before.get("headers", [])], |
| [str(item) for item in after.get("headers", [])], |
| ) |
| if added_headers: |
| sheet_notes.append(f"headers +{', '.join(added_headers[:3])}") |
| if removed_headers: |
| sheet_notes.append(f"headers -{', '.join(removed_headers[:3])}") |
| row_changes = compare_row_signatures( |
| {str(k): str(v) for k, v in dict(before.get("tracked_rows", {})).items()}, |
| {str(k): str(v) for k, v in dict(after.get("tracked_rows", {})).items()}, |
| ) |
| if row_changes: |
| sheet_notes.append(row_changes) |
| if sheet_notes: |
| notes.append(f"sheet {sheet_name}: {'; '.join(sheet_notes)}") |
| if not notes: |
| notes.append("content changed; workbook structure summary unchanged") |
| return notes |
|
|
|
|
| def summarize_change(previous_entry: dict[str, object] | None, current_payload: dict[str, object]) -> list[str]: |
| if not previous_entry: |
| return [] |
| previous_parser = str(previous_entry.get("parser", "")) |
| current_parser = str(current_payload.get("parser", "")) |
| previous_meta = dict(previous_entry.get("metadata", {})) |
| current_meta = dict(current_payload.get("metadata", {})) |
| if previous_parser == current_parser == "csv-local": |
| return summarize_csv_change(previous_meta, current_meta) |
| if previous_parser == current_parser == "xlsx-local": |
| return summarize_xlsx_change(previous_meta, current_meta) |
| return [f"summary: {previous_entry.get('summary', 'unknown')} -> {current_payload.get('summary', 'unknown')}"] |
|
|
|
|
| def summarize_file(path: Path) -> dict[str, object]: |
| ext = path.suffix.lower() |
| if ext in {".csv", ".tsv"}: |
| return summarize_csv(path) |
| if ext in {".xlsx", ".xlsm"}: |
| return summarize_xlsx(path) |
| if ext == ".xls": |
| return summarize_xls_legacy(path) |
| if ext == ".docx": |
| return summarize_docx(path) |
| if ext == ".pptx": |
| return summarize_pptx(path) |
| if ext == ".pdf": |
| return summarize_pdf(path) |
| if ext in {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".svg"}: |
| return summarize_image(path) |
| if ext in {".zip", ".rar", ".7z", ".tar", ".gz"} or path.name.endswith(".tar.gz"): |
| return summarize_archive(path) |
| return summarize_plaintext(path) |
|
|
|
|
| def load_manifest() -> list[dict[str, str]]: |
| if not MANIFEST.exists(): |
| raise FileNotFoundError(f"manifest missing: {MANIFEST}") |
| with MANIFEST.open("r", encoding="utf-8-sig", newline="") as handle: |
| reader = csv.DictReader(handle) |
| if reader.fieldnames != EXPECTED_COLUMNS: |
| raise ValueError(f"manifest columns mismatch: expected {EXPECTED_COLUMNS}, got {reader.fieldnames}") |
| return [{key: (value or "") for key, value in row.items()} for row in reader] |
|
|
|
|
| def write_manifest(rows: list[dict[str, str]]) -> None: |
| MANIFEST.parent.mkdir(parents=True, exist_ok=True) |
| with MANIFEST.open("w", encoding="utf-8", newline="") as handle: |
| writer = csv.DictWriter(handle, fieldnames=EXPECTED_COLUMNS) |
| writer.writeheader() |
| for row in rows: |
| writer.writerow({key: row.get(key, "") for key in EXPECTED_COLUMNS}) |
|
|
|
|
| def load_lock() -> dict[str, object]: |
| if not LOCK_FILE.exists(): |
| return {"files": {}} |
| try: |
| return json.loads(LOCK_FILE.read_text(encoding="utf-8")) |
| except Exception: |
| return {"files": {}} |
|
|
|
|
| def write_lock(payload: dict[str, object]) -> None: |
| LOCK_FILE.parent.mkdir(parents=True, exist_ok=True) |
| LOCK_FILE.write_text(json.dumps(payload, ensure_ascii=True, indent=2, sort_keys=True) + "\n", encoding="utf-8") |
|
|
|
|
| def next_source_id(existing_ids: set[str], content_hash: str) -> str: |
| base = f"src_{content_hash[:10]}" |
| candidate = base |
| index = 2 |
| while candidate in existing_ids: |
| candidate = f"{base}_{index}" |
| index += 1 |
| return candidate |
|
|
|
|
| def build_report( |
| raw_root: Path, |
| rows: list[dict[str, str]], |
| kinds: Counter[str], |
| new_paths: list[str], |
| changed_paths: list[str], |
| archived_paths: list[str], |
| duplicate_paths: list[str], |
| change_summaries: dict[str, list[str]], |
| ) -> str: |
| def bullets(items: list[str], *, details: dict[str, list[str]] | None = None) -> str: |
| if not items: |
| return "- none\n" |
| lines: list[str] = [] |
| for item in items[:20]: |
| detail_lines = (details or {}).get(item, []) |
| if detail_lines: |
| lines.append(f"- `{item}` — {detail_lines[0]}\n") |
| for extra in detail_lines[1:3]: |
| lines.append(f" - {extra}\n") |
| else: |
| lines.append(f"- `{item}`\n") |
| return "".join(lines) |
|
|
| lines = [ |
| "# Raw Intake Report", |
| "", |
| f"- generated_at: `{utc_now()}`", |
| f"- raw_root: `{raw_root}`", |
| f"- manifest_rows: `{len(rows)}`", |
| f"- new: `{len(new_paths)}`", |
| f"- changed: `{len(changed_paths)}`", |
| f"- archived: `{len(archived_paths)}`", |
| f"- duplicates: `{len(duplicate_paths)}`", |
| "", |
| "## Kind Summary", |
| "", |
| ] |
| if kinds: |
| for kind, count in sorted(kinds.items()): |
| lines.append(f"- `{kind}`: {count}") |
| else: |
| lines.append("- none") |
| lines.extend([ |
| "", |
| "## New Files", |
| "", |
| bullets(sorted(new_paths)), |
| "", |
| "## Changed Files", |
| "", |
| bullets(sorted(changed_paths), details=change_summaries), |
| "", |
| "## Archived Files", |
| "", |
| bullets(sorted(archived_paths)), |
| "", |
| "## Duplicate Files", |
| "", |
| bullets(sorted(duplicate_paths)), |
| ]) |
| return "\n".join(lines).rstrip() + "\n" |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description="Scan a local raw root, update the manifest, and record low-cost structural metadata.") |
| parser.add_argument("--raw-root", default=os.environ.get("PROJECT_RAW_ROOT", str(DEFAULT_RAW_ROOT)), help="Local raw root path") |
| parser.add_argument("--report-file", default=str(REPORT_FILE), help="Markdown report output path") |
| parser.add_argument("--dry-run", action="store_true", help="Preview changes without writing manifest or lock files") |
| args = parser.parse_args() |
|
|
| raw_root = Path(args.raw_root).expanduser().resolve() |
| if not raw_root.exists(): |
| print(f"ingest_raw: raw root does not exist: {raw_root}") |
| return 1 |
|
|
| rows = load_manifest() |
| previous_lock = load_lock().get("files", {}) |
| rows_by_path = {row["raw_rel_path"]: row for row in rows if row.get("raw_rel_path")} |
| existing_ids = {row["source_id"] for row in rows if row.get("source_id")} |
| seen_paths: set[str] = set() |
| kinds: Counter[str] = Counter() |
| new_paths: list[str] = [] |
| changed_paths: list[str] = [] |
| archived_paths: list[str] = [] |
| lock_entries: dict[str, object] = {} |
| hash_to_primary: dict[str, str] = {} |
| duplicate_paths: list[str] = [] |
| change_summaries: dict[str, list[str]] = {} |
|
|
| candidates: list[Path] = [] |
| for dirpath, dirnames, filenames in os.walk(raw_root): |
| dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS and not d.startswith(".")] |
| for name in filenames: |
| if name.startswith("."): |
| continue |
| path = Path(dirpath) / name |
| ext = path.suffix.lower() |
| if ext in TRACKED_EXTENSIONS or path.name.endswith(".tar.gz"): |
| candidates.append(path) |
|
|
| for path in sorted(candidates): |
| rel = path.relative_to(raw_root).as_posix() |
| seen_paths.add(rel) |
| content_hash = sha256_prefix(path) |
| kind = detect_kind(path) |
| kinds[kind] += 1 |
| summary_payload = summarize_file(path) |
| existing = rows_by_path.get(rel) |
| old_hash = None |
| previous_entry = None |
| if isinstance(previous_lock, dict) and rel in previous_lock: |
| previous_entry = previous_lock[rel] |
| old_hash = previous_entry.get("content_hash") |
|
|
| if existing is None: |
| source_id = next_source_id(existing_ids, content_hash) |
| existing_ids.add(source_id) |
| row = { |
| "source_id": source_id, |
| "company": "", |
| "vendor": "", |
| "kind": kind, |
| "filename": path.name, |
| "raw_rel_path": rel, |
| "status": "new", |
| "compiled_into": "", |
| "notes": "", |
| } |
| rows.append(row) |
| rows_by_path[rel] = row |
| new_paths.append(rel) |
| existing = row |
| else: |
| existing["kind"] = kind |
| existing["filename"] = path.name |
| if old_hash and old_hash != content_hash and existing.get("status") != "archived": |
| existing["status"] = "new" |
| changed_paths.append(rel) |
| change_summaries[rel] = summarize_change(previous_entry, summary_payload) |
|
|
| primary = hash_to_primary.setdefault(content_hash, rel) |
| duplicate_of = None if primary == rel else rows_by_path[primary]["source_id"] |
| if duplicate_of: |
| duplicate_paths.append(rel) |
|
|
| lock_entries[rel] = { |
| "source_id": existing["source_id"], |
| "filename": path.name, |
| "kind": kind, |
| "content_hash": content_hash, |
| "size_bytes": path.stat().st_size, |
| "modified_at": datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc).replace(microsecond=0).isoformat(), |
| "parser": summary_payload["parser"], |
| "summary": summary_payload["summary"], |
| "metadata": summary_payload["metadata"], |
| "duplicate_of": duplicate_of, |
| "previous_content_hash": old_hash or "", |
| "change_summary": change_summaries.get(rel, []), |
| } |
|
|
| for row in rows: |
| rel = row.get("raw_rel_path", "") |
| if rel and rel not in seen_paths and row.get("status") != "archived": |
| row["status"] = "archived" |
| archived_paths.append(rel) |
|
|
| rows.sort(key=lambda row: (row.get("status", ""), row.get("raw_rel_path", ""))) |
| report_text = build_report(raw_root, rows, kinds, new_paths, changed_paths, archived_paths, duplicate_paths, change_summaries) |
|
|
| if args.dry_run: |
| print("ingest_raw: DRY RUN") |
| print(report_text) |
| return 0 |
|
|
| write_manifest(rows) |
| write_lock({ |
| "llm_wiki_version": "1.2.2", |
| "generated_at": utc_now(), |
| "raw_root": str(raw_root), |
| "summary": { |
| "tracked_files": len(candidates), |
| "manifest_rows": len(rows), |
| "new": len(new_paths), |
| "changed": len(changed_paths), |
| "archived": len(archived_paths), |
| "duplicates": len(duplicate_paths), |
| "kinds": dict(sorted(kinds.items())), |
| }, |
| "files": lock_entries, |
| }) |
| report_path = Path(args.report_file).expanduser().resolve() |
| report_path.parent.mkdir(parents=True, exist_ok=True) |
| report_path.write_text(report_text, encoding="utf-8") |
|
|
| print(f"ingest_raw: OK ({len(candidates)} tracked file(s))") |
| print(f"- manifest: {MANIFEST}") |
| print(f"- lock: {LOCK_FILE}") |
| print(f"- report: {report_path}") |
| print(f"- new: {len(new_paths)} | changed: {len(changed_paths)} | archived: {len(archived_paths)} | duplicates: {len(duplicate_paths)}") |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
