#!/usr/bin/env python3 """Tkinter deployment helper for I Want To Heal 2.""" from __future__ import annotations import json import mimetypes import os import queue import re import shlex import subprocess import sys import threading import urllib.error import urllib.parse import urllib.request import uuid from pathlib import Path def relaunch_with_tk_python() -> None: if os.environ.get("DEPLOY_GUI_TK_RELAUNCH") == "1": return candidates = [ "/Library/Frameworks/Python.framework/Versions/3.12/bin/python3", "/usr/local/bin/python3", "/usr/bin/python3", ] for candidate in candidates: if Path(candidate) == Path(sys.executable): continue probe = subprocess.run( [candidate, "-c", "import tkinter"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) if probe.returncode == 0: env = os.environ.copy() env["DEPLOY_GUI_TK_RELAUNCH"] = "1" os.execve(candidate, [candidate, *sys.argv], env) try: from tkinter import END, StringVar, Text, Tk, messagebox from tkinter import ttk from tkinter.scrolledtext import ScrolledText except ModuleNotFoundError as error: if error.name != "_tkinter": raise relaunch_with_tk_python() raise SystemExit( "Tkinter is not available in this Python. Run scripts/deploy-gui.sh " "or install python.org Python 3 with Tk support." ) ROOT_DIR = Path(__file__).resolve().parents[1] VERSION_RE = re.compile(r"^[0-9]+[.][0-9]+[.][0-9]+$") DEFAULT_GITEA_URL = "https://git.whoagland.com" DEFAULT_GITEA_OWNER = "phenom" DEFAULT_GITEA_REPO = "i-want-to-heal-2" DEFAULT_BRANCH = "main" MAX_STAGED_FILE_BYTES = 200 * 1024 * 1024 class ApiError(RuntimeError): def __init__(self, status: int, message: str, body: str = "") -> None: super().__init__(message) self.status = status self.body = body class CommandError(RuntimeError): def __init__(self, command: list[str], return_code: int, output: str) -> None: super().__init__(f"Command failed with exit {return_code}: {shell_join(command)}") self.command = command self.return_code = return_code self.output = output def shell_join(command: list[str]) -> str: return shlex.join(command) def run_command( command: list[str], env: dict[str, str], log: callable, *, check: bool = True, ) -> int: log(f"$ {shell_join(command)}") process = subprocess.Popen( command, cwd=ROOT_DIR, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, ) assert process.stdout is not None output_lines = [] for line in process.stdout: clean_line = line.rstrip("\n") output_lines.append(clean_line) log(clean_line) return_code = process.wait() if check and return_code != 0: raise CommandError(command, return_code, "\n".join(output_lines)) return return_code def read_command(command: list[str], env: dict[str, str]) -> subprocess.CompletedProcess[str]: return subprocess.run( command, cwd=ROOT_DIR, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) def request_json( method: str, url: str, token: str, *, payload: dict | None = None, ) -> dict: data = None headers = { "Accept": "application/json", "Authorization": f"token {token}", } if payload is not None: data = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" request = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(request, timeout=60) as response: body = response.read().decode("utf-8") except urllib.error.HTTPError as error: body = error.read().decode("utf-8", errors="replace") raise ApiError(error.code, f"Gitea API returned HTTP {error.code}", body) from error if not body: return {} return json.loads(body) def upload_release_asset(url: str, token: str, apk_path: Path) -> dict: boundary = f"----codex-deploy-{uuid.uuid4().hex}" content_type = mimetypes.guess_type(apk_path.name)[0] or "application/octet-stream" file_bytes = apk_path.read_bytes() parts = [ f"--{boundary}\r\n".encode("utf-8"), ( 'Content-Disposition: form-data; name="attachment"; ' f'filename="{apk_path.name}"\r\n' ).encode("utf-8"), f"Content-Type: {content_type}\r\n\r\n".encode("utf-8"), file_bytes, f"\r\n--{boundary}--\r\n".encode("utf-8"), ] data = b"".join(parts) request = urllib.request.Request( url, data=data, headers={ "Authorization": f"token {token}", "Content-Type": f"multipart/form-data; boundary={boundary}", "Accept": "application/json", }, method="POST", ) try: with urllib.request.urlopen(request, timeout=180) as response: body = response.read().decode("utf-8") except urllib.error.HTTPError as error: body = error.read().decode("utf-8", errors="replace") raise ApiError(error.code, f"Asset upload returned HTTP {error.code}", body) from error return json.loads(body) if body else {} def api_base(gitea_url: str, owner: str, repo: str) -> str: return ( f"{gitea_url.rstrip('/')}/api/v1/repos/" f"{urllib.parse.quote(owner)}/{urllib.parse.quote(repo)}" ) def find_oversized_staged_files(env: dict[str, str]) -> list[tuple[str, int]]: result = subprocess.run( ["git", "diff", "--cached", "--name-only", "-z", "--diff-filter=ACMR"], cwd=ROOT_DIR, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) if result.returncode != 0: raise RuntimeError(result.stderr.decode("utf-8", errors="replace").strip()) oversized = [] for raw_path in result.stdout.split(b"\0"): if not raw_path: continue path = raw_path.decode("utf-8", errors="surrogateescape") file_path = ROOT_DIR / path if file_path.exists() and file_path.stat().st_size > MAX_STAGED_FILE_BYTES: oversized.append((path, file_path.stat().st_size)) return oversized def verify_remote_branch_matches(env: dict[str, str], branch: str) -> bool: local = read_command(["git", "rev-parse", branch], env) remote = read_command(["git", "ls-remote", "origin", f"refs/heads/{branch}"], env) if local.returncode != 0 or remote.returncode != 0: return False local_hash = local.stdout.strip() remote_hash = remote.stdout.split()[0] if remote.stdout.strip() else "" return bool(local_hash and remote_hash and local_hash == remote_hash) def create_or_get_release( gitea_url: str, owner: str, repo: str, branch: str, token: str, version: str, notes: str, log: callable, ) -> dict: base = api_base(gitea_url, owner, repo) tag = f"v{version}" payload = { "tag_name": tag, "target_commitish": branch, "name": tag, "body": notes, "draft": False, "prerelease": False, } log(f"Creating Gitea release {tag}") try: return request_json("POST", f"{base}/releases", token, payload=payload) except ApiError as error: if error.status != 409: raise log(f"Release {tag} already exists; using existing release") encoded_tag = urllib.parse.quote(tag, safe="") return request_json("GET", f"{base}/releases/tags/{encoded_tag}", token) def upload_apk_asset( gitea_url: str, owner: str, repo: str, token: str, release_id: int, apk_path: Path, log: callable, ) -> dict: base = api_base(gitea_url, owner, repo) asset_name = urllib.parse.quote(apk_path.name) url = f"{base}/releases/{release_id}/assets?name={asset_name}" log(f"Uploading APK asset {apk_path.name}") return upload_release_asset(url, token, apk_path) class DeployGui: def __init__(self) -> None: self.root = Tk() self.root.title("I Want To Heal 2 Deployment") self.root.geometry("980x720") self.log_queue: queue.Queue[tuple[str, str]] = queue.Queue() self.worker: threading.Thread | None = None self.version = StringVar(value=self.detect_version()) self.gitea_url = StringVar(value=DEFAULT_GITEA_URL) self.gitea_owner = StringVar(value=DEFAULT_GITEA_OWNER) self.gitea_repo = StringVar(value=DEFAULT_GITEA_REPO) self.branch = StringVar(value=DEFAULT_BRANCH) self.token = StringVar(value=os.environ.get("GITEA_TOKEN", "")) self.status = StringVar(value=f"Repo: {ROOT_DIR}") self.build_apk = StringVar(value="1") self.run_checks = StringVar(value="1") self.commit_push = StringVar(value="1") self.create_release = StringVar(value="1") self.default_notes = self.build_default_notes(self.version.get()) self.build_ui() self.root.after(100, self.drain_log_queue) def detect_version(self) -> str: build_file = ROOT_DIR / "android" / "app" / "build.gradle" if build_file.exists(): match = re.search(r'versionName\s+"([^"]+)"', build_file.read_text()) if match: return match.group(1) return "1.0.0" def build_ui(self) -> None: self.root.columnconfigure(0, weight=1) self.root.rowconfigure(3, weight=1) main = ttk.Frame(self.root, padding=14) main.grid(row=0, column=0, sticky="nsew") main.columnconfigure(1, weight=1) ttk.Label(main, text="APK version").grid(row=0, column=0, sticky="w") ttk.Entry(main, textvariable=self.version, width=20).grid(row=0, column=1, sticky="w") ttk.Label(main, text="Gitea token").grid(row=1, column=0, sticky="w", pady=(8, 0)) ttk.Entry(main, textvariable=self.token, show="*", width=48).grid( row=1, column=1, sticky="ew", pady=(8, 0) ) repo_frame = ttk.Frame(main) repo_frame.grid(row=2, column=0, columnspan=2, sticky="ew", pady=(10, 0)) for index in range(8): repo_frame.columnconfigure(index, weight=1 if index in (1, 3, 5, 7) else 0) ttk.Label(repo_frame, text="URL").grid(row=0, column=0, sticky="w") ttk.Entry(repo_frame, textvariable=self.gitea_url, width=28).grid(row=0, column=1, sticky="ew") ttk.Label(repo_frame, text="Owner").grid(row=0, column=2, sticky="w", padx=(10, 0)) ttk.Entry(repo_frame, textvariable=self.gitea_owner, width=14).grid(row=0, column=3, sticky="ew") ttk.Label(repo_frame, text="Repo").grid(row=0, column=4, sticky="w", padx=(10, 0)) ttk.Entry(repo_frame, textvariable=self.gitea_repo, width=20).grid(row=0, column=5, sticky="ew") ttk.Label(repo_frame, text="Branch").grid(row=0, column=6, sticky="w", padx=(10, 0)) ttk.Entry(repo_frame, textvariable=self.branch, width=10).grid(row=0, column=7, sticky="ew") steps = ttk.LabelFrame(main, text="Steps") steps.grid(row=3, column=0, columnspan=2, sticky="ew", pady=(12, 0)) ttk.Checkbutton( steps, text="Build Thor APK with scripts/build-thor-apk.sh", variable=self.build_apk, onvalue="1", offvalue="0", ).grid(row=0, column=0, sticky="w", padx=8, pady=4) ttk.Checkbutton( steps, text="Run npm lint and npm build", variable=self.run_checks, onvalue="1", offvalue="0", ).grid(row=0, column=1, sticky="w", padx=8, pady=4) ttk.Checkbutton( steps, text="Commit all repo changes and push", variable=self.commit_push, onvalue="1", offvalue="0", ).grid(row=1, column=0, sticky="w", padx=8, pady=4) ttk.Checkbutton( steps, text="Create Gitea release and upload APK", variable=self.create_release, onvalue="1", offvalue="0", ).grid(row=1, column=1, sticky="w", padx=8, pady=4) notes_frame = ttk.Frame(self.root, padding=(14, 0, 14, 0)) notes_frame.grid(row=1, column=0, sticky="ew") notes_frame.columnconfigure(0, weight=1) ttk.Label(notes_frame, text="Release notes").grid(row=0, column=0, sticky="w") self.notes = Text(notes_frame, height=6, wrap="word") self.notes.grid(row=1, column=0, sticky="ew") self.notes.insert("1.0", self.default_notes) buttons = ttk.Frame(self.root, padding=14) buttons.grid(row=2, column=0, sticky="ew") buttons.columnconfigure(1, weight=1) self.run_button = ttk.Button(buttons, text="Run Deployment", command=self.start_deploy) self.run_button.grid(row=0, column=0, sticky="w") ttk.Label(buttons, textvariable=self.status).grid(row=0, column=1, sticky="w", padx=(12, 0)) log_frame = ttk.Frame(self.root, padding=(14, 0, 14, 14)) log_frame.grid(row=3, column=0, sticky="nsew") log_frame.columnconfigure(0, weight=1) log_frame.rowconfigure(0, weight=1) self.log_text = ScrolledText(log_frame, height=22, wrap="word") self.log_text.grid(row=0, column=0, sticky="nsew") self.log_text.configure(state="disabled") def log(self, message: str, level: str = "info") -> None: self.log_queue.put((level, message)) def drain_log_queue(self) -> None: try: while True: level, message = self.log_queue.get_nowait() if level == "status": self.status.set(message) continue self.log_text.configure(state="normal") self.log_text.insert(END, message + "\n") self.log_text.see(END) self.log_text.configure(state="disabled") except queue.Empty: pass self.root.after(100, self.drain_log_queue) def start_deploy(self) -> None: if self.worker and self.worker.is_alive(): return version = self.version.get().strip() token = self.token.get().strip() if not VERSION_RE.match(version): messagebox.showerror("Invalid version", "Version must look like 1.0.2") return if self.create_release.get() == "1" and not token: messagebox.showerror("Missing token", "Paste a Gitea API token or export GITEA_TOKEN.") return notes = self.notes.get("1.0", END).strip() if not notes or notes == self.default_notes: notes = self.build_default_notes(version) self.log_text.configure(state="normal") self.log_text.delete("1.0", END) self.log_text.configure(state="disabled") self.run_button.configure(state="disabled") self.status.set("Deployment running") args = { "version": version, "notes": notes, "token": token, "gitea_url": self.gitea_url.get().strip(), "owner": self.gitea_owner.get().strip(), "repo": self.gitea_repo.get().strip(), "branch": self.branch.get().strip(), "build_apk": self.build_apk.get() == "1", "run_checks": self.run_checks.get() == "1", "commit_push": self.commit_push.get() == "1", "create_release": self.create_release.get() == "1", } self.worker = threading.Thread(target=self.deploy, kwargs=args, daemon=True) self.worker.start() def deploy( self, *, version: str, notes: str, token: str, gitea_url: str, owner: str, repo: str, branch: str, build_apk: bool, run_checks: bool, commit_push: bool, create_release: bool, ) -> None: try: env = os.environ.copy() if token: env["GITEA_TOKEN"] = token self.log(f"Repo: {ROOT_DIR}") self.log(f"Version: {version}") if build_apk: run_command(["scripts/build-thor-apk.sh", version], env, self.log) if run_checks: run_command(["npm", "run", "lint"], env, self.log) run_command(["npm", "run", "build"], env, self.log) if commit_push: status = read_command(["git", "status", "--short"], env) if status.stdout.strip(): self.log("Git changes before commit:") self.log(status.stdout.rstrip()) run_command(["git", "add", "."], env, self.log) diff = read_command(["git", "diff", "--cached", "--quiet"], env) if diff.returncode == 0: self.log("No staged changes; skipping commit") else: oversized = find_oversized_staged_files(env) if oversized: formatted = "\n".join( f"- {path} ({size / 1024 / 1024:.1f} MiB)" for path, size in oversized ) raise RuntimeError( "Refusing to commit oversized files. Add them to .gitignore " f"or upload them as release assets instead:\n{formatted}" ) run_command( ["git", "commit", "-m", f"I Want To Heal 2 build v{version}"], env, self.log, ) try: run_command(["git", "push", "origin", branch], env, self.log) except CommandError: if verify_remote_branch_matches(env, branch): self.log("Push reported an error, but remote branch matches local commit") else: raise if create_release: apk_path = ROOT_DIR / f"IWantToHeal2-Thor-v{version}.apk" if not apk_path.exists(): raise RuntimeError(f"APK missing: {apk_path}") release = create_or_get_release( gitea_url, owner, repo, branch, token, version, notes, self.log, ) release_id = release.get("id") if not release_id: raise RuntimeError(f"Release response did not include id: {release}") self.log(f"Release ID: {release_id}") asset = upload_apk_asset( gitea_url, owner, repo, token, int(release_id), apk_path, self.log, ) self.log(f"Uploaded asset: {asset.get('name', apk_path.name)}") self.log("Deployment complete") self.log_queue.put(("status", "Deployment complete")) except Exception as error: # noqa: BLE001 - display deployment failures in GUI. self.log(f"ERROR: {error}") if isinstance(error, ApiError) and error.body: self.log(error.body) self.log_queue.put(("status", "Deployment failed")) finally: self.root.after(0, lambda: self.run_button.configure(state="normal")) def run(self) -> None: self.root.mainloop() @staticmethod def build_default_notes(version: str) -> str: return f"I Want To Heal 2 Android and web/server build v{version}" if __name__ == "__main__": DeployGui().run()