import hashlib
import html
import io
import ipaddress
import json
import os
import re
import shutil
import sys
import threading
import time
import socket
import platform
from uuid import uuid4

import math
import requests
import simple_websocket
import websocket
from werkzeug.datastructures import FileStorage
from datetime import datetime

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

import public
from typing import Optional, Tuple, List, Callable, Union, Any, Dict
from .one_panel_api import OnePanelApiClient
from mod.project.node.dbutil import Node, NodeAPPKey

try:
    from BTPanel import cache
except:
    class _Cache:
        def __init__(self):
            self._data = {}

        def get(self, key):
            return self._data.get(key, None)

        def set(self, key, value, timeout):
            self._data[key] = value


    cache = _Cache()


def _get_os_info() -> Tuple[str, str]:
    os_data = public.readFile("/etc/os-release")
    if not os_data:
        return "", ""

    version = ["", ""]
    for line in os_data.split("\n"):
        if line.startswith("NAME="):
            name = line.split("=")[1].replace('"', "").strip()
            name_arr = name.split(" ")
            if len(name_arr) > 1:
                name = name_arr[0]
            version[0] = name
        elif line.startswith("VERSION="):
            ver = line.split("=")[1].replace('"', "").strip()
            ver_arr = ver.split(" ")
            if len(ver_arr) > 1:
                ver = ver_arr[0]
            version[1] = ver

    version = " ".join(version)
    return version, platform.machine()


