AutoRpa/codes/rpa_yisaiqi/apa_browser_adapter.py

# coding=utf-8
"""
APA (艺赛旗) BrowserAdapter 实现
将 contract_flow_steps.BrowserAdapter 协议映射到 ubpa/iedge 浏览器 API
"""

from __future__ import annotations
import json
import time
import base64


def _make_apa_attr_map(css_selector: str) -> str:
    """将 CSS 选择器转换为 APA attrMap JSON 格式"""
    hierarchy = {
        "hasReachedRelativeAncestor": "false",
        "nodeHierarchyInfo": [
            {
                "isPresentInSelector": 1,
                "otherAttributes": {},
                "selectorInfo": {
                    "attributes": {
                        "css-selector": css_selector,
                        "tag": "",
                        "xpath": ""
                    },
                    "index": 0,
                    "tagName": ""
                }
            }
        ]
    }
    return json.dumps(hierarchy, ensure_ascii=False)


class ApaBrowserAdapter:
    """APA Studio Edge 浏览器适配器，实现 BrowserAdapter 协议"""

    def __init__(self, url: str = ""):
        from ubpa.iobject.browser import Browser
        self._browser = Browser(Browser.AppName.edge)
        self._url = url
        self._is_open = False

    def _ensure_open(self):
        if not self._is_open:
            from ubpa.iobject.browser import Browser
            self._browser.open(url=self._url, timeout=30)
            self._is_open = True

    # ---- BrowserAdapter 协议实现 ----

    def open_url(self, url: str) -> None:
        self._url = url
        self._ensure_open()

    def fill_first(self, selectors: list[str], value: str) -> None:
        self._ensure_open()
        import ubpa.iedge as iedge
        last_error = None
        for sel in selectors:
            try:
                attr_map = _make_apa_attr_map(sel)
                iedge.set_element_val(
                    attrMap=attr_map, title="*", url="*",
                    text=str(value), waitfor=5, program="msedge.exe"
                )
                time.sleep(0.3)
                return
            except Exception as e:
                last_error = e
                continue
        raise RuntimeError(
            f"fill_first failed for all selectors {selectors}: {last_error}"
        )

    def click_first(self, selectors: list[str]) -> None:
        self._ensure_open()
        import ubpa.iedge as iedge
        last_error = None
        for sel in selectors:
            try:
                attr_map = _make_apa_attr_map(sel)
                iedge.do_click_pos(
                    attrMap=attr_map, title="*", url="*",
                    waitfor=5, program="msedge.exe"
                )
                time.sleep(0.5)
                return
            except Exception as e:
                last_error = e
                continue
        raise RuntimeError(
            f"click_first failed for all selectors {selectors}: {last_error}"
        )

    def wait_for_text(self, text: str, timeout_seconds: int = 15) -> None:
        import ubpa.iedge as iedge
        elapsed = 0
        interval = 1
        while elapsed < timeout_seconds:
            try:
                # 尝试获取页面文本匹配
                result = iedge.get_element_val(
                    attrMap=_make_apa_attr_map(f"text={text}"),
                    title="*", url="*", waitfor=3, program="msedge.exe"
                )
                if result:
                    return
            except Exception:
                pass
            time.sleep(interval)
            elapsed += interval
        raise TimeoutError(f"wait_for_text '{text}' timed out after {timeout_seconds}s")

    def select_first(self, selectors: list[str], value: str) -> None:
        self._ensure_open()
        import ubpa.iedge as iedge
        last_error = None
        for sel in selectors:
            try:
                attr_map = _make_apa_attr_map(sel)
                iedge.set_element_selected_item(
                    attrMap=attr_map, title="*", url="*",
                    itemText=value, itemMode="text",
                    waitfor=5, program="msedge.exe"
                )
                time.sleep(0.3)
                return
            except Exception as e:
                last_error = e
                continue
        raise RuntimeError(
            f"select_first failed for all selectors {selectors}: {last_error}"
        )

    def upload_file_first(self, selectors: list[str], file_path: str) -> None:
        self._ensure_open()
        import ubpa.iedge as iedge
        last_error = None
        for sel in selectors:
            try:
                attr_map = _make_apa_attr_map(sel)
                iedge.set_element_val(
                    attrMap=attr_map, title="*", url="*",
                    text=file_path, waitfor=5, program="msedge.exe"
                )
                time.sleep(1)
                return
            except Exception as e:
                last_error = e
                continue
        raise RuntimeError(
            f"upload_file_first failed for all selectors {selectors}: {last_error}"
        )

    def read_first(self, selectors: list[str], value_name: str = "") -> str:
        self._ensure_open()
        import ubpa.iedge as iedge
        last_error = None
        for sel in selectors:
            try:
                attr_map = _make_apa_attr_map(sel)
                result = iedge.get_element_val(
                    attrMap=attr_map, title="*", url="*",
                    waitfor=5, program="msedge.exe"
                )
                if result:
                    return str(result)
            except Exception as e:
                last_error = e
                continue
        raise RuntimeError(
            f"read_first failed for all selectors {selectors}: {last_error}"
        )

    def screenshot(self, name: str) -> str:
        import os
        save_dir = os.path.join(os.path.dirname(__file__), "screenshots")
        os.makedirs(save_dir, exist_ok=True)
        path = os.path.join(save_dir, f"{name}.png")
        # APA browser screenshot capability - placeholder
        print(f"  [screenshot] {name} -> {path}")
        return path