OJplatform/requirements/checker.py

"""项目检测引擎: 解压项目包, 执行稳定度/压力测试, 汇总报告。"""
import logging
import os
import shutil
import tarfile
import threading
import time
import zipfile
from pathlib import Path

import requests
from django.conf import settings

logger = logging.getLogger(__name__)

PROBE_PATHS = ["/", "/index.html", "/health", "/api", "/healthz", "/status"]


def _detect_kind(archive: Path) -> str:
    with open(archive, "rb") as fh:
        head = fh.read(512)
    if head[:4] == b"PK\x03\x04":
        return "zip"
    if head[:2] == b"\x1f\x8b":
        return "tar.gz"
    if b"ustar" in head[:262]:
        return "tar"
    return archive.suffix.lstrip(".") or "unknown"


def _extract(archive: Path, dest: Path) -> Path:
    kind = _detect_kind(archive)
    if kind == "zip":
        with zipfile.ZipFile(archive) as zf:
            for member in zf.namelist():
                target = (dest / member).resolve()
                if not str(target).startswith(str(dest.resolve())):
                    raise ValueError("压缩包含非法路径")
            zf.extractall(dest)
    elif kind in ("tar", "tgz", "tar.gz"):
        with tarfile.open(archive) as tf:
            for member in tf.getmembers():
                target = (dest / member.name).resolve()
                if not str(target).startswith(str(dest.resolve())):
                    raise ValueError("压缩包含非法路径")
            tf.extractall(dest, filter="data")
    else:
        raise ValueError(f"不支持的归档格式: {archive.suffix}")
    if len(list(dest.iterdir())) == 1:
        only = next(dest.iterdir())
        if only.is_dir():
            return only
    return dest


_MARKERS = {
    "requirements.txt": "python", "pyproject.toml": "python", "manage.py": "python",
    "setup.py": "python", "main.py": "python", "app.py": "python",
    "package.json": "nodejs", "pom.xml": "java", "build.gradle": "java",
    "go.mod": "go", "Cargo.toml": "rust", "tsconfig.json": "typescript",
}


def _project_language(root: Path) -> str:
    for name in ("requirements.txt", "pyproject.toml", "manage.py",
                "package.json", "pom.xml", "go.mod", "Cargo.toml"):
        if (root / name).exists():
            return _MARKERS[name]
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
        for name in filenames:
            if name in _MARKERS:
                return _MARKERS[name]
    return "unknown"


def _probe_server(base: str) -> bool:
    for path in PROBE_PATHS:
        try:
            resp = requests.get(base + path, timeout=5, verify=False)
            if resp.status_code < 500:
                return True
        except requests.RequestException:
            continue
    return False


def _stress_one(base: str, duration: float, results: list, lock: threading.Lock):
    ok = err = 0
    errors: list[str] = []
    start = time.monotonic()
    while time.monotonic() - start < duration:
        try:
            resp = requests.get(base, timeout=5, verify=False)
            if resp.status_code < 500:
                ok += 1
            else:
                err += 1
                if len(errors) < 5:
                    errors.append(f"HTTP {resp.status_code}")
        except requests.RequestException as exc:
            err += 1
            if len(errors) < 5:
                errors.append(type(exc).__name__)
    with lock:
        results.append({"ok": ok, "err": err, "errors": errors})


def _find_entry(root: Path, names: list[str]):
    if (root / names[0]).exists():
        return names[0]
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
        for name in filenames:
            if name in names:
                return os.path.relpath(Path(dirpath) / name, root)
    return None