class ServerNode:
    is_local = False

    def __init__(self, origin: str, api_key: str, app_key: str, remarks: str = "", timeout: int = 20):
        self.origin = origin
        self.api_key = api_key
        self.app_key: Optional[NodeAPPKey] = Node.parse_app_key(app_key)
        self.remarks = remarks
        self.timeout = timeout

    @property
    def node_server_ip(self):
        from urllib.parse import urlparse
        host = urlparse(self.origin).hostname
        try:
            if isinstance(host, str) and ipaddress.ip_address(host):
                return host
        except:
            pass
        try:
            ip_address = socket.gethostbyname(host)
            return ip_address
        except socket.gaierror as e:
            print(f"Error: {e}")
            return host

    def app_bind(self) -> Optional[str]:
        if self.app_key is None:
            return "app秘钥解析失败，请重新确认秘钥信息"
        version, arch = _get_os_info()
        public.print_log(self.app_key.app_token)
        pdata = {
            "bind_token": self.app_key.app_token,
            "client_brand": "BT-Panel",
            "client_model": "Linux" if not version else "Linux {} {}".format(version, arch)
        }
        header = {
            'Content-Type': 'application/x-www-form-urlencoded',
            "User-Agent": "Bt-Panel/Node Manager"
        }
        bind_url = self.app_key.origin + "/check_bind"
        try:
            res = requests.post(bind_url, data=pdata, headers=header, verify=False, timeout=self.timeout)
            if res.status_code != 200:
                return "绑定请求失败，返回状态码：%s,响应信息为：%s" % (res.status_code, res.text)
            res_str = res.text.strip()
            if res_str == "0":
                return "绑定失败，请检查您的密钥是否正确"
            elif res_str == "1":
                return None
            else:
                return "绑定失败，错误信息为：{}".format(json.loads(res_str))
        except Exception as e:
            return "绑定节点失败，错误信息为：{}".format(str(e))

    def app_bind_status(self) -> Optional[str]:
        if self.app_key is None:
            return "app秘钥解析失败，请重新确认秘钥信息"
        pdata = {"bind_token": self.app_key.app_token}
        header = {
            'Content-Type': 'application/x-www-form-urlencoded',
            "User-Agent": "Bt-Panel/Node Manager"
        }
        bind_url = self.app_key.origin + "/get_app_bind_status"
        try:
            resp = requests.post(bind_url, data=pdata, headers=header, verify=False, timeout=self.timeout)
            if not resp.status_code == 200:
                return "获取绑定状态响应状态码错误，请检查节点地址和api是否正确，目前状态码为{},返回信息为:{}".format(
                    resp.status_code, resp.text)
            res_str = resp.text.strip()
            if res_str == "0":
                return "您的设备未绑定，请尝试重新绑定"
            elif res_str == "1":
                return None
            else:
                return "请求绑定状态失败，错误信息为：{}".format(json.loads(res_str))
        except Exception as e:
            return "获取绑定状态失败，请检查节点app秘钥是否正确，错误信息为:{}".format(str(e))

    @classmethod
    def new_by_id(cls, node_id: int) -> Optional[Union['ServerNode', 'LocalNode', 'LPanelNode']]:
        data = public.M("node").where("id = ?", (node_id,)).find()
        if not data or not isinstance(data, dict):
            return None
        return cls.new_by_data(data)

    @classmethod
    def new_by_data(cls, node_data: dict):
        if node_data["api_key"] == "local" and node_data["app_key"] == "local":
            return LocalNode()

        if node_data['lpver']:
            lp = LPanelNode(
                address=node_data['address'],
                api_key=node_data['api_key'],
                lpver=node_data['lpver'],
                timeout=node_data.get('timeout', 20)
            )
            lp.remarks = node_data['remarks']
            return lp
        if not node_data["api_key"] and not node_data["app_key"]:
            return None

        return cls(
            origin=node_data['address'],
            api_key=node_data['api_key'],
            app_key=node_data['app_key'],
            remarks=node_data['remarks'],
            timeout=node_data.get('timeout', 20)
        )

    def show_name(self) -> str:
        if self.remarks:
            return "{}({})".format(self.remarks, self.origin)
        return self.origin

    @staticmethod
    def _get_node_ip(node_id: int) -> str:
        data = public.M("node").where("id = ?", (node_id,)).find()
        if not data or not isinstance(data, dict):
            return ""
        if data['address'].startswith("http"):
            from urllib.parse import urlparse
            host = urlparse(data['address']).hostname
        else:
            host = data['address']
        try:
            if isinstance(host, str) and ipaddress.ip_address(host):
                return host
        except:
            pass
        try:
            ip_address = socket.gethostbyname(host)
            return ip_address
        except socket.gaierror as e:
            print(f"Error: {e}")
            return ""

    @classmethod
    def get_node_ip(cls, node_id: int) -> Optional[str]:
        if public.M('node').where("id=? AND app_key = 'local' AND api_key = 'local'", (node_id,)).count() > 0:
            return "127.0.0.1"
        cache_key = "mod_node_server_ip_{}".format(node_id)
        c_ip = cache.get(cache_key)
        if c_ip:
            return c_ip
        else:
            ip = cls._get_node_ip(node_id)
            if ip:
                cache.set(cache_key, ip, 86400)
                return ip
            return None

    @classmethod
    def check_api_key(cls, node: Node) -> str:
        if not LPanelNode.check_api_key(node):  # 如果1panel检查可以连接上，则标记为1panel并直接返回
            return ""

        node = cls(node.address, node.api_key, node.app_key)
        data, err = node._request("/system", "GetSystemTotal")
        if err:
            return err
        if isinstance(data, dict) and "version" in data:
            return ""
        return "验证错误：%s" % str(data)

    def test_conn(self) -> str:
        data, err = self._request("/system", "GetSystemTotal")
        if err:
            return err
        if isinstance(data, dict) and "version" in data:
            return ""
        return "验证错误：%s" % str(data)

    def version(self) -> Optional[str]:
        data, err = self._request("/system", "GetSystemTotal")
        if err:
            return None
        if isinstance(data, dict) and "version" in data:
            return data["version"]
        return None

    def get_net_work(self) -> Tuple[Optional[dict], str]:
        data, err = self._request("/system", "GetNetWork")
        if err:
            return None, err
        if isinstance(data, dict) and "cpu" in data and "mem" in data:
            return data, ""
        return None, "数据格式错误: %s" % str(data)

    def get_tmp_token(self) -> Tuple[str, str]:
        data, err = self._request("/config", "get_tmp_token")
        if err:
            return "", err
        if isinstance(data, dict) and "status" in data:
            if data["status"] and "msg" in data:
                return data["msg"], ""
            return "", data.get("msg", str(data))
        return "", "数据格式错误: %s" % str(data)

    def get_bt_params(self, other_data: dict = None) -> Dict:
        now = int(time.time() * 1000)
        other_data = other_data or {}
        if self.app_key:
            md5_panel_key = hashlib.md5(self.app_key.request_token.encode()).hexdigest()
            request_token = hashlib.md5("{}{}".format(now, md5_panel_key).encode()).hexdigest()
            other_data["request_token"] = request_token
            other_data["request_time"] = str(now)
            form_data = json.dumps(other_data)
            return {
                "client_bind_token": self.app_key.app_token,
                "form_data": public.aes_encrypt(form_data, self.app_key.app_key)
            }
        else:
            md5_panel_key = hashlib.md5(self.api_key.encode()).hexdigest()
            request_token = hashlib.md5("{}{}".format(now, md5_panel_key).encode()).hexdigest()
            other_data["request_token"] = request_token
            other_data["request_time"] = str(now)
            return other_data

    def _request(self, path: str, action: str, pdata: dict = None) -> Tuple[Optional[any], str]:
        url = "{}{}".format(self.origin, path)
        pdata = pdata or {}
        bt_p = self.get_bt_params({"action": action, **pdata})
        header = {
            'Content-Type': 'application/x-www-form-urlencoded',
            "User-Agent": "Bt-Panel/Node Manager"
        }
        try:
            resp = requests.post(url, data=bt_p, headers=header, verify=False, timeout=self.timeout)
            if not resp.status_code == 200:
                return None, "响应状态码错误，请检查节点地址和api是否正确，目前响应状态码为{}".format(
                    resp.status_code)
            if self.app_key:
                real_data = public.aes_decrypt(resp.text, self.app_key.app_key)
                return json.loads(real_data), ""
            return resp.json(), ""
        except Exception as e:
            public.print_error()
            # return None, "请求节点失败，请检查节点地址和api是否正确，错误信息为:{}".format(str(e))
            if self.remarks:
                return None, "请求节点【{}】失败，请检查节点地址和api是否正确".format(self.remarks)
            return None, "请求节点失败，请检查节点地址和api是否正确"

    def get_file_body(self, path: str) -> Tuple[Optional[str], str]:
        data, err = self._request("/files", "GetFileBody", pdata={"path": path})
        if err:
            return None, err
        if not isinstance(data, dict):
            if isinstance(data, dict) and "msg" in data:
                return None, data["msg"]
            return None, "数据格式错误"
        return data["data"], ""

    def php_site_list(self) -> Tuple[Optional[List[dict]], str]:
        data, err = self._request("/mod/node/node/php_site_list", "", pdata={
            "type": -1,
            "p": 1,
            "limit": 1000,
            "table": "sites",
            "search": "",
        })
        if err:
            return None, err

        if not isinstance(data, list):
            if isinstance(data, dict) and "msg" in data:
                return None, data["msg"]
            return None, "数据格式错误"

        return data, ""

    def create_php_site(self, site_name: str, port: int, **kwargs) -> Tuple[Optional[int], str]:
        path = "/www/wwwroot/{}".format(site_name)
        if port not in (80, 443):
            path = "/www/wwwroot/{}_{}".format(site_name, port)
        webname = {
            "domain": site_name if port in (80, 443) else "{}:{}".format(site_name, port),
            "domainlist": [],
            "count": 0
        }
        data, err = self._request("/site", "AddSite", pdata={
            "path": path,
            "ftp": "false",
            "type": "PHP",
            "type_id": "0",
            "ps": "{}【负载均衡节点】".format(site_name),
            "port": str(port),
            "version": "00",
            "sql": "false",
            "webname": json.dumps(webname)
        })

        if err:
            return None, err
        if not isinstance(data, dict):
            return None, "数据格式错误"

        if isinstance(data, dict) and "msg" in data and not data.get("status", True):
            return None, data["msg"]

        site_id = data.get("siteId", None)
        if not isinstance(site_id, int):
            return None, "数据解析错误"
        return site_id, ""

    def set_firewall_open(self, port: int, protocol: str = "tcp") -> Tuple[bool, str]:
        data, err = self._request("/firewall/com/set_port_rule", "", pdata={
            "protocol": "udp" if protocol == "udp" else "tcp",
            "port": str(port),
            "choose": "all",
            "types": "accept",
            "chain": "INPUT",
            "operation": "add",
            "strategy": "accept",
        })
        if err:
            return False, err

        if isinstance(data, dict) and "msg" in data:
            msg = data["msg"]
            if "端口{}已存在".format(port) in msg:
                return True, ""
            elif "设置成功" in msg:
                return True, ""
        return False, "设置失败"

    def add_domain(self, site_id: int, site_name: str, domain: str, port: int) -> Tuple[bool, str]:
        data, err = self._request("/site", "AddDomain", pdata={
            "domain": "{}:{}".format(domain, port),
            "webname": site_name,
            "id": str(site_id)
        })
        if err:
            return False, err

        if isinstance(data, dict) and "domains" in data:
            for d in data["domains"]:
                if d["name"] == domain:
                    if "绑定过了" in d["msg"]:
                        return True, ""
                    elif "添加成功" in d["msg"]:
                        return True, ""
        return False, "添加失败,目标机返回信息为:{}".format(
            data.get("msg", str(data)) if isinstance(data, dict) else str(data))

    def has_domain(self, site_id: int, domain: str) -> bool:
        data, err = self._request("/data", "getData", pdata={
            "table": "domain",
            "list": "True",
            "search": str(site_id)
        })
        if err:
            return False
        if not isinstance(data, list):
            return False
        for d in data:
            if d["name"] == domain:
                return True
        return False

    # mode -> cover: 覆盖，ignore: 跳过，rename:重命名
    def upload_file(self, filename: str, target_path: str, mode: str = "cover",
                    call_log: Callable[[int, str], None] = None) -> str:
        if not os.path.isfile(filename):
            return "文件:{}不存在".format(filename)

        target_file = os.path.join(target_path, os.path.basename(filename))
        exits, err = self.target_file_exits(target_file)
        if err:
            return err
        if exits and mode == "ignore":
            call_log(0, "文件上传:{} -> {},目标文件已存在,跳过上传".format(filename, target_file))
            return ""
        if exits and mode == "rename":
            upload_name = "{}_{}".format(os.path.basename(filename), public.md5(filename))
            call_log(0, "文件上传:{} -> {},目标文件已存在,将重命名为{}".format(filename, target_file, upload_name))
        else:
            upload_name = os.path.basename(filename)

        if os.path.getsize(filename) > 1024 * 1024 * 5:
            return self._upload_big_file(filename, target_path, upload_name, call_log)
        else:
            return self._upload_little_file(filename, target_path, upload_name, call_log)

    def target_file_exits(self, target_file: str) -> Tuple[bool, str]:
        data, err = self._request("/files", "upload_files_exists", pdata={
            "files": target_file,
        })
        if err:
            return False, err
        if not isinstance(data, list):
            return False, "数据格式错误: %s" % str(data)
        for f in data:
            if f["filename"] == target_file and f["exists"]:
                return True, ""
        return False, ""

    def upload_check(self, target_file_list: List[str]) -> Tuple[List[dict], str]:
        data, err = self._request("/files", "upload_files_exists", pdata={
            "files": "\n".join(target_file_list),
        })
        if err:
            return [], err
        if not isinstance(data, list):
            if isinstance(data, dict):
                return [], data.get("msg", "数据格式错误")
            return [], "数据格式错误: %s" % str(data)
        return data, ""

    def _upload_big_file(self, filename: str, target_path: str, upload_name: str,
                         call_log: Callable[[int, str], None] = None) -> str:
        url = "{}/files".format(self.origin)
        bt_p = self.get_bt_params({"action": "upload"})
        header = {"User-Agent": "Bt-Panel/Node Manager"}
        try:
            fb = open(filename, 'rb')
        except Exception as e:
            public.print_error()
            return "文件{}打开失败，请检查文件权限，错误信息为:{}".format(filename, str(e))

        file_size = os.path.getsize(filename)
        for i in range(0, file_size, 1024 * 1024 * 5):
            file_data = fb.read(1024 * 1024 * 5)
            files = {'blob': ('blob', file_data, 'application/octet-stream')}
            data = {
                'f_path': target_path,
                'f_name': upload_name,
                'f_size': file_size,
                'f_start': i,
            }
            data.update(bt_p)
            try:
                resp = requests.post(url, data=data, files=files, headers=header, verify=False, timeout=self.timeout)
                if not resp.status_code == 200:
                    return "上传文件响应状态码错误，请检查节点地址和api是否正确，目前状态码为{},返回信息为:{}".format(
                        resp.status_code, resp.text)
                resp_text = resp.text
                resp_json = {}
                if self.app_key:
                    resp_text = public.aes_decrypt(resp_text, self.app_key.app_key).strip()
                    if not resp_text.isdecimal():
                        resp_json = json.loads(resp_text)
                else:
                    resp_json = resp.json()

                if resp_text.isdecimal():
                    up_p = int(resp_text)
                    if up_p != i + len(file_data):
                        return "上传文件失败，文件块大小不一致"
                    else:
                        call_log(up_p * 100 // file_size, "文件上传:{} -> {},已上传大小为:{}".format(
                            filename, upload_name, public.to_size(up_p)))
                elif "status" in resp_json:
                    if not resp_json["status"]:
                        return "上传文件失败，错误信息为:{}".format(resp_json["msg"])
                    else:
                        call_log(100, "文件上传:{} -> {},上传成功".format(filename, upload_name))
                        return ""
                else:
                    return "上传文件失败，响应信息为:{}".format(resp_text)
            except Exception as e:
                public.print_error()
                return "上传文件文件：{}失败，错误信息为:{}".format(filename, str(e))
        return ""

    def _upload_little_file(self, filename: str, target_path: str, upload_name: str,
                            call_log: Callable[[int, str], None] = None) -> str:
        url = "{}/files".format(self.origin)
        bt_p = self.get_bt_params({"action": "upload"})
        header = {"User-Agent": "Bt-Panel/Node Manager"}
        try:
            with open(filename, 'rb') as f:
                file_data = f.read()
        except Exception as e:
            public.print_error()
            return "文件{}打开失败，请检查文件权限，错误信息为:{}".format(filename, str(e))

        file_size = os.path.getsize(filename)
        files = {'blob': ('blob', file_data, 'application/octet-stream')}
        data = {
            'f_path': target_path,
            'f_name': upload_name,
            'f_size': file_size,
            'f_start': 0
        }
        data.update(bt_p)
        try:
            resp = requests.post(url, data=data, files=files, headers=header, verify=False, timeout=self.timeout)
            if not resp.status_code == 200:
                return "上传文件响应状态码错误，请检查节点地址和api是否正确，目前状态码为{},返回信息为:{}".format(
                    resp.status_code, resp.text)
            resp_json = {}
            if self.app_key:
                resp_text = public.aes_decrypt(resp.text, self.app_key.app_key).strip()
                if not resp_text.isdecimal():
                    resp_json = json.loads(resp_text)
            else:
                resp_json = resp.json()
            if not resp_json["status"]:
                return "上传文件失败，错误信息为:{}".format(resp.json()["msg"])
            return ""
        except Exception as e:
            public.print_error()
            return "上传文件文件：{}失败，错误信息为:{}".format(filename, str(e))

    def upload_proxy(self):
        try:
            from BTPanel import request
            f_name = request.form.get('f_name')
            f_path = request.form.get('f_path')
            f_size = request.form.get('f_size')
            f_start = request.form.get('f_start')
            blob_file: FileStorage = request.files.getlist('blob')[0]
            url = "{}/files".format(self.origin)
            bt_p = self.get_bt_params({"action": "upload"})
            header = {"User-Agent": "Bt-Panel/Node Manager"}
            files = {'blob': ('blob', blob_file.stream, 'application/octet-stream')}
            data = {
                'f_path': f_path,
                'f_name': f_name,
                'f_size': f_size,
                'f_start': f_start
            }
            data.update(bt_p)
            resp = requests.post(url, data=data, files=files, headers=header, verify=False, timeout=self.timeout)
            if not resp.status_code == 200:
                return "上传文件响应状态码错误，请检查节点地址和api是否正确，目前状态码为{},返回信息为:{}".format(
                    resp.status_code, resp.text)
            resp_text = resp.text
            resp_json = {}
            if self.app_key:
                resp_text = public.aes_decrypt(resp_text, self.app_key.app_key).strip()
                if not resp_text.isdecimal():
                    resp_json = json.loads(resp_text)
            else:
                resp_json = resp.json()
            if resp_text.isdecimal():
                return int(resp_text)
            elif "status" in resp_json:
                return resp_json
            else:
                return {"status": False, "msg": "上传文件失败，响应信息为:{}".format(resp_text)}
        except Exception as e:
            public.print_error()
            return {"status": False, "msg": "上传文件失败，错误信息为:{}".format(str(e))}

    def remove_file(self, filename: str, is_dir: bool) -> dict:
        action = "DeleteDir" if is_dir else "DeleteFile"
        data, err = self._request("/files", action, pdata={
            "path": filename,
        })
        if err:
            return {"status": False, "msg": "删除文件失败，错误信息为:{}".format(err)}

        if isinstance(data, dict):
            return data
        return {"status": False, "msg": "删除文件失败，响应信息为:{}".format(data)}

    def create_dir(self, path: str) -> Tuple[Optional[Dict], str]:
        data, err = self._request("/files", action="CreateDir", pdata={
            "path": path,
        })
        if err:
            return None, "创建目录失败，错误信息为:{}".format(err)
        if isinstance(data, dict):
            return data, ""
        return None, "创建目录失败，响应信息为:{}".format(data)

    def dir_walk(self, path: str) -> Tuple[List[dict], str]:
        data, err = self._request("/mod/node/file_transfer/dir_walk", "", pdata={
            "path": path,
        })
        if err:
            return [], err

        if not isinstance(data, list):
            if isinstance(data, dict) and "msg" in data:
                return [], data["msg"]
            return [], "数据格式错误: %s" + str(data)

        return data, ""

    def filetransfer_version_check(self) -> str:
        ver = self.version()
        if not ver:
            return "节点{}连接错误".format(self.show_name())

        try:
            ver_list = [int(i) for i in ver.split(".")]
            if self.app_key:  # todo: 版本号检测
                if ver_list[0] > 11 or (ver_list[0] == 11 and ver_list[1] >= 1):
                    return "节点{}版本低于【11.1.0】，无法使用app链接，请升级节点版本".format(self.show_name())

            if ver_list[0] > 10 or (ver_list[0] == 10 and ver_list[2] >= 1):
                return ""
            elif ver_list[0] == 9 and ver_list[1] >= 7:
                return ""
            return "请将面板版本升级到【10.0.1】及以上再使用"
        except:
            pass
        return "请将面板版本升级到【10.0.1】及以上再使用"


    def node_create_filetransfer_task(self, source_node: dict,
                                      target_node: dict,
                                      source_path_list: List[dict],
                                      target_path: str,
                                      created_by: str,
                                      default_mode: str = "cover") -> dict:
        data, err = self._request("/mod/node/file_transfer/node_create_filetransfer_task", "", pdata={
            "source_node": json.dumps(source_node),
            "target_node": json.dumps(target_node),
            "source_path_list": json.dumps(source_path_list),
            "target_path": target_path,
            "created_by": created_by,
            "default_mode": default_mode,
        })
        if err:
            return {"status": False, "msg": "创建文件传输任务失败，错误信息为:{}".format(err)}
        if isinstance(data, dict):
            return data
        else:
            return {"status": False, "msg": "创建文件传输任务失败，响应信息为:{}".format(data)}

    def file_list(self, path: str, p: int, row: int, search: str) -> Tuple[Dict, str]:
        data, err = self._request("/files", "GetDirNew", pdata={
            "path": path,
            "p": p,
            "showRow": row,
            "search": search,
        })
        if err:
            return {}, err
        if not isinstance(data, dict):
            return {}, "数据格式错误: %s" + str(data)
        return data, ""

    def get_transfer_status(self, task_id: int) -> dict:
        data, err = self._request("/mod/node/file_transfer/get_transfer_status", "", pdata={
            "task_id": task_id,
        })
        if err:
            return {"status": False, "msg": "获取文件传输任务状态失败，错误信息为:{}".format(err)}
        if isinstance(data, dict):
            return data
        return {"status": False, "msg": "获取文件传输任务状态失败，响应信息为:{}".format(data)}

    def proxy_transfer_status(self, task_id: int, ws: simple_websocket.Server):
        err = self._proxy_websocket(
            call_data={
                "mod_name": "node",
                "sub_mod_name": "file_transfer",
                "def_name": "node_proxy_transfer_status",
                "callback": "node_proxy_transfer_status",
                "data": {
                    "task_id": task_id,
                }
            },
            uri="ws_modsoc",
            call_back=ws.send
        )
        if err:
            ws.send(json.dumps({
                "type": "error",
                "msg": "获取文件传输任务状态失败，错误信息为:{}".format(err)
            }))

    def _proxy_websocket(self, call_data: dict, uri: str, call_back: Callable[[Any],None]) -> str:
        from urllib.parse import urlencode, urlparse
        from ssl import CERT_NONE
        try:
            bt_p = self.get_bt_params()
            header = {"User-Agent": "Bt-Panel/Node Manager"}
            u = urlparse(self.origin)
            url = (
                "{}://{}/{}?{}".format(
                    "ws" if u.scheme == "http" else "wss",
                    u.netloc, uri, urlencode(bt_p)
                )
            )
            ws_req = websocket.WebSocket(sslopt={"cert_reqs": CERT_NONE}) # 忽略证书
            ws_req.connect(url, header=header)
            ws_req.send("{}") # 跳过x-http-tokn 验证
            ws_req.send(json.dumps(call_data))  # 发送调用数据

            while True:
                data = ws_req.recv()
                if data == "{}":
                    break
                if data:
                    call_back(data)
                else:
                    break
            ws_req.close()
            return ""
        except Exception as e:
            return str(e)

    def proxy_transferfile_status(self, task_id: int, call_back: Callable[[Any],None]) -> str:
        return self._proxy_websocket(
            call_data={
                "mod_name": "node",
                "sub_mod_name": "executor",
                "def_name": "node_proxy_transferfile_status",
                "callback": "node_proxy_transferfile_status",
                "data": {
                    "task_id": task_id,
                }
            },
            uri="ws_modsoc",
            call_back=call_back
        )

    def node_create_transfer_task(self, transfer_task_data:dict) -> dict:
        data, err = self._request(
            "/mod/node/executor/node_create_transfer_task", "", pdata={
                "transfer_task_data": json.dumps(transfer_task_data),
            }
        )
        if err:
            return {"status": False, "msg": "创建文件传输任务失败，错误信息为:{}".format(err)}
        if isinstance(data, dict):
            return data
        else:
            return {"status": False, "msg": "创建文件传输任务失败，响应信息为:{}".format(data)}

    def node_transferfile_status_history(self, transfer_task_id: int, only_error=True):
        data, err = self._request(
            "/mod/node/executor/node_transferfile_status_history", "", pdata={
                "task_id": transfer_task_id,
                "only_error": 1 if only_error else 0
            }
        )
        if err:
            return {"status": False, "msg": "创建文件传输任务失败，错误信息为:{}".format(err)}
        if isinstance(data, dict):
            return data
        else:
            return {"status": False, "msg": "创建文件传输任务失败，响应信息为:{}".format(data)}

    def download_proxy(self, filename: str):
        url = "{}/download".format(self.origin)
        bt_p = self.get_bt_params()
        header = {"User-Agent": "Bt-Panel/Node Manager"}
        try:
            resp = requests.get(url, params={"filename": filename}, data=bt_p, headers=header, stream=True, verify=False, timeout=self.timeout)
            if not resp.status_code == 200:
                return "下载文件响应状态码错误，请检查节点地址和api是否正确，目前状态码为{},返回信息为:{}".format(
                    resp.status_code, resp.text)

            from flask import send_file, stream_with_context, Response
            filename = os.path.basename(filename)
            if resp.headers.get("Content-Disposition", "").find("filename=") != -1:
                filename = resp.headers.get("Content-Disposition", "").split("filename=")[1]

            def generate():
                for chunk in resp.iter_content(chunk_size=1024 * 1024 * 5):
                    if chunk:
                        yield chunk

            # 设置响应头
            headers = {
                'Content-Type': resp.headers.get('Content-Type', 'application/octet-stream'),
                'Content-Disposition': 'attachment; filename="{}"'.format(filename),
                'Content-Length': resp.headers.get('Content-Length', ''),
                'Accept-Ranges': 'bytes'
            }

            # 使用 stream_with_context 确保请求上下文在生成器运行时保持活跃
            return Response(
                stream_with_context(generate()),
                headers=headers,
                direct_passthrough=True
            )
        except Exception as e:
            public.print_error()
            return "下载文件：{}失败，错误信息为:{}".format(filename, str(e))

    def download_file(self, filename: str, target_path: str, mode: str,
                      call_log: Callable[[int, str], None] = None) -> str:
        target_file = os.path.join(target_path, os.path.basename(filename))
        exits = os.path.exists(target_file)
        if exits and mode == "ignore":
            call_log(0, "文件下载:{} -> {},目标文件已存在,跳过下载".format(filename, target_file))
            return ""
        if exits and mode == "rename":
            download_name = "{}_{}".format(os.path.basename(filename), public.md5(filename))
            call_log(0, "文件下载:{} -> {},目标文件已存在,将重命名为{}".format(filename, target_file, download_name))
        else:
            download_name = os.path.basename(filename)

        return self._download_file(filename, target_path, download_name, call_log)

    def _download_file(self, filename: str, target_path: str, download_name: str,
                       call_log: Callable[[int, str], None] = None) -> str:
        data, err = self.upload_check([filename])
        if err:
            return "请求文件：{}的状态失败，错误信息为:{}".format(filename, err)
        file_size: Optional[int] = None
        for i in data:
            if i["filename"] == filename and i["isfile"] == True:
                file_size = i["size"]
                break

        if file_size is None:
            return "文件{}不存在, 跳过下载".format(filename)
        try:
            if not os.path.isdir(target_path):
                os.makedirs(target_path)
        except  Exception as e:
            return "创建文件夹{}失败，请检查文件夹权限，错误信息为:{}".format(target_path, str(e))

        if file_size == 0:
            fp = open(os.path.join(target_path, download_name), "w")
            fp.close()
            return ""

        tmp_file = os.path.join(target_path, "{}.{}".format(download_name, uuid4().hex))
        try:
            if not os.path.exists(target_path):
                os.makedirs(target_path)
            fb = open(tmp_file, 'wb')
        except  Exception as e:
            return "创建临时文件{}失败，请检查文件夹权限，错误信息为:{}".format(tmp_file, str(e))

        url = "{}/download".format(self.origin)
        bt_p = self.get_bt_params()
        header = {"User-Agent": "Bt-Panel/Node Manager"}
        try:
            resp = requests.get(url, params={"filename": filename}, data=bt_p, headers=header, stream=True, verify=False, timeout=self.timeout)
            if not resp.status_code == 200:
                return "下载文件响应状态码错误，请检查节点地址和api是否正确，目前状态码为{},返回信息为:{}".format(
                    resp.status_code, resp.text)

            now_size = 0
            for chunk in resp.iter_content(chunk_size=1024 * 1024 * 3):
                if chunk:
                    now_size += len(chunk)
                    fb.write(chunk)
                    fb.flush()
                    call_log(now_size * 100 // file_size,
                             "下载文件{}, 已下载: {}".format(filename, public.to_size(now_size)))
            if fb.tell() != file_size:
                return "下载文件{}失败，文件大小检查错误".format(filename)
            fb.close()
            shutil.move(tmp_file, os.path.join(target_path, download_name))
            return ""
        except Exception as e:
            return "下载文件{}失败，错误信息为:{}".format(filename, str(e))
        finally:
            if not fb.closed:
                fb.close()
            if os.path.exists(tmp_file):
                os.remove(tmp_file)

    def dir_size(self, path: str) -> Tuple[Optional[int], str]:
        data, err = self._request("/files", "get_path_size", pdata={
            "path": path,
        })
        if err:
            return None, err
        if isinstance(data, dict) and "size" in data:
            return data["size"], ""
        return None, "获取目录大小失败，响应信息为:{}".format(data)

    def get_sshd_port(self) -> Optional[int]:
        data, err = self._request("/safe/ssh/GetSshInfo", "", pdata={})
        if err:
            return None
        if isinstance(data, dict) and "port" in data:
            return int(data["port"])
        return None

    def restart_bt_panel(self) -> Dict[str, Any]:
        res, err = self._request("/system", "ReWeb", pdata={})
        if err:
            return {"status": False, "msg": err}
        return {"status": True, "msg": "重启成功"}

    def server_reboot(self):
        res, err = self._request("/system", "ServiceAdmin", pdata={
            "name": "nginx",
            "type": "stop",
        })
        if err:
            return {"status": False, "msg": err}
        if not res["status"]:
            return {"status": False, "msg": "nginx停止失败, 无法继续执行重启服务器：" + res["msg"]}

        res, err = self._request("/system", "ServiceAdmin", pdata={
            "name": "mysqld",
            "type": "stop",
        })
        if err:
            return {"status": False, "msg": err}
        if not res["status"]:
            return {"status": False, "msg": "mysql服务停止失败, 无法继续执行重启服务器：" + res["msg"]}

        res, _ = self._request("/system", "RestartServer", pdata={})
        if not res["status"]:
            return {"status": False, "msg": "重启服务器失败：" + res["msg"]}

        from mod.project.node.dbutil import ServerMonitorRepo
        repo = ServerMonitorRepo()
        repo.set_wait_reboot(self.node_server_ip, True)

        def wait_for_reboot():
            # wait 等待服务器重启成功， 超时时间默认为 10 分钟
            wait_for = time.time() + 600
            time.sleep(3)
            while time.time() < wait_for:
                if self.test_conn() == "":  # 无错误时表示重启成功
                    repo.set_wait_reboot(self.node_server_ip, False)
                    from mod.project.node.dbutil import ServerNodeDB
                    node_data = ServerNodeDB().find_node(api_key=self.api_key, app_key=self.app_key.to_string())
                    if node_data:
                        monitor_node_once(node_data)
                    return {"status": True, "msg": "重启服务器成功"}
                time.sleep(3)
                public.print_log("等待服务器重启中... {}".format(wait_for - time.time()))
            repo.set_wait_reboot(self.node_server_ip, False)
            return {"status": False, "msg": "重启服务器失败, 已超过10分钟未能检测到服务器信息"}

        t = threading.Thread(target=wait_for_reboot, daemon=True)
        t.start()

        return {"status": True, "msg": "重启服务器已开始，请赖心等待重启成功"}

    def read_ssh_key(self) -> Tuple[Optional[str], str]:
        data, err = self._request("/ssh_security", "get_key", pdata={})
        if err:
            return None, err
        if not isinstance(data, dict) or "status" not in data:
            return None, "获取SSH密钥失败，请检查节点地址和api是否正确"
        if not data["status"]:
            return None, data["msg"]
        return data["msg"], ""

    def get_dir(self, path: str, search: str, disk: str):
        data, err = self._request("/files", "GetDir", pdata={
            "path": path, "disk": disk, "search": search
        })
        if err:
            return {"status": False, "msg": err}
        if not isinstance(data, dict):
            return {"status": False, "msg": "获取目录失败，请检查节点地址和api是否正确"}
        return data

    def upload_dir_check(self, target_file: str) -> str:
        data, err = self._request("/files", "upload_files_exists", pdata={"files": target_file})
        if err:
            return err
        if not isinstance(data, list):
            return "数据格式错误: %s" % str(data)
        for f in data:
            if f["filename"] == target_file and f["exists"]:
                if f["isfile"]:
                    return "该名称路径不是目录"
                else:
                    return ""
        return ""


