import base64 import json import os import pathlib import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request ROOT = "https://androidpublisher.googleapis.com/" SCOPE = "https://www.googleapis.com/auth/androidpublisher" STATUSES = ("draft", "inProgress", "halted", "completed") NOTES_LIMIT = 500 RETRY_AFTER = 5 HINTS = { "status draft may be created on draft app": "until the app's first release is published by hand in Play Console, use status: draft", "Package not found": "create the app in Play Console and upload its first bundle there by hand", "does not have permission": "invite the service account in Play Console, Users and permissions, with release rights for this app", "already been used": "every upload needs a version code higher than any uploaded before", "has not been used in project": "enable the Google Play Android Developer API in the service account's Google Cloud project", } def fail(message): print(f"::error::{message}", flush=True) sys.exit(1) def setting(name, default=""): return os.environ.get(name, "").strip() or default def b64url(data): return base64.urlsafe_b64encode(data).rstrip(b"=").decode() def explain(body): text = body.decode(errors="replace") try: error = json.loads(text) text = error.get("error_description") or error["error"]["message"] except (ValueError, KeyError, TypeError, AttributeError): text = text[:500] hints = [hint for clue, hint in HINTS.items() if clue in text] return text + "".join(f" ({hint})" for hint in hints) def send(method, url, body=None, headers=None, retry=True): where = f"{method} {urllib.parse.urlsplit(url).path}" for attempt in (1, 2, 3): try: call = urllib.request.Request(url, data=body, method=method, headers=headers or {}) with urllib.request.urlopen(call, timeout=900) as response: return response.headers, response.read() except urllib.error.HTTPError as error: if retry and attempt < 3 and error.code in (429, 500, 502, 503, 504): time.sleep(RETRY_AFTER * attempt) continue fail(f"{where} answered {error.code}: {explain(error.read())}") except (urllib.error.URLError, OSError) as error: if retry and attempt < 3: time.sleep(RETRY_AFTER * attempt) continue fail(f"{where} failed: {error}") def sign(key, data): with tempfile.TemporaryDirectory() as folder: path = pathlib.Path(folder, "key.pem") path.touch(0o600) path.write_text(key) result = subprocess.run(["openssl", "dgst", "-sha256", "-sign", str(path)], input=data, capture_output=True) if result.returncode != 0: fail("openssl could not sign with the service account's key: " + result.stderr.decode(errors="replace").strip()) return result.stdout def access_token(account): now = int(time.time()) header = b64url(json.dumps({"alg": "RS256", "typ": "JWT"}).encode()) claims = b64url(json.dumps({ "iss": account["client_email"], "scope": SCOPE, "aud": account["token_uri"], "iat": now, "exp": now + 3600, }).encode()) unsigned = f"{header}.{claims}" assertion = f"{unsigned}.{b64url(sign(account['private_key'], unsigned.encode()))}" form = urllib.parse.urlencode({"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer", "assertion": assertion}) _, body = send("POST", account["token_uri"], form.encode(), {"Content-Type": "application/x-www-form-urlencoded"}) return json.loads(body)["access_token"] class Play: def __init__(self, account, package): self.authorization = {"Authorization": f"Bearer {access_token(account)}"} self.path = f"androidpublisher/v3/applications/{urllib.parse.quote(package)}/edits" def call(self, method, path="", payload=None): headers = dict(self.authorization) body = b"" if method == "POST" else None if payload is not None: headers["Content-Type"] = "application/json" body = json.dumps(payload).encode() _, answer = send(method, f"{ROOT}{self.path}{path}", body, headers) return json.loads(answer) if answer else {} def upload(self, edit, bundle): size = bundle.stat().st_size headers = dict(self.authorization, **{ "X-Upload-Content-Type": "application/octet-stream", "X-Upload-Content-Length": str(size), }) started, _ = send("POST", f"{ROOT}upload/{self.path}/{edit}/bundles?uploadType=resumable", b"", headers) with bundle.open("rb") as content: _, answer = send("PUT", started["Location"], content, { "Content-Type": "application/octet-stream", "Content-Length": str(size), }, retry=False) return json.loads(answer) def release_notes(folder): if not folder: return [] notes = [] for file in sorted(pathlib.Path(folder).glob("*.txt")): text = file.read_text(encoding="utf-8").strip() if not text: continue if len(text) > NOTES_LIMIT: fail(f"{file} has {len(text)} characters; Google Play takes at most {NOTES_LIMIT} per language") notes.append({"language": file.stem, "text": text}) if not notes: fail(f"{folder} holds no .txt file with text in it") return notes def service_account(): try: account = json.loads(setting("PLAY_SERVICE_ACCOUNT")) except ValueError: account = None if not isinstance(account, dict) or not account.get("client_email") or not account.get("private_key"): fail("service-account must be the whole JSON key file of a Google Cloud service account") account.setdefault("token_uri", "https://oauth2.googleapis.com/token") return account def share(fraction): try: value = float(fraction) except ValueError: value = 0 if not 0 < value < 1: fail("fraction is the share of users, between 0 and 1, for example 0.1") return value def newest(releases): return max(releases, key=lambda release: max(int(code) for code in release.get("versionCodes") or ["0"])) def main(): account = service_account() package = setting("PLAY_PACKAGE") or fail("package is required, for example software.arasaka.meweb") track = setting("PLAY_TRACK", "internal") status = setting("PLAY_STATUS", "completed") fraction = setting("PLAY_FRACTION") bundle = setting("PLAY_AAB") source = setting("PLAY_FROM_TRACK") if status not in STATUSES: fail(f"status must be one of {', '.join(STATUSES)}") if (status in ("inProgress", "halted")) != bool(fraction): fail("a staged rollout takes status inProgress or halted together with fraction; other statuses take no fraction") fraction = share(fraction) if fraction else None if bool(bundle) == bool(source): fail("give either aab, to upload a bundle, or from-track, to move a release that is already there") if bundle and not pathlib.Path(bundle).is_file(): fail(f"{bundle} is not a file") notes = release_notes(setting("PLAY_NOTES")) play = Play(account, package) edit = play.call("POST")["id"] if bundle: path = pathlib.Path(bundle) uploaded = play.upload(edit, path) print(f"{path.name} uploaded as version code {uploaded['versionCode']}") release = {"versionCodes": [str(uploaded["versionCode"])]} else: releases = play.call("GET", f"/{edit}/tracks/{source}").get("releases") or [] if not releases: fail(f"the {source} track holds no release to move") chosen = newest(releases) release = {key: chosen[key] for key in ("versionCodes", "name", "releaseNotes") if chosen.get(key)} print(f"taking version code {', '.join(release['versionCodes'])} from the {source} track") release["status"] = status if fraction: release["userFraction"] = fraction if setting("PLAY_NAME"): release["name"] = setting("PLAY_NAME") if notes: release["releaseNotes"] = notes play.call("PUT", f"/{edit}/tracks/{track}", {"track": track, "releases": [release]}) play.call("POST", f"/{edit}:commit") rollout = f" for {fraction:.0%} of users" if fraction else "" print(f"version code {', '.join(release['versionCodes'])} is {status} on the {track} track{rollout}") if __name__ == "__main__": main()