tegwick
0881b9896a
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 2m43s
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
56 lines
2.5 KiB
Python
56 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Plan an existing space's original-date correction from private read-only metadata."""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from coulomb_social.apps.spaces.date_backfill import plan_date_backfill # noqa: E402
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("capture", type=Path, help="Private spaces.json/articles.json capture")
|
|
parser.add_argument("--observation", type=Path, required=True)
|
|
parser.add_argument("--selection", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
try:
|
|
output = args.output.expanduser().resolve()
|
|
repo = Path(__file__).resolve().parents[1]
|
|
if output.is_relative_to(repo):
|
|
raise ValueError("Output must be private outside checkout")
|
|
paths = {"spaces": args.capture / "spaces.json", "articles": args.capture / "articles.json",
|
|
"observation": args.observation, "selection": args.selection}
|
|
if any(path.resolve() == output for path in paths.values()):
|
|
raise ValueError("Output overlaps input")
|
|
raw = {name: path.read_bytes() for name, path in paths.items()}
|
|
inputs = {name: json.loads(value) for name, value in raw.items()}
|
|
plan = plan_date_backfill(inputs["observation"], inputs["spaces"], inputs["articles"],
|
|
inputs["selection"])
|
|
plan["source_sha256"] = {name: hashlib.sha256(value).hexdigest()
|
|
for name, value in sorted(raw.items())}
|
|
fd = os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fd, "w") as stream:
|
|
json.dump(plan, stream, indent=2, sort_keys=True)
|
|
stream.write("\n")
|
|
print(json.dumps({
|
|
"proposed_changes": len(plan["proposed_changes"]),
|
|
"page_status_counts": dict(sorted(Counter(p["status"] for p in plan["pages"]).items())),
|
|
"blockers": len(plan["blockers"]), "ready_to_apply": False,
|
|
"production_writes": False,
|
|
}))
|
|
except (ValueError, OSError, TypeError, KeyError, AttributeError, OverflowError):
|
|
print("Date plan refused; inspect private inputs/output locally.", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|