aaPanel/mod/project/project_import/taskMod.py

# coding: utf-8

import os
import signal
import subprocess
import sys

if "/www/server/panel" not in sys.path:
    sys.path.insert(0, "/www/server/panel")
if "/www/server/panel/class" not in sys.path:
    sys.path.insert(0, "/www/server/panel/class")
if "/www/server/panel/class_v2" not in sys.path:
    sys.path.insert(0, "/www/server/panel/class_v2")

import public

from mod.project.project_import.core.api_utils import error_response, parse_json_field
from mod.project.project_import.core.constants import (
    STEP_WAITING,
    TASK_CANCELLED,
    TASK_FAILED,
    TASK_QUEUED,
    TASK_RUNNING,
    logs_dir,
)
from mod.project.project_import.core.exceptions import ProjectImportError
from mod.project.project_import.core.session_store import SessionStore
from mod.project.project_import.core.task_store import TaskStore
from mod.project.project_import.sources.git import normalize_git_config


ANALYSIS_STEPS = [
    {"key": "validate_source", "title": "Validate project source", "status": STEP_WAITING},
    {"key": "fetch_source", "title": "Prepare project files", "status": STEP_WAITING},
    {"key": "detect_project", "title": "Analyze project type", "status": STEP_WAITING},
    {"key": "finalize", "title": "Save analysis result", "status": STEP_WAITING},
]

IMPORT_STEPS = [
    {"key": "preflight", "title": "Validate import settings", "status": STEP_WAITING},
    {"key": "runtime", "title": "Prepare project runtime", "status": STEP_WAITING},
    {"key": "commit_files", "title": "Commit project files", "status": STEP_WAITING},
    {"key": "create_project", "title": "Create aaPanel project", "status": STEP_WAITING},
    {"key": "database", "title": "Import database", "status": STEP_WAITING},
    {"key": "ssl", "title": "Configure SSL", "status": STEP_WAITING},
    {"key": "health", "title": "Check project health", "status": STEP_WAITING},
]


