diff --git a/.gitignore b/.gitignore index e560fd3..2e424ac 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,12 @@ state_center_web/dist/ # Local runtime / agent .runtime/ +rpa_yisaiqi/screenshots/ +rpa-chrome-profile/ +**/chrome-profile/ +.tmp/ +.grok/ +.grokignore .claude/ .superpowers/ diff --git a/rpa_yisaiqi/README.md b/rpa_yisaiqi/README.md index f858cc4..b13d101 100644 --- a/rpa_yisaiqi/README.md +++ b/rpa_yisaiqi/README.md @@ -57,10 +57,20 @@ python -m rpa_yisaiqi.contract_create_node ` ## 真实合同起草运行 -先准备合同起草输入 JSON,参考: +客商必须用合同测试系统里**真实存在**的操作码。`CUST-YYC-001` 这类占位符会在选客商时搜不到,入参校验会直接拒绝。 -- `examples/contract_draft_yyc.json` -- `examples/contract_draft_xyy.json` +测试环境当前可用: + +| 字段 | 值 | +| --- | --- | +| customerCode | `DW0100046361` | +| customerName | `重庆亿昊药业有限公司` | + +联调入参用: + +- `examples/contract_create_input_dry_run.json`:真客商 + `dry_run`,先校准页面 +- `examples/contract_draft_yyc.json`:医药城,同上客商 +- `examples/contract_draft_xyy.json`:小药药分支(不上传 PDF),客商仍用合同系统里能搜到的同一条 YYC 有多份平台 PDF 时需要合并,先安装: @@ -75,7 +85,7 @@ python -m rpa_yisaiqi.contract_create_node ` --state-center-url http://192.168.x.x:8000 ` --token dev-agent-token ` --job-id <状态中心job_id> ` - --supplier-name 重庆亿昊药业 ` + --supplier-name 重庆亿昊药业有限公司 ` --order-date 2026-08-11 ` --source-type YYC ` --draft-path rpa_yisaiqi/examples/contract_draft_yyc.json ` @@ -111,6 +121,27 @@ python -m rpa_yisaiqi.contract_create_node ` --state-center-url http://192.168.x.x:8000 ``` +## 艺赛旗画布怎么看 + +Studio 里「插入代码块」标题改不了,会一直显示「插入代码块#1」。步骤名写在节点**备注**里,画布上出现在代码块下方,例如 `S06 填写签约方`。 + +流程按四段换行,不要再排成一条长龙: + +``` +准备 开始 → S01 初始化 → S02 打开系统 → S03 登录 +进入起草 S04 进入起草 → S05 选择模板 +填写表单 S06 签约方 → S07 基础信息 → S08 金额日期 → S09 货品 → S10 付款 +提交回写 S11 上传PDF → S12 提交 → S13 读合同号 → S14 回调 +``` + +改步骤后先关掉 Studio,再生成画布: + +```powershell +python rpa_yisaiqi/build_multinode_seq.py +``` + +然后重新打开 `rpa_yisaiqi/studio` 项目并编译。 + ## 接入真实页面自动化 真实艺赛旗页面操作按流程图拆在: diff --git a/rpa_yisaiqi/_flow_state.py b/rpa_yisaiqi/_flow_state.py new file mode 100644 index 0000000..41daedf --- /dev/null +++ b/rpa_yisaiqi/_flow_state.py @@ -0,0 +1,17 @@ +# coding=utf-8 +"""多节点流程图(Main.seq)的跨节点共享状态容器。 + +艺赛旗编译 Main.seq 后,各代码节点的 PBCode 会在同一 Python 进程中执行。 +本模块以「模块级全局变量」承载 browser / ctx / client / spec 等对象, +避免依赖艺赛旗对局部变量的拼接方式,让每个节点都能可靠地取到上一步的产物。 +""" + +spec = None +client = None +ctx = None +browser = None +submitted = False +contract_no = "" +fd_id = "" +is_dry_run = False +completed = False diff --git a/rpa_yisaiqi/apa_browser_adapter.py b/rpa_yisaiqi/apa_browser_adapter.py new file mode 100644 index 0000000..e2df2ba --- /dev/null +++ b/rpa_yisaiqi/apa_browser_adapter.py @@ -0,0 +1,505 @@ +# coding=utf-8 +""" +APA (艺赛旗) Selenium Chrome BrowserAdapter 实现 +通过 Selenium WebDriver 直接操控 Chrome,不依赖 APA 浏览器模块 +""" +from __future__ import annotations + +import os +import time +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +def _is_inside_repo(path: Path, repo_root: Path) -> bool: + try: + path.resolve().relative_to(repo_root.resolve()) + except ValueError: + return False + return True + + +def resolve_chrome_profile_base(requested: str | None = None, *, repo_root: Path | None = None) -> Path: + """Chrome user-data dir must stay outside the repo so list_dir never walks 3GB.""" + root = (repo_root or REPO_ROOT).resolve() + outside = (root.parent / "rpa-chrome-profile").resolve() + raw = os.getenv("RPA_CHROME_PROFILE") if requested is None else requested + if raw: + candidate = Path(raw).expanduser().resolve() + if _is_inside_repo(candidate, root): + print(f"RPA_CHROME_PROFILE 指向仓库内 {candidate},已改用 {outside}", flush=True) + return outside + return candidate + return outside + + +class ApaBrowserAdapter: + """Selenium Chrome 浏览器适配器,实现 BrowserAdapter 协议""" + + def __init__(self, url: str = "", *, flow: str = "default"): + self._url = url + self._flow = flow + self._driver = None + self._is_open = False + + def _get_driver(self): + """创建或获取 Chrome WebDriver""" + if self._driver is not None: + return self._driver + + from selenium import webdriver + from selenium.webdriver.chrome.service import Service + from selenium.webdriver.chrome.options import Options + + profile_base = resolve_chrome_profile_base() + if _is_inside_repo(profile_base, REPO_ROOT): + raise RuntimeError(f"拒绝在仓库内创建 Chrome 用户目录: {profile_base}") + # 每次进程单独目录:上次 detach 留下的 Chrome 会锁住同一 profile + profile_dir = profile_base / str(os.getpid()) + os.makedirs(profile_dir, exist_ok=True) + profile_dir = str(profile_dir) + + options = Options() + options.page_load_strategy = "eager" + # 不自动关闭浏览器,方便调试 + options.add_experimental_option("detach", True) + # 避免被检测为自动化 + options.add_experimental_option("excludeSwitches", ["enable-automation", "enable-logging"]) + options.add_experimental_option("useAutomationExtension", False) + options.add_argument("--disable-blink-features=AutomationControlled") + options.add_argument("--no-first-run") + options.add_argument("--no-default-browser-check") + options.add_argument("--disable-gpu") + options.add_argument("--disable-extensions") + options.add_argument("--disable-popup-blocking") + options.add_argument("--start-maximized") + # 独立目录,避免和日常 Chrome 抢用户数据目录导致卡住 + options.add_argument("--user-data-dir=" + profile_dir) + + chromedriver = os.getenv("CHROMEDRIVER_PATH") + if chromedriver and os.path.exists(chromedriver): + service = Service(chromedriver) + else: + from webdriver_manager.chrome import ChromeDriverManager + service = Service(ChromeDriverManager().install()) + self._driver = webdriver.Chrome(service=service, options=options) + self._driver.set_page_load_timeout(45) + self._driver.set_script_timeout(30) + return self._driver + + def _ensure_open(self): + if self._driver is None: + driver = self._get_driver() + if self._url: + driver.get(self._url) + self._is_open = True + + # ---- 内部查找方法 ---- + + def _find(self, selector: str, timeout: int = 5): + """ + 智能查找元素:支持 CSS、XPath、text=、label=、aria=、title=、:contains() + """ + from selenium.webdriver.common.by import By + from selenium.common.exceptions import NoSuchElementException + import re + + # 1. text=xxx → 查找包含对应文本的元素 + if selector.startswith("text="): + text_val = selector[5:].strip('"').strip("'") + xpath = f'//*[contains(text(),"{text_val}")]' + return self._driver.find_element(By.XPATH, xpath) + + # 2. label=xxx → 找到对应标签的输入框 + if selector.startswith("label="): + text_val = selector[6:].strip('"').strip("'") + xpath = f'//label[contains(text(),"{text_val}")]/following-sibling::*[1][self::input or self::select or self::textarea]' + try: + return self._driver.find_element(By.XPATH, xpath) + except NoSuchElementException: + # Fallback: find the label's "for" attribute + try: + for_id = self._driver.find_element( + By.XPATH, f'//label[contains(text(),"{text_val}")]' + ).get_attribute("for") + if for_id: + return self._driver.find_element(By.ID, for_id) + except NoSuchElementException: + pass + raise + + # 3. aria=xxx → aria-label 属性 + if selector.startswith("aria="): + text_val = selector[5:].strip('"').strip("'") + return self._driver.find_element(By.CSS_SELECTOR, f'[aria-label="{text_val}"]') + + # 4. title=xxx → title 属性 + if selector.startswith("title="): + text_val = selector[6:].strip('"').strip("'") + return self._driver.find_element(By.CSS_SELECTOR, f'[title="{text_val}"]') + + # 5. :contains() → 转 XPath(支持 button:contains(文本) / a:contains("文本")) + if ":contains(" in selector: + m = re.match(r'(.+?):contains\(["\']?(.+?)["\']?\)', selector) + if m: + tag_part, text_val = m.groups() + xpath = f'//{tag_part}[contains(.,"{text_val}")]' + return self._driver.find_element(By.XPATH, xpath) + + # 6. 显式 XPath(以 / 或 // 开头)优先按 XPath 查找 + if selector.startswith("//") or selector.startswith("/"): + return self._driver.find_element(By.XPATH, selector) + + # 7. 标准 CSS selector + return self._driver.find_element(By.CSS_SELECTOR, selector) + + # ---- BrowserAdapter 协议实现 ---- + + def open_url(self, url: str) -> None: + self._url = url + driver = self._get_driver() + try: + driver.get(url) + except Exception as exc: + # eager 仍可能因长轮询资源超时;页面文档已出来就继续 + current = "" + try: + current = driver.current_url or "" + except Exception: + current = "" + if not current or current in ("data:,", "about:blank"): + raise RuntimeError(f"打开页面失败:{url} ({exc})") from exc + self._is_open = True + + def _set_value_via_js(self, el, value: str) -> None: + """通过原生 value setter + 事件注入设置值,兼容 LUI/React 受控组件""" + self._driver.execute_script( + """ + var el = arguments[0]; + var value = arguments[1]; + var proto = el.tagName === 'TEXTAREA' + ? window.HTMLTextAreaElement.prototype + : window.HTMLInputElement.prototype; + var setter = Object.getOwnPropertyDescriptor(proto, 'value').set; + setter.call(el, value); + el.dispatchEvent(new Event('input', { bubbles: true })); + el.dispatchEvent(new Event('change', { bubbles: true })); + el.dispatchEvent(new Event('blur', { bubbles: true })); + """, + el, + str(value), + ) + + def fill_first(self, selectors: list, value: str) -> None: + self._ensure_open() + last_error = None + for sel in selectors: + try: + el = self._find(sel) + # 先尝试原生输入;LUI 受控组件用 JS 注入兜底 + try: + el.clear() + el.send_keys(str(value)) + except Exception: + pass + if el.get_attribute("value") != str(value): + self._set_value_via_js(el, str(value)) + time.sleep(0.3) + return + except Exception as e: + last_error = e + continue + raise RuntimeError(f"fill_first failed: {selectors}, last: {last_error}") + + def click_native(self, selectors: list) -> None: + """真实鼠标点击,给只监听 mousedown 的 LUI 页签/工具栏用。""" + self._ensure_open() + from selenium.webdriver.common.action_chains import ActionChains + last_error = None + for sel in selectors: + try: + el = self._find(sel) + self._driver.execute_script( + "arguments[0].scrollIntoView({block:'center', inline:'center'});", + el, + ) + time.sleep(0.2) + ActionChains(self._driver).move_to_element(el).pause(0.15).click().perform() + time.sleep(0.4) + return + except Exception as e: + last_error = e + continue + raise RuntimeError(f"click_native failed: {selectors}, last: {last_error}") + + def click_first(self, selectors: list) -> None: + self._ensure_open() + last_error = None + for sel in selectors: + try: + el = self._find(sel) + # LUI/委托事件用 JS 点击最可靠 + self._driver.execute_script("arguments[0].click()", el) + time.sleep(0.5) + return + except Exception as e: + last_error = e + continue + raise RuntimeError(f"click_first failed: {selectors}, last: {last_error}") + + def double_click_first(self, selectors: list) -> None: + self._ensure_open() + from selenium.webdriver.common.action_chains import ActionChains + last_error = None + for sel in selectors: + try: + el = self._find(sel) + ActionChains(self._driver).double_click(el).perform() + time.sleep(0.5) + return + except Exception as e: + last_error = e + continue + raise RuntimeError(f"double_click_first failed: {selectors}, last: {last_error}") + + def wait_for_text(self, text: str, timeout_seconds: int = 15) -> None: + self._ensure_open() + from selenium.webdriver.support.ui import WebDriverWait + from selenium.webdriver.support import expected_conditions as EC + from selenium.webdriver.common.by import By + try: + WebDriverWait(self._driver, timeout_seconds).until( + EC.text_to_be_present_in_element((By.TAG_NAME, "body"), text) + ) + except Exception as e: + raise TimeoutError(f"wait_for_text '{text}' timed out: {e}") + + def wait_for_element(self, selectors: list, timeout_seconds: int = 15) -> None: + self._ensure_open() + deadline = time.time() + timeout_seconds + last_error = None + while time.time() < deadline: + for sel in selectors: + try: + self._find(sel, timeout=1) + return + except Exception as e: + last_error = e + time.sleep(0.5) + raise TimeoutError(f"wait_for_element timed out: {selectors}, last: {last_error}") + + def element_exists(self, selectors: list) -> bool: + self._ensure_open() + for sel in selectors: + try: + self._find(sel, timeout=1) + return True + except Exception: + continue + return False + + def switch_to_frame(self, selectors: list) -> None: + """切换到匹配到的 iframe。优先可见弹窗,避免切到上次留下的隐藏 iframe。""" + self._ensure_open() + from selenium.webdriver.common.by import By + + last_error = None + found = [] + for sel in selectors: + try: + if sel.startswith("//") or sel.startswith("/"): + found.extend(self._driver.find_elements(By.XPATH, sel)) + else: + found.extend(self._driver.find_elements(By.CSS_SELECTOR, sel)) + except Exception as e: + last_error = e + visible = [] + hidden = [] + for el in found: + try: + (visible if el.is_displayed() else hidden).append(el) + except Exception: + hidden.append(el) + for el in visible + hidden: + try: + self._driver.switch_to.default_content() + self._driver.switch_to.frame(el) + return + except Exception as e: + last_error = e + raise RuntimeError(f"switch_to_frame failed: {selectors}, last: {last_error}") + + def switch_to_default_content(self) -> None: + """切回主文档(顶层 frame)。""" + self._ensure_open() + self._driver.switch_to.default_content() + + def execute_script(self, script: str, *args): + self._ensure_open() + return self._driver.execute_script(script, *args) + + def select_xform(self, name: str, value: str) -> None: + """选中 Landray xform fancy-select(.fs-wrap / .fs-option)。""" + self._ensure_open() + result = self._driver.execute_script( + """ + const name = arguments[0]; + const wanted = String(arguments[1] || '').trim(); + const flag = document.querySelector('xformflag[property="' + name + '"]'); + if (!flag) { + return {ok: false, error: 'flag not found: ' + name}; + } + const norm = (s) => String(s || '').replace(/\\s+/g, '').trim(); + const wantedN = norm(wanted); + const opts = Array.from(flag.querySelectorAll('.fs-option')); + let target = opts.find((o) => { + const text = norm((o.querySelector('.fs-option-label') || o).textContent); + const val = norm(o.getAttribute('data-value')); + return text === wantedN || val === wantedN; + }); + if (!target && wantedN.length >= 2) { + target = opts.find((o) => { + const text = norm((o.querySelector('.fs-option-label') || o).textContent); + return text.indexOf(wantedN) >= 0; + }); + } + if (!target) { + return { + ok: false, + error: 'option not found: ' + wanted, + available: opts.map((o) => ((o.querySelector('.fs-option-label') || o).textContent || '').trim()), + }; + } + const labelWrap = flag.querySelector('.fs-label-wrap'); + if (labelWrap) { labelWrap.click(); } + target.click(); + const input = flag.querySelector('input[name="' + name + '"]'); + const selected = target.getAttribute('data-value') || ''; + if (input) { + const proto = window.HTMLInputElement.prototype; + const setter = Object.getOwnPropertyDescriptor(proto, 'value').set; + setter.call(input, selected); + input.dispatchEvent(new Event('input', { bubbles: true })); + input.dispatchEvent(new Event('change', { bubbles: true })); + if (typeof __xformDispatch === 'function') { + try { __xformDispatch(selected, input); } catch (e) {} + } + } + return { + ok: true, + value: selected, + label: ((flag.querySelector('.fs-label') || {}).textContent || '').trim(), + }; + """, + name, + str(value), + ) + if not result or not result.get("ok"): + raise RuntimeError(f"select_xform failed: {name}={value} {result}") + time.sleep(0.3) + + def select_first(self, selectors: list, value: str) -> None: + self._ensure_open() + from selenium.webdriver.support.ui import Select + last_error = None + for sel in selectors: + try: + el = self._find(sel) + # 原生