171 lines
6.7 KiB
Python
171 lines
6.7 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Build reviewed inventory snapshots from captured upstream documents."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
sys.path.insert(0, str(ROOT / "libexec"))
|
||
|
|
|
||
|
|
from mmo_inventory_snapshot import ( # noqa: E402
|
||
|
|
build_codex_installed_snapshot,
|
||
|
|
build_opencode_go_snapshot,
|
||
|
|
build_opencode_zen_snapshot,
|
||
|
|
build_openrouter_snapshot,
|
||
|
|
rehash_inventory_snapshot,
|
||
|
|
)
|
||
|
|
from mmo_util import atomic_write_json, strict_json_loads # noqa: E402
|
||
|
|
|
||
|
|
|
||
|
|
def _sha256(value: bytes) -> str:
|
||
|
|
return hashlib.sha256(value).hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def _parser() -> argparse.ArgumentParser:
|
||
|
|
parser = argparse.ArgumentParser(
|
||
|
|
description="Normalize a captured provider response into the common inventory schema"
|
||
|
|
)
|
||
|
|
subparsers = parser.add_subparsers(dest="inventory", required=True)
|
||
|
|
openrouter = subparsers.add_parser("openrouter")
|
||
|
|
openrouter.add_argument("--input", type=Path, required=True)
|
||
|
|
openrouter.add_argument("--zdr-input", type=Path, required=True)
|
||
|
|
openrouter.add_argument("--output", type=Path, required=True)
|
||
|
|
openrouter.add_argument("--as-of", required=True)
|
||
|
|
openrouter.add_argument("--retrieved-at", required=True)
|
||
|
|
openrouter.add_argument("--source-url", required=True)
|
||
|
|
openrouter.add_argument("--zdr-source-url", required=True)
|
||
|
|
openrouter.add_argument(
|
||
|
|
"--endpoint-selection",
|
||
|
|
action="append",
|
||
|
|
required=True,
|
||
|
|
metavar="MODEL_ID=ENDPOINT_TAG",
|
||
|
|
)
|
||
|
|
codex = subparsers.add_parser("codex-installed")
|
||
|
|
codex.add_argument("--input", type=Path, required=True)
|
||
|
|
codex.add_argument("--reviewed-input", type=Path, required=True)
|
||
|
|
codex.add_argument("--output", type=Path, required=True)
|
||
|
|
codex.add_argument("--as-of", required=True)
|
||
|
|
codex.add_argument("--retrieved-at", required=True)
|
||
|
|
codex.add_argument("--source-url", required=True)
|
||
|
|
zen = subparsers.add_parser("opencode-zen")
|
||
|
|
zen.add_argument("--listing-input", type=Path, required=True)
|
||
|
|
zen.add_argument("--models-dev-input", type=Path, required=True)
|
||
|
|
zen.add_argument("--docs-input", type=Path, required=True)
|
||
|
|
zen.add_argument("--output", type=Path, required=True)
|
||
|
|
zen.add_argument("--as-of", required=True)
|
||
|
|
zen.add_argument("--retrieved-at", required=True)
|
||
|
|
zen.add_argument("--listing-url", required=True)
|
||
|
|
zen.add_argument("--models-dev-url", required=True)
|
||
|
|
zen.add_argument("--docs-url", required=True)
|
||
|
|
go = subparsers.add_parser("opencode-go")
|
||
|
|
go.add_argument("--listing-input", type=Path, required=True)
|
||
|
|
go.add_argument("--models-dev-input", type=Path, required=True)
|
||
|
|
go.add_argument("--docs-input", type=Path, required=True)
|
||
|
|
go.add_argument("--output", type=Path, required=True)
|
||
|
|
go.add_argument("--as-of", required=True)
|
||
|
|
go.add_argument("--retrieved-at", required=True)
|
||
|
|
go.add_argument("--listing-url", required=True)
|
||
|
|
go.add_argument("--models-dev-url", required=True)
|
||
|
|
go.add_argument("--docs-url", required=True)
|
||
|
|
rehash = subparsers.add_parser("rehash")
|
||
|
|
rehash.add_argument("--input", type=Path, required=True)
|
||
|
|
rehash.add_argument("--output", type=Path, required=True)
|
||
|
|
return parser
|
||
|
|
|
||
|
|
|
||
|
|
def main(argv: list[str] | None = None) -> int:
|
||
|
|
args = _parser().parse_args(argv)
|
||
|
|
if args.inventory == "codex-installed":
|
||
|
|
raw = args.input.read_bytes()
|
||
|
|
reviewed_raw = args.reviewed_input.read_bytes()
|
||
|
|
snapshot = build_codex_installed_snapshot(
|
||
|
|
strict_json_loads(raw),
|
||
|
|
strict_json_loads(reviewed_raw),
|
||
|
|
as_of=args.as_of,
|
||
|
|
retrieved_at=args.retrieved_at,
|
||
|
|
response_sha256=_sha256(raw),
|
||
|
|
source_url=args.source_url,
|
||
|
|
)
|
||
|
|
input_hashes = {"codex": _sha256(raw), "reviewed": _sha256(reviewed_raw)}
|
||
|
|
elif args.inventory == "openrouter":
|
||
|
|
raw = args.input.read_bytes()
|
||
|
|
zdr_raw = args.zdr_input.read_bytes()
|
||
|
|
document = strict_json_loads(raw)
|
||
|
|
endpoint_selections: dict[str, str] = {}
|
||
|
|
for value in args.endpoint_selection:
|
||
|
|
model_id, separator, endpoint_tag = value.partition("=")
|
||
|
|
if not separator or not model_id or not endpoint_tag:
|
||
|
|
raise ValueError("--endpoint-selection must use MODEL_ID=ENDPOINT_TAG")
|
||
|
|
if model_id in endpoint_selections:
|
||
|
|
raise ValueError(f"duplicate OpenRouter endpoint selection: {model_id}")
|
||
|
|
endpoint_selections[model_id] = endpoint_tag
|
||
|
|
snapshot = build_openrouter_snapshot(
|
||
|
|
document,
|
||
|
|
strict_json_loads(zdr_raw),
|
||
|
|
as_of=args.as_of,
|
||
|
|
retrieved_at=args.retrieved_at,
|
||
|
|
response_sha256=_sha256(raw),
|
||
|
|
zdr_response_sha256=_sha256(zdr_raw),
|
||
|
|
source_url=args.source_url,
|
||
|
|
zdr_source_url=args.zdr_source_url,
|
||
|
|
endpoint_selections=endpoint_selections,
|
||
|
|
)
|
||
|
|
input_hashes = {"openrouter": _sha256(raw), "zdr": _sha256(zdr_raw)}
|
||
|
|
elif args.inventory in {"opencode-go", "opencode-zen"}:
|
||
|
|
listing_raw = args.listing_input.read_bytes()
|
||
|
|
models_dev_raw = args.models_dev_input.read_bytes()
|
||
|
|
docs_raw = args.docs_input.read_bytes()
|
||
|
|
builder = (
|
||
|
|
build_opencode_go_snapshot
|
||
|
|
if args.inventory == "opencode-go"
|
||
|
|
else build_opencode_zen_snapshot
|
||
|
|
)
|
||
|
|
snapshot = builder(
|
||
|
|
strict_json_loads(listing_raw),
|
||
|
|
strict_json_loads(models_dev_raw),
|
||
|
|
docs_raw.decode("utf-8"),
|
||
|
|
as_of=args.as_of,
|
||
|
|
retrieved_at=args.retrieved_at,
|
||
|
|
listing_sha256=_sha256(listing_raw),
|
||
|
|
models_dev_sha256=_sha256(models_dev_raw),
|
||
|
|
docs_sha256=_sha256(docs_raw),
|
||
|
|
listing_url=args.listing_url,
|
||
|
|
models_dev_url=args.models_dev_url,
|
||
|
|
docs_url=args.docs_url,
|
||
|
|
)
|
||
|
|
input_hashes = {
|
||
|
|
"docs": _sha256(docs_raw),
|
||
|
|
"models_dev": _sha256(models_dev_raw),
|
||
|
|
args.inventory.replace("-", "_"): _sha256(listing_raw),
|
||
|
|
}
|
||
|
|
else:
|
||
|
|
raw = args.input.read_bytes()
|
||
|
|
document = strict_json_loads(raw)
|
||
|
|
snapshot = rehash_inventory_snapshot(document)
|
||
|
|
input_hashes = {"snapshot": _sha256(raw)}
|
||
|
|
atomic_write_json(args.output, snapshot, 0o644)
|
||
|
|
print(
|
||
|
|
json.dumps(
|
||
|
|
{
|
||
|
|
"inventory": snapshot["inventory"],
|
||
|
|
"models": len(snapshot["models"]),
|
||
|
|
"models_sha256": snapshot["models_sha256"],
|
||
|
|
"output": str(args.output),
|
||
|
|
"input_sha256": input_hashes,
|
||
|
|
},
|
||
|
|
indent=2,
|
||
|
|
sort_keys=True,
|
||
|
|
allow_nan=False,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|