class main:
    def __init__(self):
        self.sessions = SessionStore()
        self.tasks = TaskStore()

    def start_analysis(self, get):
        try:
            session_id = str(get.get("session_id", "")).strip()
            source_type = str(get.get("source_type", "")).strip().lower()
            session = None
            if session_id:
                session = self.sessions.get(session_id)
                if source_type and source_type != session.get("source_type"):
                    raise ProjectImportError(
                        "source_type does not match the import session",
                        "SOURCE_TYPE_MISMATCH",
                    )
                source_type = str(session.get("source_type", "")).strip().lower()
            else:
                if not source_type:
                    raise ProjectImportError("source_type is required", "SOURCE_TYPE_REQUIRED")
                if source_type == "archive":
                    raise ProjectImportError(
                        "Archive analysis must use the session_id returned by upload",
                        "UPLOAD_SESSION_REQUIRED",
                    )
            source_config = parse_json_field(get.get("source_config", "{}"), "source_config")
            if not isinstance(source_config, dict):
                raise ProjectImportError("source_config must be a JSON object", "INVALID_SOURCE_CONFIG")
            if source_type == "git":
                source_config = normalize_git_config(source_config)
            if session is None:
                session = self.sessions.create(source_type)
                session_id = session["session_id"]
            self._ensure_task_inactive(session.get("analysis_task_id", ""))
            task = self.tasks.create(
                "analysis",
                session_id,
                ANALYSIS_STEPS,
                payload={"source_type": session.get("source_type", "")},
                secret={"source_config": source_config},
            )

            def update_session(data):
                data["status"] = "analyzing"
                data["analysis_task_id"] = task["task_id"]
                data["analysis"] = {}
                return data

            self.sessions.update(session_id, update_session)
            self._spawn(task["task_id"])
            if source_type:
                public.set_module_logs("project_import", "scan_project_{}".format(source_type))
            # 每次启动分析顺带清理上传区中长期残留的孤儿压缩包（> 会话 TTL 未消费）
            self.sessions.cleanup_upload_root()
            return public.return_message(0, 0, {
                "task_id": task["task_id"],
                "session_id": session_id,
            })
        except Exception as exc:
            return error_response(exc)

    def start_import(self, get):
        try:
            session_id = str(get.get("session_id", ""))
            session = self.sessions.get(session_id)
            self._ensure_task_inactive(session.get("import_task_id", ""))
            retryable_stale_import = (
                session.get("status") == "importing"
                and self._task_has_status(
                    session.get("import_task_id", ""),
                    (TASK_FAILED, TASK_CANCELLED),
                )
            )
            if session.get("status") != "analyzed" and not retryable_stale_import:
                raise ProjectImportError("Project analysis has not completed", "ANALYSIS_REQUIRED")
            project_config = parse_json_field(get.get("project_config", "{}"), "project_config")
            database_config = parse_json_field(get.get("database_config", "{}"), "database_config")
            if not isinstance(project_config, dict) or not isinstance(database_config, dict):
                raise ProjectImportError("Import configurations must be JSON objects", "INVALID_IMPORT_CONFIG")
            task = self.tasks.create(
                "import",
                session_id,
                IMPORT_STEPS,
                payload={"project_type": project_config.get("project_type", "")},
                secret={
                    "project_config": project_config,
                    "database_config": database_config,
                },
            )

            def update_session(data):
                data["status"] = "importing"
                data["import_task_id"] = task["task_id"]
                return data

            self.sessions.update(session_id, update_session)
            self._spawn(task["task_id"])
            project_type = str(project_config.get("project_type", "")).strip().lower()
            if project_type:
                public.set_module_logs("project_import", "import_project_{}".format(project_type))
            return public.return_message(0, 0, {
                "task_id": task["task_id"],
                "session_id": session_id,
            })
        except Exception as exc:
            return error_response(exc)

    def get_progress(self, get):
        try:
            return public.return_message(0, 0, self.tasks.public_progress(get.get("task_id", "")))
        except Exception as exc:
            return error_response(exc)

    def cancel_task(self, get):
        try:
            task_id = str(get.get("task_id", ""))
            task = self.tasks.request_cancel(task_id)
            pid = int(task.get("pid", 0) or 0)
            pgid = int(task.get("pgid", 0) or 0)
            if pid > 0:
                try:
                    if pgid > 0 and hasattr(os, "killpg"):
                        os.killpg(pgid, signal.SIGTERM)
                    else:
                        os.kill(pid, signal.SIGTERM)
                except (ProcessLookupError, PermissionError, OSError):
                    pass
            return public.return_message(0, 0, {
                "task_id": task_id,
                "cancel_requested": True,
            })
        except Exception as exc:
            return error_response(exc)

    def _spawn(self, task_id):
        service_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "service.py")
        python_bin = public.get_python_bin()
        if not python_bin or not os.path.isfile(python_bin):
            raise ProjectImportError("Panel Python interpreter was not found", "PYTHON_NOT_FOUND")
        os.makedirs(logs_dir(), mode=0o700, exist_ok=True)
        log_path = os.path.join(logs_dir(), task_id + ".log")
        try:
            with open(log_path, "ab", buffering=0) as output:
                process = subprocess.Popen(
                    [python_bin, "-u", service_path, task_id],
                    stdin=subprocess.DEVNULL,
                    stdout=output,
                    stderr=subprocess.STDOUT,
                    close_fds=True,
                    start_new_session=True,
                )
            self.tasks.set_process(task_id, process.pid, process.pid)
        except Exception as exc:
            self.tasks.set_failed(task_id, "Failed to start worker: {}".format(exc), "WORKER_START_FAILED")
            raise ProjectImportError("Failed to start import worker", "WORKER_START_FAILED")

    def _ensure_task_inactive(self, task_id):
        if not task_id:
            return
        try:
            task = self.tasks.get(task_id)
        except Exception:
            return
        if task.get("status") in (TASK_QUEUED, TASK_RUNNING):
            raise ProjectImportError("Another task in this stage is still running", "TASK_ALREADY_RUNNING")

    def _task_has_status(self, task_id, statuses):
        if not task_id:
            return False
        try:
            return self.tasks.get(task_id).get("status") in statuses
        except Exception:
            return False