563 lines
20 KiB
Python
Executable File
563 lines
20 KiB
Python
Executable File
#!/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()
|