tegwick
ed2c8e2f4d
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 2m59s
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
80 lines
3.5 KiB
Python
80 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Prepare private offline all-space review trees; never import or contact Bubble."""
|
|
|
|
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.bubble_bundle import prepare_bundle # noqa: E402
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("capture", type=Path)
|
|
parser.add_argument("--inventory-spaces", type=Path)
|
|
parser.add_argument("--output", type=Path, required=True, help="New directory outside checkout")
|
|
args = parser.parse_args()
|
|
output_created = False
|
|
try:
|
|
output = args.output.expanduser().resolve()
|
|
repo = Path(__file__).resolve().parents[1]
|
|
capture = args.capture.expanduser().resolve()
|
|
if output.is_relative_to(repo) or output.is_relative_to(capture) or (
|
|
capture.is_relative_to(output)
|
|
):
|
|
raise ValueError("Unsafe output location")
|
|
inputs = {name: (capture / (name + ".json")).read_bytes()
|
|
for name in ("spaces", "articles")}
|
|
if args.inventory_spaces:
|
|
if args.inventory_spaces.resolve().is_relative_to(output):
|
|
raise ValueError("Output overlaps retained inventory")
|
|
inputs["retained_spaces"] = args.inventory_spaces.read_bytes()
|
|
report, files = prepare_bundle(
|
|
json.loads(inputs["spaces"]), json.loads(inputs["articles"]),
|
|
inventory_spaces=json.loads(inputs["retained_spaces"])
|
|
if "retained_spaces" in inputs else None,
|
|
)
|
|
report["source_sha256"] = {name: hashlib.sha256(data).hexdigest()
|
|
for name, data in sorted(inputs.items())}
|
|
# Validate/compile entirely before creating output. Exclusive publication
|
|
# refuses overwrite; completion receipt is written last if interrupted.
|
|
output.mkdir(mode=0o700)
|
|
output_created = True
|
|
for name, data in sorted({**files, **{
|
|
"source-capture/" + name + ".json": data for name, data in inputs.items()
|
|
}}.items()):
|
|
dest = output / name
|
|
dest.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
fd = os.open(dest, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fd, "wb") as stream:
|
|
stream.write(data)
|
|
report["status"] = "completed_offline_preparation"
|
|
fd = os.open(output / "reconciliation.json", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fd, "w") as stream:
|
|
json.dump(report, stream, indent=2, sort_keys=True)
|
|
stream.write("\n")
|
|
print(json.dumps({
|
|
"spaces": len(report["spaces"]),
|
|
"draft_trees": sum(bool(entry["tree"]) for entry in report["spaces"]),
|
|
"exact_body_sidecars": sum(a["body_file"] is not None for a in report["articles"]),
|
|
"blocker_counts": dict(sorted(Counter(
|
|
blocker for s in report["spaces"] for blocker in s["blockers"]
|
|
).items())), "production_imported": False,
|
|
}))
|
|
except (ValueError, OSError, TypeError, KeyError):
|
|
print("Preparation refused; inspect private input/output locally. " + (
|
|
"Output may be incomplete; no completion receipt means it is unusable."
|
|
if output_created else "No output was published."
|
|
), file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|