def monitor_all_node_status():
    all_nodes = public.M("node").select()
    if not isinstance(all_nodes, list) or not all_nodes:
        return

    import threading

    threads = []
    for tmp_node_data in all_nodes:
        t = threading.Thread(target=monitor_node_once, args=(tmp_node_data,))
        t.start()
        threads.append(t)

    # 等待所有线程完成（可选）
    for t in threads:
        t.join()


def monitor_node_once(node_data: dict):
    from mod.project.node.dbutil import Node, ServerNodeDB, ServerMonitorRepo
    from mod.project.node.nodeutil.ssh_wrap import SSHApi
    from mod.base.ssh_executor import test_ssh_config
    try:
        if node_data["app_key"] == "local" and node_data["api_key"] == "local":
            return
        node_data["error"] = json.loads(node_data["error"])
        node_data["ssh_conf"] = json.loads(node_data["ssh_conf"])
        node, err = Node.from_dict(node_data)
        if err:
            public.print_log("节点数据解析错误:{}".format(err))
            return

        if node_data["lpver"]:
            srv = LPanelNode(node.address, node.api_key, node.lpver)
        elif node.api_key or node.app_key:
            srv = ServerNode(node.address, node.api_key, node.app_key)
        else:
            srv =SSHApi(**node_data["ssh_conf"])

        data, err = srv.get_net_work()
        if err:
            node.error_num += 1
            node.error = {
                "msg": err,
                "time": int(time.time())
            }
            if node.error_num >= 2:
                node.status = 0
                ServerMonitorRepo().remove_cache(node.id)

            ServerNodeDB().update_node(node)

        else:
            if node.error_num > 0:
                node.error_num = 0
                node.error = {}
                node.status = 1
                ServerNodeDB().update_node(node)

        if data:
            repo = ServerMonitorRepo()
            repo.save_server_status(node.id, data)
            if repo.is_reboot_wait(node.server_ip):
                repo.set_wait_reboot(node.server_ip, False)
        if not node_data["ssh_test"] and not node_data["ssh_conf"] and not isinstance(srv, SSHApi):
            ssh_key, err = srv.read_ssh_key()
            if err:
                public.print_log("获取SSH密钥失败:{}".format(err))
                return
            port = srv.get_sshd_port()
            if not port:
                public.print_log("获取SSH服务端口失败")
                return

            conf = {
                "host": srv.node_server_ip,
                "pkey": ssh_key,
                "port": port,
                "username": "root",
                "password": "",
                "pkey_passwd": "",
            }
            err = test_ssh_config(**conf)
            if err:
                public.print_log("SSH密钥测试失败:{}".format(err))
                ServerNodeDB().set_node_ssh_conf(node.id, {}, ssh_test=1)
                return
            else:
                ServerNodeDB().set_node_ssh_conf(node.id, conf, ssh_test=1)

    except Exception as e:
        if public.is_debug():
            public.print_error()


