Something went wrong. Try again.
A Forgejo action that submits a new version of a program to the JCOllie winget source as a pull request.
Something went wrong. Try again.
winget-submit winget_submit.py
11 kB · 282 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283# SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us># SPDX-License-Identifier: MIT
"""Make a submission to the JCOllie winget source for a program's release.
Run by the composite action in `action.yml`, with its inputs in`WINGET_SUBMIT_*` environment variables. It clones jeff/winget-pkgs at main,has `wingetsource new-version` write the new version there and`wingetsource check` judge it, then pushes a branch and opens a pull requestas the submitter. The manifest itself is never written here (ADR 0004 injeff/winget-pkgs)."""
import base64import jsonimport osimport subprocessimport sysimport tempfileimport urllib.errorimport urllib.requestfrom dataclasses import dataclass
FORGE = "https://git.jcollie.dev"MANIFEST_REPOSITORY = "jeff/winget-pkgs"BASE_BRANCH = "main"USER_AGENT = "winget-submit (+https://git.jcollie.dev/jeff/winget-submit)"
# The submitter. Built in rather than read from the forge: its token is# limited to jeff/winget-pkgs, and Forgejo lets such a token have only# repository and issue scopes, so it cannot read its own user.SUBMITTER = "winget-bot"SUBMITTER_EMAIL = "winget-bot@jcollie.dev"
class SubmitError(Exception): pass
@dataclassclass Inputs: identifier: str version: str installers: dict[str, str] product_codes: dict[str, str] release_notes_url: str release_date: str token: str dry_run: bool
def parse_pairs(name: str, what: str, text: str) -> dict[str, str]: """`<architecture> <what>`, one per line; blank lines are ignored.""" pairs = {} for number, line in enumerate(text.splitlines(), 1): fields = line.split() if not fields: continue if len(fields) != 2: raise SubmitError(f"{name} line {number}: expected '<architecture> <{what}>', got {line.strip()!r}") architecture, value = fields if architecture in pairs: raise SubmitError(f"{name} line {number}: {architecture} is given more than once") pairs[architecture] = value return pairs
def parse_installers(text: str) -> dict[str, str]: installers = parse_pairs("installers", "url", text) if not installers: raise SubmitError("installers: at least one '<architecture> <url>' line is needed") return installers
def parse_product_codes(text: str) -> dict[str, str]: """Optional: `wingetsource new-version` reads a new MSI's ProductCode from it, and one given here is written instead.""" return parse_pairs("product-codes", "code", text)
def parse_bool(name: str, text: str) -> bool: value = text.strip().lower() if value in ("", "false"): return False if value == "true": return True raise SubmitError(f"{name}: expected true or false, got {text!r}")
def read_inputs(environ) -> Inputs: def get(name: str) -> str: return environ.get(f"WINGET_SUBMIT_{name}", "").strip()
inputs = Inputs( identifier=get("IDENTIFIER"), version=get("VERSION"), installers=parse_installers(environ.get("WINGET_SUBMIT_INSTALLERS", "")), product_codes=parse_product_codes(environ.get("WINGET_SUBMIT_PRODUCT_CODES", "")), release_notes_url=get("RELEASE_NOTES_URL"), release_date=get("RELEASE_DATE"), token=get("TOKEN"), dry_run=parse_bool("dry-run", get("DRY_RUN")), ) for name in ("identifier", "version"): if not getattr(inputs, name): raise SubmitError(f"{name} is required") if not inputs.token and not inputs.dry_run: raise SubmitError("token is required, unless this is a dry run") return inputs
def branch_name(identifier: str, version: str) -> str: return f"submit/{identifier}/{version}"
def title(identifier: str, version: str) -> str: return f"{identifier} {version}"
def release_url(inputs: Inputs, environ) -> str | None: """The program's release: its release notes if given, or else the release page of the tag the workflow is running for.""" if inputs.release_notes_url: return inputs.release_notes_url server, repository, ref = (environ.get(k, "") for k in ("GITHUB_SERVER_URL", "GITHUB_REPOSITORY", "GITHUB_REF")) if server and repository and ref.startswith("refs/tags/"): return f"{server}/{repository}/releases/tag/{ref.removeprefix('refs/tags/')}" return None
def run_url(environ) -> str | None: server, repository, number = (environ.get(k, "") for k in ("GITHUB_SERVER_URL", "GITHUB_REPOSITORY", "GITHUB_RUN_NUMBER")) if server and repository and number: return f"{server}/{repository}/actions/runs/{number}" return None
def body(inputs: Inputs, copied_from: str, environ) -> str: lines = [f"Submits {inputs.identifier} {inputs.version}, written by `wingetsource new-version`.", ""] release = release_url(inputs, environ) if release: lines.append(f"- Release: {release}") run = run_url(environ) if run: lines.append(f"- Opened by: {run}") lines.append(f"- Copied from: {copied_from}") lines.append("") lines.append("Opened by [winget-submit](https://git.jcollie.dev/jeff/winget-submit).") return "\n".join(lines) + "\n"
def git_auth_environment(token: str, login: str) -> dict[str, str]: """Configuration that hands git the token as basic auth, through the environment rather than the command line or a remote URL, so that it is neither visible in the process list nor written into the clone.""" if not token: return {} credentials = base64.b64encode(f"{login}:{token}".encode()).decode() return { "GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": f"http.{FORGE}/.extraHeader", "GIT_CONFIG_VALUE_0": f"Authorization: Basic {credentials}", }
def open_pull_request_for(pulls: list[dict], branch: str) -> dict | None: for pull in pulls: head = pull.get("head") or {} if head.get("ref") == branch and (head.get("repo") or {}).get("full_name") == MANIFEST_REPOSITORY: return pull return None
class Forge: def __init__(self, token: str): self.token = token
def request(self, method: str, path: str, data: dict | None = None): request = urllib.request.Request( f"{FORGE}/api/v1{path}", method=method, data=json.dumps(data).encode() if data is not None else None, headers={ "Accept": "application/json", "Content-Type": "application/json", "Authorization": f"token {self.token}", "User-Agent": USER_AGENT, }, ) try: with urllib.request.urlopen(request, timeout=60) as response: return json.load(response) except urllib.error.HTTPError as e: raise SubmitError(f"{method} {path}: {e.code} {e.reason}: {e.read().decode(errors='replace')}") from e
def open_pulls(self) -> list[dict]: pulls, page = [], 1 while True: batch = self.request("GET", f"/repos/{MANIFEST_REPOSITORY}/pulls?state=open&limit=50&page={page}") pulls.extend(batch) if len(batch) < 50: return pulls page += 1
def run(command: list[str], cwd: str, env: dict, capture: bool = False) -> str: print(f"+ {' '.join(command)}", file=sys.stderr, flush=True) result = subprocess.run(command, cwd=cwd, env=env, text=True, stdout=subprocess.PIPE if capture else None) if result.returncode != 0: raise SubmitError(f"{command[0]} {command[1]} failed with status {result.returncode}") return result.stdout if capture else ""
def submit(inputs: Inputs, environ) -> None: forge = Forge(inputs.token) env = {**environ, **git_auth_environment(inputs.token, SUBMITTER)} branch = branch_name(inputs.identifier, inputs.version)
with tempfile.TemporaryDirectory(dir=environ.get("RUNNER_TEMP")) as work: clone = os.path.join(work, "winget-pkgs") run(["git", "clone", "--quiet", "--depth", "1", "--branch", BASE_BRANCH, f"{FORGE}/{MANIFEST_REPOSITORY}.git", clone], work, env)
command = [ "nix", "develop", "-c", "python3", "-m", "wingetsource", "new-version", "--json", "--manifests", "manifests", "--identifier", inputs.identifier, "--version", inputs.version, ] for architecture, url in inputs.installers.items(): command += ["--installer", f"{architecture}={url}"] for architecture, code in inputs.product_codes.items(): command += ["--product-code", f"{architecture}={code}"] if inputs.release_notes_url: command += ["--release-notes-url", inputs.release_notes_url] if inputs.release_date: command += ["--release-date", inputs.release_date] output = run(command, clone, env, capture=True) written = json.loads(output.strip().splitlines()[-1])
run(["git", "add", "--", written["directory"]], clone, env) identity = ["-c", f"user.name={SUBMITTER}", "-c", f"user.email={SUBMITTER_EMAIL}"] run(["git", *identity, "commit", "--quiet", "-m", title(inputs.identifier, inputs.version)], clone, env) # `check` finds the versions to download installers for by what # differs from the base, so it runs on the commit. run(["nix", "develop", "-c", "python3", "-m", "wingetsource", "check", "--manifests", "manifests", "--changed-since", f"origin/{BASE_BRANCH}"], clone, env)
if inputs.dry_run: run(["git", "--no-pager", "show", "--stat", "--patch", "HEAD"], clone, env) print(f"Dry run: would push {branch} and open '{title(inputs.identifier, inputs.version)}' with:\n") print(body(inputs, written["copied_from"], environ)) return
run(["git", "push", "--quiet", "--force", "origin", f"HEAD:refs/heads/{branch}"], clone, env)
existing = open_pull_request_for(forge.open_pulls(), branch) if existing: print(f"Updated {branch}; its pull request is already open: {existing['html_url']}") return pull = forge.request( "POST", f"/repos/{MANIFEST_REPOSITORY}/pulls", { "base": BASE_BRANCH, "head": branch, "title": title(inputs.identifier, inputs.version), "body": body(inputs, written["copied_from"], environ), }, ) print(f"Opened {pull['html_url']}")
def main() -> int: try: submit(read_inputs(os.environ), os.environ) except SubmitError as e: print(f"winget-submit: {e}", file=sys.stderr) return 1 return 0
if __name__ == "__main__": sys.exit(main())