app_for_indeed/app-connect/scripts/update_app_store_metadata.py

#!/usr/bin/env python3
"""Push app-data/*.txt to App Store Connect via the official API."""

from __future__ import annotations

import json
import os
import sys
import time
from pathlib import Path

import jwt
import requests

API_BASE = "https://api.appstoreconnect.apple.com/v1"
ROOT = Path(__file__).resolve().parents[1]
APP_DATA = ROOT / "app-data"
ENV_PATH = ROOT / ".env"
VERSION_STRING = os.environ.get("APP_STORE_VERSION", "1.0")
LOCALE = os.environ.get("APP_STORE_LOCALE", "en-US")


def load_env(path: Path) -> dict[str, str]:
    values: dict[str, str] = {}
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        values[key.strip()] = value.strip()
    return values


def read_field(name: str) -> str:
    path = APP_DATA / name
    if not path.exists():
        raise FileNotFoundError(path)
    return path.read_text(encoding="utf-8").strip()


class ASCClient:
    def __init__(self, issuer_id: str, key_id: str, private_key_path: str):
        self.issuer_id = issuer_id
        self.key_id = key_id
        self.private_key = Path(private_key_path).expanduser().read_text(encoding="utf-8")
        self.session = requests.Session()
        self._token: str | None = None
        self._token_exp = 0.0

    def _ensure_token(self) -> str:
        now = time.time()
        if self._token and now < self._token_exp - 60:
            return self._token
        issued_at = int(now)
        payload = {
            "iss": self.issuer_id,
            "iat": issued_at,
            "exp": issued_at + 1200,
            "aud": "appstoreconnect-v1",
        }
        self._token = jwt.encode(
            payload,
            self.private_key,
            algorithm="ES256",
            headers={"kid": self.key_id, "typ": "JWT"},
        )
        self._token_exp = issued_at + 1200
        return self._token

    def request(self, method: str, path: str, **kwargs) -> requests.Response:
        token = self._ensure_token()
        headers = kwargs.pop("headers", {})
        headers["Authorization"] = f"Bearer {token}"
        headers.setdefault("Content-Type", "application/json")
        url = path if path.startswith("http") else f"{API_BASE}{path}"
        response = self.session.request(method, url, headers=headers, timeout=60, **kwargs)
        return response

    def get_json(self, path: str, params: dict | None = None) -> dict:
        response = self.request("GET", path, params=params)
        if not response.ok:
            raise RuntimeError(f"GET {path} failed ({response.status_code}): {response.text}")
        return response.json()

    def patch(self, resource_type: str, resource_id: str, attributes: dict) -> dict:
        body = {
            "data": {
                "type": resource_type,
                "id": resource_id,
                "attributes": attributes,
            }
        }
        response = self.request("PATCH", f"/{resource_type}/{resource_id}", json=body)
        if not response.ok:
            raise RuntimeError(
                f"PATCH {resource_type}/{resource_id} failed ({response.status_code}): {response.text}"
            )
        return response.json()

    def post(self, resource_type: str, attributes: dict, relationships: dict) -> dict:
        body = {
            "data": {
                "type": resource_type,
                "attributes": attributes,
                "relationships": relationships,
            }
        }
        response = self.request("POST", f"/{resource_type}", json=body)
        if not response.ok:
            raise RuntimeError(
                f"POST {resource_type} failed ({response.status_code}): {response.text}"
            )
        return response.json()["data"]


def find_app(client: ASCClient, bundle_id: str) -> dict:
    data = client.get_json("/apps", params={"filter[bundleId]": bundle_id, "limit": 1})
    apps = data.get("data", [])
    if not apps:
        raise RuntimeError(f"No app found for bundle id {bundle_id!r}")
    return apps[0]


def find_mac_app_info(client: ASCClient, app_id: str) -> dict:
    data = client.get_json(f"/apps/{app_id}/appInfos")
    infos = data.get("data", [])
    if not infos:
        raise RuntimeError(f"No appInfos for app {app_id}")

    editable_states = {
        "PREPARE_FOR_SUBMISSION",
        "DEVELOPER_REJECTED",
        "REJECTED",
        "METADATA_REJECTED",
    }
    for info in infos:
        state = info.get("attributes", {}).get("appStoreState")
        if state in editable_states:
            return info

    for info in infos:
        info_id = info["id"]
        detail = client.get_json(f"/appInfos/{info_id}")
        platform = detail.get("data", {}).get("attributes", {}).get("platform")
        if platform == "MAC_OS":
            return detail["data"]

    if len(infos) == 1:
        return infos[0]
    return infos[0]


def find_app_info_localization(client: ASCClient, app_info_id: str, locale: str) -> dict:
    data = client.get_json(f"/appInfos/{app_info_id}/appInfoLocalizations")
    for loc in data.get("data", []):
        if loc.get("attributes", {}).get("locale") == locale:
            return loc
    raise RuntimeError(f"No appInfoLocalization for locale {locale!r}")


