90 lines
2.8 KiB
Python
Executable File
90 lines
2.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""Market intelligence review decision writer CLI gate.
|
|
|
|
This script only prints JSON gate status. The review_state update implementation
|
|
is intentionally disabled in this phase.
|
|
"""
|
|
|
|
import argparse
|
|
import contextlib
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
with contextlib.redirect_stdout(sys.stderr):
|
|
from services.market_intel.candidate_queue_review_decision_writer_cli import ( # noqa: E402
|
|
APPROVAL_ENV_VAR,
|
|
build_candidate_queue_review_decision_writer_cli_plan,
|
|
)
|
|
from services.market_intel.phase import MARKET_INTEL_PHASE # noqa: E402
|
|
|
|
|
|
def parse_args(argv=None):
|
|
parser = argparse.ArgumentParser(
|
|
description="Preview review decision writer execution gates."
|
|
)
|
|
parser.add_argument(
|
|
"--transaction-json",
|
|
default=None,
|
|
help="Path to a review decision transaction preview JSON file.",
|
|
)
|
|
parser.add_argument("--execute", action="store_true", help="Request execution.")
|
|
parser.add_argument(
|
|
"--apply-real-write",
|
|
action="store_true",
|
|
help="Request the guarded review_state update transaction.",
|
|
)
|
|
parser.add_argument(
|
|
"--approval-token",
|
|
default=None,
|
|
help=f"One-time approval token checked against {APPROVAL_ENV_VAR}.",
|
|
)
|
|
parser.add_argument(
|
|
"--backup-verified",
|
|
action="store_true",
|
|
help="Confirm a fresh operator backup before allowing a future write.",
|
|
)
|
|
parser.add_argument(
|
|
"--review-inventory-smoke-passed",
|
|
action="store_true",
|
|
help="Confirm read-only review inventory smoke passed.",
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def _load_transaction_preview(path):
|
|
if not path:
|
|
return {}
|
|
with open(path, encoding="utf-8") as handle:
|
|
payload = json.load(handle)
|
|
if not isinstance(payload, dict):
|
|
return {}
|
|
return payload.get("candidate_queue_review_decision_transaction", payload)
|
|
|
|
|
|
def main(argv=None):
|
|
args = parse_args(argv)
|
|
plan = build_candidate_queue_review_decision_writer_cli_plan(
|
|
transaction_preview=_load_transaction_preview(args.transaction_json),
|
|
execute_requested=args.execute,
|
|
apply_real_write=args.apply_real_write,
|
|
approval_token=args.approval_token,
|
|
approval_token_secret=os.getenv(APPROVAL_ENV_VAR),
|
|
backup_verified=args.backup_verified,
|
|
review_inventory_smoke_passed=args.review_inventory_smoke_passed,
|
|
)
|
|
plan["phase"] = MARKET_INTEL_PHASE
|
|
print(json.dumps(plan, ensure_ascii=False, indent=2, sort_keys=True))
|
|
return int(plan.get("exit_code", 2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|