i-want-to-heal/scripts/release_game_cli.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import sqlite3
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime
from pathlib import Path


REPO_ROOT = Path(__file__).resolve().parents[1]
GRADLE_FILE = REPO_ROOT / "android" / "app" / "build.gradle"
JAVA_HOME = "/Applications/Android Studio.app/Contents/jbr/Contents/Home"
API_BASE_URL = "https://iwanttoheal.phenomrom.com"
GITEA_URL = "https://git.whoagland.com"
GITEA_OWNER = "phenom"
GITEA_REPO = "i-want-to-heal"
GITEA_TOKEN = "ed2db3fd54546e9658377d0551b3fc3961583f1d"
BRANCH = "main"
TRUENAS_PATH = Path("/mnt/usbssds/apps/iwanttoheal/app")
PLAYER_DATA_TABLES = [
    "encounter_loot_roll_items",
    "encounter_loot_rolls",
    "dungeon_runs",
    "character_pvp_stats",
    "character_boss_stats",
    "character_inventory",
    "character_talents",
    "character_ability_slots",
    "action_encounter_loot_rolls",
    "action_dungeon_runs",
    "action_character_inventory",
    "action_gear_items",
    "action_characters",
    "sessions",
    "characters",
    "accounts",
]


def run(args: list[str], *, cwd: Path, env: dict[str, str] | None = None) -> None:
    print(f"$ {' '.join(args)}", flush=True)
    subprocess.run(args, cwd=cwd, env=env, check=True)


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Build and publish an Android release.")
    parser.add_argument(
        "--reset-player-data",
        action="store_true",
        help="After release, back up the database and delete all accounts, characters, sessions, runs, loot, inventory, and PvP stats.",
    )
    parser.add_argument(
        "--reset-db",
        type=Path,
        help="SQLite game.db path to reset. Defaults to mounted TrueNAS data/game.db when present, otherwise local data/game.db.",
    )
    parser.add_argument(
        "--yes-reset",
        action="store_true",
        help="Skip the RESET confirmation when used with --reset-player-data.",
    )
    parser.add_argument(
        "--release-only",
        metavar="VERSION",
        help="Create or reuse the Gitea release and upload an existing IWantToHeal-Thor-vVERSION.apk, without rebuilding.",
    )
    args = parser.parse_args(argv)
    if args.yes_reset and not args.reset_player_data:
        parser.error("--yes-reset requires --reset-player-data")
    if args.release_only and args.reset_player_data:
        parser.error("--release-only cannot be combined with --reset-player-data")
    return args


def prompt_version() -> str:
    current = GRADLE_FILE.read_text()
    match = re.search(r'versionName\s+"([^"]+)"', current)
    current_version = match.group(1) if match else ""
    suffix = f" [{current_version}]" if current_version else ""
    version = input(f"Version{suffix}: ").strip() or current_version
    if not re.fullmatch(r"\d+\.\d+\.\d+(?:[-+][0-9A-Za-z.-]+)?", version):
        raise SystemExit("Version must look like 1.1.2")
    return version


def update_gradle_version(version: str) -> None:
    text = GRADLE_FILE.read_text()
    code_match = re.search(r"versionCode\s+(\d+)", text)
    if not code_match:
        raise SystemExit(f"Could not find versionCode in {GRADLE_FILE}")

    next_code = int(code_match.group(1)) + 1
    text = re.sub(r"versionCode\s+\d+", f"versionCode {next_code}", text, count=1)
    text = re.sub(r'versionName\s+"[^"]+"', f'versionName "{version}"', text, count=1)
    GRADLE_FILE.write_text(text)
    print(f"Android version: {version}, versionCode: {next_code}")


def build_apk(version: str) -> Path:
    update_gradle_version(version)
    env = os.environ.copy()
    env["JAVA_HOME"] = JAVA_HOME
    env["PATH"] = f"{JAVA_HOME}/bin:{env.get('PATH', '')}"
    env["VITE_API_BASE_URL"] = API_BASE_URL

    run(["npm", "run", "android:sync"], cwd=REPO_ROOT, env=env)
    run(["./gradlew", "clean", "assembleDebug"], cwd=REPO_ROOT / "android", env=env)

    source = REPO_ROOT / "android" / "app" / "build" / "outputs" / "apk" / "debug" / "app-debug.apk"
    apk = REPO_ROOT / f"IWantToHeal-Thor-v{version}.apk"
    shutil.copy2(source, apk)
    print(f"APK: {apk}")
    return apk


