I Want To Heal 2 build v1.0.5 code

scripts/deploy_gui.py

@@ -0,0 +1,562 @@
+#!/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()