131 lines
3.8 KiB
Python
131 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Compile connectors/index.json + fragment.json + file hash scan → connectors.json
|
|
|
|
Usage:
|
|
python3 scripts/compile.py # writes connectors/connectors.json
|
|
python3 scripts/compile.py --verify # checks existing index is up to date
|
|
|
|
Reads:
|
|
- connectors/index.json — ordered list of connector folder ids
|
|
- connectors/<id>/fragment.json — index entry without files[]
|
|
- physical file scan — SHA-256 and size of every file in each folder
|
|
|
|
Produces:
|
|
- connectors/connectors.json — full index with files[] and SHA-256
|
|
|
|
Auto-excluded: fragment.json, connector.json, compile.sh, compile.py,
|
|
update_hashes.py, .DS_Store, and scripts/ directory.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
INDEX_DIR = os.path.join(ROOT, "connectors")
|
|
|
|
EXCLUDE_FILES = {
|
|
"fragment.json",
|
|
"connectors.json",
|
|
"index.json",
|
|
"compile.sh",
|
|
"compile.py",
|
|
"update_hashes.py",
|
|
".DS_Store",
|
|
}
|
|
|
|
EXCLUDE_DIRS = {"scripts", "__pycache__", ".git"}
|
|
|
|
|
|
def scan_files(folder_path: str) -> list[dict]:
|
|
"""Scan all files in folder_path, return [{path, sha256, size}, ...] sorted."""
|
|
files = []
|
|
for fname in sorted(os.listdir(folder_path)):
|
|
fpath = os.path.join(folder_path, fname)
|
|
if os.path.isdir(fpath):
|
|
continue
|
|
if fname in EXCLUDE_FILES:
|
|
continue
|
|
with open(fpath, "rb") as f:
|
|
data = f.read()
|
|
files.append({
|
|
"path": fname,
|
|
"sha256": hashlib.sha256(data).hexdigest(),
|
|
"size": len(data),
|
|
})
|
|
return files
|
|
|
|
|
|
def compile_index() -> list[dict]:
|
|
"""Build the connectors list from index.json + fragments + file scan."""
|
|
index_path = os.path.join(INDEX_DIR, "index.json")
|
|
if not os.path.exists(index_path):
|
|
print("❌ connectors/index.json not found. Create it first.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
with open(index_path) as f:
|
|
order: list[str] = json.load(f)
|
|
|
|
connectors = []
|
|
for i, connector_id in enumerate(order):
|
|
frag_path = os.path.join(INDEX_DIR, connector_id, "fragment.json")
|
|
if not os.path.exists(frag_path):
|
|
print(
|
|
f"⚠️ [{i+1}/{len(order)}] {connector_id} — fragment.json not found, skipping.",
|
|
file=sys.stderr,
|
|
)
|
|
continue
|
|
|
|
with open(frag_path) as f:
|
|
fragment: dict = json.load(f)
|
|
|
|
folder_path = os.path.join(INDEX_DIR, connector_id)
|
|
files = scan_files(folder_path)
|
|
fragment["files"] = files
|
|
|
|
n_files = len(files)
|
|
print(f"✓ [{i+1}/{len(order)}] {connector_id} — {n_files} file{'s' if n_files != 1 else ''}")
|
|
connectors.append(fragment)
|
|
|
|
return connectors
|
|
|
|
|
|
def write_index(connectors: list[dict]) -> None:
|
|
"""Write the compiled connectors.json."""
|
|
output = {"version": 1, "connectors": connectors}
|
|
out_path = os.path.join(INDEX_DIR, "connectors.json")
|
|
with open(out_path, "w") as f:
|
|
json.dump(output, f, indent=2)
|
|
f.write("\n")
|
|
print(f"\n✅ Written {out_path} ({len(connectors)} connectors)")
|
|
|
|
|
|
def verify_index() -> bool:
|
|
"""Verify existing connectors.json matches what compilation would produce."""
|
|
with open(os.path.join(INDEX_DIR, "connectors.json")) as f:
|
|
current = json.load(f)
|
|
|
|
fresh = compile_index()
|
|
|
|
if current["connectors"] == fresh:
|
|
print("✅ Index is up to date — no changes.")
|
|
return True
|
|
else:
|
|
print("❌ Index is stale. Run compile.py to regenerate.")
|
|
return False
|
|
|
|
|
|
def main() -> None:
|
|
if "--verify" in sys.argv:
|
|
sys.exit(0 if verify_index() else 1)
|
|
|
|
connectors = compile_index()
|
|
write_index(connectors)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|