from __future__ import annotations import argparse import json from pathlib import Path from .adapters import GitHubActionsAdapter from .anchors import AnchorStore from .ledger import SQLiteLedger from .schemas import SchemaRegistry from .sync import SyncCursor, apply_delta, prepare_delta def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="buildledger") sub = parser.add_subparsers(dest="command", required=True) sub.add_parser("schema-manifest") init = sub.add_parser("init-ledger") init.add_argument("path") append = sub.add_parser("append-sample") append.add_argument("path") verify = sub.add_parser("verify") verify.add_argument("path") anchor = sub.add_parser("anchor") anchor.add_argument("path") anchor.add_argument("reference") anchor.add_argument("--method", default="cloud-anchor") sync = sub.add_parser("sync-plan") sync.add_argument("path") sync.add_argument("--sequence", type=int, default=-1) sync.add_argument("--hash", dest="entry_hash", default="" + "0" * 64) gh = sub.add_parser("from-github") gh.add_argument("context_json") sign = sub.add_parser("sign-contract") sign.add_argument("contract_json") sign.add_argument("private_key_pem") sign.add_argument("--output") verify_contract = sub.add_parser("verify-contract") verify_contract.add_argument("contract_json") verify_contract.add_argument("public_key_pem") return parser def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) if args.command == "schema-manifest": print(json.dumps(SchemaRegistry.with_defaults().manifest(), indent=2, sort_keys=True)) return 0 if args.command == "init-ledger": SQLiteLedger(Path(args.path)).close() return 0 if args.command == "append-sample": ledger = SQLiteLedger(Path(args.path)) ledger.append("sample", {"message": "hello buildledger"}) ledger.close() return 0 if args.command == "verify": ledger = SQLiteLedger(Path(args.path)) ok, issues = ledger.verify_chain() ledger.close() print(json.dumps({"ok": ok, "issues": issues}, indent=2)) return 0 if ok else 1 if args.command == "anchor": ledger = SQLiteLedger(Path(args.path)) ref = AnchorStore(ledger).record(args.method, args.reference) ledger.close() print(ref.model_dump_json(indent=2)) return 0 if args.command == "sync-plan": ledger = SQLiteLedger(Path(args.path)) packet = prepare_delta(ledger, SyncCursor(sequence=args.sequence, entry_hash=args.entry_hash)) ledger.close() print(json.dumps({"base": packet.base_cursor.__dict__, "count": len(packet.entries), "target": packet.target_cursor.__dict__}, indent=2)) return 0 if args.command == "from-github": context = json.loads(Path(args.context_json).read_text(encoding="utf-8")) contract = GitHubActionsAdapter().contract_from_context(context) print(contract.model_dump_json(indent=2)) return 0 if args.command == "sign-contract": from .models import BuildContract contract = BuildContract.model_validate(json.loads(Path(args.contract_json).read_text(encoding="utf-8"))) signed = contract.sign(Path(args.private_key_pem).read_text(encoding="utf-8")) output = signed.model_dump_json(indent=2) if args.output: Path(args.output).write_text(output + "\n", encoding="utf-8") else: print(output) return 0 if args.command == "verify-contract": from .models import BuildContract contract = BuildContract.model_validate(json.loads(Path(args.contract_json).read_text(encoding="utf-8"))) ok = contract.verify_signature(Path(args.public_key_pem).read_text(encoding="utf-8")) print(json.dumps({"ok": ok}, indent=2)) return 0 if ok else 1 return 1 if __name__ == "__main__": raise SystemExit(main())