From 39754bef791211a4cabeaf6f4695465228cd8813 Mon Sep 17 00:00:00 2001 From: 34BFF76918A9449ABAB323149DDCEA73 Date: Fri, 9 Oct 2026 16:11:53 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=90=88=E5=90=8C=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=E6=8C=89=E5=BC=B9=E7=AA=97=E7=BB=93=E6=9E=9C=E8=AF=BB=E5=8F=B7?= =?UTF-8?q?=EF=BC=8CChrome=20=E6=8C=89=E7=8F=AD=E6=AC=A1=E5=A4=8D=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 提交后按成功 toast / 查看页判定，处理 LUI 确认框。 批跑共用一个浏览器；Chrome 按 flow 固定用户目录和调试端口，可附着已开实例。 --- rpa_contract/apa_browser_adapter.py | 195 ++++- rpa_contract/contract_flow_steps.py | 743 ++++++++++++++---- rpa_contract/run_batch.py | 6 +- .../tests/test_apa_browser_adapter.py | 21 +- rpa_contract/tests/test_flow_gates.py | 140 +++- 5 files changed, 907 insertions(+), 198 deletions(-) diff --git a/rpa_contract/apa_browser_adapter.py b/rpa_contract/apa_browser_adapter.py index 2c091ee..b9026cf 100644 --- a/rpa_contract/apa_browser_adapter.py +++ b/rpa_contract/apa_browser_adapter.py @@ -5,11 +5,18 @@ Selenium Chrome 浏览器适配。通过 WebDriver 操控 Chrome。 from __future__ import annotations import os +import socket import time from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] +_FLOW_DEBUG_PORTS = { + "CONTRACT_CREATE": 9333, + "REBATE_FIND": 9335, + "default": 9334, +} + def _is_inside_repo(path: Path, repo_root: Path) -> bool: try: @@ -33,6 +40,92 @@ def resolve_chrome_profile_base(requested: str | None = None, *, repo_root: Path return outside +def resolve_chrome_profile_dir( + flow: str = "default", + requested: str | None = None, + *, + repo_root: Path | None = None, +) -> Path: + """Stable profile per flow so later tasks reuse login cookies.""" + base = resolve_chrome_profile_base(requested, repo_root=repo_root) + safe = "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in (flow or "default").strip()) + return (base / (safe or "default")).resolve() + + +def chrome_debug_port(flow: str = "default") -> int: + raw = (os.getenv("RPA_CHROME_DEBUG_PORT") or "").strip() + if raw: + try: + return int(raw) + except ValueError: + pass + return _FLOW_DEBUG_PORTS.get((flow or "default").strip(), 9333) + + +def _port_open(port: int) -> bool: + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.4): + return True + except OSError: + return False + + +def _read_devtools_port(profile_dir: Path) -> int | None: + marker = profile_dir / "DevToolsActivePort" + if not marker.is_file(): + return None + try: + first = marker.read_text(encoding="utf-8").splitlines()[0].strip() + port = int(first) + except (OSError, ValueError, IndexError): + return None + return port if port > 0 else None + + +def _chrome_service(): + from selenium.webdriver.chrome.service import Service + + chromedriver = os.getenv("CHROMEDRIVER_PATH") + if chromedriver and os.path.exists(chromedriver): + return Service(chromedriver) + from webdriver_manager.chrome import ChromeDriverManager + + return Service(ChromeDriverManager().install()) + + +def _attach_chrome(service, port: int): + from selenium import webdriver + from selenium.webdriver.chrome.options import Options + + options = Options() + options.add_experimental_option("debuggerAddress", f"127.0.0.1:{port}") + driver = webdriver.Chrome(service=service, options=options) + driver.execute_script("return 1") + return driver + + +def _launch_chrome(service, profile_dir: Path, port: int): + from selenium import webdriver + from selenium.webdriver.chrome.options import Options + + 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") + options.add_argument("--remote-allow-origins=*") + options.add_argument(f"--remote-debugging-port={port}") + options.add_argument("--user-data-dir=" + str(profile_dir)) + return webdriver.Chrome(service=service, options=options) + + class ApaBrowserAdapter: """Selenium Chrome 浏览器适配器，实现 BrowserAdapter 协议""" @@ -43,48 +136,43 @@ class ApaBrowserAdapter: self._is_open = False def _get_driver(self): - """创建或获取 Chrome WebDriver""" + """复用本班次已打开的 Chrome；没有就按稳定 profile 新开。""" 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 / f"{os.getpid()}-{id(self)}" + profile_dir = resolve_chrome_profile_dir(self._flow) + if _is_inside_repo(profile_dir, REPO_ROOT): + raise RuntimeError(f"拒绝在仓库内创建 Chrome 用户目录: {profile_dir}") os.makedirs(profile_dir, exist_ok=True) - profile_dir = str(profile_dir) + port = _read_devtools_port(profile_dir) or chrome_debug_port(self._flow) - 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) + driver = None + if _port_open(port): + try: + driver = _attach_chrome(_chrome_service(), port) + print(f"Chrome 复用已有实例 flow={self._flow} port={port}", flush=True) + except Exception as exc: + print(f"Chrome 复用失败 port={port}：{exc}", flush=True) + driver = None + if driver is None: + launch_port = chrome_debug_port(self._flow) + try: + driver = _launch_chrome(_chrome_service(), profile_dir, launch_port) + print( + f"Chrome 新开实例 flow={self._flow} port={launch_port} profile={profile_dir}", + flush=True, + ) + except Exception as exc: + retry_port = _read_devtools_port(profile_dir) + if retry_port and _port_open(retry_port): + driver = _attach_chrome(_chrome_service(), retry_port) + print(f"Chrome 锁定后改为复用 port={retry_port}", flush=True) + else: + raise RuntimeError(f"打开 Chrome 失败：{exc}") from exc - 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) + driver.set_page_load_timeout(45) + driver.set_script_timeout(30) + self._driver = driver return self._driver def _ensure_open(self): @@ -502,3 +590,38 @@ class ApaBrowserAdapter: if self._driver: self._driver.save_screenshot(str(path)) return str(path) + + def accept_alert(self) -> str: + """Accept a native JS alert/confirm and return its text. Empty if none.""" + self._ensure_open() + try: + alert = self._driver.switch_to.alert + text = (alert.text or "").strip() + alert.accept() + return text + except Exception: + return "" + + def reopen_url(self, url: str) -> None: + """Close the current tab and open url in a fresh tab of the same Chrome.""" + driver = self._get_driver() + old = None + try: + old = driver.current_window_handle + driver.switch_to.new_window("tab") + new = driver.current_window_handle + if old and old != new: + try: + driver.switch_to.window(old) + driver.close() + except Exception: + pass + try: + driver.switch_to.window(new) + except Exception: + handles = driver.window_handles + if handles: + driver.switch_to.window(handles[-1]) + except Exception: + pass + self.open_url(url) diff --git a/rpa_contract/contract_flow_steps.py b/rpa_contract/contract_flow_steps.py index b908d95..56f91ca 100644 --- a/rpa_contract/contract_flow_steps.py +++ b/rpa_contract/contract_flow_steps.py @@ -34,6 +34,8 @@ class BrowserAdapter(Protocol): def read_first(self, selectors: list[str], value_name: str = "") -> str: ... def read_body_text(self) -> str: ... def screenshot(self, name: str) -> str: ... + def accept_alert(self) -> str: ... + def reopen_url(self, url: str) -> None: ... @dataclass(frozen=True) @@ -114,20 +116,20 @@ def run_contract_create_flow(browser: BrowserAdapter, ctx: ContractFlowContext) fill_contract_payment_info(browser, ctx) upload_contract_document_if_needed(browser, ctx) fd_id = _read_document_id(browser) - if not submit_contract_for_approval(browser, ctx): + if not submit_contract_for_approval(browser, ctx, fd_id=fd_id): if ctx.submit_mode == "dry_run": _ACCEPT.note("S13", "读合同号", {"contractNo": "(dry_run)", "mode": "dry_run"}, ok=True) return "" _ACCEPT.note("S13", "读合同号", {"contractNo": "(none)", "mode": ctx.submit_mode}, ok=False) return "" fd_id = fd_id or _read_document_id(browser) - contract_no = read_contract_number(browser, ctx, fd_id=fd_id) + contract_no, via = _read_contract_number_routed(browser, ctx, fd_id=fd_id) _ACCEPT.note( "S13", "读合同号", { "contractNo": contract_no or "(none)", - "via": "我的合同查询", + "via": via, "fdId": fd_id or "(none)", }, ok=_usable_contract_no(contract_no), @@ -162,6 +164,10 @@ VIEW_PAGE_URL = ( "http://contract-test.gzmpc.com:8080/km/agreement/km_agreement_apply/" "kmAgreementApply.do?method=view&fdId={fd_id}" ) +EDIT_PAGE_URL = ( + "http://contract-test.gzmpc.com:8080/km/agreement/km_agreement_apply/" + "kmAgreementApply.do?method=edit&fdId={fd_id}" +) # Landray/LUI 表单字段的真实 name F = { @@ -1341,55 +1347,128 @@ def _on_body_editor_tab(browser: BrowserAdapter) -> bool: return result is True +_ERROR_MARKS = ( + "不能为空", + "请填写", + "校验失败", + "提交失败", + "必填项未填", + "该文档不存在", + "操作失败", +) +_SUCCESS_MARKS = ("您的操作已成功", "操作已成功", "提交成功", "流程已提交", "成功提交") +_CONFIRM_MARKS = ("确定提交", "是否提交", "确认提交", "是否确定", "确定要提交") + + def _close_blocking_dialogs(browser: BrowserAdapter) -> None: browser.execute_script( """ const labels = ['关闭', '取消']; - const nodes = Array.from(document.querySelectorAll('div, a, button, span')); - for (const label of labels) { - const hit = nodes.find((el) => (el.textContent || '').trim() === label && el.children.length === 0); - if (hit) { hit.click(); return label; } + function clickIn(win) { + const nodes = Array.from(win.document.querySelectorAll('div, a, button, span')); + for (const label of labels) { + const hit = nodes.find((el) => (el.textContent || '').trim() === label && el.children.length === 0); + if (hit) { hit.click(); return label; } + } + for (let i = 0; i < win.frames.length; i++) { + try { const hit = clickIn(win.frames[i]); if (hit) return hit; } catch (e) {} + } + return ''; } - return ''; + return clickIn(window); """ ) time.sleep(0.4) -def submit_contract_for_approval(browser: BrowserAdapter, ctx: ContractFlowContext) -> bool: +def submit_contract_for_approval( + browser: BrowserAdapter, ctx: ContractFlowContext, *, fd_id: str = "" +) -> bool: if ctx.submit_mode == "dry_run": evidence_shot(browser, "contract-dry-run-ready") _ACCEPT.note("S12", "提交", {"mode": "dry_run", "submitted": "no"}, ok=True) return False _close_blocking_dialogs(browser) - # 测试环境直接提交会落到「该文档不存在」：先暂存落库，再提交。合同号仍从我的合同查。 + fd_id = (fd_id or "").strip() or _read_document_id(browser) + print(f"S12 fdId={fd_id or '-'}", flush=True) + # 测试环境直接提交会落到「该文档不存在」：先暂存落库，再提交。合同号从详情 #_xform_docNumber 读。 try: - _click_toolbar_button(browser, "暂存") + saved_click = _click_toolbar_button(browser, "暂存") except Exception as exc: print(f"暂存未点到：{exc}", flush=True) - saved = _wait_after_save_or_submit(browser, timeout_seconds=15) - blocked = _submit_validation_error(browser) - if blocked: + saved_click = False + save_kind, save_text = _wait_submit_outcome(browser, timeout_seconds=15) + if save_kind == "error": evidence_shot(browser, "contract-submitted", failed=True) - _ACCEPT.note("S12", "提交", {"submitted": "no", "error": blocked, "at": "暂存"}, ok=False, raise_error=False) + _ACCEPT.note( + "S12", + "提交", + {"submitted": "no", "error": save_text, "at": "暂存", "clickedSave": "yes" if saved_click else "no"}, + ok=False, + raise_error=False, + ) return False - if not saved: + if save_kind != "success": print("暂存后仍在表单页，继续点提交。", flush=True) + fd_id = fd_id or _read_document_id(browser) + if not _lui_toolbar_button_exists(browser, "提交"): + if not fd_id: + evidence_shot(browser, "contract-submitted", failed=True) + _ACCEPT.note( + "S12", + "提交", + {"submitted": "no", "error": "暂存后离开表单且无 fdId，无法再点提交", "clicked": "no"}, + ok=False, + raise_error=False, + ) + return False + print(f"暂存后打开编辑页 fdId={fd_id}", flush=True) + browser.open_url(EDIT_PAGE_URL.format(fd_id=fd_id)) + if not _wait_lui_toolbar_button(browser, "提交", timeout_seconds=12): + evidence_shot(browser, "contract-submitted", failed=True) + _ACCEPT.note( + "S12", + "提交", + {"submitted": "no", "error": "编辑页未见提交按钮 lui_widget_btn_txt", "clicked": "no", "fdId": fd_id}, + ok=False, + raise_error=False, + ) + return False try: - _click_toolbar_button(browser, "提交") + clicked = _click_toolbar_button(browser, "提交") except Exception as exc: evidence_shot(browser, "contract-submitted", failed=True) - _ACCEPT.note("S12", "提交", {"submitted": "no", "error": str(exc)[:120]}, ok=False, raise_error=False) + _ACCEPT.note("S12", "提交", {"submitted": "no", "error": str(exc)[:120], "clicked": "no"}, ok=False, raise_error=False) + return False + if not clicked: + evidence_shot(browser, "contract-submitted", failed=True) + _ACCEPT.note("S12", "提交", {"submitted": "no", "error": "未点到提交按钮 lui_widget_btn_txt", "clicked": "no"}, ok=False, raise_error=False) return False evidence_shot(browser, "contract-after-submit-click") - submitted = _wait_after_save_or_submit(browser, timeout_seconds=20) - blocked = _submit_validation_error(browser) - if blocked: + kind, dialog = _wait_submit_outcome(browser, timeout_seconds=20, trust_body_success=False) + if kind == "error": evidence_shot(browser, "contract-submitted", failed=True) - _ACCEPT.note("S12", "提交", {"submitted": "no", "error": blocked}, ok=False, raise_error=False) + _ACCEPT.note( + "S12", + "提交", + {"submitted": "no", "clicked": "yes", "error": dialog}, + ok=False, + raise_error=False, + ) return False - print("提交已点，回我的合同查询合同号。", flush=True) - _ACCEPT.note("S12", "提交", {"submitted": "yes", "via": "暂存后提交", "leftForm": "yes" if submitted else "no"}, ok=True) + print(f"提交已点 kind={kind} dialog={dialog[:80]!r}，S13 从详情读合同号。", flush=True) + _ACCEPT.note( + "S12", + "提交", + { + "submitted": "yes", + "clicked": "yes", + "via": "暂存后提交", + "signal": kind, + "dialog": dialog[:80] or "(none)", + }, + ok=True, + ) return True @@ -1402,66 +1481,256 @@ def _submit_error_from_body(body: str) -> str: return "" -def _wait_after_save_or_submit(browser: BrowserAdapter, timeout_seconds: float) -> bool: +def _classify_dialog_text(text: str) -> str: + blob = text or "" + for mark in _ERROR_MARKS: + if mark in blob: + return "error" + for mark in _SUCCESS_MARKS: + if mark in blob: + return "success" + for mark in _CONFIRM_MARKS: + if mark in blob: + return "confirm" + return "none" + + +def _current_url(browser: BrowserAdapter) -> str: + try: + return str( + browser.execute_script( + "try { return (window.top.location && window.top.location.href) || location.href; }" + " catch (e) { return location.href; }" + ) + or "" + ) + except Exception: + return "" + + +def _wait_submit_outcome( + browser: BrowserAdapter, + timeout_seconds: float, + *, + trust_body_success: bool = True, +) -> tuple[str, str]: deadline = time.time() + timeout_seconds + last_dialog = "" + saw_confirm = False while time.time() < deadline: - if _submit_validation_error(browser): - return False - _confirm_submit_dialogs(browser) - if _submit_succeeded(browser): - return True - time.sleep(0.5) - return not bool(_submit_validation_error(browser)) and _submit_succeeded(browser) + kind, text, confirm = _read_submit_signal( + browser, + trust_body_success=trust_body_success or saw_confirm, + ) + if confirm: + saw_confirm = True + if text: + last_dialog = text + if kind == "error": + return "error", text + if kind == "success": + return "success", text + time.sleep(0.4) + kind, text, _confirm = _read_submit_signal( + browser, + trust_body_success=trust_body_success or saw_confirm, + ) + if kind in ("error", "success"): + return kind, text + return "none", text or last_dialog -def _submit_succeeded(browser: BrowserAdapter) -> bool: +def _read_submit_signal( + browser: BrowserAdapter, + *, + trust_body_success: bool = True, +) -> tuple[str, str, bool]: + alert_text = _accept_native_alert(browser) + if alert_text: + kind = _classify_dialog_text(alert_text) + if kind == "none": + kind = "confirm" + print(f"S12 native alert kind={kind} text={alert_text[:80]!r}", flush=True) + if kind == "error": + return "error", alert_text, False + if kind == "success": + return "success", alert_text, False + if kind == "confirm": + return _finish_submit_signal(browser, alert_text, True, trust_body_success) + snapshot = _handle_submit_dialogs(browser) + dialog_text = str(snapshot.get("text") or "").strip() + kind = _classify_dialog_text(dialog_text) if dialog_text else "none" + if kind == "none": + kind = str(snapshot.get("kind") or "none") + if kind == "error": + return "error", dialog_text or "弹窗报错", False + if kind == "success": + return "success", dialog_text or "ok", False + return _finish_submit_signal( + browser, + dialog_text, + kind == "confirm" or bool(snapshot.get("clicked")), + trust_body_success, + ) + + +def _finish_submit_signal( + browser: BrowserAdapter, + dialog_text: str, + saw_confirm: bool, + trust_body_success: bool, +) -> tuple[str, str, bool]: + body_error = _submit_error_from_body(browser.read_body_text() or "") + if body_error: + return "error", body_error, saw_confirm + if _submit_view_page(browser): + return "success", dialog_text or "view", saw_confirm + if (trust_body_success or saw_confirm) and _submit_success_text(browser): + return "success", dialog_text or "ok", saw_confirm + return "none", dialog_text, saw_confirm + + +def _accept_native_alert(browser: BrowserAdapter) -> str: + accept = getattr(browser, "accept_alert", None) + if not callable(accept): + return "" + try: + return str(accept() or "").strip() + except Exception: + return "" + + +def _submit_view_page(browser: BrowserAdapter) -> bool: + url = _current_url(browser) + return "method=view" in url and "kmAgreementApply" in url + + +def _submit_success_text(browser: BrowserAdapter) -> bool: body = browser.read_body_text() or "" if _submit_error_from_body(body): return False - if any(mark in body for mark in ("您的操作已成功", "操作已成功", "提交成功", "流程已提交", "成功提交")): - return True - url = "" - try: - url = str(browser.execute_script("return location.href") or "") - except Exception: - url = "" - return "method=add" not in url and "kmAgreementApply" in url + return any(mark in body for mark in _SUCCESS_MARKS) + + +def _submit_succeeded(browser: BrowserAdapter) -> bool: + if _submit_error_from_body(browser.read_body_text() or ""): + return False + return _submit_success_text(browser) or _submit_view_page(browser) def _submit_validation_error(browser: BrowserAdapter) -> str: - body = browser.read_body_text() or "" - hit = _submit_error_from_body(body) - if hit: - return hit - text = browser.execute_script( - """ - const roots = Array.from(document.querySelectorAll( - '.lui_dialog, .lui_message, .lui_alert, .lui_dialog_content, .prompt_frame' - )); - return roots.map((el) => (el.innerText || '').trim()).filter(Boolean).join('\\n').slice(0, 400); - """ - ) - blob = str(text or "").strip() - if not blob: - return "" - for mark in ("不能为空", "请填写", "校验失败", "提交失败", "必填项未填", "该文档不存在", "操作失败"): - if mark in blob: - return blob[:120] + kind, text, _confirm = _read_submit_signal(browser, trust_body_success=True) + if kind == "error": + return text return "" +def _handle_submit_dialogs(browser: BrowserAdapter) -> dict: + raw = browser.execute_script( + """ + const errorMarks = ['不能为空','请填写','校验失败','提交失败','必填项未填','该文档不存在','操作失败']; + const successMarks = ['您的操作已成功','操作已成功','提交成功','流程已提交','成功提交']; + const confirmMarks = ['确定提交','是否提交','确认提交','是否确定','确定要提交']; + function classify(text) { + if (errorMarks.some((m) => text.indexOf(m) >= 0)) return 'error'; + if (successMarks.some((m) => text.indexOf(m) >= 0)) return 'success'; + if (confirmMarks.some((m) => text.indexOf(m) >= 0)) return 'confirm'; + return 'none'; + } + function dialogText(doc) { + const roots = Array.from(doc.querySelectorAll( + '.lui_dialog, .lui_message, .lui_alert, .lui_dialog_content, .prompt_frame, .lui_dialog_common' + )); + return roots.map((el) => (el.innerText || '').trim()).filter(Boolean).join('\\n'); + } + function clickLabel(doc, labels) { + const nodes = Array.from(doc.querySelectorAll('div, a, button, span, input')); + for (const label of labels) { + const hit = nodes.find((el) => { + const t = (el.textContent || el.value || '').trim(); + return t === label && (!el.children || el.children.length === 0); + }); + if (hit) { hit.click(); return label; } + } + return ''; + } + function walk(win) { + let found = {kind: 'none', text: '', clicked: ''}; + try { + const text = dialogText(win.document); + const kind = classify(text); + if (kind === 'error') return {kind: kind, text: text.slice(0, 200), clicked: ''}; + if (kind === 'confirm') { + const clicked = clickLabel(win.document, ['确定', '确认', '是', 'OK']); + return {kind: kind, text: text.slice(0, 200), clicked: clicked}; + } + if (kind === 'success') { + const clicked = clickLabel(win.document, ['确定', '关闭', '确认']); + return {kind: kind, text: text.slice(0, 200), clicked: clicked}; + } + if (text) { + const clicked = clickLabel(win.document, ['确定', '确认']); + if (clicked) return {kind: 'confirm', text: text.slice(0, 200), clicked: clicked}; + } + } catch (e) {} + for (let i = 0; i < win.frames.length; i++) { + try { + const inner = walk(win.frames[i]); + if (inner.kind !== 'none' || inner.clicked) return inner; + } catch (e) {} + } + return found; + } + return walk(window); + """ + ) + if not isinstance(raw, dict): + return {"kind": "none", "text": "", "clicked": ""} + return { + "kind": str(raw.get("kind") or "none"), + "text": str(raw.get("text") or "").strip(), + "clicked": str(raw.get("clicked") or ""), + } + + def _read_document_id(browser: BrowserAdapter) -> str: + try: + browser.switch_to_default_content() + except Exception: + pass value = browser.execute_script( """ - const href = String(location.href || ''); - const fromUrl = href.match(/[?&]fdId=([^&]+)/i); - if (fromUrl && fromUrl[1]) return decodeURIComponent(fromUrl[1]); - const input = document.querySelector('input[name="fdId"], input[name="fdid"]'); - if (input && String(input.value || '').trim()) return String(input.value).trim(); - try { - if (window.Com_Parameter && Com_Parameter.fdId) return String(Com_Parameter.fdId); - } catch (e) {} - return ''; + function fromHref(href) { + const fromUrl = String(href || '').match(/[?&]fdId=([^&]+)/i); + if (fromUrl && fromUrl[1]) return decodeURIComponent(fromUrl[1]); + return ''; + } + function fromDoc(win) { + try { + const id = fromHref(win.location && win.location.href); + if (id) return id; + } catch (e) {} + try { + const input = win.document.querySelector('input[name="fdId"], input[name="fdid"]'); + if (input && String(input.value || '').trim()) return String(input.value).trim(); + } catch (e) {} + try { + if (win.Com_Parameter && win.Com_Parameter.fdId) return String(win.Com_Parameter.fdId); + } catch (e) {} + return ''; + } + function walk(win) { + const id = fromDoc(win); + if (id) return id; + for (let i = 0; i < win.frames.length; i++) { + try { + const inner = walk(win.frames[i]); + if (inner) return inner; + } catch (e) {} + } + return ''; + } + return walk(window); """ ) return str(value or "").strip() @@ -1478,79 +1747,182 @@ def _open_submitted_view(browser: BrowserAdapter, fd_id: str) -> None: time.sleep(1.5) -def _click_toolbar_button(browser: BrowserAdapter, label: str) -> None: - clicked = browser.execute_script( - """ +def _lui_btn_xpath(label: str) -> str: + safe = (label or "").replace("'", "") + return ( + "//*[contains(@class,'lui_widget_btn_txt') and normalize-space()='%s']" % safe + ) + + +def _click_toolbar_button(browser: BrowserAdapter, label: str) -> bool: + try: + browser.switch_to_default_content() + except Exception: + pass + clicked = browser.execute_script(_LUI_TOOLBAR_CLICK_JS, label) + print(f"toolbar {label} js={clicked}", flush=True) + if isinstance(clicked, dict) and clicked.get("ok"): + return True + selectors = [_lui_btn_xpath(label)] + try: + browser.click_native(selectors) + print(f"toolbar {label} native=ok", flush=True) + return True + except Exception as exc: + print(f"toolbar {label} native={exc}", flush=True) + try: + browser.click_first(selectors) + print(f"toolbar {label} first=ok", flush=True) + return True + except Exception as exc: + print(f"toolbar {label} first={exc}", flush=True) + return False + + +_LUI_TOOLBAR_FIND_JS = """ const label = String(arguments[0] || ''); function visible(el) { const box = el.getBoundingClientRect(); return box.width > 2 && box.height > 2; } - function clickIn(win) { - const toolbar = win.document.querySelector( - '.lui_toolbar_frame, .lui_toolbar, [class*="lui_toolbar"]' - ); - const roots = [toolbar, win.document].filter(Boolean); - for (const root of roots) { - const hits = Array.from(root.querySelectorAll('div, a, button, span, li')).filter((el) => { - return (el.textContent || '').replace(/\\s+/g, '').trim() === label && visible(el); - }); - hits.sort((a, b) => a.children.length - b.children.length); - if (hits.length) { - hits[0].click(); - return true; - } + function textOf(el) { + return (el.textContent || '').replace(/\\s+/g, '').trim(); + } + function findIn(win) { + const nodes = Array.from(win.document.querySelectorAll( + 'div.lui-component.lui_widget_btn_txt, .lui_widget_btn_txt' + )); + const hit = nodes.find((el) => textOf(el) === label && visible(el)); + if (hit) { + return { + ok: true, + cid: hit.getAttribute('data-lui-cid') || '', + id: hit.id || '', + count: nodes.length + }; } for (let i = 0; i < win.frames.length; i++) { - try { if (clickIn(win.frames[i])) return true; } catch (e) {} + try { + const inner = findIn(win.frames[i]); + if (inner && inner.ok) return inner; + } catch (e) {} } - return false; + return { + ok: false, + texts: nodes.map((el) => textOf(el)).filter(Boolean).slice(0, 12) + }; } - return {ok: clickIn(window), label: label}; - """, - label, - ) - print(f"toolbar {label} js={clicked}", flush=True) - if isinstance(clicked, dict) and clicked.get("ok"): - return - browser.click_first([f"text={label}", f"button:contains({label})"]) - - -def _click_top_submit(browser: BrowserAdapter) -> None: - _click_toolbar_button(browser, "提交") - - -def _confirm_submit_dialogs(browser: BrowserAdapter) -> None: - browser.execute_script( + return findIn(window); """ - const labels = ['确定', '确认', '是', 'OK']; - const nodes = Array.from(document.querySelectorAll('div, a, button, span, input')); - for (const label of labels) { - const hit = nodes.find((el) => { - const text = (el.textContent || el.value || '').trim(); - return text === label && el.children.length === 0; + + +_LUI_TOOLBAR_CLICK_JS = """ + const label = String(arguments[0] || ''); + function visible(el) { + const box = el.getBoundingClientRect(); + return box.width > 2 && box.height > 2; + } + function textOf(el) { + return (el.textContent || '').replace(/\\s+/g, '').trim(); + } + function fire(el, win) { + const target = el.closest('[data-lui-cid]') || el; + try { target.scrollIntoView({block: 'center', inline: 'nearest'}); } catch (e) {} + ['mouseover', 'mousedown', 'mouseup', 'click'].forEach((type) => { + target.dispatchEvent(new MouseEvent(type, { + bubbles: true, cancelable: true, view: win, buttons: 1 + })); }); - if (hit) { hit.click(); return label; } + target.click(); } - return ''; + function clickIn(win) { + const nodes = Array.from(win.document.querySelectorAll( + 'div.lui-component.lui_widget_btn_txt, .lui_widget_btn_txt' + )); + const hit = nodes.find((el) => textOf(el) === label && visible(el)); + if (hit) { + fire(hit, win); + return { + ok: true, + via: 'btn_txt', + cid: hit.getAttribute('data-lui-cid') || '', + id: hit.id || '' + }; + } + for (let i = 0; i < win.frames.length; i++) { + try { + const inner = clickIn(win.frames[i]); + if (inner && inner.ok) return inner; + } catch (e) {} + } + return { + ok: false, + label: label, + texts: nodes.map((el) => textOf(el)).filter(Boolean).slice(0, 12) + }; + } + return clickIn(window); """ - ) - time.sleep(0.8) + + +def _lui_toolbar_button_exists(browser: BrowserAdapter, label: str) -> bool: + found = browser.execute_script(_LUI_TOOLBAR_FIND_JS, label) + ok = isinstance(found, dict) and bool(found.get("ok")) + print(f"toolbar {label} exists={found}", flush=True) + return ok + + +def _wait_lui_toolbar_button( + browser: BrowserAdapter, label: str, timeout_seconds: float = 10 +) -> bool: + deadline = time.time() + timeout_seconds + while time.time() < deadline: + if _lui_toolbar_button_exists(browser, label): + return True + time.sleep(0.4) + return False + + +def _wait_lui_dialog_gone(browser: BrowserAdapter, timeout_seconds: float = 8) -> None: + deadline = time.time() + timeout_seconds + while time.time() < deadline: + snapshot = _handle_submit_dialogs(browser) + kind = str(snapshot.get("kind") or "none") + text = str(snapshot.get("text") or "").strip() + if kind == "confirm": + return + if kind in ("none", "") and not text: + return + _close_blocking_dialogs(browser) + time.sleep(0.4) + _close_blocking_dialogs(browser) + + +def _click_top_submit(browser: BrowserAdapter) -> bool: + return _click_toolbar_button(browser, "提交") def read_contract_number( browser: BrowserAdapter, ctx: ContractFlowContext | None = None, *, fd_id: str = "" ) -> str: - if ctx is None: - ctx = ContractFlowContext( - contract_url="http://contract-test.gzmpc.com:8080/", - username="", - password="", - supplier_name="", - order_date="", - source_type="", - ) - return _read_contract_number_from_my_list(browser, ctx, fd_id=fd_id) + no, _via = _read_contract_number_routed(browser, ctx, fd_id=fd_id) + return no + + +def _read_contract_number_routed( + browser: BrowserAdapter, ctx: ContractFlowContext | None = None, *, fd_id: str = "" +) -> tuple[str, str]: + del ctx + doc_id = (fd_id or "").strip() or _read_document_id(browser) + if not doc_id: + print("S13 没有 fdId，无法打开详情", flush=True) + return "", "none" + print(f"S13：开详情读号 fdId={doc_id}", flush=True) + no = _read_number_by_fd_id(browser, doc_id) + if no: + return no, "docNumber" + print("S13 详情未读到合同编号", flush=True) + return "", "none" def _read_contract_number_on_page(browser: BrowserAdapter) -> str: @@ -1561,25 +1933,22 @@ def _read_contract_number_on_page(browser: BrowserAdapter) -> str: scanned = browser.execute_script( """ const skip = ['提交时自动生成', '自动生成', '==请选择==', '合同编号', '合同标题']; + function usable(raw) { + const v = String(raw || '').replace(/\\s+/g, ' ').trim(); + if (!v || skip.indexOf(v) >= 0) return ''; + return v; + } function fromDoc(doc) { - const inputs = doc.querySelectorAll( - 'input[name="fdNumber"], input[name="contractNo"], input[name="contractSystemNo"], input[name*="fdNumber"]' - ); - for (const el of inputs) { - const v = (el.value || '').trim(); - if (v && skip.indexOf(v) < 0) return v; + const node = doc.querySelector('#_xform_docNumber'); + if (node) { + const typed = node.querySelector('input, textarea'); + const v = usable(typed ? typed.value : (node.innerText || node.textContent || '')); + if (v) return v; } - const nodes = Array.from(doc.querySelectorAll('td, th, span, div, label, em, b')); - for (let i = 0; i < nodes.length; i++) { - const label = (nodes[i].innerText || '').replace(/\\s+/g, '').trim(); - if (label !== '合同编号' && label !== '合同编号：' && label !== '合同编号:') continue; - const box = nodes[i].parentElement || nodes[i]; - const typed = box.querySelector('input, textarea'); - const n = typed - ? String(typed.value || '').trim() - : ((nodes[i + 1] && (nodes[i + 1].innerText || nodes[i + 1].value || '')) || '') - .replace(/\\s+/g, ' ').trim(); - if (n && skip.indexOf(n) < 0) return n; + const input = doc.querySelector('input[name="docNumber"]'); + if (input) { + const v = usable(input.value || ''); + if (v) return v; } return ''; } @@ -1603,15 +1972,10 @@ def _read_contract_number_on_page(browser: BrowserAdapter) -> str: return scanned.strip() if scanned: print(f"S13 详情页扫到不可用值={scanned!r}", flush=True) - body = browser.read_body_text() or "" - for line in body.splitlines(): - text = line.strip() - if _usable_contract_no(text) and (text.startswith("HT") or (text.isdigit() and len(text) >= 10)): - return text return "" -def _reload_contract_home(browser: BrowserAdapter, home: str) -> None: +def _reload_contract_home(browser: BrowserAdapter, home: str) -> bool: try: browser.switch_to_default_content() except Exception: @@ -1624,18 +1988,68 @@ def _reload_contract_home(browser: BrowserAdapter, home: str) -> None: ) except Exception: pass - browser.open_url(home) - if not _wait_text_in_frames(browser, "我的合同", timeout_seconds=15): - print("主界面未出现「我的合同」，仍继续进入查询", flush=True) + try: + browser.open_url(home) + except Exception as exc: + print(f"open_url 主界面失败：{exc}", flush=True) + return _wait_text_in_frames(browser, "我的合同", timeout_seconds=12) + + +def _reopen_contract_home(browser: BrowserAdapter, ctx: ContractFlowContext, home: str) -> bool: + print("主界面刷新失败，关闭当前页后重开", flush=True) + reopen = getattr(browser, "reopen_url", None) + if callable(reopen): + reopen(home) + else: + browser.open_url(home) + if _wait_text_in_frames(browser, "我的合同", timeout_seconds=12): + return True + if not _is_logged_in(browser): + login_contract_system(browser, ctx) + browser.open_url(home) + return _wait_text_in_frames(browser, "我的合同", timeout_seconds=12) def _open_my_contracts(browser: BrowserAdapter, ctx: ContractFlowContext) -> None: home = (ctx.contract_url or "").strip() or "http://contract-test.gzmpc.com:8080/" - _reload_contract_home(browser, home) - _click_by_exact_text(browser, "我的合同") + if not _reload_contract_home(browser, home): + _reopen_contract_home(browser, ctx, home) + if not _click_my_contracts_sidebar(browser): + print("未点到侧栏「我的合同」，关闭页面后重开", flush=True) + _reopen_contract_home(browser, ctx, home) + _click_my_contracts_sidebar(browser) _wait_text_in_frames(browser, "我起草的", timeout_seconds=10) _click_by_exact_text(browser, "我起草的") _wait_text_in_frames(browser, "申请时间", timeout_seconds=10) + _wait_text_in_frames(browser, "合同编号", timeout_seconds=10) + + +def _click_my_contracts_sidebar(browser: BrowserAdapter) -> bool: + clicked = browser.execute_script( + """ + const label = '我的合同'; + function clickIn(win) { + const nodes = Array.from(win.document.querySelectorAll('a, span, div, li, p, em, td')); + const hits = nodes.filter((el) => (el.textContent || '').trim() === label); + hits.sort((a, b) => { + const ax = a.getBoundingClientRect().left; + const bx = b.getBoundingClientRect().left; + if (ax !== bx) return ax - bx; + return a.children.length - b.children.length; + }); + if (hits.length) { hits[0].click(); return true; } + for (let i = 0; i < win.frames.length; i++) { + try { if (clickIn(win.frames[i])) return true; } catch (e) {} + } + return false; + } + return clickIn(window); + """ + ) + if clicked is True: + return _wait_text_in_frames(browser, "我起草的", timeout_seconds=8) + _click_by_exact_text(browser, "我的合同") + return _wait_text_in_frames(browser, "我起草的", timeout_seconds=8) def _wait_text_in_frames(browser: BrowserAdapter, text: str, timeout_seconds: int = 15) -> bool: @@ -1851,6 +2265,16 @@ def _search_my_contracts(browser: BrowserAdapter, title: str, customer_code: str } return false; } + function fillPlaceholder(doc, hint, value) { + if (!value) return false; + const inputs = Array.from(doc.querySelectorAll('input')); + const el = inputs.find((i) => (i.getAttribute('placeholder') || '').indexOf(hint) >= 0); + if (!el) return false; + setValue(el, value); + const btn = el.parentElement && el.parentElement.querySelector('a, button, span, i, em'); + if (btn) btn.click(); + return true; + } function clickSearch(doc) { const labels = ['搜索', '查询']; const nodes = Array.from(doc.querySelectorAll('a, div, span, button, input')); @@ -1865,9 +2289,12 @@ def _search_my_contracts(browser: BrowserAdapter, title: str, customer_code: str } function walk(win) { try { - fillNearLabel(win.document, '合同标题', title) + fillPlaceholder(win.document, '合同（协议）全称', title) + || fillPlaceholder(win.document, '合同标题', title) + || fillNearLabel(win.document, '合同标题', title) || fillNearLabel(win.document, '合同（协议）全称', title); - fillNearLabel(win.document, '客户编码', customer); + fillPlaceholder(win.document, '客户编码', customer) + || fillNearLabel(win.document, '客户编码', customer); clickSearch(win.document); } catch (e) {} for (let i = 0; i < win.frames.length; i++) { @@ -2009,6 +2436,7 @@ def _read_contract_number_from_my_list( _open_my_contracts(browser, ctx) deadline = time.time() + timeout_seconds searched = False + saw_row = False while time.time() < deadline: rows = _scrape_my_contract_rows(browser) print(f"S13 列表行数={len(rows)}", flush=True) @@ -2031,14 +2459,15 @@ def _read_contract_number_from_my_list( print("从我的合同读到合同编号：", no, flush=True) return no if fd_id and _row_has_fd_id(rows, fd_id): - no = _read_number_by_fd_id(browser, fd_id) - if no: - return no - _open_my_contracts(browser, ctx) + saw_row = True if not searched: _search_my_contracts(browser, title, customer) searched = True time.sleep(1) + if saw_row and fd_id: + no = _read_number_by_fd_id(browser, fd_id) + if no: + return no print("我的合同列表未匹配到合同编号", flush=True) return "" diff --git a/rpa_contract/run_batch.py b/rpa_contract/run_batch.py index a526b64..9f77519 100644 --- a/rpa_contract/run_batch.py +++ b/rpa_contract/run_batch.py @@ -11,7 +11,7 @@ from persist import ContractOutcome from repo import OracleContractRepo -def production_runner(draft: ContractDraft) -> ContractOutcome: +def production_runner(browser: ApaBrowserAdapter, draft: ContractDraft) -> ContractOutcome: spec = ContractCreateInput( job_id="contract-batch", supplier_name=draft.supplier_name, @@ -20,7 +20,6 @@ def production_runner(draft: ContractDraft) -> ContractOutcome: submit_mode="submit", ) ctx = build_flow_context(spec, draft) - browser = ApaBrowserAdapter(url=ctx.contract_url or CONTRACT_URL, flow="CONTRACT_CREATE") def run() -> str: return run_contract_create_flow(browser, ctx) @@ -31,7 +30,8 @@ def production_runner(draft: ContractDraft) -> ContractOutcome: def main() -> int: load_runtime_env() repo = OracleContractRepo.from_env() - summary = run_batch(repo, runner=production_runner) + browser = ApaBrowserAdapter(url=CONTRACT_URL, flow="CONTRACT_CREATE") + summary = run_batch(repo, runner=lambda draft: production_runner(browser, draft)) print( f"claimed={summary.claimed} groups={summary.groups} " f"ok={summary.ok} failed={summary.failed}" diff --git a/rpa_contract/tests/test_apa_browser_adapter.py b/rpa_contract/tests/test_apa_browser_adapter.py index a817ae4..e722dc9 100644 --- a/rpa_contract/tests/test_apa_browser_adapter.py +++ b/rpa_contract/tests/test_apa_browser_adapter.py @@ -1,6 +1,11 @@ from pathlib import Path -from apa_browser_adapter import REPO_ROOT, resolve_chrome_profile_base +from apa_browser_adapter import ( + REPO_ROOT, + chrome_debug_port, + resolve_chrome_profile_base, + resolve_chrome_profile_dir, +) def test_default_chrome_profile_is_outside_repo(): @@ -25,6 +30,20 @@ def test_env_path_outside_repo_is_kept(tmp_path): assert profile == outside.resolve() +def test_chrome_profile_dir_is_stable_per_flow(): + first = resolve_chrome_profile_dir("CONTRACT_CREATE") + second = resolve_chrome_profile_dir("CONTRACT_CREATE") + assert first == second + assert first == (REPO_ROOT.parent / "rpa-chrome-profile" / "CONTRACT_CREATE").resolve() + assert not _is_relative_to(first, REPO_ROOT) + + +def test_chrome_debug_port_is_stable_per_flow(monkeypatch): + monkeypatch.delenv("RPA_CHROME_DEBUG_PORT", raising=False) + assert chrome_debug_port("CONTRACT_CREATE") == 9333 + assert chrome_debug_port("CONTRACT_CREATE") == chrome_debug_port("CONTRACT_CREATE") + + def _is_relative_to(path: Path, root: Path) -> bool: try: path.resolve().relative_to(root.resolve()) diff --git a/rpa_contract/tests/test_flow_gates.py b/rpa_contract/tests/test_flow_gates.py index 8c281cc..aab59e7 100644 --- a/rpa_contract/tests/test_flow_gates.py +++ b/rpa_contract/tests/test_flow_gates.py @@ -4,12 +4,18 @@ from contract_flow_steps import ( ContractFlowContext, begin_date_for, check_customer_diligence, + read_contract_number, sign_date_for, _as_index_list, + _classify_dialog_text, + _click_toolbar_button, + _lui_btn_xpath, _next_unused_goods_row, _pick_contract_row, + _read_contract_number_on_page, _resolve_contract_body_path, _submit_error_from_body, + _submit_succeeded, ) from errors import DiligenceError import pytest @@ -29,12 +35,29 @@ def _ctx(**overrides) -> ContractFlowContext: class _FakeBrowser: - def __init__(self, text: str): + def __init__(self, text: str, url: str = "", script_result: str = ""): self._text = text + self._url = url + self._script_result = script_result + self.opened: list[str] = [] def read_body_text(self) -> str: return self._text + def switch_to_default_content(self) -> None: + return None + + def open_url(self, url: str) -> None: + self.opened.append(url) + self._url = url + + def execute_script(self, script: str, *args): + if "_xform_docNumber" in script: + return self._script_result + if "location.href" in script: + return self._url + return "" + def test_sign_and_begin_fall_back_to_order_date(): ctx = _ctx() @@ -124,6 +147,121 @@ def test_submit_error_from_doc_missing_page(): assert _submit_error_from_body("客户编码 提交") == "" +def test_submit_succeeded_on_success_toast(): + browser = _FakeBrowser("您的操作已成功", url="kmAgreementApply.do?method=edit&fdId=1") + assert _submit_succeeded(browser) is True + + +def test_submit_succeeded_on_view_page(): + browser = _FakeBrowser("合同详情", url="kmAgreementApply.do?method=view&fdId=1") + assert _submit_succeeded(browser) is True + + +def test_submit_not_succeeded_on_edit_or_add_page(): + edit = _FakeBrowser("客户编码 暂存 提交", url="kmAgreementApply.do?method=edit&fdId=1") + add = _FakeBrowser("客户编码 暂存 提交", url="kmAgreementApply.do?method=add") + assert _submit_succeeded(edit) is False + assert _submit_succeeded(add) is False + + +def test_classify_dialog_text_kinds(): + assert _classify_dialog_text("操作失败！该文档不存在") == "error" + assert _classify_dialog_text("您确定要提交此文档吗？") == "confirm" + assert _classify_dialog_text("您的操作已成功") == "success" + assert _classify_dialog_text("客户编码") == "none" + + +def test_lui_btn_xpath_is_exact_txt(): + xpath = _lui_btn_xpath("提交") + assert "lui_widget_btn_txt" in xpath + assert "normalize-space()='提交'" in xpath + + +class _ClickBrowser(_FakeBrowser): + def __init__(self, script_result): + super().__init__("") + self.script_result = script_result + self.scripts: list[tuple] = [] + self.native: list[list[str]] = [] + self.first: list[list[str]] = [] + + def execute_script(self, script, *args): + self.scripts.append((script, args)) + return self.script_result + + def click_native(self, selectors): + self.native.append(list(selectors)) + + def click_first(self, selectors): + self.first.append(list(selectors)) + + +def test_click_toolbar_button_targets_lui_btn_txt(): + browser = _ClickBrowser({"ok": True, "via": "btn_txt", "cid": "lui-id-36"}) + assert _click_toolbar_button(browser, "提交") is True + script, args = browser.scripts[0] + assert "lui_widget_btn_txt" in script + assert args[0] == "提交" + assert browser.native == [] + assert browser.first == [] + + +def test_click_toolbar_button_fallback_uses_exact_txt(): + browser = _ClickBrowser({"ok": False, "texts": ["暂存"]}) + assert _click_toolbar_button(browser, "提交") is True + joined = " ".join(browser.native[0]) + assert "lui_widget_btn_txt" in joined + assert "normalize-space()='提交'" in joined + assert "text=提交" not in joined + assert "contains(text()" not in joined + + +def test_read_contract_number_from_xform_doc_number(): + browser = _FakeBrowser( + "手机号 13559935238 合同编号", + script_result="035012026100835435", + ) + assert _read_contract_number_on_page(browser) == "035012026100835435" + no = read_contract_number(browser, _ctx(), fd_id="abc") + assert no == "035012026100835435" + assert any("method=view" in url and "fdId=abc" in url for url in browser.opened) + assert all("我的合同" not in url for url in browser.opened) + + +def test_read_contract_number_requires_fd_id(): + browser = _FakeBrowser("合同编号") + assert read_contract_number(browser, _ctx(), fd_id="") == "" + assert browser.opened == [] + + +def test_read_contract_number_skips_placeholder_and_phone(): + browser = _FakeBrowser( + "手机号\n13559935238\n提交时自动生成", + script_result="提交时自动生成", + ) + assert _read_contract_number_on_page(browser) == "" + + +def test_pick_contract_row_skips_empty_number(): + rows = [ + { + "no": "", + "title": "重庆亿昊药业有限公司采购协议20261009", + "customer": "DW0100046361", + "href": "kmAgreementApply.do?method=view&fdId=ours", + } + ] + assert ( + _pick_contract_row( + rows, + fd_id="ours", + title="重庆亿昊药业有限公司采购协议20261009", + customer_code="DW0100046361", + ) + == "" + ) + + def test_submit_without_file_uses_placeholder_body(): ctx = _ctx(submit_mode="submit", source_type="") path = _resolve_contract_body_path(ctx)