app-stores/play/play.py

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 <language>.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()