AutoDeploy/deploy_tool/installer.py

from __future__ import annotations

import re
import hashlib
import os
import shutil
import subprocess
import tempfile
import urllib.request
import zipfile
from dataclasses import dataclass
from pathlib import Path
from typing import BinaryIO, Callable

from .scanner import ScanResult


JDK_PACKAGES = {
    "8": "EclipseAdoptium.Temurin.8.JDK",
    "11": "EclipseAdoptium.Temurin.11.JDK",
    "17": "EclipseAdoptium.Temurin.17.JDK",
    "21": "EclipseAdoptium.Temurin.21.JDK",
}

WINGET_PACKAGES = {
    "Gradle": "Gradle.Gradle",
    "Node.js": "OpenJS.NodeJS.LTS",
    "Git": "Git.Git",
    "MySQL": "Oracle.MySQL",
}

MAVEN_VERSION = "3.10.0"
MAVEN_ARCHIVE_NAME = f"apache-maven-{MAVEN_VERSION}-bin.zip"
MAVEN_DOWNLOAD_BASE = f"https://dlcdn.apache.org/maven/maven-3/{MAVEN_VERSION}/binaries"
TOMCAT_VERSION = "9.0.122"
TOMCAT_ARCHIVE_NAME = f"apache-tomcat-{TOMCAT_VERSION}-windows-x64.zip"
TOMCAT_DOWNLOAD_BASE = f"https://dlcdn.apache.org/tomcat/tomcat-9/v{TOMCAT_VERSION}/bin"

ANSI_PATTERN = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
PERCENT_PATTERN = re.compile(r"(\d{1,3})\s*%")
PROGRESS_CHARS = "█▓▒░▏▎▍▌▋▊▉■□▪▫▬─━═|/-\\"


@dataclass(frozen=True)
class JdkRecommendation:
    version: str
    reason: str


def recommend_jdk_version(scan: ScanResult | None) -> JdkRecommendation:
    if not scan or not scan.backend:
        return JdkRecommendation("8", "\u672a\u626b\u63cf\u5230\u540e\u7aef\u7248\u672c\u4fe1\u606f\uff0c\u9ed8\u8ba4\u63a8\u8350 JDK 8\u3002")

    metadata = scan.backend.metadata
    java_version = metadata.get("java_version")
    if java_version:
        mapped = _map_java_version(java_version)
        source = metadata.get("java_version_source", "\u9879\u76ee\u914d\u7f6e")
        return JdkRecommendation(mapped, f"\u6839\u636e {source}={java_version} \u63a8\u8350 JDK {mapped}\u3002")

    boot_version = metadata.get("spring_boot_version", "")
    if boot_version.startswith("3."):
        return JdkRecommendation("17", f"Spring Boot 3.x \u9700\u8981 Java 17+\uff0c\u68c0\u6d4b\u5230 Spring Boot {boot_version}\uff0c\u63a8\u8350 JDK 17\u3002")
    if boot_version.startswith("2."):
        return JdkRecommendation("8", f"\u68c0\u6d4b\u5230 Spring Boot {boot_version}\uff0c\u9ed8\u8ba4\u63a8\u8350\u517c\u5bb9\u6027\u66f4\u597d\u7684 JDK 8\u3002")

    return JdkRecommendation("8", "\u672a\u53d1\u73b0\u660e\u786e Java \u7248\u672c\uff0c\u9ed8\u8ba4\u63a8\u8350 JDK 8\u3002")


def build_install_command(tool_name: str, version: str | None = None) -> list[str]:
    if tool_name == "JDK":
        package_id = JDK_PACKAGES.get(version or "8")
        if not package_id:
            raise ValueError(f"\u4e0d\u652f\u6301\u7684 JDK \u7248\u672c: {version}")
    else:
        package_id = WINGET_PACKAGES.get(tool_name)
        if not package_id:
            raise ValueError(f"\u4e0d\u652f\u6301\u81ea\u52a8\u5b89\u88c5: {tool_name}")
    return ["winget", "install", "-e", "--id", package_id]


def default_maven_install_root() -> Path:
    local_app_data = os.environ.get("LOCALAPPDATA", "").strip()
    base = Path(local_app_data) if local_app_data else Path.home() / ".autodeploy"
    return base / "AutoDeploy" / "tools"