def monitor_node_once_with_timeout(node_data: dict, timeout: int = 5):
    if node_data["app_key"] == "local" and node_data["api_key"] == "local":
        return
    from mod.project.node.dbutil import Node, ServerNodeDB, ServerMonitorRepo
    from mod.project.node.nodeutil.ssh_wrap import SSHApi
    try:
        node_data["error"] = json.loads(node_data["error"])
        node_data["ssh_conf"] = json.loads(node_data["ssh_conf"])
        node, err = Node.from_dict(node_data)
        if err:
            public.print_log("节点数据解析错误:{}".format(err))
            return

        if node_data["app_key"] or node_data["api_key"]:
            node_data["timeout"] = timeout
            srv = ServerNode.new_by_data(node_data)
            node_data.pop("timeout")
        elif node_data["ssh_conf"]:
            srv = SSHApi(**node_data["ssh_conf"], timeout=timeout)
        else:
            return

        data, err = srv.get_net_work()
        if err:
            node.error_num += 1
            node.error = {
                "msg": err,
                "time": int(time.time())
            }
            if node.error_num >= 2:
                node.status = 0
                ServerMonitorRepo().remove_cache(node.id)

            ServerNodeDB().update_node(node)

        elif node.error_num > 0:
            node.error_num = 0
            node.error = {}
            node.status = 1
            ServerNodeDB().update_node(node)

        if data:
            repo = ServerMonitorRepo()
            repo.save_server_status(node.id, data)
            if repo.is_reboot_wait(node.server_ip):
                repo.set_wait_reboot(node.server_ip, False)
    except Exception as e:
        if public.is_debug():
            public.print_error()


