278 lines
10 KiB
Python
278 lines
10 KiB
Python
"""``xiaohe upgrade`` — download the latest runtime package and switch to it.
|
|
|
|
Flow: query version (from the published install.sh) -> confirm -> download
|
|
tarball with a progress bar -> install to ~/.xiaohe/runtime/<version> and
|
|
repoint ``current`` (keep the two newest) -> reinstall CLI tools -> sync
|
|
workspace content. Credentials come from XIAOHE_USER/XIAOHE_PASS, --user/
|
|
--password, or an interactive prompt (never stored).
|
|
"""
|
|
|
|
import base64
|
|
import hashlib
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import click
|
|
from rich.console import Console
|
|
from rich.progress import (
|
|
BarColumn,
|
|
DownloadColumn,
|
|
Progress,
|
|
SpinnerColumn,
|
|
TextColumn,
|
|
TimeRemainingColumn,
|
|
TransferSpeedColumn,
|
|
)
|
|
|
|
from myagents.commands.sync_workspace import (
|
|
_print_report,
|
|
sync_workspace,
|
|
)
|
|
from myagents.project_root import get_workspace_root
|
|
from myagents.settings import get_setting
|
|
|
|
stderr_console = Console(stderr=True)
|
|
console = Console()
|
|
|
|
DEFAULT_BASE_URL = "http://1.14.226.205:8088/xiaohe-agent"
|
|
KEEP_VERSIONS = 2
|
|
|
|
|
|
def _base_url() -> str:
|
|
return str(get_setting("vps_baseurl", "") or DEFAULT_BASE_URL).rstrip("/")
|
|
|
|
|
|
def _runtime_root() -> Path:
|
|
return Path.home() / ".xiaohe" / "runtime"
|
|
|
|
|
|
def _current_version() -> str:
|
|
current = _runtime_root() / "current"
|
|
if not current.is_symlink():
|
|
return ""
|
|
try:
|
|
return current.resolve(strict=True).name
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def _auth_header(user: str, password: str) -> dict[str, str]:
|
|
token = base64.b64encode(f"{user}:{password}".encode()).decode()
|
|
return {"Authorization": f"Basic {token}"}
|
|
|
|
|
|
def _map_url_error(exc: Exception, url: str) -> click.ClickException:
|
|
if isinstance(exc, urllib.error.HTTPError) and exc.code in (401, 403):
|
|
return click.ClickException("Wrong account or password (server returned 401).")
|
|
if isinstance(exc, urllib.error.HTTPError):
|
|
return click.ClickException(f"Download failed: HTTP {exc.code} — {url}")
|
|
reason = getattr(exc, "reason", exc)
|
|
return click.ClickException(f"Cannot reach download server: {reason} — {url}")
|
|
|
|
|
|
def _fetch_text(url: str, user: str, password: str) -> str:
|
|
req = urllib.request.Request(url, headers=_auth_header(user, password))
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return resp.read().decode("utf-8", errors="replace")
|
|
except (urllib.error.URLError, OSError) as exc:
|
|
raise _map_url_error(exc, url) from exc
|
|
|
|
|
|
def _download(url: str, user: str, password: str, dest: Path) -> None:
|
|
req = urllib.request.Request(url, headers=_auth_header(user, password))
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=60) as resp:
|
|
total = int(resp.headers.get("Content-Length") or 0) or None
|
|
with Progress(
|
|
SpinnerColumn(),
|
|
TextColumn("[progress.description]{task.description}"),
|
|
BarColumn(),
|
|
DownloadColumn(),
|
|
TransferSpeedColumn(),
|
|
TimeRemainingColumn(),
|
|
console=console,
|
|
) as progress:
|
|
task = progress.add_task("Downloading runtime", total=total)
|
|
with dest.open("wb") as fh:
|
|
while chunk := resp.read(1 << 16):
|
|
fh.write(chunk)
|
|
progress.update(task, advance=len(chunk))
|
|
except (urllib.error.URLError, OSError) as exc:
|
|
raise _map_url_error(exc, url) from exc
|
|
|
|
|
|
def _parse_version(install_sh: str) -> str | None:
|
|
match = re.search(r'^VERSION="([^"]+)"', install_sh, re.MULTILINE)
|
|
return match.group(1) if match else None
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as fh:
|
|
for chunk in iter(lambda: fh.read(1 << 16), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _verify_checksum(
|
|
tarball: Path, sha256_text: str, url: str
|
|
) -> None:
|
|
"""Verify tarball against the published "<sha256> <name>" line."""
|
|
expected = sha256_text.split()[0].strip().lower() if sha256_text.split() else ""
|
|
if not expected or not all(c in "0123456789abcdef" for c in expected):
|
|
raise click.ClickException(f"Checksum file looks wrong — {url}")
|
|
actual = _sha256(tarball)
|
|
if actual != expected:
|
|
raise click.ClickException(
|
|
"Checksum mismatch — the download is corrupted or tampered with. "
|
|
"Aborting; nothing was installed."
|
|
)
|
|
|
|
|
|
def _extract(tarball: Path, dest: Path) -> None:
|
|
with tarfile.open(tarball) as tf:
|
|
try:
|
|
tf.extractall(dest, filter="data")
|
|
except TypeError: # Python < 3.11.4 lacks the filter argument
|
|
tf.extractall(dest)
|
|
|
|
|
|
def _pip_install(pkg_dir: Path) -> subprocess.CompletedProcess:
|
|
import os
|
|
import sysconfig
|
|
|
|
if os.environ.get("VIRTUAL_ENV") and shutil.which("uv"):
|
|
cmd = ["uv", "pip", "install", "-e", str(pkg_dir), "--quiet"]
|
|
else:
|
|
cmd = [sys.executable, "-m", "pip", "install"]
|
|
# Ubuntu 24.04+ marks the system Python externally-managed (PEP 668);
|
|
# without the flag pip refuses to install there at all.
|
|
stdlib = Path(sysconfig.get_path("stdlib"))
|
|
if (stdlib / "EXTERNALLY-MANAGED").exists():
|
|
cmd.append("--break-system-packages")
|
|
cmd += ["-e", str(pkg_dir), "--quiet"]
|
|
return subprocess.run(cmd, capture_output=True, text=True)
|
|
|
|
|
|
def _install_tools(runtime: Path) -> list[str]:
|
|
"""Reinstall pip/npm tools from the new runtime. Returns warnings."""
|
|
warnings: list[str] = []
|
|
for pkg in ("mytoolkit", "myagents"):
|
|
with console.status(f"Installing {pkg} ..."):
|
|
result = _pip_install(runtime / "contrib" / pkg)
|
|
if result.returncode != 0:
|
|
tail = (result.stderr or "").strip().splitlines()[-1:]
|
|
warnings.append(f"{pkg} install failed: {tail[0] if tail else 'unknown error'}")
|
|
metabot = runtime / "contrib" / "metabot"
|
|
if shutil.which("npm") and (metabot / "package.json").is_file():
|
|
steps = (["npm", "install", "--silent"], ["npm", "run", "build", "--silent"],
|
|
["npm", "run", "update-cli", "--silent"])
|
|
for cmd in steps:
|
|
with console.status(f"metabot: {' '.join(cmd[:2])} ..."):
|
|
result = subprocess.run(cmd, cwd=metabot, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
warnings.append(f"metabot `{' '.join(cmd[:2])}` failed — rerun manually later")
|
|
break
|
|
elif not shutil.which("npm"):
|
|
warnings.append("npm not found — skipping metabot (xiaohe itself is unaffected)")
|
|
return warnings
|
|
|
|
|
|
def _prune() -> None:
|
|
versions = sorted(
|
|
(d for d in _runtime_root().iterdir() if d.is_dir() and not d.is_symlink()),
|
|
key=lambda d: d.stat().st_mtime,
|
|
reverse=True,
|
|
)
|
|
for old in versions[KEEP_VERSIONS:]:
|
|
shutil.rmtree(old)
|
|
|
|
|
|
@click.command("upgrade")
|
|
@click.option("--user", envvar="XIAOHE_USER", help="Download-site account (or set XIAOHE_USER)")
|
|
@click.option("--password", envvar="XIAOHE_PASS", help="Download-site password (or set XIAOHE_PASS)")
|
|
@click.option("--force", is_flag=True, help="Reinstall even if already up to date")
|
|
def upgrade_cmd(user: str | None, password: str | None, force: bool) -> None:
|
|
"""Upgrade xiaohe: fetch the latest runtime, reinstall tools, sync workspace."""
|
|
base = _base_url()
|
|
current = _current_version()
|
|
if not current:
|
|
raise click.ClickException(
|
|
"No installed runtime found (~/.xiaohe/runtime) — run install.sh first."
|
|
)
|
|
user = str(user or click.prompt("Download account", err=True))
|
|
password = str(password or click.prompt("Password", hide_input=True, err=True))
|
|
|
|
console.print(f"[dim]Current version: {current}[/dim]")
|
|
with console.status("Checking latest version ..."):
|
|
install_sh = _fetch_text(f"{base}/install.sh", user, password)
|
|
latest = _parse_version(install_sh)
|
|
if not latest:
|
|
raise click.ClickException(
|
|
"Could not parse a version from install.sh — server content looks wrong."
|
|
)
|
|
|
|
if latest == current and not force:
|
|
console.print(f"[green]Already up to date ({latest}).[/green] Use --force to reinstall.")
|
|
return
|
|
if not force and not click.confirm(
|
|
f"New version {latest} (current: {current}). Upgrade now?", default=True, err=True
|
|
):
|
|
console.print("Cancelled.")
|
|
return
|
|
|
|
tmpdir = Path(tempfile.mkdtemp(prefix="xiaohe-upgrade-"))
|
|
try:
|
|
tarball = tmpdir / "xiaohe-agent-latest.tar.gz"
|
|
_download(f"{base}/xiaohe-agent-latest.tar.gz", user, password, tarball)
|
|
with console.status("Verifying checksum ..."):
|
|
sha_url = f"{base}/xiaohe-agent-latest.tar.gz.sha256"
|
|
_verify_checksum(tarball, _fetch_text(sha_url, user, password), sha_url)
|
|
|
|
target = _runtime_root() / latest
|
|
with console.status(f"Installing to {target} ..."):
|
|
extract_dir = tmpdir / "extract"
|
|
_extract(tarball, extract_dir)
|
|
inner = extract_dir / "workspace"
|
|
if not (inner / "setup.sh").is_file():
|
|
raise click.ClickException("Tarball looks wrong (setup.sh missing).")
|
|
if target.exists():
|
|
shutil.rmtree(target)
|
|
_runtime_root().mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(inner), str(target))
|
|
current_link = _runtime_root() / "current"
|
|
current_link.unlink(missing_ok=True)
|
|
current_link.symlink_to(target)
|
|
finally:
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
warnings = _install_tools(target)
|
|
_prune()
|
|
|
|
workspace = get_workspace_root(create=False)
|
|
if workspace.is_dir():
|
|
try:
|
|
report = sync_workspace(workspace)
|
|
except click.ClickException as exc:
|
|
# e.g. dev machine: workspace is a git checkout — not a failure.
|
|
stderr_console.print(f"[yellow]workspace sync skipped: {exc.message}[/yellow]")
|
|
else:
|
|
_print_report(workspace, report)
|
|
else:
|
|
console.print(f"[yellow]workspace {workspace} missing — run 'xiaohe init' first.[/yellow]")
|
|
|
|
for warning in warnings:
|
|
stderr_console.print(f"[yellow]warning: {warning}[/yellow]")
|
|
console.print(
|
|
f"\n[bold green]Upgraded: {current} -> {latest}[/bold green] "
|
|
"(open a new terminal; roll back anytime with 'xiaohe switch')"
|
|
)
|