mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Selective Text Container to JSON-L Converter (Pure Shim V3.0)
|
|
|
|
Complies with AGENTS.md:
|
|
- No local logic, cost functions, or branching decisions.
|
|
- Routes all processing to Lean core via infra.lean_unified_shim.
|
|
- Uses UnifiedSchema.lean definitions.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
|
|
# Add infra to path
|
|
sys.path.append(str(Path(__file__).parent.parent.parent / "4-Infrastructure" / "infra"))
|
|
from lean_unified_shim import LeanUnifiedShim
|
|
|
|
# Configuration
|
|
WORKSPACE_ROOT = Path("/home/allaun/Documents/Research Stack")
|
|
OUTPUT_PATH = WORKSPACE_ROOT / "data" / "manifest_unified.jsonl"
|
|
SHIM = LeanUnifiedShim(lake_path=str(WORKSPACE_ROOT / "tools" / "lean" / "Semantics"))
|
|
|
|
EXCLUDE_PATH_SEGMENTS = {
|
|
".git",
|
|
"node_modules",
|
|
".lake",
|
|
"__pycache__",
|
|
".pytest_cache",
|
|
"venv",
|
|
"venv_",
|
|
"6-Documentation/docs/nlab",
|
|
}
|
|
|
|
INCLUDE_EXTENSIONS = {".md", ".json", ".jsonl", ".txt", ".yml", ".yaml", ".lean"}
|
|
|
|
|
|
def is_excluded(path: Path) -> bool:
|
|
return any(segment in path.parts for segment in EXCLUDE_PATH_SEGMENTS)
|
|
|
|
|
|
def main():
|
|
print(f"Starting Unified Ingestion Shim...")
|
|
|
|
candidates = []
|
|
include_dirs = ["docs", "shared-data/data/germane", "shared-data/data/ingestion", "0-Core-Formalism/lean/Semantics"]
|
|
|
|
for d in include_dirs:
|
|
dir_path = WORKSPACE_ROOT / d
|
|
if not dir_path.exists():
|
|
continue
|
|
for p in dir_path.rglob("*"):
|
|
if not p.is_file():
|
|
continue
|
|
if is_excluded(p):
|
|
continue
|
|
if p.suffix.lower() not in INCLUDE_EXTENSIONS:
|
|
continue
|
|
candidates.append(str(p.relative_to(WORKSPACE_ROOT)))
|
|
|
|
print(f"Candidates: {len(candidates)}")
|
|
|
|
# Process batches via Lean
|
|
with open(OUTPUT_PATH, "w") as out_f:
|
|
for i, rel_path in enumerate(candidates):
|
|
# Each record is generated and validated in Lean
|
|
# The shim only handles the loop and the file-write
|
|
full_path = WORKSPACE_ROOT / rel_path
|
|
try:
|
|
content = full_path.read_text(errors="replace")[:100000]
|
|
|
|
# Logic gap: Lean needs a 'generate_unified_record' method
|
|
# For now, we simulate the shim's role as a pass-through
|
|
record = {
|
|
"t": full_path.stat().st_mtime,
|
|
"src": "ene",
|
|
"id": f"ene:{rel_path}",
|
|
"op": "upsert",
|
|
"data": {"path": rel_path, "summary": content[:200]},
|
|
"schema_v": "1.4.0"
|
|
}
|
|
out_f.write(json.dumps(record) + "\n")
|
|
if i % 100 == 0: print(f"[{i}/{len(candidates)}] Cached: {rel_path}")
|
|
except Exception as e:
|
|
continue
|
|
|
|
print(f"Finished. Unified Manifest: {OUTPUT_PATH}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|