class LocalNode(ServerNode):
    is_local = True

    def __init__(self):
        super().__init__("", "", "")

    def create_php_site(self, site_name: str, port: int = 80, **kwargs) -> Tuple[Optional[int], str]:
        path = "/www/wwwroot/{}".format(site_name)
        webname = {
            "domain": site_name if port in (80, 443) else "{}:{}".format(site_name, port),
            "domainlist": [],
            "count": 0
        }
        pdata = {
            "path": path,
            "ftp": "false",
            "type": "PHP",
            "type_id": "0",
            "ps": kwargs.get("ps") if kwargs.get("ps", None) else "{}【负载均衡站点】".format(site_name),
            "port": str(port),
            "version": "00",
            "sql": "false",
            "webname": json.dumps(webname)
        }

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

        from panelSite import panelSite

        res = panelSite().AddSite(public.to_dict_obj(pdata))

        if "siteId" in res:
            return res["siteId"], ""
        if "status" in res and not res["status"]:
            return 0, res.get("msg", "未知错误")
        return 0, "未知错误"

    @staticmethod
    def site_proxy_list(site_name: str) -> List[dict]:
        from mod.base.web_conf.proxy import RealProxy
        data = RealProxy(config_prefix="").get_proxy_list(public.to_dict_obj({
            "sitename": site_name
        }))
        if isinstance(data, list) and data:
            return data

        from panelSite import panelSite

        data = panelSite().GetProxyList(public.to_dict_obj({
            "sitename": site_name
        }))
        if isinstance(data, list) and data:
            return data
        return []

    def show_name(self) -> str:
        return "本机节点"

    def php_site_list(self) -> Tuple[List[dict], str]:
        from mod.base.web_conf.ssl import RealSSLManger
        ssl_m = RealSSLManger()

        all_sites = public.M("sites").where("project_type=? and status = 1", "PHP").select()
        res = []
        for i in all_sites:
            domain = public.M("domain").where("pid=?", i["id"]).select()
            domains = list(set([x["name"] for x in domain]))
            port = list(set([int(x["port"]) for x in domain]))
            ssl = ssl_m.get_site_ssl_info(i["name"]) is not None
            if ssl and 443 not in port:
                port.append(443)
            res.append({
                "site_id": i["id"],
                "site_name": i["name"],
                "ports": port,
                "domains": domains,
                "ssl": ssl
            })

        return res, ""

    def add_domain(self, site_id: int, site_name: str, domain: str, port: int) -> Tuple[bool, str]:
        from panelSite import panelSite
        res = panelSite().AddDomain(public.to_dict_obj({
            "domain": "{}:{}".format(domain, port),
            "webname": site_name,
            "id": str(site_id)
        }))
        if isinstance(res, dict) and res.get("status", False):
            return True, ""
        return False, res.get("msg", "未知错误")

    def has_domain(self, site_id: int, domain: str):
        return public.M("domain").where("pid=? and name=?", site_id, domain).count() > 0

    def dir_walk(self, path: str) -> Tuple[List[dict], str]:
        if not os.path.isdir(path):
            return [], "{} 不是一个目录".format(path)
        res_file = []
        count = 0
        empty_dir = []
        for root, dirs, files in os.walk(path):
            if not files:
                empty_dir.append(root)
            for f in files:
                if count > 1000:
                    return [], "目录文件数量超过1000,请压缩后再操作"
                count += 1
                try:
                    res_file.append({
                        "path": os.path.join(root, f),
                        "size": os.path.getsize(os.path.join(root, f)),
                        "is_dir": 0
                    })
                except:
                    pass
        return [{"path": d, "size": 0, "is_dir": 1} for d in empty_dir] + res_file, ""

    def remove_file(self, filename: str, is_dir: bool) -> dict:
        from files import files
        if is_dir:
            return files().DeleteDir(public.to_dict_obj({"path": filename}))
        else:
            return files().DeleteFile(public.to_dict_obj({"path": filename}))

    def file_list(self, path: str, p: int, row: int, search: str) -> Tuple[Dict, str]:
        from files import files
        return files().GetDirNew(public.to_dict_obj({
            "path": path,
            "p": p,
            "showRow": row,
            "search": search
        })), ""

    def upload_proxy(self):
        from files import files
        return files().upload(args=public.to_dict_obj({}))

    def upload_check(self, target_file_list: List[str]) -> Tuple[List[dict], str]:
        from files import files
        return files().upload_files_exists(args=public.to_dict_obj({
            "files": "\n".join(target_file_list),
        })), ""

    def dir_size(self, path: str) -> Tuple[Optional[int], str]:
        return public.get_path_size(path), ""

    def get_sshd_port(self) -> Optional[int]:
        return public.get_sshd_port()

    def create_dir(self, path: str) -> Tuple[Optional[Dict], str]:
        from files import files
        return files().CreateDir(public.to_dict_obj({"path": path})), ""

    def get_file_body(self, path: str) -> Tuple[Optional[str], str]:
        res = public.readFile(path)
        if isinstance(res, str):
            return res, ""
        return None, "读取文件失败"

    def read_ssh_key(self) -> Tuple[Optional[str], str]:
        from ssh_security import ssh_security
        data = ssh_security().get_key(None)
        if data["status"]:
            return data["msg"], ""
        return None, data["msg"]

    def get_dir(self, path: str, search: str, disk: str):
        from files import files
        return files().GetDir(public.to_dict_obj({"path": path, "search":search, "disk": disk}))


