diff --git a/py-et/ET.py b/py-et/ET.py new file mode 100644 index 0000000..38304f1 --- /dev/null +++ b/py-et/ET.py @@ -0,0 +1,325 @@ +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 = ( + '\n\n\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 = ( + '\n\n\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 diff --git a/py-et/__pycache__/ET.cpython-313.pyc b/py-et/__pycache__/ET.cpython-313.pyc new file mode 100644 index 0000000..f9167e4 Binary files /dev/null and b/py-et/__pycache__/ET.cpython-313.pyc differ diff --git a/py-et/et.js b/py-et/et.js index b4fc9c9..3b1fa6b 100644 --- a/py-et/et.js +++ b/py-et/et.js @@ -2,11 +2,20 @@ const wpsInstance = { app: wps.EtApplication().Application, // 获取全部 sheet 页 - getSheets: () => { - const activeWorkbook = wpsInstance.app.ActiveWorkbook; - for (let i = 1; i <= activeWorkbook.Worksheets.Count; i++) { - alert(activeWorkbook.Worksheets.Item(i).Name) + getSheets: (args) => { + const workbook_name = args.workbook_name; + let workbook = null; + if (workbook_name) { + workbook = wpsInstance.app.Workbooks.Item(workbook_name); } + else { + workbook = wpsInstance.app.ActiveWorkbook; + } + const sheets = []; + for (let i = 1; i <= workbook.Worksheets.Count; i++) { + sheets.push(workbook.Worksheets.Item(i).Name); + } + return sheets; } } diff --git a/py-et/invoke-et.py b/py-et/invoke-et.py index 8032660..4952359 100644 --- a/py-et/invoke-et.py +++ b/py-et/invoke-et.py @@ -1,91 +1,12 @@ -import base64 -import uuid -import json -import http.client -import webbrowser - - -def start_http_server(): - """start debug""" - webbrowser.open("ksowpscloudsvr://start=RelayHttpServer") - - -def get_version(): - """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 invoke_wps_func(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"} - +from ET import ET if __name__ == "__main__": - start_http_server() - print(get_version()) - # print(get_version()) - path = r"C:\Users\32075\OneDrive\文档\Code\RPA_Script\wps-js-extension\XbotWps\py-et\et.js" - with open(path, "r", encoding="utf8") as f: - code = f.read() + et1 = ET() + sheets_name = et1.get_sheets() + print(f"et1: {sheets_name}") - func = "inject" - args = {"code": code} - res = invoke_wps_func(func, args) - print(res) - - func = "getSheets" - args = {"text": "123"} - res = invoke_wps_func(func, args) - print(res) + et2 = ET() + et2.workbook_name = "工作簿2" + sheets_name = et2.get_sheets() + print(f"et2:{sheets_name}") diff --git a/xbot-et/js/xbot.js b/xbot-et/js/xbot.js index 8dce0eb..1928235 100644 --- a/xbot-et/js/xbot.js +++ b/xbot-et/js/xbot.js @@ -1,7 +1,6 @@ let wpsInstance = null; - function InvokeFromXbot(param) { // 1. 解析参数, 如果不是字符串, 则转换为字符串, 并用json解析, 解析失败直接retrun if (typeof param !== 'string') { @@ -27,7 +26,7 @@ function InvokeFromXbot(param) { args = JSON.parse(param.args); } catch (e) { args = {}; - } + } console.log("method:", method); @@ -47,19 +46,19 @@ function InvokeFromXbot(param) { // 5. 判断是否有 wpsInstance, 如果没有直接return if (!wpsInstance) { - return {msg: '未注入', code: -3}; + return { msg: '未注入', code: -3 }; } // 6. 判断 wpsInstance 是否有 method 方法, 如果没有直接return if (!wpsInstance[method]) { - return {msg: '方法不存在', code: -4}; + return { msg: '方法不存在', code: -4 }; } // 7. 执行 wpsInstance 的 method 方法, 并返回结果 try { const result = wpsInstance[method](args ?? null); - return {msg: '执行成功', code: 0, data: result}; + return { msg: '执行成功', code: 0, data: result }; } catch (e) { - return {msg: '执行失败', code: -5}; + return { msg: '执行失败', code: -5 }; } }