diff --git a/examples/ziniao-login.py b/examples/ziniao-login.py index 880cf3b..1606855 100644 --- a/examples/ziniao-login.py +++ b/examples/ziniao-login.py @@ -10,4 +10,7 @@ from ziniao import Ziniao if __name__ == "__main__": ziniao = Ziniao() - ziniao.launch_ziniao() + # ziniao.launch_ziniao() + username = "17756164521" + password = "320753691Love" + ziniao.login_by_personal(username, password) diff --git a/src/__pycache__/ziniao.cpython-313.pyc b/src/__pycache__/ziniao.cpython-313.pyc index dee8144..5ef7eeb 100644 Binary files a/src/__pycache__/ziniao.cpython-313.pyc and b/src/__pycache__/ziniao.cpython-313.pyc differ diff --git a/src/xpath.py b/src/xpath.py new file mode 100644 index 0000000..e572268 --- /dev/null +++ b/src/xpath.py @@ -0,0 +1,350 @@ +import enum +import uiautomation as ui +from typing import Optional, Any, Dict, List + + +class ElementAttributes(enum.Enum): + """映射用于XPath导航的UI元素属性""" + + AutomationId = 0 + Name = 1 + ClassName = 2 + HelpText = 3 + IsPassword = 4 + FullDescription = 5 + ItemType = 6 + AcceleratorKey = 7 + AccessKey = 8 + IsEnabled = 9 + IsOffscreen = 10 + ProcessId = 11 + + +class AutomationElementXPathNavigator: + """ + 基于UIAutomation的XPath导航器,允许通过XPath语法定位UI元素 + """ + + NO_ATTRIBUTE_VALUE = -1 + + def __init__(self, root_element): + """ + 创建以指定元素为根的XPath导航器 + + Args: + root_element: 根UI元素 + """ + self._root_element = root_element + self._current_element = root_element + self._attribute_index = self.NO_ATTRIBUTE_VALUE + + @property + def is_in_attribute(self) -> bool: + """当前是否处于属性节点""" + return self._attribute_index != self.NO_ATTRIBUTE_VALUE + + @property + def has_attributes(self) -> bool: + """当前节点是否有属性""" + return not self.is_in_attribute + + @property + def value(self) -> str: + """获取当前节点的值""" + if self.is_in_attribute: + return self.get_attribute_value(self._attribute_index) + return str(self._current_element) + + @property + def underlying_object(self): + """返回当前表示的UI元素对象""" + return self._current_element + + @property + def node_type(self) -> str: + """获取当前节点类型""" + if self.is_in_attribute: + return "Attribute" + if self._current_element == self._root_element: + return "Root" + return "Element" + + @property + def local_name(self) -> str: + """获取当前节点的本地名称""" + if self.is_in_attribute: + return self.get_attribute_name(self._attribute_index) + # 获取控件类型 + control_type = self._current_element.ControlTypeName + return control_type if control_type else "Custom" + + @property + def name(self) -> str: + """获取节点名称""" + return self.local_name + + def clone(self): + """创建当前导航器的副本""" + cloned = AutomationElementXPathNavigator(self._root_element) + cloned._current_element = self._current_element + cloned._attribute_index = self._attribute_index + return cloned + + def move_to_first_attribute(self) -> bool: + """移动到第一个属性""" + if self.is_in_attribute: + return False + self._attribute_index = 0 + return True + + def move_to_next_attribute(self) -> bool: + """移动到下一个属性""" + if not self.is_in_attribute: + return False + if self._attribute_index >= len(ElementAttributes) - 1: + return False + self._attribute_index += 1 + return True + + def get_attribute(self, local_name: str, namespace_uri: str = "") -> str: + """获取指定属性的值""" + if self.is_in_attribute: + return "" + attribute_index = self.get_attribute_index_from_name(local_name) + if attribute_index != self.NO_ATTRIBUTE_VALUE: + return self.get_attribute_value(attribute_index) + return "" + + def move_to_attribute(self, local_name: str, namespace_uri: str = "") -> bool: + """移动到指定属性""" + if self.is_in_attribute: + return False + attribute_index = self.get_attribute_index_from_name(local_name) + if attribute_index != self.NO_ATTRIBUTE_VALUE: + self._attribute_index = attribute_index + return True + return False + + def move_to_root(self): + """移动到根元素""" + self._attribute_index = self.NO_ATTRIBUTE_VALUE + self._current_element = self._root_element + + def move_to_next(self) -> bool: + """移动到下一个同级元素""" + if self.is_in_attribute: + return False + next_element = self._current_element.GetNextSiblingElement() + if next_element is None: + return False + self._current_element = next_element + return True + + def move_to_previous(self) -> bool: + """移动到前一个同级元素""" + if self.is_in_attribute: + return False + prev_element = self._current_element.GetPreviousSiblingElement() + if prev_element is None: + return False + self._current_element = prev_element + return True + + def move_to_first_child(self) -> bool: + """移动到第一个子元素""" + if self.is_in_attribute: + return False + child_element = self._current_element.GetFirstChildElement() + if child_element is None: + return False + self._current_element = child_element + return True + + def move_to_parent(self) -> bool: + """移动到父元素""" + if self.is_in_attribute: + self._attribute_index = self.NO_ATTRIBUTE_VALUE + return True + if self._current_element == self._root_element: + return False + parent = self._current_element.GetParentElement() + if parent is None: + return False + self._current_element = parent + return True + + def move_to(self, other) -> bool: + """移动到另一个导航器的位置""" + if not isinstance(other, AutomationElementXPathNavigator): + return False + if self._root_element != other._root_element: + return False + self._current_element = other._current_element + self._attribute_index = other._attribute_index + return True + + def is_same_position(self, other) -> bool: + """检查是否与另一个导航器处于相同位置""" + if not isinstance(other, AutomationElementXPathNavigator): + return False + if self._root_element != other._root_element: + return False + return ( + self._current_element == other._current_element + and self._attribute_index == other._attribute_index + ) + + def get_attribute_value(self, attribute_index: int) -> str: + """获取属性值""" + try: + attr = ElementAttributes(attribute_index) + if attr == ElementAttributes.AutomationId: + return self._current_element.AutomationId or "" + elif attr == ElementAttributes.Name: + return self._current_element.Name or "" + elif attr == ElementAttributes.ClassName: + return self._current_element.ClassName or "" + elif attr == ElementAttributes.HelpText: + return self._current_element.HelpText or "" + elif attr == ElementAttributes.IsPassword: + return str(self._current_element.IsPassword).lower() + elif attr == ElementAttributes.FullDescription: + return getattr(self._current_element, "FullDescription", "") or "" + elif attr == ElementAttributes.ItemType: + return getattr(self._current_element, "ItemType", "") or "" + elif attr == ElementAttributes.AcceleratorKey: + return getattr(self._current_element, "AcceleratorKey", "") or "" + elif attr == ElementAttributes.AccessKey: + return getattr(self._current_element, "AccessKey", "") or "" + elif attr == ElementAttributes.IsEnabled: + return str(self._current_element.IsEnabled).lower() + elif attr == ElementAttributes.IsOffscreen: + return str(self._current_element.IsOffscreen).lower() + elif attr == ElementAttributes.ProcessId: + return ( + str(self._current_element.ProcessId) + if self._current_element.ProcessId + else "" + ) + else: + return "" + except Exception: + return "" + + def get_attribute_name(self, attribute_index: int) -> str: + """获取属性名""" + try: + return ElementAttributes(attribute_index).name + except ValueError: + return "" + + def get_attribute_index_from_name(self, attribute_name: str) -> int: + """从属性名获取属性索引""" + try: + return ElementAttributes[attribute_name].value + except (KeyError, ValueError): + return self.NO_ATTRIBUTE_VALUE + + +class XPathSelector: + """使用XPath表达式选择UI元素""" + + def __init__(self, root_element=None): + """ + 初始化XPath选择器 + + Args: + root_element: 根元素,默认为桌面 + """ + self.root_element = root_element or ui.GetRootControl() + self.navigator = AutomationElementXPathNavigator(self.root_element) + + def find_element(self, xpath_expr: str): + """ + 使用XPath表达式查找单个元素 + + Args: + xpath_expr: XPath表达式 + + Returns: + 找到的元素或None + """ + # 基础实现 - 在实际使用时需要扩展为真正的XPath解析和执行 + # 这里仅作为示例,实际应该使用XPath库解析表达式并遍历元素树 + return self._basic_parse_xpath(xpath_expr) + + def find_elements(self, xpath_expr: str) -> List: + """ + 使用XPath表达式查找所有匹配元素 + + Args: + xpath_expr: XPath表达式 + + Returns: + 匹配元素列表 + """ + # 完整实现应该返回所有匹配元素 + result = self.find_element(xpath_expr) + return [result] if result else [] + + def _basic_parse_xpath(self, xpath_expr: str): + """ + 基础XPath解析实现(简化版) + + Args: + xpath_expr: XPath表达式 + + Returns: + 匹配的元素或None + """ + # 这是一个极其简化的XPath解析示例 + # 仅支持类似 "//Button[@Name='确定']" 的简单表达式 + + parts = xpath_expr.strip().split("[@") + if len(parts) != 2: + return None + + control_type = parts[0].replace("//", "") + attr_part = parts[1].rstrip("]") + + if "=" not in attr_part: + return None + + attr_name, attr_value = attr_part.split("=", 1) + attr_value = attr_value.strip("'\"") + + # 简单查找实现 + def find_matching_element(element): + if element.ControlTypeName == control_type: + if attr_name == "Name" and element.Name == attr_value: + return element + elif attr_name == "AutomationId" and element.AutomationId == attr_value: + return element + + # 递归搜索子元素 + for child in element.GetChildren(): + result = find_matching_element(child) + if result: + return result + return None + + return find_matching_element(self.root_element) + + +# 使用示例 +def find_element_by_xpath(xpath_expr: str, root=None): + """使用XPath表达式查找元素的便捷函数""" + selector = XPathSelector(root) + return selector.find_element(xpath_expr) + + +if __name__ == "__main__": + # 示例:查找名为"确定"的按钮 + ui.GetRootControl().PaneControl(searchDepth=1, Name="紫鸟浏览器").SetActive() + xpath = "//PaneControl[@Name='紫鸟浏览器']" + element = find_element_by_xpath(xpath) + element.Click() + if element: + print(f"找到元素: {element}") + else: + print("未找到元素") diff --git a/src/ziniao.py b/src/ziniao.py index a5e9729..83d84fd 100644 --- a/src/ziniao.py +++ b/src/ziniao.py @@ -3,14 +3,62 @@ import time import winreg import win32gui + import subprocess +import uiautomation as auto + +auto.uiautomation.SetGlobalSearchTimeout(5) class Ziniao: def __init__(self): + pass - def login(self, username, password): + def login_by_personal(self, username, password): + """使用个人账号登录""" + auto.GetRootControl().PaneControl(searchDepth=1, Name="紫鸟浏览器").SetActive() + auto.GetRootControl().PaneControl( + searchDepth=1, Name="紫鸟浏览器" + ).DocumentControl(searchDepth=1, Name="紫鸟浏览器").TextControl( + searchDepth=8, Name="个人密码登录" + ).Click() + auto.GetRootControl().PaneControl( + searchDepth=1, Name="紫鸟浏览器" + ).DocumentControl(searchDepth=1, Name="紫鸟浏览器").EditControl( + searchDepth=20, LocalizedControlType="编辑" + ).Click() + auto.SendKeys(username) + + if auto.GetRootControl().PaneControl( + searchDepth=1, Name="紫鸟浏览器" + ).DocumentControl(searchDepth=1, Name="紫鸟浏览器").TextControl( + searchDepth=20, Name="个人密码" + ).Exists(): + auto.GetRootControl().PaneControl( + searchDepth=1, Name="紫鸟浏览器" + ).DocumentControl(searchDepth=1, Name="紫鸟浏览器").TextControl( + searchDepth=20, Name="个人密码" + ).GetParentControl().GetNextSiblingControl().Click() + else: + auto.GetRootControl().PaneControl( + searchDepth=1, Name="紫鸟浏览器" + ).DocumentControl(searchDepth=1, Name="紫鸟浏览器").TextControl( + searchDepth=20, Name="请输入个人密码" + ).GetParentControl().Click() + + auto.SendKeys(password) + + auto.GetRootControl().PaneControl( + searchDepth=1, Name="紫鸟浏览器" + ).DocumentControl(searchDepth=1, Name="紫鸟浏览器").ButtonControl( + searchDepth=20, Name="登 录" + ).Click() + + pass + + def login_by_company(self): + """使用企业账号登录""" pass def open_store(self): @@ -27,43 +75,42 @@ class Ziniao: exe_path = self.get_ziniao_exe_path() # 启动紫鸟的exe文件, 独立进程 subprocess.Popen( - exe_path, creationflags=0x01000000 | 0x00000008, close_fds=True + exe_path, + creationflags=0x01000000 | 0x00000008, + close_fds=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, ) - # 延迟导入win32gui以避免在测试收集阶段崩溃 - try: + def wait_for_foreground_window(title_keyword, timeout=60): + """等待窗口成为前台窗口""" - # 等待窗口出现 - max_wait_time = 60 # 最大等待时间(秒) - start_time = time.time() - - print("等待紫鸟窗口出现...") - - # 延迟导入 - - # 使用win32gui查找窗口 def enum_windows_callback(hwnd, found_windows): if win32gui.IsWindowVisible(hwnd): - window_title = win32gui.GetWindowText(hwnd) - if "紫鸟" in window_title: + title = win32gui.GetWindowText(hwnd) + if title_keyword in title: found_windows.append(hwnd) return True - while time.time() - start_time < max_wait_time: + start_time = time.time() + while time.time() - start_time < timeout: found_windows = [] win32gui.EnumWindows(enum_windows_callback, found_windows) + if found_windows: - print("紫鸟窗口已出现") - return - time.sleep(1) # 每秒检查一次 + for hwnd in found_windows: + fg_hwnd = win32gui.GetForegroundWindow() + if hwnd == fg_hwnd: + print(f"窗口已成为前台:{win32gui.GetWindowText(hwnd)}") + return True + else: + print("等待紫鸟窗口出现") - if time.time() - start_time >= max_wait_time: - print("紫鸟窗口未在规定时间内出现") + time.sleep(1) + return False - except ImportError: - print("警告: 未安装win32gui模块,无法检测紫鸟窗口出现") - except Exception as e: - print(f"检测窗口出错: {e}") + if not wait_for_foreground_window("紫鸟"): + raise RuntimeError("紫鸟窗口未能在规定时间内出现") def get_ziniao_exe_path(self): """返回紫鸟的安装路径""" diff --git a/src/ziniaoapi.py b/src/ziniaoapi.py new file mode 100644 index 0000000..6fd8128 --- /dev/null +++ b/src/ziniaoapi.py @@ -0,0 +1,142 @@ +from ziniao_oapi.common.OpenClient import OpenClient +from ziniao_oapi.request.AppTokenRequest import ( + AppTokenRequest, + BaseRequest, + RequestTypes, +) +import webbrowser + +## TODO: token过期处理 + + +class ZiniaoAPI: + def __init__(self, app_id, private_key): + self.app_id = app_id # 应用id + self.private_key = private_key # 应用私钥 + self.url = "https://sbappstoreapi.ziniao.com" # 请求URL + self.client = OpenClient(self.appId, self.private_key, self.url) + self.token = self.test_get_app_token() + self.company_id = self.get_company_id() + + def test_get_app_token(self): + """获取token""" + request = AppTokenRequest() + response = self.client.execute(request) + + if response.is_success(): + token = response.data.get("appAuthToken") + return token + else: + print( + "请求失败,code:%s, msg:%s, sub_code:%s, sub_msg:%s" + % (response.code, response.msg, response.sub_code, response.sub_msg) + ) + raise Exception("获取token失败") + + def get_company_id(self): + """获取公司id""" + + class CompanyIdRequest(BaseRequest): + def __init__(self): + BaseRequest.__init__(self) + + def get_method(self): + return "/app/builtin/company" + + def get_request_type(self): + return RequestTypes.GET + + # get company id + request = CompanyIdRequest() + response = self.client.execute(request, app_token=self.token) + if response.is_success(): + company_id = response.data.get("companyId") + return company_id + else: + print("失败, response: ", response) + raise Exception("获取公司id失败") + + def get_user_id(self): + class UserIdRequest(BaseRequest): + def __init__(self): + BaseRequest.__init__(self) + + def get_method(self): + return "/superbrowser/rest/v1/erp/staff/list" + + def get_request_type(self): + return RequestTypes.POST_JSON + + request = UserIdRequest() + # 按接口文档设置业务参数 + + for i in range(1, 6): + request.biz_model = { + "companyId": self.company_id, + "level": "", + "isAccurate": "", + "limit": "100", + "page": i, + "departmentIds": [], + "delflag": "", + "name ": "", + "username": "", + } + + # 发送请求 + response = self.client.execute(request, app_token=self.token) + + if response.is_success(): + + for user_info in response.data.get("data"): + # print(user_info.get("username")) + if user_info.get("username") == "Shenglei": + print(user_info) + else: + print("失败, response: ", response) + + def get_store_id_by_userid(self, user_id): + + class StoreRequest(BaseRequest): + def __init__(self): + BaseRequest.__init__(self) + + def get_method(self): + return "/superbrowser/rest/v1/erp/store/list/by/user" + + def get_request_type(self): + return RequestTypes.POST_JSON + + request = StoreRequest() + + request.biz_model = { + "companyId": self.company_id, + "limit": "30", + "storeName": "", + "page": "1", + "userId": user_id, + } + + # 发送请求 + response = self.client.execute(request, app_token=self.token) + + if response.is_success(): + print("成功, response: ", response) + for store in response.data.get("data"): + print(f"{store.get('storeId')} --> {store.get('storeName')}") + else: + print("失败, response: ", response) + raise Exception("获取店铺id失败") + + def open_store(self, user_id, store_id): + """打开店铺""" + url = f"superbrowser://OpenStrore?storeId={store_id}&openapiToken={self.token}&userId={user_id}&lanuchUrl=aHR0cHM6Ly93d3cuYmFpZHUuY29t&autoop en=true&debuggPort=60001" + webbrowser.open(url) + + def close_store(self, store_id): + """关闭店铺""" + webbrowser.open(f"superbrowser://CloseStrore?storeId={store_id}") + + def exit_ziniao(self): + """退出紫鸟""" + webbrowser.open("superbrowser://exit")