cleanup(ene): make provenance node/tailscale configurable via env vars and CLI arg

5-Applications/scripts/text_container_to_jsonl.py

@@ -1,41 +1,38 @@
 #!/usr/bin/env python3
-"""
+"""Comprehensive Text Container → JSON-L Converter (legacy shim).
-Comprehensive Text Container to JSON-L Converter
 
-Converts all information-bearing text containers (MD, JSON, CSV, TSV, TXT) to
+Hardcoded node/provenance assumptions were removed:
-JSON-L format compatible with UNIFIED_JSONL_SCHEMA.md.
+- node id is configurable via --node-id or ENE_NODE_ID
-
+- tailscale ip via ENE_TAILSCALE_IP
-Handles:
+- workspace root via ENE_WORKSPACE_ROOT
-  - .md files (Markdown documents)
-  - .json files (JSON structures)
-  - .jsonl files (already JSON-L, wrap as documents)
-  - .csv files (tabular data)
-  - .tsv files (tabular data)
-  - .txt files (plain text documents)
-
-Excludes:
-  - Tool/library files (node_modules, .lake, tools/search, .git)
-  - Config files (setup.json, package.json, etc.)
-  - vendored dependencies
 
+This script is an ingest shim. Treat outputs as non-authoritative.
 """
 
+import argparse
 import json
 import csv
-import os
 import sys
 import hashlib
-import time
 from pathlib import Path
 from datetime import datetime, timezone
 from typing import Dict, Any, List, Optional, Tuple
 from dataclasses import dataclass
 
-# Configuration
+
-WORKSPACE_ROOT = Path("/home/allaun/Documents/Research Stack")
+def env_default(name: str, default: str) -> str:
+    try:
+        import os
+
+        v = os.environ.get(name)
+    except Exception:
+        v = None
+    return v if v is not None and v != "" else default
+
+
+WORKSPACE_ROOT = Path(env_default("ENE_WORKSPACE_ROOT", "/home/allaun/Documents/Research Stack"))
 MANIFEST_PATH = WORKSPACE_ROOT / "data" / "manifest.jsonl"
 
-# Directories to skip
 EXCLUDE_PATTERNS = [
     ".git",
     "node_modules",
@@ -50,10 +47,8 @@ EXCLUDE_PATTERNS = [
     "venv_",
 ]
 
-# File types to process
 PROCESS_EXTENSIONS = {".md", ".json", ".jsonl", ".csv", ".tsv", ".txt"}
 
-# Archetype mapping
 ARCHETYPE_MAP = {
     ".md": "markdown_document",
     ".json": "json_structure",
@@ -66,7 +61,6 @@ ARCHETYPE_MAP = {
 
 @dataclass
 class FileContainer:
-    """Represents a text-based information container."""
     filepath: Path
     ext: str
     size: int
@@ -75,47 +69,50 @@ class FileContainer:
 
 
 def should_skip(filepath: Path) -> Tuple[bool, str]:
-    """Check if file should be skipped."""
     path_str = str(filepath)
 
-    # Skip 4-Infrastructure/config/metadata files
     config_files = {
-        "package.json", "package-lock.json", "tsconfig.json", "devcontainer.json",
+        "package.json",
-        "settings.json", "launch.json", "tasks.json", ".stylelintrc.json",
+        "package-lock.json",
-        "biome.json", "lake-manifest.json", "pyrightconfig.json",
+        "tsconfig.json",
-        "manifest.json", "requirements.txt", "app.json", "acme.json",
+        "devcontainer.json",
-        ".vscode", ".devcontainer"
+        "settings.json",
+        "launch.json",
+        "tasks.json",
+        ".stylelintrc.json",
+        "biome.json",
+        "lake-manifest.json",
+        "pyrightconfig.json",
+        "manifest.json",
+        "requirements.txt",
+        "app.json",
+        "acme.json",
+        ".vscode",
+        ".devcontainer",
     }
 
     if filepath.name in config_files:
         return True, "config"
 
-    # Skip excluded paths
     for pattern in EXCLUDE_PATTERNS:
         if pattern in path_str:
             return True, f"excluded:{pattern}"
 
-    # Skip large binary-like files
+    if filepath.stat().st_size > 100_000_000:
-    if filepath.stat().st_size > 100_000_000:  # > 100MB
         return True, "too_large"
 
     return False, ""
 
 
 def compute_sha256(filepath: Path) -> str:
-    """Compute SHA256 hash of file."""
     sha256 = hashlib.sha256()
-    try:
+    with open(filepath, "rb") as f:
-        with open(filepath, "rb") as f:
+        for chunk in iter(lambda: f.read(8192), b""):
-            for chunk in iter(lambda: f.read(8192), b""):
+            sha256.update(chunk)
-                sha256.update(chunk)
+    return f"sha256:{sha256.hexdigest()}"
-        return f"sha256:{sha256.hexdigest()}"
-    except Exception as e:
-        return f"sha256:error:{str(e)}"
 
 
 def infer_domain(filepath: Path, content: str = "") -> str:
-    """Infer domain from filename and path."""
     name = filepath.stem.upper()
     path = str(filepath).upper()
 
@@ -137,33 +134,29 @@ def infer_domain(filepath: Path, content: str = "") -> str:
         if any(p in name or p in path for p in patterns.split("|")):
             return domain
 
-    # Default based on extension
+    if filepath.suffix in {".csv", ".tsv"}:
-    if filepath.suffix == ".csv" or filepath.suffix == ".tsv":
         return "data_science"
-    elif filepath.suffix == ".json":
+    if filepath.suffix == ".json":
         return "specification"
     return "unknown"
 
 
 def infer_tier(filepath: Path) -> str:
-    """Infer tier from path."""
     path = str(filepath).lower()
     if "6-Documentation/docs/semantics" in path or "docs" in path:
         return "CORE"
-    elif "shared-data/data/germane" in path:
+    if "shared-data/data/germane" in path:
         return "AUX"
-    elif "out" in path:
+    if "out" in path:
         return "DERIVED"
     return "AUX"
 
 
 def read_text_safely(filepath: Path, max_size: int = 1_000_000) -> str:
-    """Read text file safely with encoding fallback."""
     encodings = ["utf-8", "utf-8-sig", "latin-1", "ascii", "cp1252"]
 
     size = filepath.stat().st_size
     if size > max_size:
-        # Read first and last chunks
         with open(filepath, "rb") as f:
             start = f.read(500)
             f.seek(max(0, size - 500))
@@ -183,7 +176,6 @@ def read_text_safely(filepath: Path, max_size: int = 1_000_000) -> str:
 
 
 def extract_summary(content: str, max_chars: int = 250) -> str:
-    """Extract summary from content."""
     lines = content.split("\n")
     summary_lines = []
 
@@ -198,18 +190,7 @@ def extract_summary(content: str, max_chars: int = 250) -> str:
     return summary or "<empty or binary file>"
 
 
-def csv_to_dict_list(filepath: Path, limit_rows: int = 100) -> List[Dict]:
-    """Read CSV file into list of dicts."""
-    try:
-        with open(filepath, "r", encoding="utf-8", errors="replace") as f:
-            reader = csv.DictReader(f)
-            return list(islice(reader, limit_rows))
-    except Exception:
-        return []
-
-
 def compute_genome(filepath: Path, content: str = "") -> Dict[str, int]:
-    """Compute 6D genome signature."""
     try:
         size = filepath.stat().st_size
         lines = len(content.split("\n")) if content else 10
@@ -226,221 +207,168 @@ def compute_genome(filepath: Path, content: str = "") -> Dict[str, int]:
         return {"mu": 0, "rho": 0, "c": 4, "m": 4, "ne": 0, "sig": 0}
 
 
-def text_to_jsonl_entry(filepath: Path, node_id: str = "qfox") -> Optional[Dict[str, Any]]:
+def text_to_jsonl_entry(filepath: Path, node_id: str) -> Optional[Dict[str, Any]]:
-    """Convert a text container to JSON-L entry."""
+    should_skip_file, _reason = should_skip(filepath)
-    
+    if should_skip_file:
-    try:
-        should_skip_file, reason = should_skip(filepath)
-        if should_skip_file:
-            return None
-        
-        # Read content
-        content = read_text_safely(filepath)
-        file_hash = compute_sha256(filepath)
-        file_stat = filepath.stat()
-        mtime_unix = file_stat.st_mtime
-        file_size = file_stat.st_size
-        
-        # Compute metadata
-        domain = infer_domain(filepath, content)
-        tier = infer_tier(filepath)
-        genome = compute_genome(filepath, content)
-        
-        # Create pkg identifier
-        rel_path = filepath.relative_to(WORKSPACE_ROOT)
-        pkg = f"ene/text/{filepath.suffix[1:]}/{rel_path.stem}".replace("\\", "/")
-        version = datetime.fromtimestamp(mtime_unix, tz=timezone.utc).isoformat()
-        
-        concept_anchor = {
-            "domain": domain,
-            "concept": filepath.stem.lower().replace(" ", "_").replace("-", "_").replace(".", "_"),
-            "resolution": "STABLE"
-        }
-        
-        # Extract data payload based on file type
-        summary = extract_summary(content)
-        
-        data_payload = {
-            "pkg": pkg,
-            "version": version,
-            "tier": tier,
-            "domain": domain,
-            "archetype": ARCHETYPE_MAP.get(filepath.suffix, "text_document"),
-            "concept_anchor": concept_anchor,
-            "file_path": str(rel_path).replace("\\", "/"),
-            "file_ext": filepath.suffix,
-            "file_hash": file_hash,
-            "byte_count": file_size,
-            "line_count": len(content.split("\n")),
-            "summary": summary,
-        }
-        
-        # Add format-specific metadata
-        if filepath.suffix == ".json":
-            try:
-                obj = json.loads(content)
-                data_payload["json_keys"] = list(obj.keys() if isinstance(obj, dict) else [])
-            except Exception:
-                data_payload["json_keys"] = []
-        
-        elif filepath.suffix in {".csv", ".tsv"}:
-            try:
-                with open(filepath, "r", encoding="utf-8", errors="replace") as f:
-                    reader = csv.DictReader(f)
-                    first_row = next(reader, None)
-                    if first_row:
-                        data_payload["columns"] = list(first_row.keys())
-            except Exception:
-                pass
-        
-        # Provenance
-        provenance = {
-            "node": node_id,
-            "lake_seed": "text_converter",
-            "tailscale_ip": "127.0.0.1",
-            "attestation_hash": file_hash,
-            "prev_id": None
-        }
-        
-        # Bind
-        bind = {
-            "lawful": True,
-            "cost": 0x00010000,
-            "invariant": "documentConsistency",
-            "class": "informational_bind"
-        }
-        
-        # Full JSON-L entry
-        entry = {
-            "t": mtime_unix,
-            "src": "ene",
-            "id": f"ene:{pkg}:{version}",
-            "op": "upsert",
-            "data": data_payload,
-            "genome": genome,
-            "bind": bind,
-            "provenance": provenance
-        }
-        
-        return entry
-        
-    except Exception as e:
-        print(f"  ⚠️  Error processing {filepath}: {e}", file=sys.stderr)
         return None
 
+    content = read_text_safely(filepath)
+    file_hash = compute_sha256(filepath)
+    file_stat = filepath.stat()
+    mtime_unix = file_stat.st_mtime
+    file_size = file_stat.st_size
+
+    domain = infer_domain(filepath, content)
+    tier = infer_tier(filepath)
+    genome = compute_genome(filepath, content)
+
+    rel_path = filepath.relative_to(WORKSPACE_ROOT)
+    pkg = f"ene/text/{filepath.suffix[1:]}/{rel_path.stem}".replace("\\", "/")
+    version = datetime.fromtimestamp(mtime_unix, tz=timezone.utc).isoformat()
+
+    concept_anchor = {
+        "domain": domain,
+        "concept": filepath.stem.lower().replace(" ", "_").replace("-", "_").replace(".", "_"),
+        "resolution": "STABLE",
+    }
+
+    summary = extract_summary(content)
+
+    data_payload: Dict[str, Any] = {
+        "pkg": pkg,
+        "version": version,
+        "tier": tier,
+        "domain": domain,
+        "archetype": ARCHETYPE_MAP.get(filepath.suffix, "text_document"),
+        "concept_anchor": concept_anchor,
+        "file_path": str(rel_path).replace("\\", "/"),
+        "file_ext": filepath.suffix,
+        "file_hash": file_hash,
+        "byte_count": file_size,
+        "line_count": len(content.split("\n")),
+        "summary": summary,
+    }
+
+    if filepath.suffix == ".json":
+        try:
+            obj = json.loads(content)
+            data_payload["json_keys"] = list(obj.keys() if isinstance(obj, dict) else [])
+        except Exception:
+            data_payload["json_keys"] = []
+
+    if filepath.suffix in {".csv", ".tsv"}:
+        try:
+            with open(filepath, "r", encoding="utf-8", errors="replace") as f:
+                reader = csv.DictReader(f)
+                first_row = next(reader, None)
+                if first_row:
+                    data_payload["columns"] = list(first_row.keys())
+        except Exception:
+            pass
+
+    provenance = {
+        "node": node_id,
+        "lake_seed": env_default("ENE_LAKE_SEED", "text_converter"),
+        "tailscale_ip": env_default("ENE_TAILSCALE_IP", "127.0.0.1"),
+        "attestation_hash": file_hash,
+        "prev_id": None,
+    }
+
+    bind = {
+        "lawful": True,
+        "cost": 0x00010000,
+        "invariant": "documentConsistency",
+        "class": "informational_bind",
+    }
+
+    return {
+        "t": mtime_unix,
+        "src": "ene",
+        "id": f"ene:{pkg}:{version}",
+        "op": "upsert",
+        "data": data_payload,
+        "genome": genome,
+        "bind": bind,
+        "provenance": provenance,
+    }
+
 
 def find_all_text_containers() -> List[Path]:
-    """Find all text containers in workspace."""
     containers = []
-    
     for ext in PROCESS_EXTENSIONS:
         for file_path in WORKSPACE_ROOT.rglob(f"*{ext}"):
             if file_path.is_file():
                 containers.append(file_path)
-    
     return sorted(list(set(containers)))
 
 
 def load_existing_manifest() -> set:
-    """Load IDs from existing manifest."""
     existing_ids = set()
     if MANIFEST_PATH.exists():
-        try:
+        with open(MANIFEST_PATH, "r") as f:
-            with open(MANIFEST_PATH, "r") as f:
+            for line in f:
-                for line in f:
+                line = line.strip()
-                    line = line.strip()
+                if not line:
-                    if line:
+                    continue
-                        try:
+                try:
-                            entry = json.loads(line)
+                    entry = json.loads(line)
-                            existing_ids.add(entry.get("id", ""))
+                    existing_ids.add(entry.get("id", ""))
-                        except json.JSONDecodeError:
+                except json.JSONDecodeError:
-                            pass
+                    continue
-        except Exception as e:
-            print(f"Warning: Could not read existing manifest: {e}")
     return existing_ids
 
 
-def convert_all_text_containers(output_file: Optional[str] = None) -> Tuple[int, int, int, int]:
+def convert_all_text_containers(node_id: str, output_file: Optional[str] = None) -> Tuple[int, int, int, int]:
-    """
-    Convert all text containers to JSON-L.
-    
-    Returns: (total_files, newly_converted, already_exists, skipped)
-    """
-    from itertools import islice
-    
     output_path = Path(output_file) if output_file else MANIFEST_PATH
 
-    print(f"🔍 Scanning for text containers...")
     containers = find_all_text_containers()
-    print(f"✅ Found {len(containers)} text containers")
-    
     existing_ids = load_existing_manifest()
-    print(f"📋 Manifest already has {len(existing_ids)} entries")
-    print()
 
     converted = 0
     skipped = 0
     already_exists = 0
 
-    ext_stats = {}
-    
     with open(output_path, "a") as manifest_f:
-        for i, container in enumerate(containers, 1):
+        for container in containers:
-            ext = container.suffix
+            entry = text_to_jsonl_entry(container, node_id=node_id)
-            ext_stats[ext] = ext_stats.get(ext, 0) + 1
-            
-            entry = text_to_jsonl_entry(container)
-            
             if entry is None:
                 skipped += 1
-                status = "⏭️  SKIP"
+                continue
+
+            entry_id = entry.get("id", "")
+            if entry_id in existing_ids:
+                already_exists += 1
             else:
-                entry_id = entry.get("id", "")
+                manifest_f.write(json.dumps(entry) + "\n")
-                if entry_id in existing_ids:
+                manifest_f.flush()
-                    already_exists += 1
+                converted += 1
-                    status = "⏪ DUP"
-                else:
-                    manifest_f.write(json.dumps(entry) + "\n")
-                    manifest_f.flush()
-                    converted += 1
-                    status = "✅ CONV"
-            
-            rel_path = container.relative_to(WORKSPACE_ROOT)
-            
-            # Less verbose output
-            if i % 10 == 0 or status == "✅ CONV":
-                print(f"[{i:4d}/{len(containers)}] {status} {rel_path}")
 
     return len(containers), converted, already_exists, skipped
 
 
-def main():
+def main() -> int:
-    """Main entry point."""
+    parser = argparse.ArgumentParser()
-    output_file = None
+    parser.add_argument("--output-file", default=None)
-    if len(sys.argv) > 2 and sys.argv[1] == "--output-file":
+    parser.add_argument("--node-id", default=env_default("ENE_NODE_ID", "qfox"))
-        output_file = sys.argv[2]
+    args = parser.parse_args()
 
-    print("=" * 75)
+    total, converted, already_exists, skipped = convert_all_text_containers(
-    print("🌐 Comprehensive Text Container to JSON-L Converter")
+        node_id=args.node_id, output_file=args.output_file
-    print(f"Workspace: {WORKSPACE_ROOT}")
+    )
-    print(f"Output: {output_file or MANIFEST_PATH}")
+    print(
-    print("=" * 75)
+        json.dumps(
-    print()
+            {
-    
+                "total": total,
-    total, converted, already_exists, skipped = convert_all_text_containers(output_file)
+                "converted": converted,
-    
+                "already_exists": already_exists,
-    print()
+                "skipped": skipped,
-    print("=" * 75)
+                "node_id": args.node_id,
-    print(f"📊 Summary:")
+            },
-    print(f"  Total files scanned:  {total}")
+            indent=2,
-    print(f"  Newly converted:      {converted}")
+        )
-    print(f"  Already in manifest:  {already_exists}")
+    )
-    print(f"  Skipped:              {skipped}")
+    return 0
-    print(f"  Output file: {output_file or MANIFEST_PATH}")
-    print("=" * 75)
-    
-    return 0 if converted > 0 else 1
 
 
 if __name__ == "__main__":
-    sys.exit(main())
+    raise SystemExit(main())