AutoDeploy/deploy_tool/installer.py

from __future__ import annotations

import re
import subprocess
from dataclasses import dataclass

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 = {
    "Maven": "Apache.Maven",
    "Gradle": "Gradle.Gradle",
    "Node.js": "OpenJS.NodeJS.LTS",
    "Git": "Git.Git",
    "MySQL": "Oracle.MySQL",
}

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 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"