publish_issue.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | #!/usr/bin/env python3 |
| 2 | """Publish a single staging ticket's current content to the production repo. |
| 3 | |
| 4 | #186: gabriel shares tickets on social media as part of developing in public, |
| 5 | but doesn't want to send people to staging infrastructure. This is |
| 6 | deliberately one-way and on-demand — not a tracker sync. Given this same |
| 7 | session found a phantom-success write bug and an issue-numbering collision |
| 8 | in staging's own issue tracking (#182/#183/#184), an always-on bidirectional |
| 9 | mirror would just add a second place for that class of bug to bite. Instead: |
| 10 | copy one ticket's current content to production, on request, idempotently. |
| 11 | |
| 12 | Explicitly out of scope: comments, assignees, state transitions, and any |
| 13 | kind of two-way sync — see #186 for the full rationale. |
| 14 | |
| 15 | Usage:: |
| 16 | |
| 17 | python3 publish_issue.py 185 --json |
| 18 | python3 publish_issue.py 185 --production-hub https://musehub.ai --json |
| 19 | """ |
| 20 | from __future__ import annotations |
| 21 | |
| 22 | import argparse |
| 23 | import json |
| 24 | import re |
| 25 | import subprocess |
| 26 | import sys |
| 27 | from typing import Protocol |
| 28 | |
| 29 | _STAGING_URL_RE = re.compile(r"https?://staging\.musehub\.ai\S*") |
| 30 | # Bare cross-references like "#182" — these are staging-only numbers and |
| 31 | # won't resolve to the same ticket (or anything at all) on production. |
| 32 | # Requires the digits immediately after '#' with no intervening whitespace, |
| 33 | # so markdown headings ("## Summary") never match. |
| 34 | _BARE_ISSUE_REF_RE = re.compile(r"(?<!\w)#\d+\b") |
| 35 | |
| 36 | DEFAULT_STAGING_HUB = "https://staging.musehub.ai" |
| 37 | DEFAULT_PRODUCTION_HUB = "https://musehub.ai" |
| 38 | |
| 39 | |
| 40 | class StagingUrlLeakError(RuntimeError): |
| 41 | """Raised when a ticket body would publish an internal staging.musehub.ai |
| 42 | URL publicly. A hard block, not a warning: staging is a shared, |
| 43 | disposable-by-design environment (see #185/#180) that was never meant |
| 44 | for public traffic, and a warning is exactly the thing that gets skimmed |
| 45 | past at the moment someone's about to post a link on social media.""" |
| 46 | |
| 47 | |
| 48 | def build_mirror_marker(staging_number: int) -> str: |
| 49 | """A stable, greppable marker appended to every published body. |
| 50 | |
| 51 | Used to find an existing production mirror on re-publish, so re-running |
| 52 | this tool updates in place instead of creating a duplicate issue. |
| 53 | """ |
| 54 | return f"_Mirrored from staging#{staging_number}._" |
| 55 | |
| 56 | |
| 57 | class HubClient(Protocol): |
| 58 | def read_issue(self, number: int, hub: str) -> dict: ... |
| 59 | def list_issues(self, hub: str, state: str = "all") -> list[dict]: ... |
| 60 | def create_issue(self, hub: str, *, title: str, body: str, labels: list[str]) -> dict: ... |
| 61 | def update_issue(self, hub: str, number: int, *, title: str, body: str) -> dict: ... |
| 62 | def set_labels(self, hub: str, number: int, labels: list[str]) -> None: ... |
| 63 | |
| 64 | |
| 65 | class MuseHubCliClient: |
| 66 | """Real implementation — shells out to `muse hub issue ...`.""" |
| 67 | |
| 68 | def _run(self, args: list[str]) -> dict: |
| 69 | proc = subprocess.run( |
| 70 | ["muse", "hub", *args, "--json"], capture_output=True, text=True, |
| 71 | ) |
| 72 | if proc.returncode != 0: |
| 73 | raise RuntimeError(f"muse hub {' '.join(args)} failed: {proc.stderr.strip()}") |
| 74 | return json.loads(proc.stdout) |
| 75 | |
| 76 | def read_issue(self, number: int, hub: str) -> dict: |
| 77 | return self._run(["issue", "read", str(number), "--hub", hub]) |
| 78 | |
| 79 | def list_issues(self, hub: str, state: str = "all") -> list[dict]: |
| 80 | return self._run(["issue", "list", "--hub", hub, "--state", state]).get("issues", []) |
| 81 | |
| 82 | def create_issue(self, hub: str, *, title: str, body: str, labels: list[str]) -> dict: |
| 83 | args = ["issue", "create", "--hub", hub, "--title", title, "--body", body] |
| 84 | for label in labels: |
| 85 | args += ["--label", label] |
| 86 | return self._run(args) |
| 87 | |
| 88 | def update_issue(self, hub: str, number: int, *, title: str, body: str) -> dict: |
| 89 | return self._run([ |
| 90 | "issue", "update", str(number), "--hub", hub, |
| 91 | "--title", title, "--body", body, |
| 92 | ]) |
| 93 | |
| 94 | def set_labels(self, hub: str, number: int, labels: list[str]) -> None: |
| 95 | if not labels: |
| 96 | return |
| 97 | self._run(["issue", "label", str(number), "--hub", hub, "--set", *labels]) |
| 98 | |
| 99 | |
| 100 | def find_existing_mirror(client: HubClient, production_hub: str, staging_number: int) -> dict | None: |
| 101 | marker = build_mirror_marker(staging_number) |
| 102 | for issue in client.list_issues(production_hub, state="all"): |
| 103 | if marker in (issue.get("body") or ""): |
| 104 | return issue |
| 105 | return None |
| 106 | |
| 107 | |
| 108 | def publish_issue( |
| 109 | client: HubClient, staging_number: int, *, staging_hub: str, production_hub: str, |
| 110 | ) -> dict: |
| 111 | """Copy staging issue ``staging_number``'s current content to ``production_hub``. |
| 112 | |
| 113 | Idempotent: a second call for the same ``staging_number`` updates the |
| 114 | previously-created mirror in place rather than creating a duplicate. |
| 115 | """ |
| 116 | src = client.read_issue(staging_number, staging_hub) |
| 117 | |
| 118 | body_text = src.get("body") or "" |
| 119 | if _STAGING_URL_RE.search(body_text): |
| 120 | raise StagingUrlLeakError( |
| 121 | f"staging#{staging_number}'s body contains a staging.musehub.ai URL — " |
| 122 | "refusing to publish. Remove the internal link before publishing." |
| 123 | ) |
| 124 | |
| 125 | warnings: list[str] = [] |
| 126 | bare_refs = sorted(set(_BARE_ISSUE_REF_RE.findall(body_text)), key=lambda r: int(r[1:])) |
| 127 | if bare_refs: |
| 128 | warnings.append( |
| 129 | f"Source body contains bare issue references ({', '.join(bare_refs)}) that may " |
| 130 | "not resolve to the same tickets on production — review before sharing publicly." |
| 131 | ) |
| 132 | |
| 133 | marker = build_mirror_marker(staging_number) |
| 134 | mirrored_body = f"{src['body']}\n\n{marker}" |
| 135 | labels = src.get("labels", []) |
| 136 | |
| 137 | existing = find_existing_mirror(client, production_hub, staging_number) |
| 138 | if existing is not None: |
| 139 | client.update_issue(production_hub, existing["number"], title=src["title"], body=mirrored_body) |
| 140 | client.set_labels(production_hub, existing["number"], labels) |
| 141 | return { |
| 142 | "action": "updated", "number": existing["number"], |
| 143 | "url": existing.get("url"), "warnings": warnings, |
| 144 | } |
| 145 | |
| 146 | created = client.create_issue(production_hub, title=src["title"], body=mirrored_body, labels=labels) |
| 147 | return { |
| 148 | "action": "created", "number": created["number"], |
| 149 | "url": created.get("url"), "warnings": warnings, |
| 150 | } |
| 151 | |
| 152 | |
| 153 | def main(argv: list[str] | None = None) -> int: |
| 154 | parser = argparse.ArgumentParser(description=__doc__) |
| 155 | parser.add_argument("staging_number", type=int, help="Issue number on the staging hub to publish.") |
| 156 | parser.add_argument("--staging-hub", default=DEFAULT_STAGING_HUB) |
| 157 | parser.add_argument("--production-hub", default=DEFAULT_PRODUCTION_HUB) |
| 158 | parser.add_argument("--json", action="store_true", help="Emit JSON (the only supported output form).") |
| 159 | args = parser.parse_args(argv) |
| 160 | |
| 161 | client = MuseHubCliClient() |
| 162 | try: |
| 163 | result = publish_issue( |
| 164 | client, args.staging_number, |
| 165 | staging_hub=args.staging_hub, production_hub=args.production_hub, |
| 166 | ) |
| 167 | except (RuntimeError, KeyError) as e: |
| 168 | print(f"❌ {e}", file=sys.stderr) |
| 169 | return 1 |
| 170 | |
| 171 | for w in result["warnings"]: |
| 172 | print(f"⚠️ {w}", file=sys.stderr) |
| 173 | print(json.dumps(result)) |
| 174 | return 0 |
| 175 | |
| 176 | |
| 177 | if __name__ == "__main__": |
| 178 | sys.exit(main()) |