121 lines
5.2 KiB
Python
121 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Move duplicate same-day autonomous review jobs out of the startup queue.
|
|
|
|
The archive is lossless: every moved job directory is retained with a hashed
|
|
JSONL manifest. One best-evidence job remains in the active queue per persona
|
|
and event_id. Non-review jobs are never selected.
|
|
"""
|
|
import argparse
|
|
import collections
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
RANK = {"ACCEPT": 0, "HOLD": 1, "REJECT": 2, "QUEUED": 3,
|
|
"RETRY_WAIT": 4, "PROCESSING": 5, "ERROR": 6}
|
|
|
|
|
|
def sha256(path):
|
|
value = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
value.update(chunk)
|
|
return value.hexdigest()
|
|
|
|
|
|
def load(path):
|
|
with path.open(encoding="utf-8") as stream:
|
|
return json.load(stream)
|
|
|
|
|
|
def atomic_json(path, value):
|
|
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
|
|
temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
os.chmod(temporary, 0o600)
|
|
os.replace(temporary, path)
|
|
|
|
|
|
def choose(group):
|
|
return min(group, key=lambda item: (RANK.get(item["state"], 99),
|
|
item.get("completed_at", "9999"), item["job_id"]))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--root", required=True)
|
|
parser.add_argument("--batch", required=True)
|
|
parser.add_argument("--apply", action="store_true")
|
|
args = parser.parse_args()
|
|
root = Path(args.root).resolve()
|
|
if root != Path("/var/lib/guanghu/tcs-mother-body").resolve() and "archive-review-test" not in str(root):
|
|
raise SystemExit("REFUSE_UNREGISTERED_STATE_ROOT")
|
|
lock_path = root / ".daily-review-archive.lock"
|
|
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with lock_path.open("w") as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
selected = []
|
|
before = collections.Counter()
|
|
nonreview = 0
|
|
personas = root / "personas"
|
|
for persona in sorted(personas.iterdir()):
|
|
jobs = persona / "jobs"
|
|
if not jobs.is_dir():
|
|
continue
|
|
groups = collections.defaultdict(list)
|
|
for job in jobs.iterdir():
|
|
if not job.is_dir() or len(job.name) != 64:
|
|
continue
|
|
try:
|
|
event = load(job / "input.json")
|
|
state = load(job / "state.json")
|
|
except (OSError, ValueError):
|
|
continue
|
|
if event.get("source_type") != "AUTONOMOUS_REVIEW_OPPORTUNITY" or not str(event.get("event_id", "")).startswith("DAILY-REVIEW-"):
|
|
nonreview += 1
|
|
continue
|
|
item = {"persona_id": persona.name, "event_id": event["event_id"],
|
|
"job_id": job.name, "path": job, "state": state.get("status", "UNKNOWN"),
|
|
"completed_at": state.get("completed_at", "")}
|
|
groups[event["event_id"]].append(item)
|
|
before[item["state"]] += 1
|
|
for group in groups.values():
|
|
keeper = choose(group)
|
|
for item in group:
|
|
if item is not keeper:
|
|
item["kept_job_id"] = keeper["job_id"]
|
|
selected.append(item)
|
|
summary = {"schema": "guanghu.daily-review-duplicate-archive/v1", "batch": args.batch,
|
|
"root": str(root), "mode": "APPLY" if args.apply else "DRY_RUN",
|
|
"review_jobs_before": sum(before.values()), "review_state_counts_before": dict(before),
|
|
"duplicate_jobs_selected": len(selected), "active_review_jobs_after": sum(before.values()) - len(selected),
|
|
"nonreview_jobs_untouched": nonreview, "lossless_move": True}
|
|
if not args.apply:
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
return
|
|
archive_root = root / "review-job-archive" / args.batch
|
|
if archive_root.exists():
|
|
raise SystemExit("REFUSE_EXISTING_ARCHIVE_BATCH")
|
|
archive_root.mkdir(parents=True, mode=0o700)
|
|
manifest = archive_root / "MANIFEST.jsonl"
|
|
with manifest.open("x", encoding="utf-8") as output:
|
|
os.chmod(manifest, 0o600)
|
|
for item in selected:
|
|
source = item.pop("path")
|
|
destination = archive_root / item["persona_id"] / item["job_id"]
|
|
destination.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
files = {p.name: sha256(p) for p in sorted(source.iterdir()) if p.is_file()}
|
|
record = {**item, "source": str(source), "archive": str(destination), "file_sha256": files}
|
|
os.rename(source, destination)
|
|
output.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
|
|
output.flush()
|
|
os.fsync(output.fileno())
|
|
summary["manifest"] = str(manifest)
|
|
summary["manifest_sha256"] = sha256(manifest)
|
|
atomic_json(archive_root / "RECEIPT.json", summary)
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|