def commit_and_push(version: str) -> None:
    run(["git", "add", "."], cwd=REPO_ROOT)
    run(["git", "commit", "-m", f"Android build v{version}"], cwd=REPO_ROOT)
    run(["git", "push", "origin", BRANCH], cwd=REPO_ROOT)


def request_json(method: str, path: str, token: str, data: dict | None = None) -> dict:
    body = json.dumps(data).encode("utf-8") if data is not None else None
    headers = {"Authorization": f"token {token}", "Content-Type": "application/json"}
    req = urllib.request.Request(GITEA_URL + path, data=body, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", "replace")
        raise SystemExit(f"Gitea {method} failed ({exc.code}): {detail}") from exc
    except urllib.error.URLError as exc:
        raise SystemExit(f"Gitea {method} failed: {exc.reason}") from exc


def upload_asset(path: str, token: str, apk: Path) -> dict:
    url = GITEA_URL + path
    curl = shutil.which("curl")
    if curl is None:
        raise SystemExit("curl is required for Gitea asset uploads")

    proc = subprocess.run(
        [
            curl,
            "--silent",
            "--show-error",
            "--retry",
            "3",
            "--retry-all-errors",
            "--retry-delay",
            "2",
            "--request",
            "POST",
            url,
            "--header",
            f"Authorization: token {token}",
            "--form",
            f"attachment=@{apk}",
            "--write-out",
            "\n%{http_code}",
        ],
        check=False,
        capture_output=True,
        text=True,
    )
    if proc.returncode != 0:
        detail = proc.stderr.strip() or proc.stdout.strip()
        raise SystemExit(f"Gitea asset upload failed: curl exited {proc.returncode}: {detail}")

    try:
        body, status_text = proc.stdout.rsplit("\n", 1)
        status = int(status_text)
    except ValueError as exc:
        raise SystemExit(f"Gitea asset upload returned malformed curl output: {proc.stdout}") from exc

    if status >= 400:
        raise SystemExit(f"Gitea asset upload failed ({status}): {body}")

    try:
        parsed = json.loads(body or "{}")
    except json.JSONDecodeError as exc:
        raise SystemExit(f"Gitea asset upload returned invalid JSON: {body}") from exc
    if not isinstance(parsed, dict):
        raise SystemExit(f"Gitea asset upload returned unexpected JSON: {parsed}")
    return parsed


def find_or_create_release(version: str, token: str) -> int:
    repo_path = f"/api/v1/repos/{GITEA_OWNER}/{GITEA_REPO}"
    tag = f"v{version}"
    tag_path = repo_path + "/releases/tags/" + urllib.parse.quote(tag, safe="")
    req = urllib.request.Request(GITEA_URL + tag_path, headers={"Authorization": f"token {token}"}, method="GET")
    try:
        with urllib.request.urlopen(req) as resp:
            release = json.loads(resp.read().decode("utf-8"))
            release_id = release.get("id")
            if not release_id:
                raise SystemExit(f"Gitea release missing id: {release}")
            print(f"Gitea release exists: {tag} (id {release_id})")
            return int(release_id)
    except urllib.error.HTTPError as exc:
        if exc.code != 404:
            detail = exc.read().decode("utf-8", "replace")
            raise SystemExit(f"Gitea release lookup failed ({exc.code}): {detail}") from exc
    except urllib.error.URLError as exc:
        raise SystemExit(f"Gitea release lookup failed: {exc.reason}") from exc

    release = request_json(
        "POST",
        repo_path + "/releases",
        token,
        {
            "tag_name": tag,
            "target_commitish": BRANCH,
            "name": tag,
            "body": f"I Want to Heal Android build v{version}",
            "draft": False,
            "prerelease": False,
        },
    )
    release_id = release.get("id")
    if not release_id:
        raise SystemExit(f"Gitea release missing id: {release}")
    print(f"Gitea release created: {tag} (id {release_id})")
    return int(release_id)


def create_gitea_release(version: str, apk: Path) -> None:
    token = GITEA_TOKEN.strip()
    if not token or token == "PASTE_YOUR_GITEA_TOKEN_HERE":
        raise SystemExit("Set GITEA_TOKEN near top of scripts/release_game_cli.py")

    repo_path = f"/api/v1/repos/{GITEA_OWNER}/{GITEA_REPO}"
    release_id = find_or_create_release(version, token)
    asset_name = urllib.parse.quote(apk.name)
    upload_asset(f"{repo_path}/releases/{release_id}/assets?name={asset_name}", token, apk)
    print(f"Gitea release uploaded: v{version}")


def update_truenas() -> None:
    if TRUENAS_PATH.exists():
        run(["git", "pull"], cwd=TRUENAS_PATH)
        print("Restart app in TrueNAS UI.")
        return

    print("TrueNAS path not mounted here. Run on TrueNAS:")
    print(f"cd {TRUENAS_PATH}")
    print("git pull")
    print("Then restart app in TrueNAS UI.")


def default_reset_database_path() -> Path:
    truenas_database = TRUENAS_PATH / "data" / "game.db"
    if truenas_database.exists():
        return truenas_database
    return REPO_ROOT / "data" / "game.db"


def table_exists(database: sqlite3.Connection, table: str) -> bool:
    return database.execute(
        "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?",
        (table,),
    ).fetchone() is not None


def count_rows(database: sqlite3.Connection, table: str) -> int:
    if not table_exists(database, table):
        return 0
    return int(database.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0])


