cloudinit-bot
a8ec7b80de
---ci--- project: inboxos phase: 1 milestone: v1.4 status: execute wave: 2 tasks: [T3, T4, T5, T6, T7, T8] ---/ci---
62 lines
1.9 KiB
Python
Executable File
62 lines
1.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Replay failed/skipped MIDs.
|
|
|
|
v1.4 REQ-INB-45.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
|
|
def load_state(path: str) -> dict:
|
|
if not os.path.exists(path):
|
|
return {"pending": [], "completed": []}
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def select_mids(state: dict, last: int | None, status: str) -> list[str]:
|
|
if status == "all":
|
|
items = state.get("pending", [])
|
|
elif status == "errored":
|
|
items = [m for m in state.get("pending", []) if m.get("status") == "errored"]
|
|
elif status == "skipped":
|
|
items = [m for m in state.get("pending", []) if m.get("status") == "skipped"]
|
|
else:
|
|
items = state.get("pending", [])
|
|
if last is not None:
|
|
items = items[-last:]
|
|
return [m.get("mid") if isinstance(m, dict) else m for m in items]
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser(description="Replay MIDs from state file")
|
|
g = p.add_mutually_exclusive_group()
|
|
g.add_argument("--last", type=int, help="Replay last N MIDs")
|
|
g.add_argument("--state", help="State file path (default: state.json)")
|
|
p.add_argument("--status", choices=["all", "errored", "skipped"], default="all")
|
|
p.add_argument("--dry-run", action="store_true", default=True)
|
|
p.add_argument("--confirm", action="store_true", help="Actually replay")
|
|
args = p.parse_args()
|
|
|
|
state_path = args.state or os.environ.get("INBOXOS_STATE_FILE", "state.json")
|
|
state = load_state(state_path)
|
|
mids = select_mids(state, args.last, args.status)
|
|
|
|
if not args.confirm and not args.dry_run:
|
|
sys.stderr.write("Refusing to run without --dry-run or --confirm.\n")
|
|
return 1
|
|
|
|
dry_run = not args.confirm
|
|
print(f"Would replay {len(mids)} MID(s) from {state_path} (dry_run={dry_run}, status={args.status})")
|
|
for mid in mids:
|
|
print(f" {mid}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|