def find_version(client: ASCClient, app_id: str, version_string: str, *, create: bool = False) -> dict:
    data = client.get_json(
        f"/apps/{app_id}/appStoreVersions",
        params={"filter[platform]": "MAC_OS", "limit": 50},
    )
    matches = [
        v
        for v in data.get("data", [])
        if v.get("attributes", {}).get("versionString") == version_string
    ]
    if not matches:
        if not create:
            raise RuntimeError(f"No MAC_OS appStoreVersion {version_string!r}")
        created = client.post(
            "appStoreVersions",
            {"platform": "MAC_OS", "versionString": version_string},
            {"app": {"data": {"type": "apps", "id": app_id}}},
        )
        print(f"Created version {version_string}: {created['id']}")
        return created
    # Prefer editable state if multiple.
    for v in matches:
        state = v.get("attributes", {}).get("appStoreState")
        if state in ("PREPARE_FOR_SUBMISSION", "DEVELOPER_REJECTED", "REJECTED", "METADATA_REJECTED"):
            return v
    return matches[0]


def find_version_localization(
    client: ASCClient, version_id: str, locale: str, *, create: bool = False
) -> dict:
    data = client.get_json(f"/appStoreVersions/{version_id}/appStoreVersionLocalizations")
    for loc in data.get("data", []):
        if loc.get("attributes", {}).get("locale") == locale:
            return loc
    if not create:
        raise RuntimeError(f"No appStoreVersionLocalization for locale {locale!r}")
    created = client.post(
        "appStoreVersionLocalizations",
        {"locale": locale},
        {"appStoreVersion": {"data": {"type": "appStoreVersions", "id": version_id}}},
    )
    print(f"Created version localization ({locale}): {created['id']}")
    return created


def main() -> int:
    if not ENV_PATH.exists():
        print(f"Missing {ENV_PATH}", file=sys.stderr)
        return 1

    env = load_env(ENV_PATH)
    issuer = env.get("APP_STORE_CONNECT_ISSUER_ID", "")
    key_id = env.get("APP_STORE_CONNECT_KEY_ID", "")
    key_path = env.get("APP_STORE_CONNECT_PRIVATE_KEY_PATH", "")
    bundle_id = env.get("APP_STORE_CONNECT_BUNDLE_ID", "")

    for key, value in [
        ("APP_STORE_CONNECT_ISSUER_ID", issuer),
        ("APP_STORE_CONNECT_KEY_ID", key_id),
        ("APP_STORE_CONNECT_PRIVATE_KEY_PATH", key_path),
        ("APP_STORE_CONNECT_BUNDLE_ID", bundle_id),
    ]:
        if not value:
            print(f"Set {key} in {ENV_PATH}", file=sys.stderr)
            return 1

    client = ASCClient(issuer, key_id, key_path)
    app = find_app(client, bundle_id)
    app_id = app["id"]
    print(f"App: {app['attributes'].get('name')} ({app_id})")

    app_info = find_mac_app_info(client, app_id)
    app_info_id = app_info["id"]
    print(f"App Information (MAC_OS): {app_info_id}")

    info_loc = find_app_info_localization(client, app_info_id, LOCALE)
    info_loc_id = info_loc["id"]

    app_name = read_field("app-name.txt")
    subtitle = read_field("subtitle.txt")
    privacy_url = read_field("link-privacy.txt")

    info_patch = {
        "name": app_name,
        "subtitle": subtitle,
        "privacyPolicyUrl": privacy_url,
    }
    client.patch("appInfoLocalizations", info_loc_id, info_patch)
    print(f"Updated App Information localization ({LOCALE}): name, subtitle, privacyPolicyUrl")

    create_version = os.environ.get("APP_STORE_CREATE_VERSION", "").lower() in ("1", "true", "yes")
    version = find_version(client, app_id, VERSION_STRING, create=create_version)
    version_id = version["id"]
    version_state = version.get("attributes", {}).get("appStoreState")
    print(f"Version {VERSION_STRING}: {version_id} (state: {version_state})")

    copyright_text = read_field("copyright.txt")
    client.patch("appStoreVersions", version_id, {"copyright": copyright_text})
    print(f"Updated version copyright")

    version_loc = find_version_localization(client, version_id, LOCALE, create=create_version)
    version_loc_id = version_loc["id"]

    version_patch = {
        "description": read_field("description.txt"),
        "keywords": read_field("keywords.txt"),
        "promotionalText": read_field("promotional-text.txt"),
        "marketingUrl": read_field("link-marketing.txt"),
        "supportUrl": read_field("link-support.txt"),
    }
    whats_new_path = APP_DATA / "whats-new.txt"
    if whats_new_path.exists():
        version_patch["whatsNew"] = whats_new_path.read_text(encoding="utf-8").strip()
    client.patch("appStoreVersionLocalizations", version_loc_id, version_patch)
    fields = "description, keywords, promotionalText, marketingUrl, supportUrl"
    if "whatsNew" in version_patch:
        fields += ", whatsNew"
    print(f"Updated version {VERSION_STRING} localization ({LOCALE}): {fields}")

    # Custom EULA URL (Terms) — attach to app if not using standard Apple EULA.
    terms_url = read_field("link-terms.txt")
    try:
        eula_data = client.get_json(f"/apps/{app_id}/endUserLicenseAgreement")
        eula = eula_data.get("data")
        if eula:
            # Custom EULA resource uses agreementText, not URL in all cases; try app-level license.
            print("App has custom EULA resource; update Terms URL in App Store Connect if needed:", terms_url)
        else:
            print("Using standard Apple EULA; Terms URL for reference:", terms_url)
    except RuntimeError:
        print("EULA lookup skipped; Terms URL:", terms_url)

    print("Done.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())