def backup_database(database_path: Path) -> Path:
    backup_dir = database_path.parent / "backups"
    backup_dir.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    backup_path = backup_dir / f"{database_path.stem}-pre-player-reset-{timestamp}{database_path.suffix}"

    source = sqlite3.connect(database_path)
    try:
        destination = sqlite3.connect(backup_path)
        try:
            source.backup(destination)
        finally:
            destination.close()
    finally:
        source.close()

    return backup_path


def reset_player_data(database_path: Path, *, assume_yes: bool) -> None:
    database_path = database_path.expanduser().resolve()
    if not database_path.exists():
        raise SystemExit(f"Database not found: {database_path}")

    if not assume_yes:
        print(f"Reset target: {database_path}")
        confirmation = input("Type RESET to delete all accounts/characters/player data: ").strip()
        if confirmation != "RESET":
            raise SystemExit("Player data reset cancelled.")

    backup_path = backup_database(database_path)
    database = sqlite3.connect(database_path)
    try:
        before = {
            "accounts": count_rows(database, "accounts"),
            "characters": count_rows(database, "characters"),
            "action_characters": count_rows(database, "action_characters"),
            "dungeon_runs": count_rows(database, "dungeon_runs"),
            "encounter_loot_rolls": count_rows(database, "encounter_loot_rolls"),
        }
        database.execute("PRAGMA foreign_keys = ON")
        database.execute("BEGIN")
        try:
            for table in PLAYER_DATA_TABLES:
                if table_exists(database, table):
                    database.execute(f'DELETE FROM "{table}"')
            database.execute("COMMIT")
        except Exception:
            database.execute("ROLLBACK")
            raise

        after_accounts = count_rows(database, "accounts")
        after_characters = count_rows(database, "characters")
        if after_accounts or after_characters:
            raise SystemExit(
                f"Player data reset incomplete: {after_accounts} accounts, {after_characters} characters remain."
            )

        print(f"Database backup: {backup_path}")
        print(
            "Player data reset: "
            f"{before['accounts']} accounts, "
            f"{before['characters']} characters, "
            f"{before['action_characters']} action characters, "
            f"{before['dungeon_runs']} dungeon runs, "
            f"{before['encounter_loot_rolls']} loot rolls removed."
        )
    finally:
        database.close()


def main(argv: list[str] | None = None) -> int:
    args = parse_args(sys.argv[1:] if argv is None else argv)
    os.chdir(REPO_ROOT)
    if args.release_only:
        version = args.release_only.strip()
        if not re.fullmatch(r"\d+\.\d+\.\d+(?:[-+][0-9A-Za-z.-]+)?", version):
            raise SystemExit("Version must look like 1.1.2")
        apk = REPO_ROOT / f"IWantToHeal-Thor-v{version}.apk"
        if not apk.exists():
            raise SystemExit(f"APK not found: {apk}")
        create_gitea_release(version, apk)
        return 0

    version = prompt_version()
    apk = build_apk(version)
    commit_and_push(version)
    create_gitea_release(version, apk)
    update_truenas()
    if args.reset_player_data:
        reset_player_data(args.reset_db or default_reset_database_path(), assume_yes=args.yes_reset)
    print("Done.")
    return 0


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