aaPanel/mod/project/mail/billionmailMod.py

# coding: utf-8
# -------------------------------------------------------------------
# aaPanel
# -------------------------------------------------------------------

import copy
import hashlib
import hmac
import json
import os
import re
import requests
import shlex
import sys
import threading
import time
import uuid

try:
    from urllib.parse import urlencode
except ImportError:
    from urllib import urlencode

if "/www/server/panel/class" not in sys.path:
    sys.path.append("/www/server/panel/class")

import public


class main:
    APP_NAME = "billionmail"
    APP_TYPE = "Email"
    PROXY_PREFIX = "/billionmail"
    DEFAULT_SERVICE_NAME = "billionmail"
    DEFAULT_ADMIN_USERNAME = "billion"
    LEGACY_MAIL_PLUGIN_PATH = "/www/server/panel/plugin/mail_sys"
    VMAIL_PATH = "/www/vmail"
    HOST_MAIL_SERVICES = ("postfix", "dovecot")
    INSTALL_STATUS_FILE = "/tmp/billionmail_install.log"
    INSTALL_LOCK_FILE = "/tmp/billionmail_install.lock"
    DOCKER_INSTALL_TIMEOUT = 1800
    BILLIONMAIL_START_TIMEOUT = 420
    DEFAULT_PORTS = {
        "HTTPS_PORT": "41443",
        "HTTP_PORT": "8080",
        "SMTP_PORT": "25",
        "SMTPS_PORT": "465",
        "SUBMISSION_PORT": "587",
        "IMAP_PORT": "143",
        "IMAPS_PORT": "993",
        "POP_PORT": "110",
        "POPS_PORT": "995",
    }
    MAIL_SERVICE_PORTS = {
        "postfix": (25, 465, 587),
        "dovecot": (110, 143, 993, 995),
    }
    SECRET_WORDS = ("PASSWORD", "PASS", "TOKEN", "SECRET", "KEY", "SALT")

    def _ok(self, data):
        return public.return_message(0, 0, data)

    def _err(self, message):
        return public.return_message(-1, 0, message)

    @staticmethod
    def _as_text(value):
        if value is None:
            return ""
        return str(value).strip()

    @staticmethod
    def _bool_int(value, default=1):
        if value is None or value == "":
            return default
        if isinstance(value, bool):
            return 1 if value else 0
        value = str(value).strip().lower()
        if value in ("1", "true", "yes", "on"):
            return 1
        if value in ("0", "false", "no", "off"):
            return 0
        return default

    @staticmethod
    def _valid_service_name(service_name):
        return re.match(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$", service_name or "") is not None

    def _is_secret_key(self, key):
        key = (key or "").upper()
        return any(word in key for word in self.SECRET_WORDS)

    def _mask_value(self, key, value):
        if value in (None, ""):
            return value
        return "******" if self._is_secret_key(key) else value

    def _mask_text(self, text):
        if not text:
            return ""
        for key in ("ADMIN_PASSWORD", "DBPASS", "REDISPASS", "API_TOKEN", "SECRET_KEY", "JWT_SECRET"):
            text = re.sub(r"({}=).*".format(re.escape(key)), r"\1******", text)
        return text

    def _project_path(self):
        dk_project_path = "/www/dk_project"
        project_path_file = "{}/class_v2/btdockerModelV2/config/project_path.pl".format(public.get_panel_path())
        path = public.readFile(project_path_file)
        if path:
            dk_project_path = path.strip()
        return os.path.join(dk_project_path, "dk_app")

    def _installed_json_file(self):
        return os.path.join(self._project_path(), "installed.json")

    def _confirm_file(self):
        return os.path.join(public.get_panel_path(), "data", "billionmail_vmail_confirm.json")

    def _read_json(self, filename, default=None):
        if default is None:
            default = {}
        try:
            data = public.readFile(filename)
            if not data:
                return default
            return json.loads(data)
        except Exception:
            return default

    def _write_json(self, filename, data):
        try:
            dirname = os.path.dirname(filename)
            if dirname and not os.path.exists(dirname):
                os.makedirs(dirname)
            return public.writeFile(filename, json.dumps(data, indent=2))
        except Exception:
            return False

    def _all_installed_apps(self):
        installed_json = self._read_json(self._installed_json_file(), {})
        apps = []
        if isinstance(installed_json, dict):
            for app_list in installed_json.values():
                if isinstance(app_list, list):
                    apps.extend(app_list)
        return apps

    def _find_installed_app(self, service_name=None, app_id=None):
        service_name = self._as_text(service_name)
        app_id = self._as_text(app_id)
        matches = []
        for app in self._all_installed_apps():
            if app.get("appname") != self.APP_NAME:
                continue
            if service_name and app.get("service_name") != service_name:
                continue
            if app_id and app.get("id") != app_id:
                continue
            matches.append(app)
        if not matches:
            return None
        matches.sort(key=lambda item: int(item.get("createat", 0) or 0), reverse=True)
        return copy.deepcopy(matches[0])

    def _service_path(self, app):
        return os.path.join(app.get("path", ""), app.get("service_name", ""))

    def _compose_file(self, app):
        return os.path.join(self._service_path(app), "docker-compose.yml")

    def _env_file(self, app):
        return os.path.join(self._service_path(app), ".env")

    def _app_info_map(self, app):
        info = {}
        for item in app.get("appinfo") or []:
            key = item.get("fieldKey")
            if key:
                info[key] = item.get("fieldValue")
        return info

    def _read_env(self, app, mask=False):
        env = {}
        env_file = self._env_file(app)
        content = public.readFile(env_file)
        if not content:
            return env
        for raw_line in content.splitlines():
            line = raw_line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            key, value = line.split("=", 1)
            key = key.strip()
            value = value.strip().strip('"').strip("'")
            env[key] = self._mask_value(key, value) if mask else value
        return env

    def _write_env_value(self, app, key, value):
        env_file = self._env_file(app)
        content = public.readFile(env_file)
        if content is None:
            content = ""

        lines = content.splitlines()
        updated = False
        new_lines = []
        for raw_line in lines:
            line = raw_line.strip()
            if line and not line.startswith("#") and "=" in line:
                env_key = line.split("=", 1)[0].strip()
                if env_key == key:
                    new_lines.append("{}={}".format(key, value))
                    updated = True
                    continue
            new_lines.append(raw_line)

        if not updated:
            if new_lines and new_lines[-1].strip():
                new_lines.append("")
            new_lines.append("{}={}".format(key, value))

        return public.writeFile(env_file, "\n".join(new_lines) + "\n")

    @staticmethod
    def _first_value(*values):
        for value in values:
            if value not in (None, ""):
                return str(value)
        return ""

    def _port_from_app(self, app, key):
        info = self._app_info_map(app)
        env = self._read_env(app, mask=False)
        return self._first_value(info.get(key), env.get(key))

    def _direct_console_url(self, app):
        https_port = self._port_from_app(app, "HTTPS_PORT")
        http_port = self._port_from_app(app, "HTTP_PORT")
        ip_addr = public.GetLocalIp()
        if https_port:
            return "https://{}:{}".format(ip_addr, https_port)
        if http_port:
            return "http://{}:{}".format(ip_addr, http_port)
        return ""

    def _proxy_target(self, app):
        https_port = self._port_from_app(app, "HTTPS_PORT")
        http_port = self._port_from_app(app, "HTTP_PORT")
        if https_port:
            return {
                "scheme": "https",
                "host": "127.0.0.1",
                "port": https_port,
                "base_url": "https://127.0.0.1:{}".format(https_port),
            }
        if http_port:
            return {
                "scheme": "http",
                "host": "127.0.0.1",
                "port": http_port,
                "base_url": "http://127.0.0.1:{}".format(http_port),
            }
        return {}

    def _masked_appinfo(self, app):
        appinfo = []
        for item in app.get("appinfo") or []:
            row = dict(item)
            row["fieldValue"] = self._mask_value(row.get("fieldKey"), row.get("fieldValue"))
            appinfo.append(row)
        return appinfo

    def _format_app(self, app, include_secret=False, include_env=False):
        if not app:
            return {
                "installed": False,
                # "console_url": self.PROXY_PREFIX + "/",
                # "api_prefix": self.PROXY_PREFIX + "/api",
            }
        info = self._app_info_map(app)
        env = self._read_env(app, mask=False)
        ports = {}
        for key in self.DEFAULT_PORTS:
            ports[key] = self._port_from_app(app, key)

        admin_username = self._first_value(env.get("ADMIN_USERNAME"), info.get("ADMIN_USERNAME"))
        admin_password = self._first_value(env.get("ADMIN_PASSWORD"), info.get("ADMIN_PASSWORD"))
        sso_secret = self._first_value(env.get("AAPANEL_SSO_SECRET"), info.get("AAPANEL_SSO_SECRET"))
        direct_console_url = self._direct_console_url(app)
        data = {
            "installed": True,
            "id": app.get("id"),
            "service_name": app.get("service_name"),
            "appname": app.get("appname"),
            "apptitle": app.get("apptitle"),
            "status": app.get("status"),
            "version": app.get("version"),
            "m_version": app.get("m_version"),
            "s_version": app.get("s_version"),
            "createat": app.get("createat"),
            "path": self._service_path(app),
            "ports": ports,
            "admin": {
                "username": admin_username,
                "password": admin_password if include_secret else self._mask_value("ADMIN_PASSWORD", admin_password),
            },
            "sso": {
                "enabled": bool(sso_secret),
                "endpoint": "/v2/mod/mail/billionmail/sso",
            },
            "console_url": self.PROXY_PREFIX + "/",
            "api_prefix": self.PROXY_PREFIX + "/api",
            "direct_console_url": direct_console_url,
            "webmail_url": direct_console_url.rstrip("/") + "/roundcube" if direct_console_url else "",
            "proxy_target": self._proxy_target(app),
            "installed_log": os.path.join("/tmp", "{}.log".format(app.get("service_name", ""))),
            "appinfo": app.get("appinfo") if include_secret else self._masked_appinfo(app),
        }
        if include_env:
            data["env"] = self._read_env(app, mask=not include_secret)
        return data

    def _manager_app(self, service_name=None):
        try:
            from mod.project.docker.app.appManageMod import AppManage
            args = public.to_dict_obj({"app_type": self.APP_TYPE, "row": 10000, "p": 1})
            result = AppManage().get_installed_apps(args)
            message = result.get("message", {})
            data = message.get("data", []) if isinstance(message, dict) else []
            for app in data:
                if app.get("appname") != self.APP_NAME:
                    continue
                if service_name and app.get("service_name") != service_name:
                    continue
                return app
        except Exception:
            pass
        return None

    def _validate_port(self, key, value):
        value = self._as_text(value)
        try:
            port = int(value)
        except Exception:
            return None, "{} must be a valid port".format(key)
        if port < 1 or port > 65535:
            return None, "{} must be between 1 and 65535".format(key)
        return str(port), None

    def _ports_from_get(self, get):
        ports = {}
        for key, default_value in self.DEFAULT_PORTS.items():
            value = get.get(key, get.get(key.lower(), default_value))
            value, err = self._validate_port(key, value)
            if err:
                return None, err
            ports[key] = value
        return ports, None

    def _guess_mail_service(self, port, process_name=""):
        process_name = (process_name or "").lower()
        if "postfix" in process_name:
            return "postfix"
        if "dovecot" in process_name:
            return "dovecot"
        try:
            port = int(port)
        except Exception:
            return ""
        for service, service_ports in self.MAIL_SERVICE_PORTS.items():
            if port in service_ports:
                if service == "postfix" and process_name in ("master", "smtpd", "pickup", "qmgr"):
                    return service
                if service == "dovecot" and process_name in ("dovecot", "imap", "imap-login", "pop3", "pop3-login"):
                    return service
        return ""

    def _decorate_port_conflict(self, conflict):
        row = dict(conflict)
        service = self._guess_mail_service(row.get("port"), row.get("process_name"))
        row["service"] = service
        row["can_stop"] = row.get("source") == "process" and service in self.HOST_MAIL_SERVICES
        return row

    def _port_conflicts(self, ports):
        conflicts = []
        port_set = set(int(port) for port in ports.values())
        for app in self._all_installed_apps():
            for item in app.get("port") or []:
                try:
                    item_port = int(item)
                except Exception:
                    continue
                if item_port in port_set:
                    conflicts.append({
                        "port": item_port,
                        "source": "docker_app",
                        "service_name": app.get("service_name"),
                        "appname": app.get("appname"),
                    })
        try:
            import psutil
            for conn in psutil.net_connections("tcp4"):
                if not conn.laddr:
                    continue
                if conn.laddr.port not in port_set:
                    continue
                process_name = ""
                if conn.pid:
                    try:
                        process_name = psutil.Process(conn.pid).name()
                    except Exception:
                        process_name = ""
                conflicts.append({
                    "port": conn.laddr.port,
                    "source": "process",
                    "pid": conn.pid,
                    "process_name": process_name,
                    })
        except Exception:
            pass
        return [self._decorate_port_conflict(conflict) for conflict in conflicts]

    def _port_precheck(self, ports):
        conflicts = self._port_conflicts(ports)
        return {
            "ok": len(conflicts) == 0,
            # "checked_ports": dict((key, int(value)) for key, value in ports.items()),
            "conflicts": conflicts,
        }

    @staticmethod
    def _parse_os_release(content):
        data = {}
        if not content:
            return data
        for raw_line in content.splitlines():
            line = raw_line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            key, value = line.split("=", 1)
            data[key] = value.strip().strip('"').strip("'")
        return data

    @staticmethod
    def _version_tuple(version):
        parts = re.findall(r"\d+", version or "")
        return tuple(int(item) for item in parts[:3])

    @staticmethod
    def _version_gte(version, target):
        value = list(main._version_tuple(version))
        target_value = list(target)
        length = max(len(value), len(target_value))
        value += [0] * (length - len(value))
        target_value += [0] * (length - len(target_value))
        return tuple(value) >= tuple(target_value)

    def _system_precheck(self):
        """
        检查系统信息
        """
        release = self._parse_os_release(public.readFile("/etc/os-release"))
        os_id = (release.get("ID") or "").lower()
        version_id = release.get("VERSION_ID", "")
        is_ubuntu_24 = os_id == "ubuntu" and self._version_gte(version_id, (24, 4))
        is_debian_13 = os_id == "debian" and self._version_gte(version_id, (13,))
        billionmail_only = is_ubuntu_24 or is_debian_13
        warning = ""
        if billionmail_only:
            warning = (
                "The current OS version is not compatible with the legacy mail_sys plugin. "
                "BillionMail installation is recommended."
            )
        return {
            # "id": os_id,
            "name": release.get("NAME", ""),
            "version": release.get("VERSION", ""),
            # "version_id": version_id,
            # "major": self._version_tuple(version_id)[0] if self._version_tuple(version_id) else 0,
            "billionmail_only": billionmail_only,
            "warning": warning,
            # "raw": release,
        }

    def _docker_status(self):
        docker = {
            "installed": None,
            "running": None,
            "error": "",
        }
        try:
            from btdockerModelV2 import setupModel as ds
            setup = ds.main()
            docker["running"] = bool(setup.get_service_status())
            docker["installed"] = bool(setup.check_docker_service())
        except Exception as ex:
            docker["error"] = str(ex)
        return docker

    def _docker_setup_model(self):
        from btdockerModelV2 import setupModel as ds
        return ds.main()

    def _is_success(self, result):
        return isinstance(result, dict) and result.get("status") == 0

    def _ensure_docker_ready(self, install_if_missing=False):
        try:
            setup = self._docker_setup_model()
            installed = bool(setup.check_docker_service())
            running = bool(setup.get_service_status())
        except Exception as ex:
            return False, self._err("Failed to check Docker service: {}".format(str(ex)))

        if not installed:
            if not install_if_missing:
                return False, self._err("Docker is not installed.")
            try:
                installing = public.M('tasks').where('name=? and status=?', ("Install Docker Service", "-1")).count()
            except Exception:
                installing = 0
            if installing:
                return False, self._ok({
                    "docker_installing": True,
                    "docker": self._docker_status(),
                    "result": "Docker installation task already exists.",
                    "notice": "Docker is being installed. Please retry BillionMail installation after Docker is ready.",
                })
            try:
                result = setup.install_docker_program(public.to_dict_obj({
                    "url": "",
                    "type": 0,
                }))
            except Exception as ex:
                return False, self._err("Failed to add Docker installation task: {}".format(str(ex)))

            message = result.get("message", result) if isinstance(result, dict) else result
            if self._is_success(result) or "installation task already exists" in str(message).lower():
                return False, self._ok({
                    "docker_installing": True,
                    "docker": self._docker_status(),
                    "result": message,
                    "notice": "Docker installation task has been queued. Please retry BillionMail installation after Docker is ready.",
                })
            return False, self._err("Failed to add Docker installation task: {}".format(message))

        if not running:
            try:
                result = setup.docker_service(public.to_dict_obj({"act": "start"}))
            except Exception as ex:
                return False, self._err("Failed to start Docker service: {}".format(str(ex)))
            if not self._is_success(result):
                message = result.get("message", result) if isinstance(result, dict) else result
                return False, self._err("Failed to start Docker service: {}".format(message))
            for _ in range(5):
                if setup.get_service_status():
                    return True, None
                time.sleep(1)
            return False, self._err("Docker service start command succeeded, but Docker is not running yet.")

        return True, None

    def _docker_install_task_exists(self):
        try:
            count = public.M('tasks').where('name=? and status=?', ("Install Docker Service", "-1")).count()
            if count:
                return True
            count = public.M('tasks').where('name=? and status=?', ("Install Docker Service", "0")).count()
            return bool(count)
        except Exception:
            return False

    def _install_progress_template(self):
        return {
            "status": 0,
            "progress": 0,
            "result": {},
            "type": self.APP_NAME,
            "steps": [
                {
                    "status": 2,
                    "error": "",
                    "ps": "Waiting to install Docker...",
                    "title": "Install Docker",
                },
                {
                    "status": 2,
                    "error": "",
                    "ps": "Waiting to install Docker Compose...",
                    "title": "Install Docker Compose",
                },
                {
                    "status": 2,
                    "error": "",
                    "ps": "Waiting to install BillionMail...",
                    "title": "Install BillionMail",
                },
            ],
        }

    def _read_install_progress(self):
        data = self._read_json(self.INSTALL_STATUS_FILE, None)
        if not isinstance(data, dict) or "steps" not in data:
            return self._install_progress_template()
        return data

    def _write_install_progress(self, data):
        public.writeFile(self.INSTALL_STATUS_FILE, json.dumps(data, ensure_ascii=False))

    def _clear_install_progress(self):
        try:
            if os.path.exists(self.INSTALL_STATUS_FILE):
                os.remove(self.INSTALL_STATUS_FILE)
        except Exception:
            pass

    def _compact_install_result(self, result):
        if not isinstance(result, dict):
            return {"message": result} if result not in (None, "") else {}

        result = dict(result)
        result.pop("billionmail", None)

        create_result = result.pop("result", None)
        if "message" not in result and create_result not in (None, ""):
            if isinstance(create_result, dict) and set(create_result.keys()) == {"result"}:
                create_result = create_result.get("result")
            result["message"] = create_result
        return result

    def _install_result_app_summary(self, app_data):
        if not isinstance(app_data, dict) or not app_data.get("installed"):
            return {}

        summary = {}
        for key in ("status", "version", "path", "installed_log"):
            value = app_data.get(key)
            if value not in (None, ""):
                summary[key] = value
        return summary

    def _current_install_status(self):
        data = self._read_install_progress()
        try:
            progress = int(data.get("progress", 0) or 0)
        except Exception:
            progress = 0
        try:
            status = int(data.get("status", 0) or 0)
        except Exception:
            status = 0

        # steps = data.get("steps", [])
        # if not isinstance(steps, list):
        #     steps = []

        # def step_status(index):
        #     if 0 <= index < len(steps) and isinstance(steps[index], dict):
        #         try:
        #             return int(steps[index].get("status", 2))
        #         except Exception:
        #             return 2
        #     return 2

        lock_exists = os.path.exists(self.INSTALL_LOCK_FILE)
        installing = status == 0 and (lock_exists or 0 < progress < 100)
        # docker_installing = installing and (
        #     progress < 55 or step_status(0) == 0 or step_status(1) == 0
        # )
        # billionmail_installing = installing and (
        #     progress >= 55 or step_status(2) == 0
        # )


        # return {
        #     "installing": bool(installing),
        #     "docker": bool(docker_installing),
        #     "billionmail": bool(billionmail_installing),
        #     "progress": progress,
        #     "status": status,
        #     "completed": bool(status == 1 or progress >= 100),
        #     "failed": bool(status == -1),
        #     "progress_url": "/v2/mod/mail/billionmail/install_progress",
        # }


        return bool(installing)

    def _init_install_progress(self):
        data = self._install_progress_template()
        self._write_install_progress(data)
        return data

    def _update_install_progress(self, step_index, ps, progress, status=0, error_msg="", result=None):
        data = self._read_install_progress()
        steps = data.get("steps", [])
        if 0 <= step_index < len(steps):
            steps[step_index]["status"] = status
            steps[step_index]["ps"] = str(ps)
            steps[step_index]["error"] = str(error_msg or "")
            if status == 1 and step_index + 1 < len(steps) and steps[step_index + 1].get("status") == 2:
                steps[step_index + 1]["status"] = 0
        data["progress"] = max(0, min(100, int(progress)))
        if status == -1:
            data["status"] = -1
        if result is not None:
            data["result"] = result
        if all(item.get("status") == 1 for item in steps):
            data["status"] = 1
            data["progress"] = 100
        self._write_install_progress(data)
        return data

    def _queue_docker_install_if_needed(self, setup):
        if self._docker_install_task_exists():
            return True, "Docker installation task already exists."
        result = setup.install_docker_program(public.to_dict_obj({
            "url": "",
            "type": 0,
        }))
        message = result.get("message", result) if isinstance(result, dict) else result
        if self._is_success(result):
            return True, message
        return False, message

    def _wait_until(self, checker, timeout, interval=3, on_tick=None):
        start_time = time.time()
        while time.time() - start_time <= timeout:
            try:
                if checker():
                    return True
            except Exception:
                pass
            if on_tick:
                on_tick(time.time() - start_time, timeout)
            time.sleep(interval)
        return False

    def _prepare_docker_for_install(self):
        setup = self._docker_setup_model()

        self._update_install_progress(0, "Checking Docker service...", 5, 0)
        if not setup.check_docker_service():
            ok, message = self._queue_docker_install_if_needed(setup)
            if not ok:
                self._update_install_progress(0, "Docker installation failed.", 5, -1, message)
                return False, message

            def docker_tick(elapsed, timeout):
                progress = 5 + int(min(25, elapsed * 25 / max(timeout, 1)))
                self._update_install_progress(0, "Installing Docker...", progress, 0)

            if not self._wait_until(
                    lambda: setup.check_docker_service() and not self._docker_install_task_exists(),
                    self.DOCKER_INSTALL_TIMEOUT,
                    5,
                    docker_tick
            ):
                message = "Docker installation timed out."
                self._update_install_progress(0, message, 30, -1, message)
                return False, message
        elif self._docker_install_task_exists():
            def docker_task_tick(elapsed, timeout):
                progress = 5 + int(min(25, elapsed * 25 / max(timeout, 1)))
                self._update_install_progress(0, "Waiting for Docker installation task to finish...", progress, 0)

            if not self._wait_until(
                    lambda: not self._docker_install_task_exists(),
                    self.DOCKER_INSTALL_TIMEOUT,
                    5,
                    docker_task_tick
            ):
                message = "Docker installation task timed out."
                self._update_install_progress(0, message, 30, -1, message)
                return False, message

        self._update_install_progress(0, "Docker installation successful!", 35, 1)

        self._update_install_progress(1, "Checking Docker Compose and starting Docker service...", 40, 0)
        compose_ok, compose_path = setup.check_docker_compose_service()
        if not compose_ok:
            ok, message = self._queue_docker_install_if_needed(setup)
            if not ok:
                self._update_install_progress(1, "Docker Compose installation failed.", 40, -1, message)
                return False, message

            def compose_tick(elapsed, timeout):
                progress = 40 + int(min(10, elapsed * 10 / max(timeout, 1)))
                self._update_install_progress(1, "Installing Docker Compose...", progress, 0)

            if not self._wait_until(
                    lambda: setup.check_docker_compose_service()[0] and not self._docker_install_task_exists(),
                    self.DOCKER_INSTALL_TIMEOUT,
                    5,
                    compose_tick
            ):
                message = "Docker Compose installation timed out."
                self._update_install_progress(1, message, 50, -1, message)
                return False, message
            compose_ok, compose_path = setup.check_docker_compose_service()
        elif self._docker_install_task_exists():
            def compose_task_tick(elapsed, timeout):
                progress = 40 + int(min(10, elapsed * 10 / max(timeout, 1)))
                self._update_install_progress(1, "Waiting for Docker Compose installation task to finish...", progress, 0)

            if not self._wait_until(
                    lambda: not self._docker_install_task_exists(),
                    self.DOCKER_INSTALL_TIMEOUT,
                    5,
                    compose_task_tick
            ):
                message = "Docker Compose installation task timed out."
                self._update_install_progress(1, message, 50, -1, message)
                return False, message

        if not setup.get_service_status():
            last_message = ""

            def start_docker_service():
                nonlocal last_message
                if setup.get_service_status():
                    return True
                result = setup.docker_service(public.to_dict_obj({"act": "start"}))
                if not self._is_success(result):
                    last_message = result.get("message", result) if isinstance(result, dict) else result
                    return False
                return setup.get_service_status()

            def service_tick(elapsed, timeout):
                progress = 45 + int(min(5, elapsed * 5 / max(timeout, 1)))
                self._update_install_progress(1, "Starting Docker service...", progress, 0, last_message)

            if not self._wait_until(start_docker_service, 180, 5, service_tick):
                message = last_message or "Docker service start command succeeded, but Docker is not running yet."
                self._update_install_progress(1, "Docker service startup failed.", 50, -1, message)
                return False, message

        self._update_install_progress(1, "Docker Compose installation successful!", 55, 1)
        return True, compose_path

    def _wait_billionmail_started(self, service_name):
        log_file = os.path.join("/tmp", "{}.log".format(service_name))

        def checker():
            content = public.readFile(log_file) or ""
            if "bt_failed" in content:
                raise RuntimeError(self._mask_text(content[-2000:]))
            return "bt_successful" in content

        start_time = time.time()
        while time.time() - start_time <= self.BILLIONMAIL_START_TIMEOUT:
            try:
                if checker():
                    return True, ""
            except RuntimeError as ex:
                return False, str(ex)
            elapsed = time.time() - start_time
            progress = 75 + int(min(20, elapsed * 20 / max(self.BILLIONMAIL_START_TIMEOUT, 1)))
            self._update_install_progress(2, "Waiting for BillionMail containers to start...", progress, 0)
            time.sleep(3)
        return False, "BillionMail startup timed out. Please check {}.".format(log_file)

    def _install_worker(self, args_data, started=None):
        public.writeFile(self.INSTALL_LOCK_FILE, str(threading.get_ident()))
        if started:
            started.set()
        self._init_install_progress()
        try:
            ok, message = self._prepare_docker_for_install()
            if not ok:
                return

            args = public.to_dict_obj(args_data)
            self._update_install_progress(2, "Creating BillionMail application...", 60, 0)
            from mod.project.docker.app.appManageMod import AppManage
            result = AppManage().create_app(args)
            if result.get("status") == -1:
                message = result.get("message", result)
                self._update_install_progress(2, "BillionMail installation failed.", 70, -1, message)
                return

            ok, message = self._wait_billionmail_started(args.service_name)
            if not ok:
                self._update_install_progress(2, "BillionMail startup failed.", 95, -1, message)
                return

            app = self._manager_app(args.service_name) or self._find_installed_app(service_name=args.service_name)
            app_summary = self._install_result_app_summary(self._format_app(app))
            create_message = self._compact_install_result({
                "result": result.get("message", result),
            }).get("message")
            final_result = {
                "service_name": args.service_name,
                "console_url": self.PROXY_PREFIX + "/",
                "api_prefix": self.PROXY_PREFIX + "/api",
                "admin": {
                    "username": args.ADMIN_USERNAME,
                    "password": args.ADMIN_PASSWORD,
                },
                "message": create_message,
            }
            final_result.update(app_summary)
            final_result["docker"] = self._docker_status()
            self._update_install_progress(2, "BillionMail installation successful!", 100, 1, result=final_result)
        except Exception as ex:
            self._update_install_progress(2, "BillionMail installation failed.", 95, -1, str(ex))
        finally:
            public.progress_release_lock(self.INSTALL_LOCK_FILE)

    def _vmail_snapshot(self):
        path = self.VMAIL_PATH
        snapshot = {
            "path": path,
            "exists": os.path.exists(path),
            "is_dir": os.path.isdir(path),
            "empty": True,
            "has_mail_data": False,
            "file_count": 0,
            "dir_count": 0,
            "size": 0,
            "mtime": 0,
            "scan_limited": False,
            "top_entries": [],
            "error": "",
        }
        if not snapshot["exists"]:
            return snapshot
        try:
            snapshot["mtime"] = int(os.path.getmtime(path))
            if not snapshot["is_dir"]:
                snapshot["empty"] = False
                snapshot["has_mail_data"] = True
                snapshot["file_count"] = 1
                snapshot["size"] = os.path.getsize(path)
                return snapshot

            top_entries = os.listdir(path)
            snapshot["top_entries"] = top_entries[:20]
            max_scan = 10000
            for root, dirs, files in os.walk(path):
                snapshot["dir_count"] += len(dirs)
                snapshot["file_count"] += len(files)
                for filename in files:
                    try:
                        snapshot["size"] += os.path.getsize(os.path.join(root, filename))
                    except Exception:
                        pass
                if snapshot["dir_count"] + snapshot["file_count"] >= max_scan:
                    snapshot["scan_limited"] = True
                    break
            snapshot["empty"] = snapshot["dir_count"] == 0 and snapshot["file_count"] == 0
            snapshot["has_mail_data"] = not snapshot["empty"]
        except Exception as ex:
            snapshot["error"] = str(ex)
            snapshot["empty"] = False
            snapshot["has_mail_data"] = True
        return snapshot

    @staticmethod
    def _vmail_signature(snapshot):
        payload = {
            "path": snapshot.get("path", ""),
            "exists": snapshot.get("exists", False),
            "is_dir": snapshot.get("is_dir", False),
            "file_count": snapshot.get("file_count", 0),
            "dir_count": snapshot.get("dir_count", 0),
            "size": snapshot.get("size", 0),
            "mtime": snapshot.get("mtime", 0),
        }
        return hashlib.sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()

    def _vmail_confirmation(self, snapshot):
        record = self._read_json(self._confirm_file(), {})
        if not isinstance(record, dict):
            record = {}
        signature = self._vmail_signature(snapshot)
        confirmed = bool(record.get("confirmed_empty")) and record.get("signature") == signature
        return confirmed, record if confirmed else {}

    def _legacy_mail_precheck(self):
        """
        检查是否新用户(未安装旧版邮局)
        """
        vmail = self._vmail_snapshot()
        confirmed_empty, confirmation = self._vmail_confirmation(vmail)
        mail_sys_installed = os.path.exists(self.LEGACY_MAIL_PLUGIN_PATH)
        is_new_user = (not mail_sys_installed) and (
            (not vmail.get("exists")) or (not vmail.get("has_mail_data")) or confirmed_empty
        )
        return {
            "mail_sys_installed": mail_sys_installed,
            # "mail_sys_path": self.LEGACY_MAIL_PLUGIN_PATH,
            # "vmail_exists": vmail.get("exists"),
            # "vmail_path": self.VMAIL_PATH,
            # "vmail_empty": vmail.get("empty"),
            # "vmail_size": vmail.get("size"),
            # "vmail_file_count": vmail.get("file_count"),
            # "vmail_dir_count": vmail.get("dir_count"),
            # "has_mail_data": vmail.get("has_mail_data"),
            # "confirmed_empty": confirmed_empty,
            # "confirmation": confirmation,
            "is_new_user": is_new_user,
            # "snapshot": vmail,
        }

    def _install_args(self, get):
        service_name = self._as_text(get.get("service_name", self.DEFAULT_SERVICE_NAME))
        if not self._valid_service_name(service_name):
            return None, self._err("Invalid service_name. Use letters, numbers, dot, underscore or hyphen.")

        ports, err = self._ports_from_get(get)
        if err:
            return None, self._err(err)

        conflicts = self._port_conflicts(ports)
        if conflicts:
            return None, self._err({"error": "Port conflict detected", "conflicts": conflicts})

        admin_username = self._as_text(
            get.get("ADMIN_USERNAME", get.get("admin_username", self.DEFAULT_ADMIN_USERNAME))
        ) or self.DEFAULT_ADMIN_USERNAME
        admin_password = self._as_text(get.get("ADMIN_PASSWORD", get.get("admin_password", "")))
        if not admin_password or admin_password == "billion":
            admin_password = public.GetRandomString(20)
        sso_secret = self._as_text(get.get("AAPANEL_SSO_SECRET", get.get("aapanel_sso_secret", "")))
        if not sso_secret:
            sso_secret = public.GetRandomString(48)

        args = {
            "app_name": self.APP_NAME,
            "service_name": service_name,
            "m_version": self._as_text(get.get("m_version", "latest")) or "latest",
            "s_version": self._as_text(get.get("s_version", "")),
            "allow_access": str(self._bool_int(get.get("allow_access", 1), 1)),
            "disable_domain": str(self._bool_int(get.get("disable_domain", 1), 1)),
            "domain": self._as_text(get.get("domain", "")),
            "cpus": self._as_text(get.get("cpus", "0")) or "0",
            "memory_limit": self._as_text(get.get("memory_limit", "0")) or "0",
            "ADMIN_USERNAME": admin_username,
            "ADMIN_PASSWORD": admin_password,
            "AAPANEL_SSO_SECRET": sso_secret,
        }
        args.update(ports)
        return public.to_dict_obj(args), None

    def _target_app(self, get):
        service_name = get.get("service_name", None)
        app_id = get.get("id", None)
        return self._find_installed_app(service_name=service_name, app_id=app_id)

    def _control_app(self):
        return self._manager_app() or self._find_installed_app()

    def _billionmail_default_app_info(self):
        try:
            from mod.project.docker.app.appManageMod import AppManage
            args = public.to_dict_obj({
                "p": 1,
                "row": 20000,
                "query": "BillionMail",
                "app_type": self.APP_TYPE,
            })
            result = AppManage().get_apps(args)
            if not isinstance(result, dict) or result.get("status") != 0:
                return None

            message = result.get("message", {})
            data = message.get("data", []) if isinstance(message, dict) else result.get("data", [])
            if not isinstance(data, list):
                return None

            for app in data:
                if isinstance(app, dict) and app.get("appname") == self.APP_NAME:
                    return app
            for app in data:
                if not isinstance(app, dict):
                    continue
                appname = self._as_text(app.get("appname")).lower()
                apptitle = self._as_text(app.get("apptitle")).lower()
                if self.APP_NAME in (appname, apptitle):
                    return app
        except Exception:
            pass
        return None

    def check_env(self, get):
        app = self._manager_app(get.get("service_name", None)) or self._target_app(get)
        return self._ok({
            "docker": self._docker_status(),
            "billionmail": self._format_app(app),
            "routes": {
                "control": "/v2/mod/mail/billionmail/<fun>",
                "sso": "/v2/mod/mail/billionmail/sso",
                "proxy": self.PROXY_PREFIX + "/",
                "api": self.PROXY_PREFIX + "/api",
            },
        })

    def status(self, get):
        app = self._manager_app(get.get("service_name", None)) or self._target_app(get)
        return self._ok(self._format_app(app))

    def precheck(self, get):
        ports, err = self._ports_from_get(get)
        if err:
            return self._err(err)

        docker = self._docker_status()
        system = self._system_precheck()
        legacy_mail = self._legacy_mail_precheck()
        app = self._manager_app(get.get("service_name", None)) or self._target_app(get)

        data = {
            "system": system,
            "docker": docker,
            "billionmail": self._format_app(app),
            "legacy_mail": legacy_mail,
            "install_status": self._current_install_status(),

        }
        return self._ok(data)

    def check_install(self, get):
        """
        安装前检查端口占用情况
        """
        ports, err = self._ports_from_get(get)
        if err:
            return self._err(err)
        port_result = self._port_precheck(ports)
        # 兜底数据
        port_result['app_info'] = {
  "appid": 226,
  "appname": "billionmail",
  "apptitle": "BillionMail",
  "apptype": "Email",
  "appTypeCN": "Email",
  "appversion": [
    {
      "m_version": "latest",
      "s_version": []
    }
  ],
  "appdesc": "BillionMail provides you with an open-source mail server, email marketing – fully self-hosted, developer-friendly, and free.",
  "appstatus": 1,
  "icon": "",
  "sort": 17,
  "home": "",
  "help": "https://www.billionmail.com/",
  "cpu": 0,
  "mem": 0,
  "disk": 10240,
  "updateat": 1722408250,
  "installed": False,
  "reuse": True,
  "depend": None,
  "field": [
    {
      "attr": "HTTPS_PORT",
      "name": "web port(Htpps)",
      "type": "number",
      "default": 41443,
      "suffix": "Once deployed, access via 'https://ip:port/'",
      "unit": ""
    },
    {
      "attr": "ADMIN_USERNAME",
      "name": "web user",
      "type": "text",
      "default": "billion",
      "suffix": "",
      "unit": ""
    },
    {
      "attr": "ADMIN_PASSWORD",
      "name": "web password",
      "type": "text",
      "default": "billion",
      "suffix": "",
      "unit": ""
    },
    {
      "attr": "allow_access",
      "name": "Allow external access",
      "type": "checkbox",
      "default": True,
      "suffix": "Allow direct access through the host IP+ port",
      "unit": ""
    },
    {
      "attr": "SMTP_PORT",
      "name": "SMTP_PORT",
      "type": "number",
      "default": 25,
      "suffix": "Make sure the port is not occupied",
      "unit": ""
    },
    {
      "attr": "SMTPS_PORT",
      "name": "SMTPS_PORT",
      "type": "number",
      "default": 465,
      "suffix": "Make sure the port is not occupied",
      "unit": ""
    },
    {
      "attr": "SUBMISSION_PORT",
      "name": "SUBMISSION_PORT",
      "type": "number",
      "default": 587,
      "suffix": "Make sure the port is not occupied",
      "unit": ""
    },
    {
      "attr": "IMAP_PORT",
      "name": "IMAP_PORT",
      "type": "number",
      "default": 143,
      "suffix": "Make sure the port is not occupied",
      "unit": ""
    },
    {
      "attr": "IMAPS_PORT",
      "name": "IMAPS_PORT",
      "type": "number",
      "default": 993,
      "suffix": "Make sure the port is not occupied",
      "unit": ""
    },
    {
      "attr": "POP_PORT",
      "name": "POP_PORT",
      "type": "number",
      "default": 110,
      "suffix": "Make sure the port is not occupied",
      "unit": ""
    },
    {
      "attr": "POPS_PORT",
      "name": "POPS_PORT",
      "type": "number",
      "default": 995,
      "suffix": "Make sure the port is not occupied",
      "unit": ""
    },
    {
      "attr": "HTTP_PORT",
      "name": "web port",
      "type": "number",
      "default": 8080,
      "suffix": "",
      "unit": ""
    }
  ],
  "env": [
    {
      "key": "version",
      "type": "string",
      "default": None,
      "desc": "version"
    },
    {
      "key": "SMTP_PORT",
      "type": "number",
      "default": None,
      "desc": "SMTP_PORT"
    },
    {
      "key": "SMTPS_PORT",
      "type": "number",
      "default": None,
      "desc": "SMTPS_PORT"
    },
    {
      "key": "SUBMISSION_PORT",
      "type": "number",
      "default": None,
      "desc": "SUBMISSION_PORT"
    },
    {
      "key": "IMAP_PORT",
      "type": "number",
      "default": None,
      "desc": "IMAP_PORT"
    },
    {
      "key": "IMAPS_PORT",
      "type": "number",
      "default": None,
      "desc": "IMAPS_PORT"
    },
    {
      "key": "POP_PORT",
      "type": "number",
      "default": None,
      "desc": "POP_PORT"
    },
    {
      "key": "POPS_PORT",
      "type": "number",
      "default": None,
      "desc": "POPS_PORT"
    },
    {
      "key": "HTTP_PORT",
      "type": "number",
      "default": None,
      "desc": "HTTP_PORT"
    },
    {
      "key": "HTTPS_PORT",
      "type": "number",
      "default": None,
      "desc": "HTTPS_PORT"
    },
    {
      "key": "ADMIN_USERNAME",
      "type": "string",
      "default": None,
      "desc": "ADMIN_USERNAME"
    },
    {
      "key": "ADMIN_PASSWORD",
      "type": "string",
      "default": None,
      "desc": "ADMIN_PASSWORD"
    },
    {
      "key": "app_path",
      "type": "path",
      "default": None,
      "desc": "app path"
    },
    {
      "key": "host_ip",
      "type": "string",
      "default": None,
      "desc": "host IP"
    }
  ],
  "volumes": {
    "data": {
      "type": "path",
      "desc": ""
    },
    "conf": {
      "type": "path",
      "desc": ""
    },
    "ssl-self-signed": {
      "type": "path",
      "desc": ""
    }
  },
  "installedCount": 0
}
        docker_app_info = self._billionmail_default_app_info()
        if docker_app_info:
            port_result["app_info"] = docker_app_info

        return self._ok(port_result)


    def confirm_vmail_empty(self, get):
        if self._bool_int(get.get("confirm", 0), 0) != 1:
            return self._err("confirm=1 is required.")

        snapshot = self._vmail_snapshot()
        signature = self._vmail_signature(snapshot)
        record = {
            "confirmed_empty": True,
            "signature": signature,
            "time": int(time.time()),
            "vmail_path": self.VMAIL_PATH,
            "snapshot": snapshot,
        }
        if not self._write_json(self._confirm_file(), record):
            return self._err("Failed to save vmail confirmation.")

        return self._ok({
            "confirmed_empty": True,
            "confirm_file": self._confirm_file(),
            "legacy_mail": self._legacy_mail_precheck(),
        })

    def resolve_ports(self, get):
        action = self._as_text(get.get("action", "stop_host_mail_services"))
        if action != "stop_host_mail_services":
            return self._err("Unsupported action: {}".format(action))
        if self._bool_int(get.get("confirm", 0), 0) != 1:
            return self._err("confirm=1 is required.")

        services_text = self._as_text(get.get("services", ",".join(self.HOST_MAIL_SERVICES)))
        requested_services = []
        for service in re.split(r"[,;\s]+", services_text):
            service = service.strip().lower()
            if not service:
                continue
            if service not in self.HOST_MAIL_SERVICES:
                return self._err("Unsupported service: {}".format(service))
            if service not in requested_services:
                requested_services.append(service)
        if not requested_services:
            return self._err("No services were specified.")

        ports, err = self._ports_from_get(get)
        if err:
            return self._err(err)

        ports_before = self._port_precheck(ports)
        stoppable_services = []
        for conflict in ports_before.get("conflicts", []):
            service = conflict.get("service")
            if conflict.get("can_stop") and service in requested_services and service not in stoppable_services:
                stoppable_services.append(service)

        stopped = []
        for service in stoppable_services:
            stdout, stderr = public.ExecShell("systemctl stop {}".format(service))
            if stderr and (
                "not found" in stderr.lower()
                or "could not be found" in stderr.lower()
                or "system has not been booted" in stderr.lower()
                or "failed to connect to bus" in stderr.lower()
            ):
                stdout2, stderr2 = public.ExecShell("service {} stop".format(service))
                stdout = "{}\n{}".format(stdout, stdout2).strip()
                stderr = "{}\n{}".format(stderr, stderr2).strip()
            stopped.append({
                "service": service,
                "success": not bool(stderr),
                "stdout": self._mask_text(stdout),
                "stderr": self._mask_text(stderr),
            })

        ports_after = self._port_precheck(ports)
        for row in stopped:
            service = row.get("service")
            still_occupied = any(
                item.get("service") == service and item.get("source") == "process"
                for item in ports_after.get("conflicts", [])
            )
            row["success"] = row.get("success") and not still_occupied

        return self._ok({
            "action": action,
            "requested_services": requested_services,
            "stoppable_services": stoppable_services,
            "ports_before": ports_before,
            "stopped": stopped,
            "ports_after": ports_after,
            "message": "No stoppable host mail services were found." if not stoppable_services else "",
        })

    def proxy_config(self, get):
        app = self._target_app(get)
        if not app:
            return self._ok(self._format_app(None))
        return self._ok(self.get_proxy_config(app.get("service_name")))

    def get_proxy_config(self, service_name=None):
        app = self._find_installed_app(service_name=service_name)
        if not app:
            return self._format_app(None)
        data = self._format_app(app)
        data["target_base"] = data.get("proxy_target", {}).get("base_url", "")
        return data

    def _sso_bridge_url(self, service_name=None, redirect=None):
        params = {}
        service_name = self._as_text(service_name)
        redirect = self._as_text(redirect)
        if service_name:
            params["service_name"] = service_name
        if redirect:
            params["redirect"] = redirect
        query = urlencode(params)
        url = self.PROXY_PREFIX + "/__aapanel_sso__"
        return url + ("?" + query if query else "")

    def _sso_secret(self, app):
        info = self._app_info_map(app)
        env = self._read_env(app, mask=False)
        return self._first_value(env.get("AAPANEL_SSO_SECRET"), info.get("AAPANEL_SSO_SECRET"))

    def _ensure_sso_secret(self, app):
        env = self._read_env(app, mask=False)
        secret = self._first_value(env.get("AAPANEL_SSO_SECRET"))
        if secret:
            return secret, False

        info = self._app_info_map(app)
        secret = self._first_value(info.get("AAPANEL_SSO_SECRET"))
        if secret:
            self._write_env_value(app, "AAPANEL_SSO_SECRET", secret)
            return secret, False

        secret = public.GetRandomString(48)
        if not self._write_env_value(app, "AAPANEL_SSO_SECRET", secret):
            return "", False
        return secret, True

    @staticmethod
    def _sso_signature(secret, timestamp, nonce, username):
        payload = "{}\n{}\n{}".format(timestamp, nonce, username)
        return hmac.new(secret.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).hexdigest()

    @staticmethod
    def _read_response_json(response):
        try:
            return response.json()
        except Exception:
            return {
                "success": False,
                "code": response.status_code,
                "msg": response.text,
            }

    def _post_sso(self, target_base, username, secret):
        timestamp = str(int(time.time()))
        nonce = uuid.uuid4().hex
        payload = {"username": username}
        headers = {
            "Content-Type": "application/json",
            "X-Aapanel-Timestamp": timestamp,
            "X-Aapanel-Nonce": nonce,
            "X-Aapanel-Signature": self._sso_signature(secret, timestamp, nonce, username),
        }
        response = requests.post(
            target_base.rstrip("/") + "/api/aapanel/sso",
            data=json.dumps(payload, separators=(",", ":")),
            headers=headers,
            verify=False,
            timeout=30,
        )
        return self._read_response_json(response), response.status_code

    def _post_password_login(self, target_base, username, password):
        response = requests.post(
            target_base.rstrip("/") + "/api/login",
            json={"username": username, "password": password},
            verify=False,
            timeout=30,
        )
        return self._read_response_json(response), response.status_code

    def sso(self, get):
        app = self._target_app(get)
        if not app and self._as_text(get.get("service_name", "")) == self.DEFAULT_SERVICE_NAME and not get.get("id"):
            app = self._find_installed_app()
        if not app:
            return self._err("BillionMail is not installed")

        config = self.get_proxy_config(app.get("service_name"))
        target_base = config.get("target_base")
        if not target_base:
            return self._err("BillionMail web port is not configured")

        app_secret, secret_created = self._ensure_sso_secret(app)
        app_data = self._format_app(app, include_secret=True)
        username = self._as_text(get.get("username", app_data.get("admin", {}).get("username", "")))
        mode = "aapanel_sso"

        try:
            if app_secret:
                body, status_code = self._post_sso(target_base, username, app_secret)
                if status_code == 404:
                    mode = "password_login"
                    password = app_data.get("admin", {}).get("password", "")
                    if not username or not password:
                        return self._err("aaPanel SSO endpoint is unavailable and admin credential is unavailable")
                    body, status_code = self._post_password_login(target_base, username, password)
            else:
                mode = "password_login"
                password = app_data.get("admin", {}).get("password", "")
                if not username or not password:
                    return self._err("AAPANEL_SSO_SECRET is not configured and admin credential is unavailable")
                body, status_code = self._post_password_login(target_base, username, password)
        except Exception as ex:
            return self._err("Failed to request BillionMail SSO: {}".format(str(ex)))

        if status_code >= 400 or not body.get("success"):
            return self._err({
                "status_code": status_code,
                "response": body,
                "mode": mode,
            })

        data = body.get("data", {})
        iframe_url = self._sso_bridge_url(app.get("service_name"), get.get("redirect", ""))
        return self._ok({
            "mode": mode,
            "token": data.get("token", ""),
            "refreshToken": data.get("refreshToken", ""),
            "ttl": data.get("ttl", 0),
            "accountInfo": data.get("accountInfo", {}),
            "api_prefix": self.PROXY_PREFIX + "/api",
            "proxy_prefix": self.PROXY_PREFIX,
            "console_url": self.PROXY_PREFIX + "/",
            "login_url": iframe_url,
            "iframe_url": iframe_url,
            "login_url_type": "aapanel_iframe_bridge",
            "sso_secret_created": secret_created,
        })


    def install(self, get):
        exists = self._target_app(get)
        if exists and self._bool_int(get.get("force", 0), 0) != 1:
            return self._ok({
                "installed": True,
                "message": "BillionMail is already installed",
                "billionmail": self._format_app(exists),
            })

        args, err = self._install_args(get)
        if err:
            return err

        if not public.progress_acquire_lock(self.INSTALL_LOCK_FILE):
            return self._err("BillionMail installation task is already running.")

        # 清理上次安装记录
        self._clear_install_progress()

        started = threading.Event()
        args_data = dict(args.get_items())
        worker = threading.Thread(target=self._install_worker, args=(args_data, started))
        worker.daemon = True
        worker.start()
        started.wait(timeout=5)
        public.set_module_logs('BillionMail', 'install')
        return self._ok({
            "result": "Successful startup!",
            "service_name": args.service_name,
            "progress_url": "/v2/mod/mail/billionmail/install_progress",
            "notice": "BillionMail installation has started. Please poll install_progress for details.",
        })

    def install_progress(self, get):
        """
        获取 docker,docker-compose,billionmail 安装进度
        2 = waiting
        0 = running
        1 = success
        -1 = failed
        """

        data = self._read_install_progress()
        billionmail = self._format_app(self._control_app())
        data["result"] = self._compact_install_result(data.get("result", {}))
        data["result"].update(self._install_result_app_summary(billionmail))
        data["docker"] = self._docker_status()
        return self._ok(data)

    def uninstall(self, get):
        app = self._control_app()
        if not app:
            return self._err("BillionMail is not installed")
        args = public.to_dict_obj({
            "id": app.get("id"),
            "delete_data": get.get("delete_data", 0),
        })
        try:
            from mod.project.docker.app.appManageMod import AppManage
            result = AppManage().remove_app(args)
            if self._is_success(result):
                self._clear_install_progress()
            return result
        except Exception as ex:
            return self._err("Failed to uninstall BillionMail: {}".format(str(ex)))

    def _set_status(self, get, status):
        app = self._control_app()
        if not app:
            return self._err("BillionMail is not installed")
        args = public.to_dict_obj({
            "app_name": self.APP_NAME,
            "service_name": app.get("service_name"),
            "status": status,
        })
        try:
            from mod.project.docker.app.appManageMod import AppManage
            aa = AppManage().set_app_status(args)
            time.sleep(2)
            return aa
        except Exception as ex:
            return self._err("Failed to {} BillionMail: {}".format(status, str(ex)))

    def start(self, get):
        docker_ready, docker_result = self._ensure_docker_ready(install_if_missing=False)
        if not docker_ready:
            return docker_result
        return self._set_status(get, "start")

    def stop(self, get):
        return self._set_status(get, "stop")

    def restart(self, get):
        docker_ready, docker_result = self._ensure_docker_ready(install_if_missing=False)
        if not docker_ready:
            return docker_result
        return self._set_status(get, "restart")

    def rebuild(self, get):
        return self._set_status(get, "rebuild")

    def upgrade(self, get):
        app = self._target_app(get)
        if not app:
            return self._err("BillionMail is not installed")

        m_version = self._as_text(get.get("m_version", ""))
        s_version = self._as_text(get.get("s_version", ""))
        if m_version and s_version:
            args = public.to_dict_obj({
                "id": app.get("id"),
                "m_version": m_version,
                "s_version": s_version,
                "backup": get.get("backup", False),
                "pull": get.get("pull", True),
            })
            try:
                from mod.project.docker.app.appManageMod import AppManage
                return AppManage().update_app(args)
            except Exception as ex:
                return self._err("Failed to upgrade BillionMail: {}".format(str(ex)))

        compose_file = self._compose_file(app)
        if not os.path.exists(compose_file):
            return self._err("docker-compose.yml was not found")
        compose_file = shlex.quote(compose_file)
        cmd = "docker-compose -f {0} pull && docker-compose -f {0} up -d".format(compose_file)
        stdout, stderr = public.ExecShell(cmd)
        return self._ok({
            "stdout": self._mask_text(stdout),
            "stderr": self._mask_text(stderr),
            "message": "BillionMail images are pulled and compose services are updated.",
        })

    def _tail_file(self, filename, lines=200):
        content = public.readFile(filename)
        if not content:
            return ""
        try:
            lines = int(lines)
        except Exception:
            lines = 200
        lines = max(20, min(lines, 1000))
        return self._mask_text("\n".join(content.splitlines()[-lines:]))

    def logs(self, get):
        app = self._target_app(get)
        if not app:
            return self._err("BillionMail is not installed")
        lines = get.get("lines", 200)
        service_path = self._service_path(app)
        log_files = {
            "install": os.path.join("/tmp", "{}.log".format(app.get("service_name"))),
            "core": os.path.join(service_path, "data/logs/core/latest"),
            "postfix": os.path.join(service_path, "data/logs/postfix/mail.log"),
            "dovecot": os.path.join(service_path, "data/logs/dovecot/mail.log"),
        }
        return self._ok({
            "service_name": app.get("service_name"),
            "logs": dict((name, self._tail_file(path, lines)) for name, path in log_files.items()),
        })

    def installed_apps(self, get):
        apps = []
        for app in self._all_installed_apps():
            if app.get("appname") == self.APP_NAME:
                apps.append(self._format_app(app))
        return self._ok(apps)