732642856
1326 lines
52 KiB
Python
1326 lines
52 KiB
Python
#!/usr/bin/env python3
|
|
"""Build a lightweight character-level inventory manifest for Jyotish sources.
|
|
|
|
This is an index and governance manifest, not a semantic promotion pass. It
|
|
hashes every in-scope file, extracts text metadata when cheap, and queues heavy
|
|
PDF/image/document extraction instead of running expensive OCR by default.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import zipfile
|
|
from datetime import datetime, timezone
|
|
from xml.etree import ElementTree
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
REPORT_JSON = ROOT / "docs" / "research" / "character_level_inventory_manifest_latest.json"
|
|
REPORT_MD = ROOT / "docs" / "research" / "character_level_inventory_manifest_latest.md"
|
|
EXTERNAL_REPORT_JSON = ROOT / "docs" / "research" / "character_level_external_manifest_latest.json"
|
|
EXTERNAL_REPORT_MD = ROOT / "docs" / "research" / "character_level_external_manifest_latest.md"
|
|
EXTRACTION_QUEUE_JSON = ROOT / "docs" / "research" / "character_level_extraction_queue_latest.json"
|
|
EXTRACTION_QUEUE_MD = ROOT / "docs" / "research" / "character_level_extraction_queue_latest.md"
|
|
EXTRACTION_RESULTS_JSON = ROOT / "docs" / "research" / "character_level_extraction_results_latest.json"
|
|
EXTRACTION_RESULTS_MD = ROOT / "docs" / "research" / "character_level_extraction_results_latest.md"
|
|
EXTRACTION_REVIEW_JSON = ROOT / "docs" / "research" / "character_level_extraction_review_latest.json"
|
|
EXTRACTION_REVIEW_MD = ROOT / "docs" / "research" / "character_level_extraction_review_latest.md"
|
|
SOURCE_GRADING_BATCH1_JSON = ROOT / "docs" / "research" / "character_level_source_grading_batch1_latest.json"
|
|
SOURCE_GRADING_BATCH1_MD = ROOT / "docs" / "research" / "character_level_source_grading_batch1_latest.md"
|
|
VISION_OCR_SOURCE = r'''
|
|
import Foundation
|
|
import Vision
|
|
import AppKit
|
|
|
|
let args = CommandLine.arguments
|
|
if args.count < 2 {
|
|
fputs("usage: vision_ocr image\n", stderr)
|
|
exit(2)
|
|
}
|
|
let url = URL(fileURLWithPath: args[1])
|
|
guard let image = NSImage(contentsOf: url), let tiff = image.tiffRepresentation,
|
|
let bitmap = NSBitmapImageRep(data: tiff), let cgImage = bitmap.cgImage else {
|
|
fputs("cannot load image\n", stderr)
|
|
exit(3)
|
|
}
|
|
let request = VNRecognizeTextRequest { request, error in
|
|
if let error = error {
|
|
fputs("vision error: \(error)\n", stderr)
|
|
exit(4)
|
|
}
|
|
let observations = request.results as? [VNRecognizedTextObservation] ?? []
|
|
for obs in observations {
|
|
if let top = obs.topCandidates(1).first {
|
|
print(top.string)
|
|
}
|
|
}
|
|
}
|
|
request.recognitionLevel = .accurate
|
|
request.usesLanguageCorrection = true
|
|
if #available(macOS 11.0, *) {
|
|
request.recognitionLanguages = ["zh-Hans", "en-US"]
|
|
}
|
|
let handler = VNImageRequestHandler(cgImage: cgImage, options: [:])
|
|
do {
|
|
try handler.perform([request])
|
|
} catch {
|
|
fputs("perform error: \(error)\n", stderr)
|
|
exit(5)
|
|
}
|
|
'''
|
|
|
|
PROJECT_SCAN_ROOTS = [
|
|
"references",
|
|
"references/open_source_sources",
|
|
"docs/research",
|
|
"SKILL.md",
|
|
"AGENTS.md",
|
|
]
|
|
|
|
EXTERNAL_SCAN_ROOTS = [
|
|
Path.home() / "Downloads",
|
|
Path.home() / "Desktop",
|
|
Path.home() / "文件仓库" / "印度占星文章",
|
|
Path.home() / ".workbuddy" / "skills" / "jyotish-vedic-astrology",
|
|
Path.home() / "WorkBuddy" / "2026-06-09-20-03-34" / "jyotish-fragments",
|
|
Path.home() / "WorkBuddy" / "2026-06-10-21-30-47",
|
|
Path.home() / "WorkBuddy" / "2026-06-12-15-22-12",
|
|
Path.home() / "WorkBuddy" / "20260422235041",
|
|
Path.home() / "WorkBuddy" / "20260503121822",
|
|
Path.home() / "WorkBuddy" / "engines-repo" / "jyotish",
|
|
Path.home() / "engines-repo" / "jyotish",
|
|
Path.home() / "Documents" / "ObsidianVault" / "03_研究_术数占星",
|
|
]
|
|
|
|
EXTERNAL_KEYWORDS = [
|
|
"印度占星",
|
|
"占星",
|
|
"jyotish",
|
|
"vedic",
|
|
"astrology",
|
|
"jhora",
|
|
"pyjhora",
|
|
"vedastro",
|
|
"dasha",
|
|
"shadbala",
|
|
"tajika",
|
|
"jaimini",
|
|
]
|
|
|
|
EXCLUDED_PARTS = {
|
|
".git",
|
|
".pytest_cache",
|
|
".ruff_cache",
|
|
"__pycache__",
|
|
"node_modules",
|
|
"venv",
|
|
"venv_vedastro",
|
|
".venv",
|
|
"build",
|
|
"dist",
|
|
}
|
|
|
|
TEXT_EXTENSIONS = {
|
|
".css",
|
|
".csv",
|
|
".html",
|
|
".ini",
|
|
".js",
|
|
".json",
|
|
".jsonl",
|
|
".jsx",
|
|
".mjs",
|
|
".md",
|
|
".py",
|
|
".sh",
|
|
".toml",
|
|
".ts",
|
|
".tsx",
|
|
".txt",
|
|
".yaml",
|
|
".yml",
|
|
}
|
|
PDF_EXTENSIONS = {".pdf"}
|
|
IMAGE_EXTENSIONS = {".bmp", ".gif", ".jpeg", ".jpg", ".png", ".tif", ".tiff", ".webp"}
|
|
DOCUMENT_EXTENSIONS = {".doc", ".docx", ".epub", ".mobi", ".pages", ".ppt", ".pptx", ".rtf", ".xls", ".xlsx"}
|
|
QUEUED_EXTRACTION_STATUSES = {
|
|
"pdf_text_extraction_queued",
|
|
"image_ocr_queued",
|
|
"document_text_extraction_queued",
|
|
}
|
|
|
|
|
|
def _relative(path: Path) -> str:
|
|
return str(path.relative_to(ROOT))
|
|
|
|
|
|
def _should_skip(path: Path) -> bool:
|
|
try:
|
|
rel_parts = path.relative_to(ROOT).parts
|
|
except ValueError:
|
|
rel_parts = path.parts
|
|
return any(part in EXCLUDED_PARTS for part in rel_parts)
|
|
|
|
|
|
def _iter_project_files() -> list[Path]:
|
|
files: set[Path] = set()
|
|
for root_name in PROJECT_SCAN_ROOTS:
|
|
root = ROOT / root_name
|
|
if not root.exists() or _should_skip(root):
|
|
continue
|
|
if root.is_file():
|
|
files.add(root)
|
|
else:
|
|
files.update(path for path in root.rglob("*") if path.is_file() and not _should_skip(path))
|
|
return sorted(files)
|
|
|
|
|
|
def _is_external_relevant(path: Path) -> bool:
|
|
lowered = str(path).lower()
|
|
return any(keyword.lower() in lowered for keyword in EXTERNAL_KEYWORDS)
|
|
|
|
|
|
def _iter_external_files() -> list[Path]:
|
|
files: set[Path] = set()
|
|
for root in EXTERNAL_SCAN_ROOTS:
|
|
if not root.exists() or _should_skip(root):
|
|
continue
|
|
if root.is_file():
|
|
if _is_external_relevant(root):
|
|
files.add(root)
|
|
continue
|
|
for path in root.rglob("*"):
|
|
if path.is_file() and not _should_skip(path) and _is_external_relevant(path):
|
|
files.add(path)
|
|
return sorted(files)
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _text_stats(path: Path) -> dict[str, Any]:
|
|
raw = path.read_bytes()
|
|
decoded = raw.decode("utf-8", errors="replace")
|
|
replacement_count = decoded.count("\ufffd")
|
|
return {
|
|
"extraction_status": "text_decode_lossy" if replacement_count else "text_indexed",
|
|
"character_count": len(decoded),
|
|
"line_count": decoded.count("\n") + (1 if decoded else 0),
|
|
"decode_replacement_count": replacement_count,
|
|
}
|
|
|
|
|
|
def _extraction_status(path: Path) -> dict[str, Any]:
|
|
suffix = path.suffix.lower()
|
|
if suffix in TEXT_EXTENSIONS:
|
|
return _text_stats(path)
|
|
if suffix in PDF_EXTENSIONS:
|
|
return {
|
|
"extraction_status": "pdf_text_extraction_queued",
|
|
"character_count": None,
|
|
"line_count": None,
|
|
"decode_replacement_count": None,
|
|
}
|
|
if suffix in IMAGE_EXTENSIONS:
|
|
return {
|
|
"extraction_status": "image_ocr_queued",
|
|
"character_count": None,
|
|
"line_count": None,
|
|
"decode_replacement_count": None,
|
|
}
|
|
if suffix in DOCUMENT_EXTENSIONS:
|
|
return {
|
|
"extraction_status": "document_text_extraction_queued",
|
|
"character_count": None,
|
|
"line_count": None,
|
|
"decode_replacement_count": None,
|
|
}
|
|
return {
|
|
"extraction_status": "binary_indexed",
|
|
"character_count": None,
|
|
"line_count": None,
|
|
"decode_replacement_count": None,
|
|
}
|
|
|
|
|
|
def _classify_path(path: str) -> dict[str, str]:
|
|
if path.startswith("references/open_source_sources/rishi-ai-mcp/"):
|
|
return {
|
|
"classification": "open_source_reference",
|
|
"priority": "priority_1",
|
|
"promotion_status": "reference_layer_candidate",
|
|
"reason": "User-prioritized open-source workflow corpus; requires license-aware review.",
|
|
}
|
|
if path.startswith("references/open_source_sources/vedic-astro-skills/"):
|
|
return {
|
|
"classification": "open_source_reference",
|
|
"priority": "priority_1",
|
|
"promotion_status": "reference_layer_candidate",
|
|
"reason": "User-prioritized Jyotish skill corpus; requires selective source promotion.",
|
|
}
|
|
if path.startswith("references/open_source_sources/"):
|
|
return {
|
|
"classification": "open_source_reference",
|
|
"priority": "priority_2",
|
|
"promotion_status": "reference_layer_candidate",
|
|
"reason": "Open-source corpus; index first, promote only after license and conflict review.",
|
|
}
|
|
if path.startswith("references/real_case_studies/"):
|
|
return {
|
|
"classification": "real_case_calibration",
|
|
"priority": "priority_1",
|
|
"promotion_status": "reference_layer_candidate",
|
|
"reason": "Real-case material for calibration; not direct rule truth.",
|
|
}
|
|
if path.startswith("references/oracle/"):
|
|
return {
|
|
"classification": "oracle_artifact",
|
|
"priority": "priority_2",
|
|
"promotion_status": "oracle_evidence_only",
|
|
"reason": "Oracle artifact; use for parity evidence, not interpretive rules.",
|
|
}
|
|
if path.startswith("references/"):
|
|
return {
|
|
"classification": "reference_candidate",
|
|
"priority": "priority_1",
|
|
"promotion_status": "reference_layer_candidate",
|
|
"reason": "Local reference source candidate; requires content-level review before promotion.",
|
|
}
|
|
if path.startswith("docs/research/local_drafts/"):
|
|
return {
|
|
"classification": "quarantined_draft",
|
|
"priority": "priority_3",
|
|
"promotion_status": "not_truth_source",
|
|
"reason": "Historical draft; do not promote without explicit review.",
|
|
}
|
|
if path.startswith("docs/research/"):
|
|
return {
|
|
"classification": "research_governance",
|
|
"priority": "priority_3",
|
|
"promotion_status": "governance_or_history",
|
|
"reason": "Research history or governance record; index separately from rule truth.",
|
|
}
|
|
return {
|
|
"classification": "project_governance",
|
|
"priority": "priority_3",
|
|
"promotion_status": "governance_or_history",
|
|
"reason": "Project-level governance file.",
|
|
}
|
|
|
|
|
|
def _root_bucket(path: str) -> str:
|
|
if path.startswith("references/open_source_sources/"):
|
|
return "references/open_source_sources"
|
|
if path.startswith("references/"):
|
|
return "references"
|
|
if path.startswith("docs/research/"):
|
|
return "docs/research"
|
|
return path.split("/", 1)[0]
|
|
|
|
|
|
def _external_path(path: Path) -> str:
|
|
return str(path)
|
|
|
|
|
|
def _classify_external_path(path: str) -> dict[str, str]:
|
|
suffix = Path(path).suffix.lower()
|
|
if "/.workbuddy/skills/jyotish-vedic-astrology/" in path:
|
|
return {
|
|
"classification": "external_skill_fragment",
|
|
"priority": "priority_1",
|
|
"promotion_status": "external_candidate",
|
|
"reason": "External Jyotish skill copy; index only until diffed against repo truth.",
|
|
}
|
|
if "/engines-repo/jyotish/" in path or "/WorkBuddy/engines-repo/jyotish/" in path:
|
|
return {
|
|
"classification": "external_engine_fragment",
|
|
"priority": "priority_1",
|
|
"promotion_status": "external_candidate",
|
|
"reason": "External engine fragment; requires source diff and license boundary review.",
|
|
}
|
|
if "/WorkBuddy/" in path or "/Desktop/" in path:
|
|
return {
|
|
"classification": "external_historical_report",
|
|
"priority": "priority_2",
|
|
"promotion_status": "external_reference_only",
|
|
"reason": "Historical report or work artifact; do not promote without privacy review.",
|
|
}
|
|
if suffix in PDF_EXTENSIONS or suffix in DOCUMENT_EXTENSIONS or "/文件仓库/印度占星文章/" in path:
|
|
return {
|
|
"classification": "external_book_or_document",
|
|
"priority": "priority_1",
|
|
"promotion_status": "external_reference_only",
|
|
"reason": "External book/document source; requires extraction and source grading before use.",
|
|
}
|
|
return {
|
|
"classification": "external_archive_or_binary",
|
|
"priority": "priority_3",
|
|
"promotion_status": "external_index_only",
|
|
"reason": "External relevant asset; indexed without content promotion.",
|
|
}
|
|
|
|
|
|
def _external_root_bucket(path: str) -> str:
|
|
home = str(Path.home())
|
|
if path.startswith(home + "/Downloads/"):
|
|
return "~/Downloads"
|
|
if path.startswith(home + "/Desktop/"):
|
|
return "~/Desktop"
|
|
if path.startswith(home + "/文件仓库/"):
|
|
return "~/文件仓库"
|
|
if path.startswith(home + "/.workbuddy/"):
|
|
return "~/.workbuddy"
|
|
if path.startswith(home + "/WorkBuddy/"):
|
|
return "~/WorkBuddy"
|
|
if path.startswith(home + "/engines-repo/"):
|
|
return "~/engines-repo"
|
|
if path.startswith(home + "/Documents/ObsidianVault/"):
|
|
return "~/Documents/ObsidianVault"
|
|
return path
|
|
|
|
|
|
def _build_index(
|
|
*,
|
|
files: list[Path],
|
|
path_label,
|
|
classify,
|
|
root_bucket,
|
|
) -> tuple[dict[str, dict[str, Any]], dict[str, Any], dict[str, dict[str, int]]]:
|
|
by_path: dict[str, dict[str, Any]] = {}
|
|
classification_counts: dict[str, int] = {}
|
|
extraction_counts: dict[str, int] = {}
|
|
root_summary: dict[str, dict[str, int]] = {}
|
|
unclassified_files = 0
|
|
unknown_extraction_status = 0
|
|
|
|
for file_path in files:
|
|
rel = path_label(file_path)
|
|
file_class = classify(rel)
|
|
extraction = _extraction_status(file_path)
|
|
if not file_class.get("classification"):
|
|
unclassified_files += 1
|
|
if not extraction.get("extraction_status"):
|
|
unknown_extraction_status += 1
|
|
byte_count = file_path.stat().st_size
|
|
item = {
|
|
"path": rel,
|
|
"suffix": file_path.suffix.lower(),
|
|
"byte_count": byte_count,
|
|
"sha256": _sha256(file_path),
|
|
**extraction,
|
|
**file_class,
|
|
}
|
|
by_path[rel] = item
|
|
classification_counts[item["classification"]] = classification_counts.get(item["classification"], 0) + 1
|
|
extraction_counts[item["extraction_status"]] = extraction_counts.get(item["extraction_status"], 0) + 1
|
|
bucket = root_bucket(rel)
|
|
root_summary.setdefault(bucket, {"files": 0, "bytes": 0})
|
|
root_summary[bucket]["files"] += 1
|
|
root_summary[bucket]["bytes"] += byte_count
|
|
|
|
summary = {
|
|
"total_files": len(by_path),
|
|
"unhashed_files": sum(1 for item in by_path.values() if not item.get("sha256")),
|
|
"unclassified_files": unclassified_files,
|
|
"unknown_extraction_status": unknown_extraction_status,
|
|
"classification_counts": dict(sorted(classification_counts.items())),
|
|
"extraction_status_counts": dict(sorted(extraction_counts.items())),
|
|
}
|
|
return by_path, summary, dict(sorted(root_summary.items()))
|
|
|
|
|
|
def build_manifest(*, scope: str = "project", write: bool = True) -> dict[str, Any]:
|
|
if scope == "project":
|
|
files = _iter_project_files()
|
|
path_label = _relative
|
|
classify = _classify_path
|
|
root_bucket = _root_bucket
|
|
scan_roots = PROJECT_SCAN_ROOTS
|
|
json_report = REPORT_JSON
|
|
markdown_report = REPORT_MD
|
|
title = "Character-Level Inventory Manifest"
|
|
mode = {
|
|
"heavy_ocr": False,
|
|
"whole_machine_scan": False,
|
|
"external_high_relevance_scan": False,
|
|
"semantic_promotion": False,
|
|
"boundary": "Fast manifest only: hashes and cheap text stats now; heavy OCR and whole-machine scan are queued.",
|
|
}
|
|
elif scope == "external":
|
|
files = _iter_external_files()
|
|
path_label = _external_path
|
|
classify = _classify_external_path
|
|
root_bucket = _external_root_bucket
|
|
scan_roots = [str(path) for path in EXTERNAL_SCAN_ROOTS]
|
|
json_report = EXTERNAL_REPORT_JSON
|
|
markdown_report = EXTERNAL_REPORT_MD
|
|
title = "External High-Relevance Inventory Manifest"
|
|
mode = {
|
|
"heavy_ocr": False,
|
|
"whole_machine_scan": False,
|
|
"external_high_relevance_scan": True,
|
|
"semantic_promotion": False,
|
|
"boundary": "External scan is high-relevance only. It records hashes and extraction states without copying private source text into repo truth.",
|
|
}
|
|
elif scope == "extraction-queue":
|
|
return build_extraction_queue(write=write)
|
|
elif scope == "extraction-results":
|
|
return build_extraction_results(write=write)
|
|
elif scope == "extraction-review":
|
|
return build_extraction_review(write=write)
|
|
elif scope == "source-grading-batch1":
|
|
return build_source_grading_batch1(write=write)
|
|
else:
|
|
raise ValueError(f"Unsupported scope: {scope}")
|
|
|
|
by_path, summary, root_summary = _build_index(
|
|
files=files,
|
|
path_label=path_label,
|
|
classify=classify,
|
|
root_bucket=root_bucket,
|
|
)
|
|
status = (
|
|
"pass"
|
|
if summary["unhashed_files"] == 0
|
|
and summary["unclassified_files"] == 0
|
|
and summary["unknown_extraction_status"] == 0
|
|
else "fail"
|
|
)
|
|
report = {
|
|
"scope": scope,
|
|
"status": status,
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"mode": mode,
|
|
"scan_roots": scan_roots,
|
|
"summary": summary,
|
|
"root_summary": root_summary,
|
|
"by_path": by_path,
|
|
"artifacts": {
|
|
"json_report": _relative(json_report),
|
|
"markdown_report": _relative(markdown_report),
|
|
},
|
|
"title": title,
|
|
}
|
|
if write:
|
|
_write_reports(report, json_report=json_report, markdown_report=markdown_report)
|
|
return report
|
|
|
|
|
|
def _source_grading_batch1_items(reviews: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
candidates = [item for item in reviews if item["decision"] == "promote_review_candidate"]
|
|
doc_pdf = [
|
|
item
|
|
for item in candidates
|
|
if item["extraction_method"] in {"docx", "pdfplumber", "pypdf"}
|
|
]
|
|
high_signal_ocr = [
|
|
item
|
|
for item in candidates
|
|
if item["extraction_method"] == "macos_vision" and item["extracted_character_count"] >= 2500
|
|
]
|
|
selected = doc_pdf[:10] + high_signal_ocr[:2]
|
|
return selected[:12]
|
|
|
|
|
|
def _source_grade(item: dict[str, Any]) -> tuple[str, str]:
|
|
path = str(item["path"])
|
|
if "/Desktop/" in path or "/WorkBuddy/" in path:
|
|
return "quarantine_or_private", "Private or historical personal report; keep out of reference pack."
|
|
if item["extraction_method"] == "macos_vision":
|
|
if item["extracted_character_count"] >= 2500:
|
|
return "reference_only", "OCR text is high-signal but still requires manual verification against image."
|
|
return "ocr_low_confidence_reference_only", "OCR text too short or noisy for promotion review."
|
|
if "jyotish_training" in path:
|
|
return "reference_only", "Training material can inform style/workflow but cannot outrank source rules."
|
|
if "/文件仓库/印度占星文章/" in path or path.endswith("印度占星.pdf") or path.endswith("private_chart_reference.pdf"):
|
|
return "promote_to_reference_pack_candidate", "Source-like document; candidate for reference pack after conflict arbitration."
|
|
return "reference_only", "Useful context, not first-order rule source."
|
|
|
|
|
|
def _arbitrate_batch1_source(item: dict[str, Any], grade: str) -> dict[str, Any]:
|
|
if grade == "promote_to_reference_pack_candidate":
|
|
pack_decision = "reference_pack_candidate"
|
|
status = "batch1_arbitrated"
|
|
reason = (
|
|
"Document/PDF source lineage is clear enough for reference/source pack candidacy; "
|
|
"still cannot enter runtime truth until source-pack visibility tests bind it explicitly."
|
|
)
|
|
else:
|
|
pack_decision = "reference_only"
|
|
status = "batch1_arbitrated"
|
|
reason = "Kept as reference-only because the source is OCR-derived, training-oriented, private, or not first-order rule material."
|
|
return {
|
|
"status": status,
|
|
"pack_decision": pack_decision,
|
|
"reason": reason,
|
|
"runtime_truth_allowed": False,
|
|
"runtime_truth_reason": "Batch 1 arbitrates source-pack eligibility only; it does not promote runtime truth.",
|
|
}
|
|
|
|
|
|
def build_source_grading_batch1(*, write: bool = True) -> dict[str, Any]:
|
|
review_report = build_extraction_review(write=False)
|
|
total_review_candidates = sum(1 for item in review_report["reviews"] if item["decision"] == "promote_review_candidate")
|
|
selected = _source_grading_batch1_items(review_report["reviews"])
|
|
graded_sources: list[dict[str, Any]] = []
|
|
for item in selected:
|
|
grade, reason = _source_grade(item)
|
|
arbitration = _arbitrate_batch1_source(item, grade)
|
|
graded_sources.append(
|
|
{
|
|
"path": item["path"],
|
|
"source_scope": item["source_scope"],
|
|
"suffix": item["suffix"],
|
|
"byte_count": item["byte_count"],
|
|
"sha256": item["sha256"],
|
|
"extraction_method": item["extraction_method"],
|
|
"extracted_character_count": item["extracted_character_count"],
|
|
"text_sha256": item.get("text_sha256"),
|
|
"review_decision": item["decision"],
|
|
"source_grade": grade,
|
|
"grade_reason": reason,
|
|
"conflict_arbitration": arbitration,
|
|
"pack_decision": arbitration["pack_decision"],
|
|
"requires_conflict_arbitration": False,
|
|
"requires_source_pack_visibility_test": True,
|
|
"runtime_truth_allowed": False,
|
|
}
|
|
)
|
|
grade_counts: dict[str, int] = {}
|
|
pack_decision_counts: dict[str, int] = {}
|
|
for item in graded_sources:
|
|
grade_counts[item["source_grade"]] = grade_counts.get(item["source_grade"], 0) + 1
|
|
pack_decision_counts[item["pack_decision"]] = pack_decision_counts.get(item["pack_decision"], 0) + 1
|
|
stored_text_payload_fields = sum(1 for item in graded_sources if "text" in item or "text_preview" in item)
|
|
reference_pack_ready_count = pack_decision_counts.get("reference_pack_candidate", 0)
|
|
report = {
|
|
"scope": "source-grading-batch1",
|
|
"status": "pass",
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"mode": {
|
|
"runtime_truth_promotion": False,
|
|
"semantic_promotion": False,
|
|
"boundary": "Batch 1 grades extracted candidates only; no source enters runtime truth without a later source-pack test.",
|
|
},
|
|
"summary": {
|
|
"total_review_candidates": total_review_candidates,
|
|
"selected_count": len(graded_sources),
|
|
"graded_files": len(graded_sources),
|
|
"runtime_promotions": 0,
|
|
"stored_text_payload_fields": stored_text_payload_fields,
|
|
},
|
|
"grade_counts": dict(sorted(grade_counts.items())),
|
|
"pack_decision_counts": dict(sorted(pack_decision_counts.items())),
|
|
"conflict_arbitration": {
|
|
"status": "batch1_arbitrated",
|
|
"scope": "source_pack_eligibility_only",
|
|
"runtime_truth_allowed": False,
|
|
"reference_pack_ready_count": reference_pack_ready_count,
|
|
"runtime_truth_ready_count": 0,
|
|
"reference_pack_next_steps": [
|
|
"wire selected candidates into interpretation_source_inventory/source pack",
|
|
"add strict workflow visibility tests",
|
|
"keep content-level precedence below core rules until line-level rule review exists",
|
|
],
|
|
},
|
|
"graded_sources": graded_sources,
|
|
"artifacts": {
|
|
"json_report": _relative(SOURCE_GRADING_BATCH1_JSON),
|
|
"markdown_report": _relative(SOURCE_GRADING_BATCH1_MD),
|
|
},
|
|
"title": "Source Grading Batch 1 Manifest",
|
|
}
|
|
if write:
|
|
_write_reports(report, json_report=SOURCE_GRADING_BATCH1_JSON, markdown_report=SOURCE_GRADING_BATCH1_MD)
|
|
return report
|
|
|
|
|
|
def _review_decision(item: dict[str, Any]) -> tuple[str, str]:
|
|
path = str(item["path"])
|
|
if item["extraction_result"] == "extraction_failed":
|
|
return "blocked_extraction_failure", "Extraction failed; keep blocked until repaired."
|
|
if item["post_extraction_classification"] == "extracted_private_reference_only":
|
|
return "private_reference_only", "Private or historical personal report; do not promote into runtime truth."
|
|
if item["extraction_result"] == "text_empty":
|
|
return "asset_or_empty_reference_only", "OCR/extraction produced empty text; keep as indexed asset only."
|
|
if "/文件仓库/印度占星文章/" in path or "/Downloads/" in path:
|
|
return "promote_review_candidate", "Extracted source-like material; requires human source grading before promotion."
|
|
if item["post_extraction_classification"] == "extracted_candidate_for_review":
|
|
return "promote_review_candidate", "Candidate text extracted; requires conflict arbitration and source grading."
|
|
return "reference_only", "Useful as context or asset evidence, not direct runtime truth."
|
|
|
|
|
|
def build_extraction_review(*, write: bool = True) -> dict[str, Any]:
|
|
results_report = build_extraction_results(write=False)
|
|
reviews: list[dict[str, Any]] = []
|
|
for item in results_report["results"]:
|
|
decision, reason = _review_decision(item)
|
|
reviews.append(
|
|
{
|
|
"path": item["path"],
|
|
"source_scope": item["source_scope"],
|
|
"suffix": item["suffix"],
|
|
"byte_count": item["byte_count"],
|
|
"sha256": item["sha256"],
|
|
"extraction_result": item["extraction_result"],
|
|
"extraction_method": item["extraction_method"],
|
|
"extracted_character_count": item["extracted_character_count"],
|
|
"text_sha256": item.get("text_sha256"),
|
|
"post_extraction_classification": item["post_extraction_classification"],
|
|
"decision": decision,
|
|
"decision_reason": reason,
|
|
"runtime_truth_allowed": False,
|
|
}
|
|
)
|
|
decision_counts: dict[str, int] = {}
|
|
for item in reviews:
|
|
decision_counts[item["decision"]] = decision_counts.get(item["decision"], 0) + 1
|
|
stored_text_payload_fields = sum(1 for item in reviews if "text" in item or "text_preview" in item)
|
|
report = {
|
|
"scope": "extraction-review",
|
|
"status": "pass",
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"mode": {
|
|
"whole_machine_scan": False,
|
|
"semantic_promotion": False,
|
|
"runtime_truth_promotion": False,
|
|
"boundary": "Review decisions are governance labels only. No extracted source is promoted to runtime truth here.",
|
|
},
|
|
"summary": {
|
|
"reviewed_files": len(reviews),
|
|
"runtime_promotions": 0,
|
|
"stored_text_payload_fields": stored_text_payload_fields,
|
|
},
|
|
"decision_counts": dict(sorted(decision_counts.items())),
|
|
"reviews": reviews,
|
|
"artifacts": {
|
|
"json_report": _relative(EXTRACTION_REVIEW_JSON),
|
|
"markdown_report": _relative(EXTRACTION_REVIEW_MD),
|
|
},
|
|
"title": "Extraction Review Manifest",
|
|
}
|
|
if write:
|
|
_write_reports(report, json_report=EXTRACTION_REVIEW_JSON, markdown_report=EXTRACTION_REVIEW_MD)
|
|
return report
|
|
|
|
|
|
def _hash_text(text: str) -> str:
|
|
return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest()
|
|
|
|
|
|
def _extract_docx_text(path: Path) -> tuple[str, str | None]:
|
|
try:
|
|
from docx import Document
|
|
|
|
document = Document(str(path))
|
|
paragraphs = [paragraph.text for paragraph in document.paragraphs]
|
|
table_text: list[str] = []
|
|
for table in document.tables:
|
|
for row in table.rows:
|
|
table_text.append("\t".join(cell.text for cell in row.cells))
|
|
return "\n".join(paragraphs + table_text), None
|
|
except Exception as exc: # pragma: no cover - exercised by real files.
|
|
fallback_text, fallback_error = _extract_docx_xml_text(path)
|
|
if fallback_text.strip():
|
|
return fallback_text, f"python-docx fallback used after {type(exc).__name__}: {exc}"
|
|
return "", fallback_error or f"{type(exc).__name__}: {exc}"
|
|
|
|
|
|
def _extract_docx_xml_text(path: Path) -> tuple[str, str | None]:
|
|
try:
|
|
with zipfile.ZipFile(path) as archive:
|
|
xml_bytes = archive.read("word/document.xml")
|
|
root = ElementTree.fromstring(xml_bytes)
|
|
text_nodes = []
|
|
namespace = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}"
|
|
for node in root.iter(f"{namespace}t"):
|
|
if node.text:
|
|
text_nodes.append(node.text)
|
|
return "\n".join(text_nodes), None
|
|
except Exception as exc:
|
|
return "", f"docx_xml_fallback_failed {type(exc).__name__}: {exc}"
|
|
|
|
|
|
def _extract_pdf_text(path: Path) -> tuple[str, str, str | None]:
|
|
try:
|
|
import pdfplumber
|
|
|
|
parts: list[str] = []
|
|
with pdfplumber.open(str(path)) as pdf:
|
|
for page in pdf.pages:
|
|
parts.append(page.extract_text() or "")
|
|
text = "\n".join(parts)
|
|
if text.strip():
|
|
return text, "pdfplumber", None
|
|
except Exception as exc:
|
|
pdfplumber_error = f"{type(exc).__name__}: {exc}"
|
|
else:
|
|
pdfplumber_error = None
|
|
try:
|
|
from pypdf import PdfReader
|
|
|
|
reader = PdfReader(str(path))
|
|
text = "\n".join(page.extract_text() or "" for page in reader.pages)
|
|
return text, "pypdf", None
|
|
except Exception as exc: # pragma: no cover - depends on local PDF shape.
|
|
errors = "; ".join(error for error in [pdfplumber_error, f"{type(exc).__name__}: {exc}"] if error)
|
|
return "", "pypdf", errors
|
|
|
|
|
|
def _extract_image_text(path: Path) -> tuple[str, str, str | None]:
|
|
if not shutil.which("tesseract"):
|
|
return _extract_image_text_with_vision(path)
|
|
try:
|
|
from PIL import Image
|
|
import pytesseract
|
|
|
|
with Image.open(path) as image:
|
|
return pytesseract.image_to_string(image, lang="+".join(_ocr_languages())), "pytesseract", None
|
|
except Exception as exc: # pragma: no cover - depends on local OCR install/languages.
|
|
return "", "pytesseract", f"{type(exc).__name__}: {exc}"
|
|
|
|
|
|
def _vision_ocr_binary() -> Path | None:
|
|
if sys.platform != "darwin" or not shutil.which("swiftc"):
|
|
return None
|
|
cache_dir = Path.home() / ".cache" / "jyotish-ocr"
|
|
cache_dir.mkdir(parents=True, exist_ok=True)
|
|
source = cache_dir / "vision_ocr.swift"
|
|
binary = cache_dir / "vision_ocr"
|
|
if binary.exists() and binary.stat().st_mtime >= source.stat().st_mtime if source.exists() else False:
|
|
return binary
|
|
source.write_text(VISION_OCR_SOURCE, encoding="utf-8")
|
|
completed = subprocess.run(
|
|
["swiftc", str(source), "-o", str(binary)],
|
|
text=True,
|
|
capture_output=True,
|
|
timeout=60,
|
|
check=False,
|
|
)
|
|
if completed.returncode != 0:
|
|
return None
|
|
return binary
|
|
|
|
|
|
def _extract_image_text_with_vision(path: Path) -> tuple[str, str, str | None]:
|
|
binary = _vision_ocr_binary()
|
|
if not binary:
|
|
return "", "none", "no OCR backend available: tesseract missing and macOS Vision unavailable"
|
|
completed = subprocess.run(
|
|
[str(binary), str(path)],
|
|
text=True,
|
|
capture_output=True,
|
|
timeout=90,
|
|
check=False,
|
|
)
|
|
if completed.returncode != 0:
|
|
return "", "macos_vision", completed.stderr.strip() or f"vision_ocr exit {completed.returncode}"
|
|
return completed.stdout, "macos_vision", None
|
|
|
|
|
|
def _ocr_cache_path(item: dict[str, Any], method: str) -> Path:
|
|
cache_dir = Path.home() / ".cache" / "jyotish-ocr" / "results"
|
|
cache_dir.mkdir(parents=True, exist_ok=True)
|
|
return cache_dir / f"{item['sha256']}.{method}.json"
|
|
|
|
|
|
def _load_ocr_cached_item(item: dict[str, Any]) -> dict[str, Any] | None:
|
|
for method in ["macos_vision", "pytesseract"]:
|
|
path = _ocr_cache_path(item, method)
|
|
if not path.exists():
|
|
continue
|
|
try:
|
|
cached = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
continue
|
|
if cached.get("source_sha256") != item.get("sha256"):
|
|
continue
|
|
return {
|
|
**item,
|
|
"extraction_result": cached["extraction_result"],
|
|
"extraction_method": cached["extraction_method"],
|
|
"blocked_reason": cached.get("blocked_reason"),
|
|
"extracted_character_count": cached["extracted_character_count"],
|
|
"extracted_line_count": cached["extracted_line_count"],
|
|
"text_sha256": cached.get("text_sha256"),
|
|
"post_extraction_classification": cached["post_extraction_classification"],
|
|
"ocr_engine_available": cached["ocr_engine_available"],
|
|
"ocr_requested_languages": cached["ocr_requested_languages"],
|
|
"ocr_available_languages": cached["ocr_available_languages"],
|
|
"ocr_backend": cached["ocr_backend"],
|
|
"ocr_cache_status": "hit",
|
|
}
|
|
return None
|
|
|
|
|
|
def _write_ocr_cached_item(item: dict[str, Any]) -> None:
|
|
method = str(item.get("ocr_backend") or item.get("extraction_method") or "unknown")
|
|
path = _ocr_cache_path(item, method)
|
|
cache_payload = {
|
|
"source_sha256": item["sha256"],
|
|
"extraction_result": item["extraction_result"],
|
|
"extraction_method": item["extraction_method"],
|
|
"blocked_reason": item.get("blocked_reason"),
|
|
"extracted_character_count": item["extracted_character_count"],
|
|
"extracted_line_count": item["extracted_line_count"],
|
|
"text_sha256": item.get("text_sha256"),
|
|
"post_extraction_classification": item["post_extraction_classification"],
|
|
"ocr_engine_available": item.get("ocr_engine_available"),
|
|
"ocr_requested_languages": item.get("ocr_requested_languages", ["chi_sim", "eng"]),
|
|
"ocr_available_languages": item.get("ocr_available_languages", []),
|
|
"ocr_backend": item.get("ocr_backend", method),
|
|
}
|
|
path.write_text(json.dumps(cache_payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
|
|
def _ocr_available_languages() -> list[str]:
|
|
if not shutil.which("tesseract"):
|
|
return []
|
|
try:
|
|
import pytesseract
|
|
|
|
return sorted(str(lang) for lang in pytesseract.get_languages(config=""))
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def _ocr_languages() -> list[str]:
|
|
available = set(_ocr_available_languages())
|
|
preferred = ["chi_sim", "eng"]
|
|
selected = [lang for lang in preferred if lang in available]
|
|
return selected or (["eng"] if "eng" in available else preferred)
|
|
|
|
|
|
def _post_extraction_classification(item: dict[str, Any]) -> str:
|
|
path = str(item["path"])
|
|
if "/Desktop/" in path or "/WorkBuddy/" in path:
|
|
return "extracted_private_reference_only"
|
|
if item["classification"] in {"external_book_or_document", "reference_candidate"}:
|
|
return "extracted_candidate_for_review"
|
|
return "extracted_reference_only"
|
|
|
|
|
|
def _extract_queued_item(item: dict[str, Any]) -> dict[str, Any]:
|
|
path = Path(item["path"])
|
|
status = item["extraction_status"]
|
|
if status == "document_text_extraction_queued" and path.suffix.lower() == ".docx":
|
|
text, error = _extract_docx_text(path)
|
|
method = "docx"
|
|
elif status == "pdf_text_extraction_queued":
|
|
text, method, error = _extract_pdf_text(path)
|
|
elif status == "image_ocr_queued":
|
|
cached = _load_ocr_cached_item(item)
|
|
if cached:
|
|
return cached
|
|
text, method, error = _extract_image_text(path)
|
|
if error and "no OCR backend available" in error:
|
|
blocked_item = {
|
|
**item,
|
|
"extraction_result": "ocr_blocked_missing_engine",
|
|
"extraction_method": method,
|
|
"blocked_reason": error,
|
|
"extracted_character_count": 0,
|
|
"extracted_line_count": 0,
|
|
"text_sha256": None,
|
|
"post_extraction_classification": "extracted_reference_only",
|
|
"ocr_engine_available": False,
|
|
"ocr_requested_languages": ["chi_sim", "eng"],
|
|
"ocr_available_languages": [],
|
|
"ocr_backend": "none",
|
|
"ocr_cache_status": "miss",
|
|
}
|
|
_write_ocr_cached_item(blocked_item)
|
|
return blocked_item
|
|
else:
|
|
text = ""
|
|
method = "unsupported"
|
|
error = "unsupported extraction target"
|
|
|
|
normalized = text or ""
|
|
if error and not normalized.strip():
|
|
result = "extraction_failed"
|
|
elif normalized.strip():
|
|
result = "text_extracted"
|
|
else:
|
|
result = "text_empty"
|
|
result_item = {
|
|
**item,
|
|
"extraction_result": result,
|
|
"extraction_method": method,
|
|
"blocked_reason": error,
|
|
"extracted_character_count": len(normalized),
|
|
"extracted_line_count": normalized.count("\n") + (1 if normalized else 0),
|
|
"text_sha256": _hash_text(normalized) if normalized else None,
|
|
"post_extraction_classification": _post_extraction_classification(item),
|
|
**(
|
|
{
|
|
"ocr_engine_available": shutil.which("tesseract") is not None,
|
|
"ocr_requested_languages": ["chi_sim", "eng"],
|
|
"ocr_available_languages": _ocr_available_languages(),
|
|
"ocr_backend": "tesseract" if method == "pytesseract" else method,
|
|
"ocr_cache_status": "miss",
|
|
}
|
|
if status == "image_ocr_queued"
|
|
else {}
|
|
),
|
|
}
|
|
if status == "image_ocr_queued":
|
|
_write_ocr_cached_item(result_item)
|
|
return result_item
|
|
|
|
|
|
def build_extraction_results(*, write: bool = True) -> dict[str, Any]:
|
|
queue_report = build_extraction_queue(write=False)
|
|
results = [_extract_queued_item(item) for item in queue_report["queue"]]
|
|
result_counts: dict[str, int] = {}
|
|
method_counts: dict[str, int] = {}
|
|
classification_counts: dict[str, int] = {}
|
|
for item in results:
|
|
result_counts[item["extraction_result"]] = result_counts.get(item["extraction_result"], 0) + 1
|
|
method_counts[item["extraction_method"]] = method_counts.get(item["extraction_method"], 0) + 1
|
|
classification_counts[item["post_extraction_classification"]] = (
|
|
classification_counts.get(item["post_extraction_classification"], 0) + 1
|
|
)
|
|
stored_text_payload_fields = sum(1 for item in results if "text" in item or "text_preview" in item)
|
|
report = {
|
|
"scope": "extraction-results",
|
|
"status": "pass",
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"mode": {
|
|
"heavy_ocr": shutil.which("tesseract") is not None,
|
|
"macos_vision_available": _vision_ocr_binary() is not None,
|
|
"ocr_backend_policy": "prefer_tesseract_else_macos_vision_else_blocked",
|
|
"whole_machine_scan": False,
|
|
"external_high_relevance_scan": True,
|
|
"semantic_promotion": False,
|
|
"boundary": "Extraction results store text hashes and counts only. Extracted private text is not copied into the repo truth chain.",
|
|
},
|
|
"summary": {
|
|
"total_files": len(results),
|
|
"unhashed_files": sum(1 for item in results if not item.get("sha256")),
|
|
"stored_text_payload_fields": stored_text_payload_fields,
|
|
},
|
|
"result_counts": dict(sorted(result_counts.items())),
|
|
"method_counts": dict(sorted(method_counts.items())),
|
|
"post_extraction_classification_counts": dict(sorted(classification_counts.items())),
|
|
"results": results,
|
|
"artifacts": {
|
|
"json_report": _relative(EXTRACTION_RESULTS_JSON),
|
|
"markdown_report": _relative(EXTRACTION_RESULTS_MD),
|
|
},
|
|
"title": "Extraction Results Manifest",
|
|
}
|
|
if write:
|
|
_write_reports(report, json_report=EXTRACTION_RESULTS_JSON, markdown_report=EXTRACTION_RESULTS_MD)
|
|
return report
|
|
|
|
|
|
def build_extraction_queue(*, write: bool = True) -> dict[str, Any]:
|
|
project = build_manifest(scope="project", write=False)
|
|
external = build_manifest(scope="external", write=False)
|
|
queue: list[dict[str, Any]] = []
|
|
for source_scope, manifest in [("project", project), ("external", external)]:
|
|
for item in manifest["by_path"].values():
|
|
if item["extraction_status"] not in QUEUED_EXTRACTION_STATUSES:
|
|
continue
|
|
queue.append(
|
|
{
|
|
"source_scope": source_scope,
|
|
"path": item["path"],
|
|
"suffix": item["suffix"],
|
|
"byte_count": item["byte_count"],
|
|
"sha256": item["sha256"],
|
|
"extraction_status": item["extraction_status"],
|
|
"classification": item["classification"],
|
|
"priority": item["priority"],
|
|
"promotion_status": item["promotion_status"],
|
|
}
|
|
)
|
|
queue.sort(key=lambda item: (item["extraction_status"], item["source_scope"], item["path"]))
|
|
queue_counts: dict[str, int] = {}
|
|
source_counts: dict[str, int] = {}
|
|
for item in queue:
|
|
queue_counts[item["extraction_status"]] = queue_counts.get(item["extraction_status"], 0) + 1
|
|
source_counts[item["source_scope"]] = source_counts.get(item["source_scope"], 0) + 1
|
|
report = {
|
|
"scope": "extraction-queue",
|
|
"status": "pass",
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"mode": {
|
|
"heavy_ocr": False,
|
|
"whole_machine_scan": False,
|
|
"external_high_relevance_scan": True,
|
|
"semantic_promotion": False,
|
|
"boundary": "Queue only. It identifies PDF/image/document extraction work without performing OCR or promoting extracted text.",
|
|
},
|
|
"summary": {
|
|
"queued_files": len(queue),
|
|
"unhashed_files": sum(1 for item in queue if not item.get("sha256")),
|
|
"source_counts": dict(sorted(source_counts.items())),
|
|
},
|
|
"queue_counts": dict(sorted(queue_counts.items())),
|
|
"queue": queue,
|
|
"artifacts": {
|
|
"json_report": _relative(EXTRACTION_QUEUE_JSON),
|
|
"markdown_report": _relative(EXTRACTION_QUEUE_MD),
|
|
},
|
|
"title": "Extraction Queue Manifest",
|
|
}
|
|
if write:
|
|
_write_reports(report, json_report=EXTRACTION_QUEUE_JSON, markdown_report=EXTRACTION_QUEUE_MD)
|
|
return report
|
|
|
|
|
|
def _write_reports(report: dict[str, Any], *, json_report: Path, markdown_report: Path) -> None:
|
|
json_report.parent.mkdir(parents=True, exist_ok=True)
|
|
json_report.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
markdown_report.write_text(_render_markdown(report), encoding="utf-8")
|
|
|
|
|
|
def _render_markdown(report: dict[str, Any]) -> str:
|
|
summary = report["summary"]
|
|
if report["scope"] == "extraction-queue":
|
|
return _render_extraction_queue_markdown(report)
|
|
if report["scope"] == "extraction-results":
|
|
return _render_extraction_results_markdown(report)
|
|
if report["scope"] == "extraction-review":
|
|
return _render_extraction_review_markdown(report)
|
|
if report["scope"] == "source-grading-batch1":
|
|
return _render_source_grading_batch1_markdown(report)
|
|
lines = [
|
|
f"# {report.get('title', 'Character-Level Inventory Manifest')}",
|
|
"",
|
|
f"- status: {report['status']}",
|
|
f"- scope: {report['scope']}",
|
|
f"- generated_at: {report['generated_at']}",
|
|
f"- total_files: {summary['total_files']}",
|
|
f"- unhashed_files: {summary['unhashed_files']}",
|
|
f"- unclassified_files: {summary['unclassified_files']}",
|
|
f"- unknown_extraction_status: {summary['unknown_extraction_status']}",
|
|
f"- Heavy OCR: {'enabled' if report['mode']['heavy_ocr'] else 'disabled'}",
|
|
f"- Whole-machine scan: {'enabled' if report['mode']['whole_machine_scan'] else 'disabled'}",
|
|
"",
|
|
"## Root Summary",
|
|
"",
|
|
"| Root | Files | Bytes |",
|
|
"| --- | ---: | ---: |",
|
|
]
|
|
for root, item in report["root_summary"].items():
|
|
lines.append(f"| `{root}` | {item['files']} | {item['bytes']} |")
|
|
lines.extend(["", "## Extraction Status Counts", ""])
|
|
for name, count in summary["extraction_status_counts"].items():
|
|
lines.append(f"- `{name}`: {count}")
|
|
lines.extend(["", "## Classification Counts", ""])
|
|
for name, count in summary["classification_counts"].items():
|
|
lines.append(f"- `{name}`: {count}")
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Boundary",
|
|
"",
|
|
report["mode"]["boundary"],
|
|
"",
|
|
"This manifest proves indexing, hashing, and extraction-state classification. It does not by itself promote any source into the runtime truth chain.",
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _render_source_grading_batch1_markdown(report: dict[str, Any]) -> str:
|
|
lines = [
|
|
f"# {report.get('title', 'Source Grading Batch 1 Manifest')}",
|
|
"",
|
|
f"- status: {report['status']}",
|
|
f"- scope: {report['scope']}",
|
|
f"- generated_at: {report['generated_at']}",
|
|
f"- total_review_candidates: {report['summary']['total_review_candidates']}",
|
|
f"- selected_count: {report['summary']['selected_count']}",
|
|
f"- graded_files: {report['summary']['graded_files']}",
|
|
f"- runtime_promotions: {report['summary']['runtime_promotions']}",
|
|
f"- stored_text_payload_fields: {report['summary']['stored_text_payload_fields']}",
|
|
"",
|
|
"## Grade Counts",
|
|
"",
|
|
]
|
|
for name, count in report["grade_counts"].items():
|
|
lines.append(f"- `{name}`: {count}")
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Pack Decisions",
|
|
"",
|
|
]
|
|
)
|
|
for name, count in report["pack_decision_counts"].items():
|
|
lines.append(f"- `{name}`: {count}")
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Conflict Arbitration",
|
|
"",
|
|
f"- status: {report['conflict_arbitration']['status']}",
|
|
f"- scope: {report['conflict_arbitration']['scope']}",
|
|
f"- reference_pack_ready_count: {report['conflict_arbitration']['reference_pack_ready_count']}",
|
|
f"- runtime_truth_ready_count: {report['conflict_arbitration']['runtime_truth_ready_count']}",
|
|
f"- runtime_truth_allowed: {report['conflict_arbitration']['runtime_truth_allowed']}",
|
|
"",
|
|
"## Boundary",
|
|
"",
|
|
report["mode"]["boundary"],
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _render_extraction_review_markdown(report: dict[str, Any]) -> str:
|
|
lines = [
|
|
f"# {report.get('title', 'Extraction Review Manifest')}",
|
|
"",
|
|
f"- status: {report['status']}",
|
|
f"- scope: {report['scope']}",
|
|
f"- generated_at: {report['generated_at']}",
|
|
f"- reviewed_files: {report['summary']['reviewed_files']}",
|
|
f"- runtime_promotions: {report['summary']['runtime_promotions']}",
|
|
f"- stored_text_payload_fields: {report['summary']['stored_text_payload_fields']}",
|
|
"",
|
|
"## Decision Counts",
|
|
"",
|
|
]
|
|
for name, count in report["decision_counts"].items():
|
|
lines.append(f"- `{name}`: {count}")
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Boundary",
|
|
"",
|
|
report["mode"]["boundary"],
|
|
"",
|
|
"Promotion still requires human review, source grading, conflict arbitration, and explicit source-pack tests.",
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _render_extraction_results_markdown(report: dict[str, Any]) -> str:
|
|
lines = [
|
|
f"# {report.get('title', 'Extraction Results Manifest')}",
|
|
"",
|
|
f"- status: {report['status']}",
|
|
f"- scope: {report['scope']}",
|
|
f"- generated_at: {report['generated_at']}",
|
|
f"- total_files: {report['summary']['total_files']}",
|
|
f"- unhashed_files: {report['summary']['unhashed_files']}",
|
|
f"- stored_text_payload_fields: {report['summary']['stored_text_payload_fields']}",
|
|
f"- Heavy OCR: {'enabled' if report['mode']['heavy_ocr'] else 'disabled'}",
|
|
f"- macos_vision_available: {report['mode'].get('macos_vision_available', False)}",
|
|
f"- ocr_backend_policy: {report['mode'].get('ocr_backend_policy', 'unknown')}",
|
|
"",
|
|
"## Result Counts",
|
|
"",
|
|
]
|
|
for name, count in report["result_counts"].items():
|
|
lines.append(f"- `{name}`: {count}")
|
|
lines.extend(["", "## Method Counts", ""])
|
|
for name, count in report["method_counts"].items():
|
|
lines.append(f"- `{name}`: {count}")
|
|
lines.extend(["", "## Post-Extraction Classification Counts", ""])
|
|
for name, count in report["post_extraction_classification_counts"].items():
|
|
lines.append(f"- `{name}`: {count}")
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Boundary",
|
|
"",
|
|
report["mode"]["boundary"],
|
|
"",
|
|
"No extracted text or OCR text is stored in this report; only hashes, counts, methods, and statuses are persisted.",
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _render_extraction_queue_markdown(report: dict[str, Any]) -> str:
|
|
lines = [
|
|
f"# {report.get('title', 'Extraction Queue Manifest')}",
|
|
"",
|
|
f"- status: {report['status']}",
|
|
f"- scope: {report['scope']}",
|
|
f"- generated_at: {report['generated_at']}",
|
|
f"- queued_files: {report['summary']['queued_files']}",
|
|
f"- unhashed_files: {report['summary']['unhashed_files']}",
|
|
f"- Heavy OCR: {'enabled' if report['mode']['heavy_ocr'] else 'disabled'}",
|
|
f"- Whole-machine scan: {'enabled' if report['mode']['whole_machine_scan'] else 'disabled'}",
|
|
"",
|
|
"## Queue Counts",
|
|
"",
|
|
]
|
|
for name, count in report["queue_counts"].items():
|
|
lines.append(f"- `{name}`: {count}")
|
|
lines.extend(["", "## Source Counts", ""])
|
|
for name, count in report["summary"]["source_counts"].items():
|
|
lines.append(f"- `{name}`: {count}")
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Boundary",
|
|
"",
|
|
report["mode"]["boundary"],
|
|
"",
|
|
"Queued files are indexed by path, size, hash, and extraction status only. Private source text is not copied into this report.",
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _summary_view(report: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
key: value
|
|
for key, value in report.items()
|
|
if key not in {"by_path"}
|
|
}
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--scope",
|
|
default="project",
|
|
choices=[
|
|
"project",
|
|
"external",
|
|
"extraction-queue",
|
|
"extraction-results",
|
|
"extraction-review",
|
|
"source-grading-batch1",
|
|
],
|
|
)
|
|
parser.add_argument("--no-write", action="store_true", help="Print the manifest without writing report artifacts.")
|
|
parser.add_argument("--summary-only", action="store_true", help="Print only summary fields; still writes full artifacts unless --no-write is set.")
|
|
args = parser.parse_args(argv)
|
|
|
|
report = build_manifest(scope=args.scope, write=not args.no_write)
|
|
printable = _summary_view(report) if args.summary_only else report
|
|
print(json.dumps(printable, ensure_ascii=False, indent=2))
|
|
return 0 if report["status"] == "pass" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|