AutoDeploy/deploy_tool/scanner.py

from __future__ import annotations

import json
import os
import re
from collections.abc import Iterator
from dataclasses import dataclass, field
from pathlib import Path


IGNORED_DIRECTORIES = {"node_modules", "target", "build", "dist", ".git", ".idea", ".vscode"}


@dataclass(frozen=True)
class ProjectComponent:
    kind: str
    path: Path
    detail: str
    metadata: dict[str, str] = field(default_factory=dict)


@dataclass(frozen=True)
class ScanResult:
    root: Path
    backend: ProjectComponent | None
    frontend: ProjectComponent | None
    sql_files: list[Path]
    frontends: list[ProjectComponent] = field(default_factory=list)

    @property
    def detected_frontends(self) -> list[ProjectComponent]:
        """Return every detected frontend while preserving the old single-frontend API."""
        if self.frontends:
            return self.frontends
        return [self.frontend] if self.frontend else []


def scan_project(root: str | Path) -> ScanResult:
    project_root = Path(root).resolve()
    if not project_root.exists():
        raise FileNotFoundError(f"目录不存在: {project_root}")
    if not project_root.is_dir():
        raise NotADirectoryError(f"不是文件夹: {project_root}")

    backend = _find_backend(project_root)
    frontends = _find_frontends(project_root)
    frontend = frontends[0] if frontends else None
    sql_files = list(_walk_project_files(project_root, suffix=".sql"))
    return ScanResult(project_root, backend, frontend, sql_files, frontends)


def _find_backend(root: Path) -> ProjectComponent | None:
    candidates: list[tuple[int, Path, str]] = []
    for build_file in _walk_project_files(root, names={"pom.xml", "build.gradle", "build.gradle.kts"}):
        project_dir = build_file.parent
        score = 1
        detail_parts = [build_file.name]
        text = _read_text(build_file)
        metadata = _extract_backend_metadata(text)
        metadata.update(_extract_backend_runtime_metadata(project_dir))
        if "spring-boot" in text or "org.springframework.boot" in text:
            score += 3
            detail_parts.append("Spring Boot dependency")
        if java_version := metadata.get("java_version"):
            detail_parts.append(f"Java {java_version}")
        if boot_version := metadata.get("spring_boot_version"):
            detail_parts.append(f"Spring Boot {boot_version}")
        if (project_dir / "src" / "main" / "java").exists():
            score += 2
            detail_parts.append("src/main/java")
        candidates.append((score, project_dir, ", ".join(detail_parts), metadata))

    if not candidates:
        return None
    score, path, detail, metadata = sorted(candidates, key=lambda item: (-item[0], len(item[1].parts)))[0]
    if score < 2:
        return None
    return ProjectComponent("Spring Boot", path, detail, metadata)


def _find_frontends(root: Path) -> list[ProjectComponent]:
    candidates: list[tuple[int, Path, str, dict[str, str]]] = []
    for package_file in _walk_project_files(root, names={"package.json"}):
        project_dir = package_file.parent
        score = 1
        detail_parts = ["package.json"]
        package = _load_json(package_file)
        deps = {}
        for key in ("dependencies", "devDependencies"):
            value = package.get(key)
            if isinstance(value, dict):
                deps.update(value)
        scripts = package.get("scripts") if isinstance(package.get("scripts"), dict) else {}
        if "vue" in deps:
            score += 3
            detail_parts.append("Vue dependency")
        if "build" in scripts:
            score += 1
            detail_parts.append("build script")
        if any((project_dir / name).exists() for name in ("vue.config.js", "vite.config.js", "vite.config.ts")):
            score += 1
            detail_parts.append("Vue/Vite config")
        metadata = _classify_frontend(project_dir)
        for script_name in ("serve", "dev", "start"):
            if isinstance(scripts.get(script_name), str):
                metadata["start_script"] = script_name
                break
        configured_port = _extract_frontend_port(project_dir)
        if configured_port:
            metadata["port"] = configured_port
            detail_parts.append(f"port {configured_port}")
        candidates.append((score, project_dir, ", ".join(detail_parts), metadata))

    if not candidates:
        return []
    detected = [
        ProjectComponent("Vue", path, detail, metadata)
        for score, path, detail, metadata in candidates
        if score >= 3
    ]
    role_order = {"user": 0, "admin": 1, "frontend": 2}
    return sorted(
        detected,
        key=lambda item: (role_order.get(item.metadata.get("role", "frontend"), 2), len(item.path.parts), str(item.path)),
    )


def _find_frontend(root: Path) -> ProjectComponent | None:
    """Compatibility helper for callers that still expect one frontend."""
    frontends = _find_frontends(root)
    return frontends[0] if frontends else None