def install_maven(
    on_output: Callable[[str], None],
    install_root: Path | None = None,
    opener: Callable[..., BinaryIO] | None = None,
) -> Path:
    root = Path(install_root) if install_root is not None else default_maven_install_root()
    archive_url = f"{MAVEN_DOWNLOAD_BASE}/{MAVEN_ARCHIVE_NAME}"
    return _install_verified_apache_archive(
        product_name="Maven",
        version=MAVEN_VERSION,
        archive_name=MAVEN_ARCHIVE_NAME,
        archive_url=archive_url,
        extracted_dir_name=f"apache-maven-{MAVEN_VERSION}",
        executable_relative=Path("bin") / "mvn.cmd",
        on_output=on_output,
        install_root=root,
        opener=opener,
    )


def install_tomcat(
    on_output: Callable[[str], None],
    install_root: Path | None = None,
    opener: Callable[..., BinaryIO] | None = None,
) -> Path:
    root = Path(install_root) if install_root is not None else default_maven_install_root()
    archive_url = f"{TOMCAT_DOWNLOAD_BASE}/{TOMCAT_ARCHIVE_NAME}"
    return _install_verified_apache_archive(
        product_name="Tomcat",
        version=TOMCAT_VERSION,
        archive_name=TOMCAT_ARCHIVE_NAME,
        archive_url=archive_url,
        extracted_dir_name=f"apache-tomcat-{TOMCAT_VERSION}",
        executable_relative=Path("bin") / "catalina.bat",
        on_output=on_output,
        install_root=root,
        opener=opener,
    )


def _install_verified_apache_archive(
    *,
    product_name: str,
    version: str,
    archive_name: str,
    archive_url: str,
    extracted_dir_name: str,
    executable_relative: Path,
    on_output: Callable[[str], None],
    install_root: Path,
    opener: Callable[..., BinaryIO] | None,
) -> Path:
    product_home = install_root / extracted_dir_name
    executable = product_home / executable_relative
    if executable.exists():
        on_output(f"{product_name} {version} 已安装: {executable}")
        return executable

    open_url = opener or urllib.request.urlopen
    checksum_url = f"{archive_url}.sha512"
    install_root.mkdir(parents=True, exist_ok=True)

    with tempfile.TemporaryDirectory(prefix=f"autodeploy-{product_name.lower()}-") as temporary:
        temp_dir = Path(temporary)
        archive_path = temp_dir / archive_name

        on_output(f"正在从 Apache 官方地址下载 {product_name} {version}...")
        _download_to_file(archive_url, archive_path, open_url, on_output)
        expected_checksum = _download_checksum(checksum_url, open_url)
        actual_checksum = _sha512(archive_path)
        if actual_checksum.lower() != expected_checksum.lower():
            raise RuntimeError(f"{product_name} 安装包 SHA-512 校验失败，已停止安装。")
        on_output(f"{product_name} 安装包校验通过，正在解压...")

        extract_root = temp_dir / "extracted"
        _safe_extract_zip(archive_path, extract_root)
        extracted_home = extract_root / extracted_dir_name
        extracted_executable = extracted_home / executable_relative
        if not extracted_executable.exists():
            raise RuntimeError(f"{product_name} 安装包结构异常，未找到 {executable_relative}。")
        shutil.copytree(extracted_home, product_home, dirs_exist_ok=True)

    if not executable.exists():
        raise RuntimeError(f"{product_name} 解压完成，但未找到 {executable_relative}。")
    on_output(f"{product_name} {version} 安装完成: {executable}")
    return executable


