XbotWps/py-et/ET.py

import base64
import uuid
import json
import http.client
import webbrowser
import os
import platform
import xml.dom.minidom as minidom
import sys
import shutil
import zipfile
import time


class Client:
    def __init__(self):
        self.addon_name = "XbotWps"
        self.addon_type = "et"
        self.addon_version = "1.0.0"

    def invoke_func(self, func, args={}):

        # 可能触发是否允许访问
        invoke_param = {
            "method": func,
            "args": json.dumps(args),
            "AppType": "et",
        }

        req_uui = str(uuid.uuid4())

        info = f"{json.dumps(invoke_param)});\
            var xhr = new XMLHttpRequest();\
            xhr.open('POST', 'http://127.0.0.1:58890/transferEcho/runParams');\
            xhr.send(JSON.stringify({{id: '{req_uui}', response: res}}));void(0"

        data = {
            "name": "xbot-et",
            "function": "var res = InvokeFromXbot",
            "info": info,
            "showToFront": True,
            "jsPluginsXml": "",
        }

        payload = {
            "id": req_uui,
            "app": "et",
            "data": "ksowebstartupet://"
            + base64.b64encode(json.dumps(data).encode()).decode(),
            "mode": False,
        }

        conn = http.client.HTTPConnection("127.0.0.1", 58890)
        payload = json.dumps(payload)

        conn.request("POST", "/transfer/runParams", payload)

        http_response = conn.getresponse()
        data = http_response.read()
        conn.close()

        try:
            return json.loads(data.decode("utf-8"))
        except Exception as e:
            print(data.decode("utf-8"))
            return {"message": "failed to parse response"}

    def install_addon(self):
        """安装插件"""
        addon_path = r"C:\Users\32075\OneDrive\文档\Code\RPA_Script\wps-js-extension\XbotWps\scripts\XbotWps_1.0.0.zip"
        install_dir = self._get_install_directory()
        print(f"Installation directory: {install_dir}")

        # Create the installation directory if it doesn't exist
        if not os.path.exists(install_dir):
            os.makedirs(install_dir)
            print(f"Created installation directory: {install_dir}")

        # Create the required XML files
        self._create_authwebsite_xml(install_dir, self.addon_name)
        self._create_publish_xml(
            install_dir,
            self.addon_name,
            verson=self.addon_version,
            addon_type=self.addon_type,
        )

        # Extract the addon to the installation directory
        if self._extract_addon(
            addon_path, install_dir, self.addon_name, self.addon_version
        ):
            print("Addon extracted successfully.")
        else:
            print("Failed to extract addon.")

        print("Installation completed successfully.")

    def check_addon_install(self):
        """通过文件夹是否存在判断插件是否安装"""
        install_dir = self._get_install_directory()
        addon_name_with_version = f"{self.addon_name}_{self.addon_version}"
        file_path = os.path.join(install_dir, addon_name_with_version)

        if os.path.exists(file_path):
            return True
        else:
            return False

    def check_http_status(self):
        for _ in range(10):
            if self._get_version():
                print("Wps server started")
                return
            self._start_http_server()
            time.sleep(1)
        raise ("Wps server not started")

    def _start_http_server(self):
        """start debug"""
        webbrowser.open("ksowpscloudsvr://start=RelayHttpServer")

    def _get_version(self):
        """check http status by get version"""
        try:
            conn = http.client.HTTPConnection("127.0.0.1", 58890)
            conn.request("POST", "/version", body="{}", headers={})
            response = conn.getresponse()
            response.read()
            conn.close()
            return response.status == 200
        except:
            return False

    def _get_install_directory(self):
        """Get the correct installation directory based on the platform."""
        system = platform.system()
        if system == "Windows":
            return os.path.join(
                os.environ.get("APPDATA"), "kingsoft", "wps", "jsaddons"
            )
        elif system == "Linux":
            return os.path.join(
                os.path.expanduser("~"),
                ".local",
                "share",
                "Kingsoft",
                "wps",
                "jsaddons",
            )
        else:
            print(f"Unsupported platform: {system}")
            sys.exit(1)

    def _create_authwebsite_xml(self, directory, node_name="xbotwps"):
        """Create the authwebsite.xml file if it doesn't exist or update it if it exists."""
        file_path = os.path.join(directory, "authwebsite.xml")

        # Create the basic XML structure if the file doesn't exist
        if not os.path.exists(file_path):
            # Create base XML structure
            content = (
                '<?xml version="1.0" encoding="UTF-8"?>\n<websites>\n</websites>\n'
            )
            with open(file_path, "w", encoding="utf-8") as f:
                f.write(content)
            print(f"Created basic authwebsite.xml structure at {file_path}")

        try:
            # Parse the XML file
            dom = minidom.parse(file_path)

            # Get the websites element
            websites_elements = dom.getElementsByTagName("websites")
            if not websites_elements:
                # If no websites element exists, create it
                websites = dom.createElement("websites")
                dom.appendChild(websites)
            else:
                websites = websites_elements[0]

            # Check if a website node with the specified name already exists
            existing_nodes = []
            for website in websites.getElementsByTagName("website"):
                if website.getAttribute("name") == node_name:
                    existing_nodes.append(website)

            # Remove existing nodes with the same name
            for node in existing_nodes:
                websites.removeChild(node)

            # Create and append the new website node
            new_website = dom.createElement("website")
            new_website.setAttribute("origin", "null")
            new_website.setAttribute("status", "enable")
            new_website.setAttribute("name", node_name)
            websites.appendChild(new_website)

            # Save the updated XML file
            with open(file_path, "w", encoding="utf-8") as f:
                f.write(dom.toprettyxml(indent="    "))

            print(f"Updated authwebsite.xml at {file_path}")

        except Exception as e:
            raise Exception(f"Error working with authwebsite.xml: {e}")

    def _create_publish_xml(
        delf, directory, node_name="xbotwps", verson="1.0.0", addon_type="wps"
    ):
        """Create the publish.xml file if it doesn't exist or update it if it exists."""
        file_path = os.path.join(directory, "publish.xml")

        # Create the basic XML structure if the file doesn't exist
        if not os.path.exists(file_path):
            # Create base XML structure
            content = (
                '<?xml version="1.0" encoding="UTF-8"?>\n<jsplugins>\n</jsplugins>\n'
            )
            with open(file_path, "w", encoding="utf-8") as f:
                f.write(content)
            print(f"Created basic publish.xml structure at {file_path}")

        try:
            # Parse the XML file
            dom = minidom.parse(file_path)

            # Get the jsplugins element
            jsplugins_elements = dom.getElementsByTagName("jsplugins")
            if not jsplugins_elements:
                # If no jsplugins element exists, create it
                jsplugins = dom.createElement("jsplugins")
                dom.appendChild(jsplugins)
            else:
                jsplugins = jsplugins_elements[0]

            # Check if a jsplugin node with the specified name already exists
            existing_nodes = []
            for jsplugin in jsplugins.getElementsByTagName("jsplugin"):
                if jsplugin.getAttribute("name") == node_name:
                    existing_nodes.append(jsplugin)

            # Remove existing nodes with the same name
            for node in existing_nodes:
                jsplugins.removeChild(node)

            # Create and append the new jsplugin node
            new_jsplugin = dom.createElement("jsplugin")
            new_jsplugin.setAttribute("enable", "enable_dev")
            new_jsplugin.setAttribute("url", "null")
            new_jsplugin.setAttribute("install", "null")
            new_jsplugin.setAttribute("version", verson)
            new_jsplugin.setAttribute("type", addon_type)
            new_jsplugin.setAttribute("customDomain", "")
            new_jsplugin.setAttribute("name", node_name)
            jsplugins.appendChild(new_jsplugin)

            # Save the updated XML file
            with open(file_path, "w", encoding="utf-8") as f:
                f.write(dom.toprettyxml(indent="    "))

            print(f"Updated publish.xml at {file_path}")

        except Exception as e:
            raise Exception(f"Error working with publish.xml: {e}")

    def _extract_addon(self, addon_path, install_dir, addon_name, addon_version):
        """Extract the addon archive to the installation directory."""
        addon_name_with_version = f"{addon_name}_{addon_version}"
        extract_dir = os.path.join(install_dir, addon_name_with_version)

        # Clean the existing directory if it exists
        if os.path.exists(extract_dir):
            print(f"Removing existing addon directory: {extract_dir}")
            shutil.rmtree(extract_dir)

        # Create the extraction directory
        os.makedirs(extract_dir)

        # Determine the archive type and extract accordingly
        file_ext = os.path.splitext(addon_path)[1].lower()

        try:
            print(f"Extracting ZIP archive to {extract_dir}")
            with zipfile.ZipFile(addon_path, "r") as zip_ref:
                zip_ref.extractall(extract_dir)
            print("Extraction completed successfully")

            return True
        except Exception as e:
            print(f"Error extracting archive: {e}")
            return False


class ET(Client):
    def __init__(self):
        super().__init__()
        self.workbook_name = ""  # 用于判断当前打开的表格
        self.check_http_status()
        # self.check_addon_install()  # 调试模式暂时不判断
        self._inject_js_code()

    def open(self, file_path):
        """open et"""
        pass

    def get_sheets(self) -> list:
        """get all sheets"""
        func = "getSheets"
        sheets_name = self._wrap_invoke_func(func)
        return sheets_name

    def _wrap_invoke_func(self, func, args={}):
        """将工作簿的名称自动带进去"""
        if self.workbook_name:
            args["workbook_name"] = self.workbook_name
        return self.invoke_func(func, args)

    def _inject_js_code(self):
        path = r"C:\Users\32075\OneDrive\文档\Code\RPA_Script\wps-js-extension\XbotWps\py-et\et.js"
        with open(path, encoding="utf8") as f:
            code = f.read()
        func = "inject"
        args = {"code": code}
        res = self.invoke_func(func, args)
        return res