def _classify_frontend(project_dir: Path) -> dict[str, str]:
    name = project_dir.name.lower()
    if any(token in name for token in ("manage", "manager", "admin", "backend", "backstage", "后台", "管理")):
        return {"role": "admin", "display_name": "管理端"}
    if any(token in name for token in ("client", "user", "portal", "front", "web", "用户", "前台")):
        return {"role": "user", "display_name": "用户端"}
    return {"role": "frontend", "display_name": project_dir.name}


def _extract_frontend_port(project_dir: Path) -> str | None:
    for name in ("vue.config.js", "vite.config.js", "vite.config.ts"):
        config_file = project_dir / name
        if not config_file.is_file():
            continue
        match = re.search(r"\bport\s*:\s*['\"]?(\d{1,5})", _read_text(config_file), re.IGNORECASE)
        if match and 1 <= int(match.group(1)) <= 65535:
            return match.group(1)
    return None


def _is_ignored(path: Path) -> bool:
    return any(part in IGNORED_DIRECTORIES for part in path.parts)


def _walk_project_files(
    root: Path,
    *,
    names: set[str] | None = None,
    suffix: str | None = None,
) -> Iterator[Path]:
    """Yield matching files without descending into generated or dependency directories."""
    for current_dir, directories, filenames in os.walk(root, topdown=True, onerror=lambda _error: None):
        directories[:] = sorted(name for name in directories if name not in IGNORED_DIRECTORIES)
        for filename in sorted(filenames):
            if names is not None and filename not in names:
                continue
            if suffix is not None and not filename.lower().endswith(suffix.lower()):
                continue
            yield Path(current_dir) / filename


def _read_text(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8", errors="ignore").lower()
    except OSError:
        return ""


def _load_json(path: Path) -> dict:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}


def _extract_backend_metadata(text: str) -> dict[str, str]:
    metadata: dict[str, str] = {}
    java_patterns = [
        r"<maven\.compiler\.source>\s*([^<\s]+)\s*</maven\.compiler\.source>",
        r"<maven\.compiler\.target>\s*([^<\s]+)\s*</maven\.compiler\.target>",
        r"<java\.version>\s*([^<\s]+)\s*</java\.version>",
        r"sourceCompatibility\s*=\s*['\"]?([^'\"\s]+)",
        r"targetCompatibility\s*=\s*['\"]?([^'\"\s]+)",
        r"java\s*\{\s*sourceCompatibility\s*=\s*JavaVersion\.VERSION_([^'\"\s}]+)",
    ]
    for pattern in java_patterns:
        match = re.search(pattern, text, re.IGNORECASE | re.MULTILINE)
        if match:
            metadata["java_version"] = _normalize_java_version(match.group(1))
            metadata["java_version_source"] = _java_source_label(pattern)
            break

    boot_patterns = [
        r"<artifactId>\s*spring-boot-starter-parent\s*</artifactId>\s*<version>\s*([^<\s]+)\s*</version>",
        r"org\.springframework\.boot['\"]?\s*:\s*spring-boot[^:'\"]*['\"]?\s*:\s*([^'\"\s]+)",
        r"id\s+['\"]org\.springframework\.boot['\"]\s+version\s+['\"]([^'\"]+)['\"]",
    ]
    for pattern in boot_patterns:
        match = re.search(pattern, text, re.IGNORECASE | re.MULTILINE)
        if match:
            metadata["spring_boot_version"] = match.group(1)
            break
    return metadata


def _extract_backend_runtime_metadata(project_dir: Path) -> dict[str, str]:
    resources = project_dir / "src" / "main" / "resources"
    if not resources.is_dir():
        return {}
    for pattern in ("application*.properties", "application*.yml", "application*.yaml"):
        for config_file in sorted(resources.glob(pattern)):
            try:
                text = config_file.read_text(encoding="utf-8", errors="ignore")
            except OSError:
                continue
            match = re.search(
                r"(?:server\.(?:servlet\.)?)?context-path\s*[:=]\s*['\"]?([^'\"\s#]+)",
                text,
                re.IGNORECASE,
            )
            if match:
                context_path = match.group(1).strip()
                if not context_path.startswith("/"):
                    context_path = "/" + context_path
                return {"context_path": context_path}
    return {}


def _normalize_java_version(value: str) -> str:
    clean = value.strip().replace("version_", "").replace("_", ".").lower()
    if clean.startswith("1."):
        return clean.split(".", 1)[1]
    return clean.split(".", 1)[0]


def _java_source_label(pattern: str) -> str:
    if "maven\\.compiler\\.source" in pattern:
        return "maven.compiler.source"
    if "maven\\.compiler\\.target" in pattern:
        return "maven.compiler.target"
    if "java\\.version" in pattern:
        return "java.version"
    if "sourceCompatibility" in pattern:
        return "sourceCompatibility"
    if "targetCompatibility" in pattern:
        return "targetCompatibility"
    return "项目配置"