def run_stability_check(root: Path, requirements_text: str) -> dict:
    """启动项目服务 → 压测基准/加压两阶段 → 返回稳定度指标"""
    lang = _project_language(root)
    env = {**os.environ, "PORT": "8321", "FLASK_RUN_PORT": "8321"}
    interpreter = None
    venv_python = root / ".venv" / "bin" / "python"
    if venv_python.exists():
        interpreter = str(venv_python)
    elif shutil.which("python"):
        interpreter = shutil.which("python")
    start_cmd = None
    if lang == "python" and interpreter:
        entry = _find_entry(root, ["manage.py", "app.py", "main.py"])
        if entry and entry == "manage.py":
            start_cmd = [interpreter, "manage.py", "runserver", "0.0.0.0:8321"]
        elif entry:
            start_cmd = [interpreter, entry]
    elif lang == "nodejs":
        entry = _find_entry(root, ["server.js", "index.js", "app.js"])
        if entry and shutil.which("node"):
            start_cmd = ["node", entry]

    if not start_cmd:
        return {
            "pass": None,
            "reason": f"无法自动启动项目(语言: {lang}), 未执行压力测试",
            "base_error_rate": None, "rampup_crash_rate": None, "total_requests": 0,
            "details": {},
        }

    import subprocess
    log_path = str(root / ".server.log")
    try:
        logfh = open(log_path, "ab")
        server = subprocess.Popen(
            start_cmd, cwd=str(root),
            stdout=logfh, stderr=subprocess.STDOUT,
            env=env,
            start_new_session=True,
        )
    except OSError as exc:
        return {"pass": None, "reason": f"启动失败: {exc}", "base_error_rate": None,
                "rampup_crash_rate": None, "total_requests": 0, "details": {}}

    base = "http://127.0.0.1:8321"
    ready = False
    for _ in range(30):
        if _probe_server(base):
            ready = True
            break
        time.sleep(1)
    if not ready:
        server.terminate()
        return {
            "pass": None, "reason": "服务 30s 内未就绪, 未执行压力测试",
            "base_error_rate": None, "rampup_crash_rate": None, "total_requests": 0, "details": {},
        }

    try:
        lock = threading.Lock()
        n = settings.STABILITY_CONCURRENCY
        base_res: list = []
        threads = [threading.Thread(target=_stress_one, args=(base, settings.STABILITY_BASE_DURATION, base_res, lock)) for _ in range(n)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        ramp_res: list = []
        threads = [threading.Thread(target=_stress_one, args=(base, settings.STABILITY_RAMPUP_DURATION, ramp_res, lock)) for _ in range(n + n // 2)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        base_ok = sum(r["ok"] for r in base_res)
        base_err = sum(r["err"] for r in base_res)
        ramp_ok = sum(r["ok"] for r in ramp_res)
        ramp_err = sum(r["err"] for r in ramp_res)
        base_total, ramp_total = base_ok + base_err, ramp_ok + ramp_err
        base_error_rate = round(base_err / base_total, 4) if base_total else None
        rampup_crash_rate = round(ramp_err / ramp_total, 4) if ramp_total else None

        pass_flag = base_error_rate is not None and base_error_rate <= 0.01
        if rampup_crash_rate is not None:
            pass_flag = pass_flag and rampup_crash_rate <= 0.005

        error_samples = []
        for r in base_res + ramp_res:
            error_samples.extend(r["errors"])

        return {
            "pass": pass_flag,
            "requirements": requirements_text,
            "base_error_rate": base_error_rate,
            "rampup_crash_rate": rampup_crash_rate,
            "total_requests": base_total + ramp_total,
            "details": {
                "base_total": base_total, "base_err": base_err,
                "ramp_total": ramp_total, "ramp_err": ramp_err,
                "error_samples": error_samples[:10],
            },
        }
    finally:
        import signal
        try:
            os.killpg(os.getpgid(server.pid), signal.SIGTERM)
        except (ProcessLookupError, PermissionError):
            server.terminate()
        try:
            server.wait(timeout=5)
        except subprocess.TimeoutExpired:
            try:
                os.killpg(os.getpgid(server.pid), signal.SIGKILL)
            except (ProcessLookupError, PermissionError):
                server.kill()


def run_checks(archive_path: Path, spec: dict) -> tuple[dict, Path]:
    """对提交的项目包执行全部启用的检测, 返回 (report, project_root)"""
    workdir = Path(settings.WORKSPACE_DIR) / f"{archive_path.stem}-{threading.get_ident()}"
    workdir.mkdir(parents=True, exist_ok=True)
    report = {
        "language": None,
        "stability": {"enabled": False},
        "performance": {"enabled": False},
        "security": {"enabled": False},
        "functional": {"enabled": False, "points": []},
    }
    project_root = _extract(archive_path, workdir)
    report["language"] = _project_language(project_root)

    if spec.get("stability", {}).get("enabled"):
        report["stability"] = {
            "enabled": True,
            "requirements": spec["stability"].get("requirements", ""),
            **run_stability_check(project_root, spec["stability"].get("requirements", "")),
        }
    if spec.get("performance", {}).get("enabled"):
        report["performance"] = {
            "enabled": True,
            "requirements": spec["performance"].get("requirements", ""),
            "note": "性能检测依赖稳定度压测数据(P95 等), 当前版本在稳定度结果中体现",
        }
    if spec.get("security", {}).get("enabled"):
        report["security"] = {
            "enabled": True,
            "requirements": spec["security"].get("requirements", ""),
            "note": "安全检测当前未接入扫描器, 交由 AI 检测项覆盖",
        }
    func = spec.get("functional", {})
    report["functional"] = {"enabled": bool(func.get("enabled")), "points": func.get("points", [])}
    return report, project_root