完善插件功能

py-et/.vscode/settings.json

@@ -0,0 +1,8 @@
+{
+    "python.testing.pytestArgs": [
+        "tests"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true,
+    "python.testing.autoTestDiscoverOnSaveEnabled": true,
+}

py-et/README

@@ -0,0 +1 @@
+# WPS 表格自动化(插件版)

py-et/invoke-et.py

@@ -1,12 +0,0 @@
-from ET import ET
-
-if __name__ == "__main__":
-
-    et1 = ET()
-    sheets_name = et1.get_sheets()
-    print(f"et1: {sheets_name}")
-
-    et2 = ET()
-    et2.workbook_name = "工作簿2"
-    sheets_name = et2.get_sheets()
-    print(f"et2:{sheets_name}")

py-et/src/ET.py

@@ -0,0 +1,46 @@
+import os
+
+
+from .client import Client
+
+
+class ET(Client):
+    def __init__(self):
+        super().__init__()
+        self.workbook_name = ""  # 用于判断当前打开的表格
+        # self.check_addon_install()  # 调试模式暂时不判断
+        self.inject_flag = 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
+        res = self.invoke_func(func, args)
+        if res.get("code") != 0:
+            raise Exception(f"调用函数 {func} 失败, 错误信息: {res.get('msg')}")
+        return res.get("data")
+
+    def _inject_js_code(self) -> bool:
+        """注入js脚本"""
+        current_path = os.path.abspath(__file__)
+        js_code_path = os.path.join(os.path.dirname(current_path), "et.js")
+        if not os.path.exists(js_code_path):
+            raise Exception(f"js code not found: {js_code_path}")
+        # 读取js代码
+        with open(js_code_path, encoding="utf8") as f:
+            code = f.read()
+        func = "inject"
+        args = {"code": code}
+        res = self.invoke_func(func, args)
+        return res.get("code") == 0

py-et/src/client.py

@@ -1,22 +1,124 @@
-import base64
+import os
+import sys
 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
+import shutil
+import base64
+import zipfile
+import platform
+import webbrowser
+import logging
+
+logger = logging.getLogger(__file__)
+
+import http.client
+import xml.dom.minidom as minidom
+
+from xml.etree import ElementTree
 
 
 class Client:
     def __init__(self):
-        self.addon_name = "XbotWps"
+        self.addon_name = "xbot-et"
         self.addon_type = "et"
         self.addon_version = "1.0.0"
+        self.url = "http://127.0.0.1:3889/"
+        self.start_http_server()
+
+    def start_http_server(self) -> bool:
+        for _ in range(10):
+            if self._get_version():
+                print("Wps server started")
+                return True
+            self._start_http_server()
+            time.sleep(1)
+        raise ("Wps server not started")
+
+    def get_addon_list(self) -> list:
+        """获取插件列表"""
+        path = "/publishlist"
+        conn = http.client.HTTPConnection("127.0.0.1", 58890)
+        conn.request("GET", path, body="{}", headers={})
+        response = conn.getresponse()
+        data = response.read()
+        conn.close()
+        json_data = json.loads(data.decode("utf-8"))
+        return json_data
+
+    def verify_addon_status(self) -> bool:
+        """验证插件状态"""
+        url = self.url + "ribbon.xml"
+        data = {"method": "get", "url": url, "data": ""}
+        payload = base64.b64encode(json.dumps(data).encode("utf-8")).decode()
+        print(f"payload: {payload}")
+        conn = http.client.HTTPConnection("127.0.0.1", 58890)
+        conn.request("POST", "/redirect/runParams", payload, {})
+        response = conn.getresponse()
+        data = response.read()
+        conn.close()
+        try:
+            root = ElementTree.fromstring(data)
+            # 检查根标签名是否包含 customUI（包含命名空间）
+            return root.tag.endswith("customUI")
+        except ElementTree.ParseError:
+            return False
+
+    def enable_addon(self) -> bool:
+        """开启插件"""
+        data = {
+            "cmd": "enable",
+            "name": self.addon_name,
+            "url": self.url,
+            "addonType": "et",
+            "online": "true",
+            "time": int(time.time()),
+        }
+        path = "/deployaddons/runParams"
+        payload = base64.b64encode(json.dumps(data).encode("utf-8")).decode()
+        conn = http.client.HTTPConnection("127.0.0.1", 58890)
+        conn.request("POST", path, payload, {})
+        response = conn.getresponse()
+        data = response.read()
+        conn.close()
+        return data.decode() == "OK"
+
+    def disable_addon(self) -> bool:
+        """禁用插件"""
+        data = {
+            "cmd": "disable",
+            "name": self.addon_name,
+            "url": self.url,
+            "addonType": "et",
+            "online": "true",
+            "time": int(time.time()),
+        }
+        path = "/deployaddons/runParams"
+        payload = base64.b64encode(json.dumps(data).encode("utf-8")).decode()
+        conn = http.client.HTTPConnection("127.0.0.1", 58890)
+        conn.request("POST", path, payload, {})
+
+        response = conn.getresponse()
+        data = response.read()
+        conn.close()
+        return data.decode() == "OK"
+
+    def is_client_running(self) -> bool:
+        """检查客户端是否运行"""
+        # --data-raw '{"app":"et"}'
+        path = "/isRunning"
+        data = {
+            "app": self.addon_type,
+        }
+        payload = json.dumps(data)
+
+        conn = http.client.HTTPConnection("127.0.0.1", 58890)
+        conn.request("POST", path, payload, {})
+
+        response = conn.getresponse()
+        data = response.read()
+        conn.close()
+        return data.decode() == "Client is running."
 
     def invoke_func(self, func, args={}):
 
@@ -35,7 +137,7 @@ class Client:
             xhr.send(JSON.stringify({{id: '{req_uui}', response: res}}));void(0"
 
         data = {
-            "name": "xbot-et",
+            "name": self.addon_name,
             "function": "var res = InvokeFromXbot",
             "info": info,
             "showToFront": True,
@@ -95,26 +197,6 @@ class Client:
 
         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")
@@ -289,37 +371,3 @@ class Client:
         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

py-et/tests/test_client.py

@@ -0,0 +1,70 @@
+# test src/et.py client
+
+from src.ET import Client
+import pytest
+
+
+@pytest.fixture(scope="module")
+def test_create_client():
+    """创建测试用的client"""
+    client = Client()
+    yield client
+
+
+def test_client_get_addon_list(test_create_client):
+    """测试获取插件列表"""
+    client = test_create_client
+    print("Getting addon list...")
+    addon_list = client.get_addon_list()
+    print(f"Retrieved {len(addon_list)} addons")
+    assert isinstance(addon_list, list)
+    assert len(addon_list) > 0
+    assert all(isinstance(addon, dict) for addon in addon_list)
+    # 遍历所有插件, 并输出
+    for i, addon in enumerate(addon_list):
+        print(f"Addon {i}: {addon}")
+    print("First addon in list:", addon_list[0] if addon_list else "None")
+    print("Addon list test passed!")
+
+
+def test_client_verify_addon_status(test_create_client):
+    """测试插件状态"""
+    client = test_create_client
+    print("Verifying addon status...")
+    status = client.verify_addon_status()
+    print(f"Addon status: {status}")
+    assert status is True, "Addon status is not verified successfully"
+    print("Addon status test passed!")
+
+
+def test_client_enable_addon(test_create_client):
+    """测试启用插件"""
+    client = test_create_client
+    print("Enabling addon...")
+    status = client.enable_addon()
+    print(f"Addon enabled: {status}")
+    assert status is True, "Failed to enable addon"
+    print("Enable addon test passed!")
+
+
+def test_client_disable_addon(test_create_client):
+    """测试禁用插件"""
+    client = test_create_client
+    print("Disabling addon...")
+    status = client.disable_addon()
+    print(f"Addon disabled: {status}")
+    assert status is True, "Failed to disable addon"
+    print("Disable addon test passed!")
+
+def test_client_is_client_running(test_create_client):
+    """测试客户端是否运行"""
+    client = test_create_client
+    print("Checking if client is running...")
+    is_running = client.is_client_running()
+    print(f"Client is running: {is_running}")
+    assert isinstance(is_running, bool), "Client running status is not a boolean"
+    print("Client running status test passed!")
+
+if __name__ == "__main__":
+    # Add direct test execution for better debugging
+    pass

py-et/tests/test_et.py

@@ -0,0 +1,39 @@
+import pytest
+
+from src.ET import ET
+
+
+@pytest.fixture(scope="module")
+def test_create_client():
+    """创建测试用的client"""
+    client = ET()
+    yield client
+
+
+def test_et_check_inject():
+    """测试注入"""
+    et = ET()
+    print("Injecting...")
+    assert et.inject_flag is True, "Injection failed"
+    print("Inject test passed!")
+
+
+def test_et_get_sheets():
+    """测试获取工作表"""
+    et = ET()
+    print("Getting sheets...")
+    sheets_name = et.get_sheets()
+    print(f"Retrieved {len(sheets_name)} sheets")
+    print(f"Sheets_name: {sheets_name}")
+    assert isinstance(sheets_name, list)
+    assert len(sheets_name) > 0
+    assert all(isinstance(sheet, str) for sheet in sheets_name)
+    # 遍历所有工作表, 并输出
+    for i, sheet in enumerate(sheets_name):
+        print(f"Sheet {i}: {sheet}")
+    print("First sheet in list:", sheets_name[0] if sheets_name else "None")
+    print("Get sheets test passed!")
+
+
+if __name__ == "__main__":
+    pass