class LPanelNode(ServerNode):
    _TIME_REGEXP = re.compile(r":\d{2}(?P<err>\.\d{6,})[+-]\d{2}:\d{2}")

    def __init__(self, address: str, api_key: str, lpver: str = "v2", timeout: int = 20):
        super().__init__(address, api_key, "",  timeout=timeout)
        self.ver = lpver
        self.lpc = OnePanelApiClient(address, api_key, lpver, self.timeout)

    @classmethod
    def check_api_key(cls, node: Node) -> str:
        public.print_log(node.address, node.api_key)
        lpc = OnePanelApiClient(node.address, node.api_key)
        if lpc.test_ver():
            node.lpver = lpc.ver
            return ""
        return "1Panel节点测试连接失败"

    def test_conn(self) -> str:
        if self.lpc.test_ver():
            return ""
        return "1Panel节点测试连接失败"

    def get_net_work(self) -> Tuple[Optional[dict], str]:
        """{
    "code": 200,
    "message": "",
    "data": {
        "path": "/",
        "name": "/",
        "user": "root",
        "group": "root",
        "uid": "0",
        "gid": "0",
        "extension": "",
        "content": "",
        "size": 4096,
        "isDir": true,
        "isSymlink": false,
        "isHidden": false,
        "linkPath": "",
        "type": "",
        "mode": "0555",
        "mimeType": "",
        "updateTime": "0001-01-01T00:00:00Z",
        "modTime": "2023-12-16T18:21:10.369198018+08:00",
        "items": [],
        "itemTotal": 22,
        "favoriteID": 0,
        "isDetail": true
    }
}
API Response Status: 200
{
    "code": 200,
    "message": "",
    "data": {
        "uptime": 8060394,
        "timeSinceUptime": "2025-03-05 10:02:04",
        "procs": 173,
        "load1": 1.05,
        "load5": 1.05,
        "load15": 1.05,
        "loadUsagePercent": 17.5,
        "cpuPercent": [
            100,
            0,
            0,
            0
        ],
        "cpuUsedPercent": 25.000000232830644,
        "cpuUsed": 1.0000000093132257,
        "cpuTotal": 4,
        "memoryTotal": 1038336000,
        "memoryAvailable": 298635264,
        "memoryUsed": 572526592,
        "memoryUsedPercent": 55.13885601577909,
        "swapMemoryTotal": 4160745472,
        "swapMemoryAvailable": 3816017920,
        "swapMemoryUsed": 344727552,
        "swapMemoryUsedPercent": 8.285235285836778,
        "ioReadBytes": 0,
        "ioWriteBytes": 0,
        "ioCount": 0,
        "ioReadTime": 0,
        "ioWriteTime": 0,
        "diskData": [
            {
                "path": "/",
                "type": "xfs",
                "device": "/dev/mapper/centos-root",
                "total": 37688381440,
                "free": 9805086720,
                "used": 27883294720,
                "usedPercent": 73.98379461954417,
                "inodesTotal": 18411520,
                "inodesUsed": 729745,
                "inodesFree": 17681775,
                "inodesUsedPercent": 3.963523924151836
            },
            {
                "path": "/www/monitor_data",
                "type": "ext4",
                "device": "/dev/vdb1",
                "total": 10432565248,
                "free": 8118497280,
                "used": 1760526336,
                "usedPercent": 17.82085360286682,
                "inodesTotal": 655360,
                "inodesUsed": 10029,
                "inodesFree": 645331,
                "inodesUsedPercent": 1.530303955078125
            }
        ],
        "netBytesSent": 0,
        "netBytesRecv": 0,
        "gpuData": null,
        "xpuData": null,
        "shotTime": "2025-06-06T17:01:59.156659566+08:00"
    }
}"""
        try:
            data = self.lpc.system_status()
            system_data = data.get("data", {})
            return {
                "cpu": [round(system_data["cpuUsedPercent"], 2), system_data["cpuTotal"]],
                "mem": {
                    "memRealUsed": system_data["memoryUsed"] / 1024 / 1024,
                    "memTotal": system_data["memoryTotal"] / 1024 / 1024,
                    "memNewTotal": public.to_size(system_data["memoryTotal"])
                },
                "version": "1Panel"
            }, ""

        except Exception as e:
            public.print_error()
            return None, str(e)

    def get_tmp_token(self) -> Tuple[Optional[str], str]:
        return None, "1Panel不支持api临时访问"

    def php_site_list(self) -> Tuple[List[dict], str]:
        """{
          "code": 200,
          "message": "",
          "data": [
            {
              "id": 2,
              "createdAt": "2025-06-06T14:49:00.687860227+08:00",
              "updatedAt": "2025-06-06T14:49:00.687860227+08:00",
              "protocol": "HTTP",
              "primaryDomain": "www.halotest.com",
              "type": "deployment",
              "alias": "www.halotest.com",
              "remark": "",
              "status": "Running",
              "httpConfig": "",
              "expireDate": "9999-12-31T00:00:00Z",
              "proxy": "127.0.0.1:8090",
              "proxyType": "",
              "errorLog": true,
              "accessLog": true,
              "defaultServer": false,
              "IPV6": false,
              "rewrite": "",
              "webSiteGroupId": 1,
              "webSiteSSLId": 0,
              "runtimeID": 0,
              "appInstallId": 5,
              "ftpId": 0,
              "parentWebsiteID": 0,
              "user": "",
              "group": "",
              "dbType": "",
              "dbID": 0,
              "favorite": false,
              "domains": [
                {
                  "id": 2,
                  "createdAt": "2025-06-06T14:49:00.69011631+08:00",
                  "updatedAt": "2025-06-06T14:49:00.69011631+08:00",
                  "websiteId": 2,
                  "domain": "www.halotest.com",
                  "ssl": false,
                  "port": 7080
                },
                {
                  "id": 3,
                  "createdAt": "2025-06-06T14:51:47.932141706+08:00",
                  "updatedAt": "2025-06-06T14:51:47.932141706+08:00",
                  "websiteId": 2,
                  "domain": "qwww.com",
                  "ssl": false,
                  "port": 7080
                }
              ],
              "webSiteSSL": {
                "id": 0,
                "createdAt": "0001-01-01T00:00:00Z",
                "updatedAt": "0001-01-01T00:00:00Z",
                "primaryDomain": "",
                "privateKey": "",
                "pem": "",
                "domains": "",
                "certURL": "",
                "type": "",
                "provider": "",
                "organization": "",
                "dnsAccountId": 0,
                "acmeAccountId": 0,
                "caId": 0,
                "autoRenew": false,
                "expireDate": "0001-01-01T00:00:00Z",
                "startDate": "0001-01-01T00:00:00Z",
                "status": "",
                "message": "",
                "keyType": "",
                "pushDir": false,
                "dir": "",
                "description": "",
                "skipDNS": false,
                "nameserver1": "",
                "nameserver2": "",
                "disableCNAME": false,
                "execShell": false,
                "shell": "",
                "acmeAccount": {
                  "id": 0,
                  "createdAt": "0001-01-01T00:00:00Z",
                  "updatedAt": "0001-01-01T00:00:00Z",
                  "email": "",
                  "url": "",
                  "type": "",
                  "eabKid": "",
                  "eabHmacKey": "",
                  "keyType": "",
                  "useProxy": false,
                  "caDirURL": ""
                },
                "dnsAccount": {
                  "id": 0,
                  "createdAt": "0001-01-01T00:00:00Z",
                  "updatedAt": "0001-01-01T00:00:00Z",
                  "name": "",
                  "type": ""
                },
                "websites": null
              },
              "errorLogPath": "",
              "accessLogPath": "",
              "sitePath": "",
              "appName": "",
              "runtimeName": "",
              "runtimeType": "",
              "siteDir": ""
            }
          ]
        }"""
        try:
            data = self.lpc.get_websites()
            res = []
            for i in data["data"]:
                res.append({
                    "site_id": i["id"],
                    "site_name": i["alias"],
                    "ports": list(set([i["port"] for i in i["domains"]])),
                    "domains": list(set([i["domain"] for i in i["domains"]])),
                    "ssl": any(i["ssl"] for i in i["domains"])
                })
            return res, ""
        except Exception as e:
            return [], str(e)

    def create_php_site(self, site_name: str, port: int, **kwargs) -> Tuple[Optional[int], str]:
        try:
            dat = self.lpc.add_website(site_name, port, **kwargs)
            public.print_log(dat)
            if dat["code"] == 200:
                time.sleep(0.5)
                site_id = self.lpc.check_site_create(site_name)
                if isinstance(site_id, int):
                    return site_id, ""
                return None, "网站创建失败"
            return None, dat["message"]
        except Exception as e:
            return None, str(e)

    def set_firewall_open(self, port: int, protocol: str = "tcp") -> Tuple[bool, str]:
        try:
            dat = self.lpc.open_port(port, protocol)
            if dat["code"] == 200:
                return True, ""
            return False, dat["message"]
        except Exception as e:
            return False, str(e)

    def add_domain(self, site_id: int, site_name: str, domain: str, port: int) -> Tuple[bool, str]:
        try:
            dat = self.lpc.add_website_domain(site_id, domain, port)
            if dat["code"] == 200:
                return True, ""
            return False, dat["message"]
        except Exception as e:
            return False, str(e)

    def has_domain(self, site_id: int, domain: str) -> bool:
        try:
            dat = self.lpc.website_domains(site_id)
            if dat["code"] == 200:
                for i in dat["data"]:
                    if i["domain"] == domain:
                        return True
                return False
            return False
        except Exception as e:
            return False

    def target_file_exits(self, target_file: str) -> Tuple[bool, str]:
        data = self.lpc.files_exits([target_file])
        if not isinstance(data, dict):
            return False, "请求文件：{}的状态失败".format(target_file)
        for i in data["data"]:
            if i["path"] == target_file:
                return True, ""
        return False, ""

    def upload_dir_check(self, target_dir: str) -> str:
        data = self.lpc.files_exits([target_dir])
        if not isinstance(data, dict):
            return "请求文件：{}的状态失败".format(target_dir)
        for i in data["data"]:
            if i["path"] == target_dir:
                if not i.get("isDir", True):
                    return "该名称路径不是目录"
                return ""
        return ""

    def _upload_big_file(self, filename: str, target_path: str, upload_name: str,
                         call_log: Callable[[int, str], None] = None) -> str:
        try:
            fb = open(filename, 'rb')
        except Exception as e:
            public.print_error()
            return "文件{}打开失败，请检查文件权限，错误信息为:{}".format(filename, str(e))

        file_size = os.path.getsize(filename)
        count = math.ceil(file_size / (1024 * 1024 * 5))
        idx = 0
        for i in range(0, file_size, 1024 * 1024 * 5):
            file_data = fb.read(1024 * 1024 * 5)
            err, data = self.lpc.chunkupload(upload_name, target_path, file_data, idx, count)
            idx += 1
            if err:
                return "上传文件{}失败，错误信息为:{}".format(filename, str(err))
            if data:
                call_log(100, "文件上传:{} -> {},上传成功".format(filename, upload_name))
            else:
                up_d = (i + len(file_data)) // file_size
                call_log(up_d, "文件上传:{} -> {},已上传大小为:{}".format(
                    filename, upload_name, public.to_size(i + len(file_data))))
        return ""

    def _upload_little_file(self, filename: str, target_path: str, upload_name: str,
                            call_log: Callable[[int, str], None] = None) -> str:
        return self.lpc.upload(filename, target_path, upload_name)

    def download_file(self, filename: str, target_path: str, mode: str,
                      call_log: Callable[[int, str], None] = None) -> str:
        target_file = os.path.join(target_path, os.path.basename(filename))
        exits = os.path.exists(target_file)
        if exits and mode == "ignore":
            call_log(0, "文件下载:{} -> {},目标文件已存在,跳过下载".format(filename, target_file))
            return ""
        if exits and mode == "rename":
            download_name = "{}_{}".format(os.path.basename(filename), public.md5(filename))
            call_log(0, "文件下载:{} -> {},目标文件已存在,将重命名为{}".format(filename, target_file, download_name))
        else:
            download_name = os.path.basename(filename)

        return self.lpc.download_file(filename, target_path, download_name, call_log=call_log)

    def dir_walk(self, path: str) -> Tuple[List[dict], str]:
        return self.lpc.dir_walk(path)

    def upload_proxy(self):
        try:
            from BTPanel import request, cache
            f_name = request.form.get('f_name')
            f_path = request.form.get('f_path')
            f_size = request.form.get('f_size')
            f_start = request.form.get('f_start')
            cache_key = "upload_file_{}_{}_{}".format(f_name, f_path, f_size)
            num = cache.get(cache_key)
            if num is None:
                num = 0
                cache.set(cache_key, num, 86400)
            else:
                num = int(num)
                cache.set(cache_key, num + 1, 86400)
            blob_file: FileStorage = request.files.getlist('blob')[0]
            file_data = blob_file.read()
            next_size = int(f_start) + len(file_data)
            if next_size == int(f_size):
                chunk_num = num + 1
            else:
                chunk_num = num + 2
            err, data = self.lpc.chunkupload(f_name, f_path, file_data, num, chunk_num)
            if err:
                return {"status": False, "msg": "上传文件失败，错误信息为:{}".format(str(err))}
            elif data is None:
                return str(next_size)
            elif isinstance(data, dict) and data["code"] == 200:
                return {"status": True, "msg": "上传成功"}
            return {"status": False, "msg": "上传文件失败，错误信息为:{}".format(data["message"])}
        except Exception as e:
            public.print_error()
            return {"status": False, "msg": "上传文件失败，错误信息为:{}".format(str(e))}

    def remove_file(self, filename: str, is_dir: bool) -> dict:
        try:
            res = self.lpc.remove_file(filename, is_dir)
            if isinstance(res, dict):
                return {"status": res["code"] == 200, "msg": res["message"]}
            return {"status": False, "msg": "删除文件失败，请求远程失败"}
        except Exception as e:
            return {"status": False, "msg": "删除文件失败，错误信息为:{}".format(str(e))}

    def file_list(self, path: str, p: int, row: int, search: str) -> Tuple[Dict, str]:
        try:
            res, err = self.lpc.files_search(path, p, row, search)
            if err:
                return {}, "获取文件列表失败，错误信息为:{}".format(err)
            count = res["itemTotal"]
            path = res["path"]
            dirs = []
            files = []
            sub_items = [] if res["items"] is None else res["items"]
            for i in sub_items:
                # res = self._TIME_REGEXP.findall(i["modTime"])
                # public.print_log(res)
                mt_str = self._TIME_REGEXP.sub(lambda x: x.group().replace(x.group("err"), ""), i["modTime"])
                for f_str in ("%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%dT%H:%M:%S.%f%z"):
                    try:
                        mt = datetime.strptime(mt_str, f_str)
                        break
                    except:
                        pass
                        # public.print_log(i["modTime"], mt_str)
                        # public.print_error()
                else:
                    continue

                if i["isDir"]:
                    dirs.append({
                        "nm": html.unescape(i["name"]),
                        "sz": i["size"],
                        "mt": int(mt.timestamp()),
                        "acc": i["mode"][1:],
                        "user": i["user"],
                        "lnk": "" if not i["linkPath"] else " -> " + i["linkPath"],
                        "durl": "",
                        "cmp": 0,
                        "fav": "0",
                        "rmk": "",
                        "top": 0,
                        "sn": i["name"]
                    })
                else:
                    files.append({
                        "nm": html.unescape(i["name"]),
                        "sz": i["size"],
                        "mt": int(mt.timestamp()),
                        "acc": i["mode"][1:],
                        "user": i["user"],
                        "lnk": "" if not i["linkPath"] else " -> " + i["linkPath"],
                        "durl": "",
                        "cmp": 0,
                        "fav": "0",
                        "rmk": "",
                        "top": 0,
                        "sn": i["name"]
                    })

            return {
                "path": path,
                "page": public.get_page(count, p, row)["page"],
                "dir": dirs,
                "files": files
            }, ""
        except Exception as e:
            return {}, "获取文件列表失败，错误信息为:{}".format(str(e))

    def download_proxy(self, filename: str):
        return self.lpc.download_proxy(filename)

    def upload_check(self, target_file_list: List[str]) -> Tuple[List[dict], str]:
        try:
            res = self.lpc.files_exits(target_file_list)
            if res is None:
                return [], "请求远程1Panel失败"
            data = res.get("data", [])
            res_data = []
            for i in data:
                try:
                    mt = datetime.strptime(i["modTime"], "%Y-%m-%dT%H:%M:%S%z")
                except:
                    try:
                        mt = datetime.strptime(i["modTime"], "%Y-%m-%dT%H:%M:%S.%f%z")
                    except:
                        continue
                res_data.append({
                    'filename': i["path"],
                    'exists': True,
                    'size': i["size"],
                    'mtime': int(mt.timestamp()),
                    'isfile': False
                })

            return res_data, ""
        except Exception as e:
            return [], "请求远程1Panel失败，错误信息为:{}".format(str(e))

    def dir_size(self, path: str) -> Tuple[Optional[int], str]:
        try:
            res = self.lpc.dir_size(path)
            if res is None:
                return None, "请求远程1Panel失败"
            if not isinstance(res, dict):
                return None, "请求远程1Panel失败, 响应数据：%s" % str(res)
            return res["data"].get("size", 0), ""
        except Exception as e:
            return None, "请求远程1Panel失败: %s" % str(e)

    def get_sshd_port(self) -> Optional[int]:
        try:
            data = self.lpc.get_sshd_config()
            return int(data["port"]) if data else None
        except:
            return None

    def create_dir(self, path: str) -> Tuple[Optional[Dict], str]:
        try:
            data = self.lpc.create_dir(path)
            if not data:
                return None, "创建目录失败, 1Panel节点连接失败"
            status = data["code"] == 200
            return {"status": status, "msg": data["message"] if not status else "创建成功"}, ""
        except Exception as e:
            return None, "请求远程1Panel节点失败： %s" % str(e)

    def restart_bt_panel(self) -> Dict[str, Any]:
        self.lpc.restart_panel()
        return {"status": True, "msg": "重启成功"}

    def server_reboot(self) -> Dict[str, Any]:
        self.lpc.server_reboot()

        from mod.project.node.dbutil import ServerMonitorRepo
        repo = ServerMonitorRepo()
        repo.set_wait_reboot(self.node_server_ip, True)

        def wait_for_reboot():
            # wait 等待服务器重启成功， 超时时间默认为 10 分钟
            wait_for = time.time() + 600
            time.sleep(3)
            while time.time() < wait_for:
                if self.test_conn() == "":  # 无错误时表示重启成功
                    repo.set_wait_reboot(self.node_server_ip, False)
                    from mod.project.node.dbutil import ServerNodeDB
                    node_data = ServerNodeDB().find_node(api_key=self.api_key, app_key=self.app_key.to_string())
                    if node_data:
                        monitor_node_once(node_data)
                    return {"status": True, "msg": "重启服务器成功"}
                time.sleep(3)
                public.print_log("等待服务器重启中... {}".format(wait_for - time.time()))
            repo.set_wait_reboot(self.node_server_ip, False)
            return {"status": False, "msg": "重启服务器失败, 已超过10分钟未能检测到服务器信息"}

        t = threading.Thread(target=wait_for_reboot, daemon=True)
        t.start()

        return {"status": True, "msg": "重启服务器已开始，请赖心等待重启成功"}

    def get_file_body(self, path: str) -> Tuple[Optional[str], str]:
        data_dict, err = self.lpc.get_file_body(path)
        if err != "":
            return None, err
        return data_dict["content"], ""

    def read_ssh_key(self) -> Tuple[Optional[str], str]:
        key, err = self.lpc.get_file_body("/root/.ssh/id_ed25519_1panel")
        if err:
            return None, "读取SSH密钥失败, 错误信息为: %s" % err
        return key, ""