"""项目检测引擎: 解压项目包, 执行稳定度/压力测试, 汇总报告。""" 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