def _download_to_file(
    url: str,
    destination: Path,
    opener: Callable[..., BinaryIO],
    on_output: Callable[[str], None],
) -> None:
    with opener(url, timeout=60) as response, destination.open("wb") as output:
        headers = getattr(response, "headers", {})
        total_text = headers.get("Content-Length", "") if hasattr(headers, "get") else ""
        total = int(total_text) if str(total_text).isdigit() else 0
        downloaded = 0
        last_percent = -10
        while True:
            chunk = response.read(1024 * 256)
            if not chunk:
                break
            output.write(chunk)
            downloaded += len(chunk)
            if total:
                percent = min(downloaded * 100 // total, 100)
                if percent >= last_percent + 10 or percent == 100:
                    on_output(f"下载进度: {percent}%")
                    last_percent = percent


def _download_checksum(url: str, opener: Callable[..., BinaryIO]) -> str:
    with opener(url, timeout=30) as response:
        text = response.read().decode("ascii", errors="strict").strip()
    checksum = text.split()[0] if text else ""
    if not re.fullmatch(r"[0-9a-fA-F]{128}", checksum):
        raise RuntimeError("Apache Maven 校验文件格式无效。")
    return checksum


def _sha512(path: Path) -> str:
    digest = hashlib.sha512()
    with path.open("rb") as source:
        for chunk in iter(lambda: source.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _safe_extract_zip(archive_path: Path, destination: Path) -> None:
    destination.mkdir(parents=True, exist_ok=True)
    destination_resolved = destination.resolve()
    with zipfile.ZipFile(archive_path) as archive:
        for member in archive.infolist():
            target = (destination / member.filename).resolve()
            if target != destination_resolved and destination_resolved not in target.parents:
                raise RuntimeError("Maven 安装包包含不安全的文件路径。")
        archive.extractall(destination)


def run_install_command(command: list[str], on_output) -> int:
    output_lines: list[str] = []
    process = subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        shell=False,
        bufsize=0,
    )
    assert process.stdout is not None
    last_line = ""
    for raw_line in process.stdout:
        for line in format_install_output(raw_line):
            if line and line != last_line:
                on_output(line)
                output_lines.append(line)
                last_line = line
    exit_code = process.wait()
    failure_hint = explain_install_failure(_tool_name_from_command(command), exit_code, output_lines)
    if failure_hint:
        on_output(failure_hint)
    return exit_code


def explain_install_failure(tool_name: str | None, exit_code: int, output_lines: list[str]) -> str:
    if exit_code == 0:
        return ""

    output_text = "\n".join(output_lines)
    if tool_name == "Maven" and ("找不到与输入条件匹配的程序包" in output_text or "No package found" in output_text):
        return (
            "winget 当前源找不到 Maven 包。请手动下载 Apache Maven，解压后在工具里选择 "
            "apache-maven 的 bin\\mvn.cmd，或把该 bin 目录加入 PATH。"
        )

    return ""


def format_install_output(raw: bytes) -> list[str]:
    text = _decode_process_bytes(raw)
    text = ANSI_PATTERN.sub("", text).replace("\b", "")
    lines: list[str] = []
    for part in re.split(r"[\r\n]+", text):
        cleaned = _clean_install_line(part)
        if cleaned:
            lines.append(cleaned)
    return lines


def _decode_process_bytes(raw: bytes) -> str:
    for encoding in ("utf-8", "utf-16", "gb18030", "mbcs"):
        try:
            return raw.decode(encoding)
        except (LookupError, UnicodeDecodeError):
            continue
    return raw.decode("utf-8", errors="replace")


def _clean_install_line(line: str) -> str:
    text = line.strip()
    if not text:
        return ""

    percent = PERCENT_PATTERN.search(text)
    progress_only = _looks_like_progress_line(text)
    if percent and progress_only:
        value = min(int(percent.group(1)), 100)
        return f"\u5b89\u88c5\u8fdb\u5ea6: {value}%"

    if progress_only:
        return ""

    return text


def _looks_like_progress_line(text: str) -> bool:
    if not text:
        return False
    progress_count = sum(1 for char in text if char in PROGRESS_CHARS)
    percent = bool(PERCENT_PATTERN.search(text))
    if percent and progress_count:
        return True
    visible = [char for char in text if not char.isspace()]
    if not visible:
        return False
    return progress_count / len(visible) > 0.45


def _tool_name_from_command(command: list[str]) -> str | None:
    package_ids = {package_id: name for name, package_id in WINGET_PACKAGES.items()}
    package_ids.update({package_id: "JDK" for package_id in JDK_PACKAGES.values()})
    for part in command:
        name = package_ids.get(part)
        if name:
            return name
    return None


def _map_java_version(version: str) -> str:
    try:
        major = int(version.split(".", 1)[0])
    except ValueError:
        return "8"
    if major <= 8:
        return "8"
    if major <= 11:
        return "11"
    if major <= 17:
        return "17"
    return "21"