"""合同系统页面步骤：登录、起草、填表、提交、读合同号。""" from __future__ import annotations import time from dataclasses import dataclass, field from pathlib import Path from typing import Protocol from accept import AcceptanceReport, accept_log_path, evidence_shot from errors import ( LoginFailed, NodeError, PdfMergeUnsupported, SupplierContractDisabled, ) class BrowserAdapter(Protocol): def open_url(self, url: str) -> None: ... def fill_first(self, selectors: list[str], value: str) -> None: ... def click_first(self, selectors: list[str]) -> None: ... def click_native(self, selectors: list[str]) -> None: ... def double_click_first(self, selectors: list[str]) -> None: ... def wait_for_text(self, text: str, timeout_seconds: int = 15) -> None: ... def wait_for_element(self, selectors: list[str], timeout_seconds: int = 15) -> None: ... def element_exists(self, selectors: list[str]) -> bool: ... def switch_to_frame(self, selectors: list[str]) -> None: ... def switch_to_default_content(self) -> None: ... def select_first(self, selectors: list[str], value: str) -> None: ... def select_xform(self, name: str, value: str) -> None: ... def execute_script(self, script: str, *args): ... def upload_file_first(self, selectors: list[str], file_path: str) -> None: ... def read_first(self, selectors: list[str], value_name: str = "") -> str: ... def read_body_text(self) -> str: ... def screenshot(self, name: str) -> str: ... @dataclass(frozen=True) class ContractGoodsLine: operation_code: str goods_code: str goods_name: str approval_no: str spec: str unit: str quantity: float mdm_code: str = "" unit_price: str = "" @dataclass(frozen=True) class ContractFlowContext: contract_url: str username: str password: str supplier_name: str order_date: str source_type: str template_type: str = "FORMAT" template_name: str = "" customer_code: str = "" customer_name: str = "" contract_category: str = "" settlement_type: str = "付款" contract_background: str = "" contract_nature: str = "主合同" amount_currency: str = "人民币" amount_type: str = "合同标的金额" amount: str = "" contract_title: str = "" contract_end_date: str = "" is_major_contract: str = "否" is_prepayment: str = "否" belong_year: str = "" payment_method: str = "账期支付" payment_days: str = "30" contract_status: str = "正式合同" contains_blood_products: str = "否" submit_mode: str = "submit" goods_lines: list[ContractGoodsLine] = field(default_factory=list) merged_pdf_path: str | None = None contract_document_path: str = "" order_no_list: list[str] = field(default_factory=list) _ACCEPT = AcceptanceReport(flow="CONTRACT_CREATE") def _expected_title(ctx: ContractFlowContext) -> str: title = (ctx.contract_title or "").strip() if title: return title who = (ctx.customer_name or ctx.supplier_name or "").strip() cate = (ctx.contract_category or "采购协议").strip() date = (ctx.order_date or "").replace("-", "") return f"{who}{cate}{date}" def run_contract_create_flow(browser: BrowserAdapter, ctx: ContractFlowContext) -> str: _ACCEPT.reset() try: open_contract_system(browser, ctx) login_contract_system(browser, ctx) enter_contract_create(browser) select_contract_template(browser, ctx) fill_contract_signatory_info(browser, ctx) fill_contract_basic_info(browser, ctx) fill_contract_amount_and_dates(browser, ctx) fill_contract_goods_lines(browser, ctx) fill_contract_payment_info(browser, ctx) upload_contract_document_if_needed(browser, ctx) if not submit_contract_for_approval(browser, ctx): 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 "" contract_no = read_contract_number(browser, ctx) _ACCEPT.note( "S13", "读合同号", {"contractNo": contract_no or "(none)", "via": "我的合同/申请时间/第一行"}, ok=_usable_contract_no(contract_no), ) return contract_no finally: print(_ACCEPT.summary_text(), end="", flush=True) _ACCEPT.dump() # 合同起草页（模板选择）直达 URL，绕过 SPA 菜单/iframe 复杂性 DRAFT_PAGE_URL = ( "http://contract-test.gzmpc.com:8080/km/agreement/km_agreement_apply/draft/" "draft.jsp?isSimpleCategory=false" "&mainModelName=com.landray.kmss.km.agreement.model.KmAgreementApply" "&modelName=com.landray.kmss.km.agreement.model.KmAgreementTmpl" ) # 非格式-采购协议（2025集团版）模板 ID（测试环境固定值） # 直接跳转表单页，绕过「点模板」这一依赖前端委托事件、Selenium 点击无效的步骤 TEMPLATE_ID = "198d10b04325279b8ffeb8b4c6493b5a" FORM_PAGE_URL = ( "http://contract-test.gzmpc.com:8080/km/agreement/km_agreement_apply/" "kmAgreementApply.do?method=add" f"&fdTemplateId={TEMPLATE_ID}" f"&.fdTemplate={TEMPLATE_ID}" f"&i.docTemplate={TEMPLATE_ID}" "&s_css=default" ) VIEW_PAGE_URL = ( "http://contract-test.gzmpc.com:8080/km/agreement/km_agreement_apply/" "kmAgreementApply.do?method=view&fdId={fd_id}" ) # Landray/LUI 表单字段的真实 name F = { "customer_code": 'input[name="extendDataFormInfo.value(fd_customeropcode)"]', "customer_name": 'input[name="extendDataFormInfo.value(fd_customername)"]', "contract_category": 'input[name="extendDataFormInfo.value(fd_contract_cate_type)"]', "settlement_type": 'select#fdSettype', "contract_background": 'textarea[name="extendDataFormInfo.value(fd_desc_info)"]', "nature_main": 'input[name="fdType"][value="00"]', "amount": 'input[name="fdAccount"]', "sign_date": 'input[name="fdSigndate"]', "begin_date": 'input[name="fdEffedate"]', "end_date": 'input[name="fdEnddate"]', "is_major_contract": 'input[name="extendDataFormInfo.value(fd_is_major_contract)"]', "is_prepayment": 'input[name="extendDataFormInfo.value(fd_advance_amout)"]', "belong_year": 'input[name="extendDataFormInfo.value(fd_contract_year_text)"]', "payment_method": 'input[name="extendDataFormInfo.value(fd_paytype)"]', "payment_days": 'input[name="extendDataFormInfo.value(fd_pay_time_limit)"]', "contract_status": 'input[name="extendDataFormInfo.value(fd_suconstatus)"]', "contains_blood": 'input[name="extendDataFormInfo.value(fd_3d1ce3953f1162)"]', "doc_subject": 'input[name="docSubject"]', } # xform fancy-select 的 property 名（select_xform 用） X = { "contract_category": "extendDataFormInfo.value(fd_contract_cate_type)", "cover_template": "extendDataFormInfo.value(fd_contract_cover_type)", "amount_type": "extendDataFormInfo.value(fd_all_amount_type)", "is_major_contract": "extendDataFormInfo.value(fd_is_major_contract)", "is_prepayment": "extendDataFormInfo.value(fd_advance_amout)", "payment_method": "extendDataFormInfo.value(fd_paytype)", "contract_status": "extendDataFormInfo.value(fd_suconstatus)", "contains_blood": "extendDataFormInfo.value(fd_3d1ce3953f1162)", } _YES_NO = { "1": "是", "2": "否", "Y": "是", "N": "否", "YES": "是", "NO": "否", "TRUE": "是", "FALSE": "否", } _CATEGORY_ALIASES = { "采购协议": "其他协议（采购）", "采购": "其他协议（采购）", } _PAYMENT_ALIASES = { "账期支付": "电汇", "账期": "电汇", "承兑": "银行承兑汇票", } def _yes_no(value: str) -> str: raw = (value or "").strip() return _YES_NO.get(raw.upper(), raw) def _category_option(value: str) -> str: raw = (value or "").strip() return _CATEGORY_ALIASES.get(raw, raw) def _payment_option(value: str) -> str: raw = (value or "").strip() return _PAYMENT_ALIASES.get(raw, raw) # 货品明细行的字段后缀；行索引从 0 开始（fd_detail1.0.* 为第 1 行） GOODS_FIELD_SUFFIXES = { "goods_code": "fd_opcode", "goods_name": "fd_goodsname", "goods_mdm": "fd_mdmCode", "goods_approval": "fd_registdocno", "goods_spec": "fd_goodstype", "goods_unit": "fd_goodsunit", "goods_qty": "fd_goodsqty", "goods_price": "fd_unitprice", } ADD_ROW_SELECTORS = [ "text=添加行", "text=新增明细", "text=新增行", "text=新增", "button:contains(添加行)", ] # 货品选择窗口：点「选择商品」弹出 EopBasedataMaterial 列表，按操作码/名称搜索回填。 # lui-id-* 每次加载会变，不要写死；用 title / 按钮文本定位。 GOODS_PICKER = { "trigger": [ 'div[title="选择商品"]', 'div.lui-component.lui_widget_btn_txt', "text=选择商品", ], "frame": [ 'iframe[src*="EopBasedataMaterial"]', 'iframe[src*="dialog_fdMaterial"]', 'iframe[src*="common-template"]', ], "search_input": ["input.search-box-text", "input[placeholder*='关键字']"], "search_button": [ "input.search-box-btn", "a.search-box-btn", "div.search-box-btn", "span.search-box-btn", "input[value='搜索']", "input[value='查询']", "[class*='search-box'] [class*='btn']", ], "result_checkbox": ['input[name="List_Selected"]'], } # 客商选择窗口（点击客户名称旁「选择」按钮弹出）。若真实页面改版，仅需调整此处。 CUSTOMER_PICKER = { "trigger": [ 'div.inputselectsgl[onclick*="fdCustomerEdit"]', 'div[class*="inputselect"][onclick*="fdCustomer"]', ], "frame": [ 'iframe[src*="common-template"]', 'iframe[src*="KmRelativeMain"]', 'iframe[src*="relative"]', ], "search_input": ["input.search-box-text"], "search_button": ["input.search-box-btn"], "result_checkbox": ['input[name="List_Selected"]'], "confirm_button": ["div.btn.ok", "div.lui_toolbar_btn_def"], } def open_contract_system(browser: BrowserAdapter, ctx: ContractFlowContext) -> None: browser.open_url(ctx.contract_url) _ACCEPT.note("S02", "打开合同系统", {"url": ctx.contract_url}, ok=True) _PORTAL_MARKERS = ("切换门户", "待办", "待阅", "合同管理", "合同起草") def _is_logged_in(browser: BrowserAdapter) -> bool: body = browser.read_body_text() or "" return any(marker in body for marker in _PORTAL_MARKERS) def login_contract_system(browser: BrowserAdapter, ctx: ContractFlowContext) -> None: try: if _is_logged_in(browser): evidence_shot(browser, "contract-already-logged-in") _ACCEPT.note("S03", "登录", {"user": ctx.username, "via": "already"}, ok=True) return browser.fill_first(['input[name="j_username"]', 'input[placeholder="用户名"]'], ctx.username) browser.fill_first(['input[name="j_password"]', 'input[placeholder="密码"]'], ctx.password) browser.click_first(['a[class*="login_button"]', 'a:contains("登录")', 'input[name="btn_submit"]']) deadline = time.time() + 20 while time.time() < deadline: if _is_logged_in(browser): evidence_shot(browser, "contract-login-ok") _ACCEPT.note("S03", "登录", {"user": ctx.username, "portal": "yes"}, ok=True) return time.sleep(0.5) raise TimeoutError("登录后未进入门户（未看到切换门户/待办/待阅）") except Exception as exc: evidence_shot(browser, "contract-login-failed", failed=True) _ACCEPT.note("S03", "登录", {"user": ctx.username, "error": exc}, ok=False, raise_error=False) raise LoginFailed(f"合同系统登录失败：{exc}") from exc def enter_contract_create(browser: BrowserAdapter) -> None: # 登录后直接导航到起草表单页，跳过模板选择（前端委托事件 Selenium 点击无效） browser.open_url(FORM_PAGE_URL) browser.wait_for_text("客户编码", timeout_seconds=20) on_form = "客户编码" in (browser.read_body_text() or "") _ACCEPT.check( "S04", "进入起草", [("formMarker", "客户编码" if on_form else (browser.read_body_text() or "")[:40], "客户编码")], browser=browser, screenshot_on_fail="contract-login-ok", ) def select_contract_template(browser: BrowserAdapter, ctx: ContractFlowContext) -> None: # 已通过 URL 直接指定模板，本步骤仅兜底：若尚未进入表单则再次导航 if "客户编码" not in (browser.read_body_text() or ""): browser.open_url(FORM_PAGE_URL) browser.wait_for_text("客户编码", timeout_seconds=20) on_form = "客户编码" in (browser.read_body_text() or "") _ACCEPT.check( "S05", "选择模板", [ ("formMarker", "客户编码" if on_form else "", "客户编码"), ("templateType", contract_version_label(ctx), contract_version_label(ctx)), ], browser=browser, screenshot_on_fail="contract-login-ok", ) def fill_contract_signatory_info(browser: BrowserAdapter, ctx: ContractFlowContext) -> None: expected_name = (ctx.customer_name or ctx.supplier_name).strip() if not ctx.customer_code: raise SupplierContractDisabled("客商操作码(customerCode)为空，无法加载客商") from contract_draft import is_placeholder_customer_code if is_placeholder_customer_code(ctx.customer_code): raise SupplierContractDisabled( f"customerCode「{ctx.customer_code}」是占位符，请换成合同系统真实客商操作码" "（测试环境：DW0100046361 / 重庆亿昊药业有限公司）" ) # 点击客户名称字段旁的「选择」按钮打开客商选择窗口，按编码搜索并带回客商 browser.click_first(CUSTOMER_PICKER["trigger"]) _pick_customer_from_dialog(browser, ctx.customer_code) loaded_name = _wait_customer_name(browser, timeout=8) if not loaded_name: evidence_shot(browser, "contract-customer-load-failed", failed=True) raise SupplierContractDisabled( f"客商加载失败，请确认输入的客商操作码是否有误：{ctx.customer_code}" ) if expected_name and loaded_name != expected_name: evidence_shot(browser, "contract-customer-name-mismatch", failed=True) raise SupplierContractDisabled( f"客商名称校验不通过：期望「{expected_name}」，实际「{loaded_name}」" ) evidence_shot(browser, "contract-signatory-info-filled") loaded_code = _try_read(browser, [F["customer_code"]]) _ACCEPT.check( "S06", "签约方", [ ("customerName", loaded_name, expected_name), ("customerCode", loaded_code, ctx.customer_code), ], browser=browser, screenshot_on_fail="contract-customer-name-mismatch", ) def _wait_customer_name(browser: BrowserAdapter, timeout: int) -> str: deadline = time.time() + timeout while time.time() < deadline: body = browser.read_body_text() if "自动加载客商失败" in body or "客商操作码是否有误" in body: return "" name = _try_read(browser, [F["customer_name"]]) if name and name not in ("==请选择==", "请选择"): return name time.sleep(0.5) return "" def _pick_customer_from_dialog(browser: BrowserAdapter, code: str) -> None: # 客商选择窗口是一个 iframe 弹窗：切入后搜索编码 → 按编码点对应行 → 切回主文档。 browser.wait_for_element(CUSTOMER_PICKER["frame"], timeout_seconds=10) browser.switch_to_frame(CUSTOMER_PICKER["frame"]) try: browser.fill_first(CUSTOMER_PICKER["search_input"], code) browser.click_first(CUSTOMER_PICKER["search_button"]) time.sleep(1.5) browser.wait_for_element(CUSTOMER_PICKER["result_checkbox"], timeout_seconds=10) picked = browser.execute_script( """ const code = String(arguments[0] || ''); const rows = Array.from(document.querySelectorAll('tr')); for (const row of rows) { const text = (row.innerText || ''); if (text.indexOf(code) < 0) continue; const box = row.querySelector('input[name="List_Selected"]'); if (box) { box.click(); return {ok: true, text: text.replace(/\\s+/g, ' ').slice(0, 80)}; } } return {ok: false}; """, code, ) if not (isinstance(picked, dict) and picked.get("ok")): browser.click_first(CUSTOMER_PICKER["result_checkbox"]) time.sleep(1) finally: browser.switch_to_default_content() def fill_contract_basic_info(browser: BrowserAdapter, ctx: ContractFlowContext) -> None: if ctx.contract_category: browser.select_xform(X["contract_category"], _category_option(ctx.contract_category)) _select_cover_template_if_needed(browser) if ctx.settlement_type: _select_settlement_type(browser, ctx.settlement_type) if ctx.contract_background: browser.fill_first([F["contract_background"]], ctx.contract_background) # 合同性质默认主合同（页面已勾选），仅在需要从合同时切换 if ctx.contract_nature and ctx.contract_nature != "主合同": browser.click_first(['input[name="fdType"][value="01"]']) _fill_contract_title(browser, ctx) evidence_shot(browser, "contract-basic-info-filled") _ACCEPT.check( "S07", "基础信息", [ ("category", _read_xform_label(browser, X["contract_category"]), _category_option(ctx.contract_category) if ctx.contract_category else ""), ("settlement", _try_read(browser, [F["settlement_type"]]), ctx.settlement_type), ("background", _try_read(browser, [F["contract_background"]]), ctx.contract_background), ("title", _try_read(browser, [F["doc_subject"]]), _expected_title(ctx)), ], browser=browser, screenshot_on_fail="contract-basic-info-filled", ) def _fill_contract_title(browser: BrowserAdapter, ctx: ContractFlowContext) -> None: title = (ctx.contract_title or "").strip() if not title: who = (ctx.customer_name or ctx.supplier_name or "").strip() cate = (ctx.contract_category or "采购协议").strip() date = (ctx.order_date or "").replace("-", "") title = f"{who}{cate}{date}" if title: browser.fill_first([F["doc_subject"]], title) def _select_cover_template_if_needed(browser: BrowserAdapter) -> None: # 常用模板/已审核版本：优先选「否」，避免连带「原合同申请编号」变必填 current = _read_xform_label(browser, X["cover_template"]) if current and current not in ("==请选择==", "请选择"): return picked = browser.execute_script( """ const name = arguments[0]; const flag = document.querySelector('xformflag[property="' + name + '"]'); if (!flag) return ''; const opts = Array.from(flag.querySelectorAll('.fs-option')).map((o) => ({ value: (o.getAttribute('data-value') || '').trim(), text: ((o.querySelector('.fs-option-label') || o).textContent || '').trim(), })).filter((o) => o.value || (o.text && o.text.indexOf('请选择') < 0)); const preferNo = opts.find((o) => o.text === '否' || o.value === '2' || o.value === '0'); const hit = preferNo || opts[0]; return hit ? (hit.text || hit.value) : ''; """, X["cover_template"], ) if picked: browser.select_xform(X["cover_template"], picked) def _read_xform_label(browser: BrowserAdapter, name: str) -> str: try: return ( browser.execute_script( """ const flag = document.querySelector('xformflag[property="' + arguments[0] + '"]'); return flag ? ((flag.querySelector('.fs-label') || {}).textContent || '').trim() : ''; """, name, ) or "" ).strip() except Exception: return "" def _select_settlement_type(browser: BrowserAdapter, value: str) -> None: browser.select_first([F["settlement_type"]], value) selected = _try_read(browser, [F["settlement_type"]]) if not selected or selected in ("==请选择==", "请选择"): # LUI 自定义下拉兜底：点击展开后按可见文本选择 browser.click_first([F["settlement_type"]]) browser.click_first([f"//option[contains(text(),'{value}')]", f"text={value}"]) selected = _try_read(browser, [F["settlement_type"]]) if not selected or selected in ("==请选择==", "请选择"): raise NodeError(f"结算方式下拉未选中，仍停留在「==请选择==」，期望「{value}」") def fill_contract_amount_and_dates(browser: BrowserAdapter, ctx: ContractFlowContext) -> None: amount_type = (ctx.amount_type or "合同标的金额").strip() if amount_type: browser.select_xform(X["amount_type"], amount_type) if ctx.amount: browser.fill_first([F["amount"]], ctx.amount) browser.fill_first([F["sign_date"]], ctx.order_date) browser.fill_first([F["begin_date"]], ctx.order_date) if ctx.contract_end_date: browser.fill_first([F["end_date"]], ctx.contract_end_date) evidence_shot(browser, "contract-amount-and-dates-filled") _ACCEPT.check( "S08", "金额日期", [ ("amount", _try_read(browser, [F["amount"]]), ctx.amount), ("signDate", _try_read(browser, [F["sign_date"]]), ctx.order_date), ("beginDate", _try_read(browser, [F["begin_date"]]), ctx.order_date), ("endDate", _try_read(browser, [F["end_date"]]), ctx.contract_end_date), ], browser=browser, screenshot_on_fail="contract-amount-and-dates-filled", ) def fill_contract_goods_lines(browser: BrowserAdapter, ctx: ContractFlowContext) -> None: if not ctx.goods_lines: evidence_shot(browser, "contract-goods-lines-filled") _ACCEPT.note("S09", "货品明细", {"lines": 0}, ok=True) return for line in ctx.goods_lines: _pick_goods_from_dialog(browser, line) _fill_goods_required_qty_price(browser, line) _delete_empty_goods_rows(browser) evidence_shot(browser, "contract-goods-lines-filled") for index, line in enumerate(ctx.goods_lines): row = _find_goods_row(browser, line) code = _try_read(browser, [_goods_selector("goods_code", row)]) if row >= 0 else "" qty = _try_read(browser, [_goods_selector("goods_qty", row)]) if row >= 0 else "" price = _try_read(browser, [_goods_selector("goods_price", row)]) if row >= 0 else "" _ACCEPT.check( "S09", f"货品明细#{index + 1}", [ ("goodsCode", code, line.goods_code), ("qty", qty, line.quantity), ("price", price, line.unit_price or "1.00"), ], browser=browser, screenshot_on_fail="contract-goods-pick-failed", ) def _pick_goods_from_dialog(browser: BrowserAdapter, line: ContractGoodsLine) -> None: keyword = (line.goods_code or line.goods_name or "").strip() if not keyword: raise NodeError("货品缺少 goodsCode/goodsName，无法在「选择商品」中搜索") browser.click_first(GOODS_PICKER["trigger"]) browser.wait_for_element(GOODS_PICKER["frame"], timeout_seconds=15) browser.switch_to_frame(GOODS_PICKER["frame"]) try: _search_goods_in_dialog(browser, keyword) if not _dialog_row_has_keyword(browser, keyword) and line.goods_name: _search_goods_in_dialog(browser, line.goods_name.strip()) picked = _select_dialog_row_by_keyword(browser, keyword) if not picked and line.goods_name: picked = _select_dialog_row_by_keyword(browser, line.goods_name.strip()) if not picked: browser.wait_for_element(GOODS_PICKER["result_checkbox"], timeout_seconds=10) browser.click_first(GOODS_PICKER["result_checkbox"]) time.sleep(0.5) evidence_shot(browser, "contract-goods-picker") except Exception as exc: browser.switch_to_default_content() _scroll_to_goods_table(browser) evidence_shot(browser, "contract-goods-pick-failed", failed=True) raise NodeError(f"选择商品失败，未找到「{keyword}」：{exc}") from exc browser.switch_to_default_content() _click_dialog_ok(browser) deadline = time.time() + 12 while time.time() < deadline: if _find_goods_row(browser, line) >= 0: return _click_dialog_ok(browser) time.sleep(0.6) _scroll_to_goods_table(browser) evidence_shot(browser, "contract-goods-pick-failed", failed=True) raise NodeError(f"选择商品后未在明细表找到「{keyword}」") def _search_goods_in_dialog(browser: BrowserAdapter, keyword: str) -> None: browser.fill_first(GOODS_PICKER["search_input"], keyword) browser.execute_script( """ const el = document.querySelector('input.search-box-text, input[placeholder*="关键字"]'); if (!el) return false; el.dispatchEvent(new KeyboardEvent('keydown', {key:'Enter', code:'Enter', keyCode:13, which:13, bubbles:true})); el.dispatchEvent(new KeyboardEvent('keyup', {key:'Enter', code:'Enter', keyCode:13, which:13, bubbles:true})); if (el.form) { el.form.dispatchEvent(new Event('submit', {bubbles:true, cancelable:true})); } return true; """ ) try: browser.click_native(GOODS_PICKER["search_button"]) except Exception: try: browser.click_first(GOODS_PICKER["search_button"]) except Exception: pass deadline = time.time() + 6 while time.time() < deadline: if _dialog_row_has_keyword(browser, keyword): return time.sleep(0.4) def _dialog_row_has_keyword(browser: BrowserAdapter, keyword: str) -> bool: result = browser.execute_script( """ const keyword = String(arguments[0] || '').replace(/\\s+/g, ''); if (!keyword) return false; return Array.from(document.querySelectorAll('tr')).some((row) => { const text = (row.innerText || '').replace(/\\s+/g, ''); return text.indexOf(keyword) >= 0 && !!row.querySelector( 'input[name="List_Selected"], input[type="checkbox"], input[type="radio"]' ); }); """, keyword, ) return result is True def _select_dialog_row_by_keyword(browser: BrowserAdapter, keyword: str) -> bool: picked = browser.execute_script( """ const keyword = String(arguments[0] || '').replace(/\\s+/g, ''); if (!keyword) return {ok: false}; const rows = Array.from(document.querySelectorAll('tr')); for (const row of rows) { const text = (row.innerText || '').replace(/\\s+/g, ''); if (text.indexOf(keyword) < 0) continue; const box = row.querySelector( 'input[name="List_Selected"], input[type="checkbox"], input[type="radio"]' ); if (box) { if (!box.checked) { box.click(); } return {ok: true, text: (row.innerText || '').replace(/\\s+/g, ' ').slice(0, 80)}; } row.click(); return {ok: true, text: 'row-click'}; } return {ok: false, rows: rows.length}; """, keyword, ) return isinstance(picked, dict) and bool(picked.get("ok")) def _scroll_to_goods_table(browser: BrowserAdapter) -> None: try: browser.execute_script( """ const hit = document.querySelector('div[title="选择商品"]') || Array.from(document.querySelectorAll('div, a, span, button')) .find((el) => (el.textContent || '').trim() === '选择商品'); if (hit) hit.scrollIntoView({block: 'center'}); """ ) time.sleep(0.3) except Exception: pass def _click_dialog_ok(browser: BrowserAdapter) -> None: try: browser.click_native( [ "//*[contains(@class,'lui_dialog')]//*[normalize-space()='确定']", ".lui_dialog_main div.lui_widget_btn_txt", "text=确定", ] ) return except Exception: pass result = browser.execute_script( """ const root = document.querySelector('.lui_dialog_main') || document; const btns = Array.from(root.querySelectorAll('div, a, span, button')).filter((el) => { const text = (el.textContent || '').trim(); return text === '确定' && el.children.length === 0; }); if (!btns.length) return {ok: false}; btns[btns.length - 1].click(); return {ok: true}; """ ) if isinstance(result, dict) and result.get("ok"): return browser.click_first([".lui_dialog_main div.lui_widget_btn_txt", "text=确定"]) def _find_goods_row(browser: BrowserAdapter, line: ContractGoodsLine) -> int: row = browser.execute_script( """ const keys = [String(arguments[0] || '').trim(), String(arguments[1] || '').trim()].filter(Boolean); for (let i = 0; i < 30; i++) { const fields = ['fd_opcode', 'fd_goodsname', 'fd_mdmCode', 'fd_registdocno']; const values = fields.map((suffix) => { const el = document.querySelector( 'input[name="extendDataFormInfo.value(fd_detail1.' + i + '.' + suffix + ')"]' ); return el ? String(el.value || '').trim() : ''; }).filter(Boolean); if (!values.length) continue; const joined = values.join(' '); for (const keyword of keys) { if (values.some((v) => v === keyword || v.indexOf(keyword) >= 0) || joined.indexOf(keyword) >= 0) { return i; } } } return -1; """, line.goods_code, line.goods_name, ) try: return int(row) except (TypeError, ValueError): return -1 def _fill_goods_required_qty_price(browser: BrowserAdapter, line: ContractGoodsLine) -> None: row = _find_goods_row(browser, line) if row < 0: raise NodeError(f"选择商品后未在明细表找到「{line.goods_code or line.goods_name}」") qty = str(line.quantity if line.quantity else "") if qty: browser.fill_first([_goods_selector("goods_qty", row)], qty) price = (line.unit_price or "").strip() or "1.00" browser.fill_first([_goods_selector("goods_price", row)], price) def _delete_empty_goods_rows(browser: BrowserAdapter) -> None: marked = browser.execute_script( """ const marked = []; for (let i = 0; i < 30; i++) { const code = document.querySelector('input[name="extendDataFormInfo.value(fd_detail1.' + i + '.fd_opcode)"]'); const name = document.querySelector('input[name="extendDataFormInfo.value(fd_detail1.' + i + '.fd_goodsname)"]'); if (!code && !name) continue; const empty = !(code && String(code.value || '').trim()) && !(name && String(name.value || '').trim()); if (!empty) continue; const row = (code || name).closest('tr'); const box = row ? row.querySelector('input[type="checkbox"]') : null; if (box && !box.checked) { box.click(); } marked.push(i); } return marked; """ ) if marked: try: browser.click_first(["text=删除行", "a:contains(删除行)"]) time.sleep(0.5) except Exception: pass def _goods_selector(field_key: str, row_index: int) -> str: suffix = GOODS_FIELD_SUFFIXES[field_key] return f'input[name="extendDataFormInfo.value(fd_detail1.{row_index}.{suffix})"]' def fill_contract_payment_info(browser: BrowserAdapter, ctx: ContractFlowContext) -> None: browser.select_xform(X["is_major_contract"], _yes_no(ctx.is_major_contract)) browser.select_xform(X["is_prepayment"], _yes_no(ctx.is_prepayment)) if ctx.belong_year: browser.fill_first([F["belong_year"]], ctx.belong_year) if ctx.payment_method: browser.select_xform(X["payment_method"], _payment_option(ctx.payment_method)) if ctx.payment_days: browser.fill_first([F["payment_days"]], ctx.payment_days) if ctx.contract_status: browser.select_xform(X["contract_status"], ctx.contract_status) if ctx.contains_blood_products: browser.select_xform(X["contains_blood"], _yes_no(ctx.contains_blood_products)) _fill_concession_defaults(browser) evidence_shot(browser, "contract-payment-filled") _ACCEPT.check( "S10", "付款信息", [ ("major", _read_xform_label(browser, X["is_major_contract"]), _yes_no(ctx.is_major_contract)), ("prepay", _read_xform_label(browser, X["is_prepayment"]), _yes_no(ctx.is_prepayment)), ("year", _try_read(browser, [F["belong_year"]]), ctx.belong_year), ("payMethod", _read_xform_label(browser, X["payment_method"]), _payment_option(ctx.payment_method) if ctx.payment_method else ""), ("payDays", _try_read(browser, [F["payment_days"]]), ctx.payment_days), ("status", _read_xform_label(browser, X["contract_status"]), ctx.contract_status), ], browser=browser, screenshot_on_fail="contract-payment-filled", ) def _fill_concession_defaults(browser: BrowserAdapter) -> None: # 让步情形等是/否单选未填会挡住提交；默认点「否」 browser.execute_script( """ const clicked = []; document.querySelectorAll('tr').forEach((tr) => { const radios = Array.from(tr.querySelectorAll('input[type="radio"]')); if (radios.length < 2) return; const labeled = radios.map((r) => ({ el: r, text: ((r.closest('label') || r.parentElement || {}).textContent || '') .replace(/\\s+/g, '').trim(), })); if (!labeled.some((x) => x.text === '是') || !labeled.some((x) => x.text === '否')) return; if (radios.some((r) => r.checked)) return; const no = labeled.find((x) => x.text === '否'); if (no) { no.el.click(); clicked.push(radios[0].name); } }); return clicked; """ ) def upload_contract_document_if_needed(browser: BrowserAdapter, ctx: ContractFlowContext) -> None: path = (ctx.contract_document_path or "").strip() if not path and ctx.source_type == "YYC": path = (ctx.merged_pdf_path or "").strip() if not path: if ctx.submit_mode == "dry_run": evidence_shot(browser, "contract-pdf-upload-skipped") _ACCEPT.note("S11", "上传正文", {"skipped": "dry_run_no_file"}, ok=True) return if ctx.source_type == "YYC": raise PdfMergeUnsupported("缺少合同正文/平台合同文件，无法上传") _ACCEPT.note("S11", "上传正文", {"skipped": ctx.source_type or "no_file"}, ok=True) return try: _upload_contract_body(browser, path) except Exception as exc: _ACCEPT.note("S11", "上传正文", {"attached": "no", "error": exc}, ok=False, raise_error=False) raise evidence_shot(browser, "contract-body-uploaded") attached = _contract_body_attached(browser, path) _ACCEPT.check( "S11", "上传正文", [("attached", "yes" if attached else "no", "yes"), ("file", Path(path).name, Path(path).name)], browser=browser, screenshot_on_fail="contract-body-uploaded", ) def _upload_contract_body(browser: BrowserAdapter, file_path: str) -> None: from pathlib import Path if not Path(file_path).is_file(): raise NodeError(f"合同正文文件不存在：{file_path}") _close_blocking_dialogs(browser) _reveal_contract_body_section(browser) evidence_shot(browser, "contract-body-tab") if _contract_body_attached(browser, file_path): evidence_shot(browser, "contract-body-uploaded") return _click_add_contract_body(browser) time.sleep(0.8) try: browser.click_native(["text=打开本地初稿", "text=本地初稿", "text=上传本地文件"]) time.sleep(0.8) except Exception: pass browser.upload_file_first( [ 'input[name="_apply_filemainOnline"]', 'input[name="__landray_filemainOnline"]', 'input[name*="filemainOnline"]', 'input[name*="fdMainOnline"]', '.lui_dialog input[type="file"]', 'input[type="file"]', ], file_path, ) deadline = time.time() + 12 while time.time() < deadline: if _contract_body_attached(browser, file_path) or _on_body_editor_tab(browser): evidence_shot(browser, "contract-body-uploaded") return time.sleep(0.6) evidence_shot(browser, "contract-body-uploaded", failed=True) raise NodeError("合同正文未挂上：页面仍是「+添加合同」，提交会被系统拒绝") def _reveal_contract_body_section(browser: BrowserAdapter) -> None: browser.execute_script( """ const labelOf = (el) => (el.textContent || '').replace(/\\s+/g, '').trim(); const addBtn = Array.from(document.querySelectorAll('a, span, div, td')).find((el) => { const text = labelOf(el); return text === '+添加合同' || text === '添加合同'; }); if (addBtn) { addBtn.scrollIntoView({block: 'center'}); return {ok: true, via: 'add'}; } const hits = Array.from(document.querySelectorAll('div, li, span, a')).filter((el) => labelOf(el) === '合同正文'); for (const el of hits) { const group = (el.parentElement && el.parentElement.parentElement) || el; if (labelOf(group).indexOf('签约方信息') >= 0) { (el.closest('li') || el).click(); return {ok: true, via: 'tabbar'}; } } return {ok: false}; """ ) time.sleep(0.6) def _click_add_contract_body(browser: BrowserAdapter) -> None: clicked = browser.execute_script( """ const labelOf = (el) => (el.textContent || '').replace(/\\s+/g, '').trim(); const hit = Array.from(document.querySelectorAll('a, span, div, td, button')).find((el) => { const text = labelOf(el); return (text === '+添加合同' || text === '添加合同') && el.children.length <= 1; }); if (!hit) return {ok: false}; hit.scrollIntoView({block: 'center'}); hit.click(); return {ok: true}; """ ) if isinstance(clicked, dict) and clicked.get("ok"): return browser.click_native(["text=+添加合同", "text=添加合同"]) def _contract_body_attached(browser: BrowserAdapter, file_path: str) -> bool: from pathlib import Path stem = Path(file_path).name result = browser.execute_script( """ const wanted = String(arguments[0] || ''); const wantedStem = wanted.replace(/\\.[^.]+$/, ''); const fileInputs = Array.from(document.querySelectorAll('input[type="file"]')); if (fileInputs.some((el) => el.files && el.files.length > 0)) { return {ok: true, via: 'input'}; } const frames = Array.from(document.querySelectorAll('iframe, object, embed')); const office = frames.find((el) => { const hint = ((el.src || '') + ' ' + (el.id || '') + ' ' + (el.className || '')).toLowerCase(); const box = el.getBoundingClientRect(); return box.width > 80 && box.height > 80 && /office|ntko|wps|kingsoft|weboffice|filemain|online|pdf/.test(hint); }); if (office) return {ok: true, via: 'iframe'}; const rows = Array.from(document.querySelectorAll('tr')); for (const row of rows) { const cells = Array.from(row.querySelectorAll('td, th')); if (!cells.length) continue; const head = (cells[0].innerText || '').replace(/\\s+/g, '').trim(); if (head !== '合同正文') continue; const body = (row.innerText || '').replace(/\\s+/g, ' ').trim(); if (wanted && body.indexOf(wanted) >= 0) return {ok: true, via: 'name'}; if (wantedStem && body.indexOf(wantedStem) >= 0) return {ok: true, via: 'stem'}; if (/\\.(pdf|doc|docx|wps)/i.test(body)) return {ok: true, via: 'ext'}; if (body.indexOf('添加合同') >= 0 && body.indexOf('打开本地初稿') < 0) return {ok: false, via: 'empty'}; } const page = (document.body.innerText || ''); if (wanted && page.indexOf(wanted) >= 0) return {ok: true, via: 'page'}; return {ok: false, via: 'missing'}; """, stem, ) return isinstance(result, dict) and bool(result.get("ok")) def _on_body_editor_tab(browser: BrowserAdapter) -> bool: """底部「合同正文」在线编辑页：文件进编辑器后不一定回显文件名。""" result = browser.execute_script( """ const page = (document.body.innerText || '').replace(/\\s+/g, ''); return page.indexOf('打开本地初稿') >= 0 && page.indexOf('添加合同') < 0; """ ) return result is True 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; } } return ''; """ ) time.sleep(0.4) def submit_contract_for_approval(browser: BrowserAdapter, ctx: ContractFlowContext) -> 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 = _read_document_id(browser) try: browser.click_native( [ "//*[normalize-space()='暂存']/following::*[normalize-space()='提交'][1]", "text=提交", ] ) except Exception: _click_top_submit(browser) evidence_shot(browser, "contract-after-submit-click") _confirm_submit_dialogs(browser) deadline = time.time() + 5 while time.time() < deadline: if _submit_succeeded(browser): evidence_shot(browser, "contract-submitted") print("提交成功，回我的合同取编号。", flush=True) fd_id = fd_id or _read_document_id(browser) _ACCEPT.note("S12", "提交", {"submitted": "yes"}, ok=True) return True _confirm_submit_dialogs(browser) time.sleep(0.5) evidence_shot(browser, "contract-submitted", failed=True) print("提交后仍在起草页，浏览器保持打开，请手工检查。", flush=True) _ACCEPT.note("S12", "提交", {"submitted": "no", "page": "draft"}, ok=False, raise_error=False) return False def _submit_succeeded(browser: BrowserAdapter) -> bool: body = browser.read_body_text() or "" 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 def _read_document_id(browser: BrowserAdapter) -> str: 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 ''; """ ) return str(value or "").strip() def _open_submitted_view(browser: BrowserAdapter, fd_id: str) -> None: doc_id = (fd_id or "").strip() or _read_document_id(browser) if not doc_id: print("成功页没有 fdId，无法自动打开详情", flush=True) return url = VIEW_PAGE_URL.format(fd_id=doc_id) print("打开合同详情：", url, flush=True) browser.open_url(url) time.sleep(1.5) def _click_top_submit(browser: BrowserAdapter) -> None: clicked = browser.execute_script( """ const toolbar = document.querySelector('.lui_toolbar_frame, .lui_toolbar, [class*="lui_toolbar"]'); const root = toolbar || document; const hits = Array.from(root.querySelectorAll('div, a, button, span')).filter((el) => { return (el.textContent || '').trim() === '提交' && el.children.length === 0; }); if (!hits.length) return {ok: false}; const save = Array.from(document.querySelectorAll('div, a, span')).find((el) => { return (el.textContent || '').trim() === '暂存' && el.children.length === 0; }); const target = (save && hits.find((el) => { const a = save.getBoundingClientRect(); const b = el.getBoundingClientRect(); return Math.abs(a.top - b.top) < 40; })) || hits[0]; target.click(); return {ok: true}; """ ) if isinstance(clicked, dict) and clicked.get("ok"): return browser.click_first(["text=提交", "text=提交送审", "text=送审"]) def _confirm_submit_dialogs(browser: BrowserAdapter) -> None: browser.execute_script( """ 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; }); if (hit) { hit.click(); return label; } } return ''; """ ) time.sleep(0.8) def read_contract_number(browser: BrowserAdapter, ctx: ContractFlowContext | None = None) -> 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) def _read_contract_number_on_page(browser: BrowserAdapter) -> str: try: browser.switch_to_default_content() except Exception: pass try: value = (browser.read_first( [ 'input[name="fdNumber"]', 'input[name="contractNo"]', 'input[name="contractSystemNo"]', "label=合同编号", ], "contract_number", ) or "").strip() if _usable_contract_no(value): return value except Exception: pass scanned = browser.execute_script( """ const skip = ['提交时自动生成', '自动生成', '==请选择==']; const inputs = document.querySelectorAll( 'input[name="fdNumber"], input[name="contractNo"], input[name*="fdNumber"]' ); for (const el of inputs) { const v = (el.value || '').trim(); if (v && skip.indexOf(v) < 0) return v; } const nodes = Array.from(document.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 !== '合同编号：') continue; const next = nodes[i + 1]; const n = next ? ((next.innerText || next.value || '').replace(/\\s+/g, ' ').trim()) : ''; if (n && skip.indexOf(n) < 0) return n; } return ''; """ ) if isinstance(scanned, str) and _usable_contract_no(scanned.strip()): return scanned.strip() 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: try: browser.switch_to_default_content() except Exception: pass print("刷新回合同主界面：", home, flush=True) try: browser.execute_script( "try { window.top.location.href = arguments[0]; } catch (e) { location.href = arguments[0]; } return true;", home, ) except Exception: pass browser.open_url(home) if not _wait_text_in_frames(browser, "我的合同", timeout_seconds=15): print("主界面未出现「我的合同」，仍继续进入查询", flush=True) 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, "我的合同") _wait_text_in_frames(browser, "我起草的", timeout_seconds=10) _click_by_exact_text(browser, "我起草的") _wait_text_in_frames(browser, "申请时间", timeout_seconds=10) def _wait_text_in_frames(browser: BrowserAdapter, text: str, timeout_seconds: int = 15) -> bool: deadline = time.time() + timeout_seconds while time.time() < deadline: if _page_has_text(browser, text): return True time.sleep(0.3) return False def _page_has_text(browser: BrowserAdapter, text: str) -> bool: found = browser.execute_script( """ const wanted = String(arguments[0] || ''); function hasTextInFrames(win) { try { const body = (win.document && win.document.body && win.document.body.innerText) || ''; if (body.indexOf(wanted) >= 0) return true; } catch (e) {} for (let i = 0; i < win.frames.length; i++) { try { if (hasTextInFrames(win.frames[i])) return true; } catch (e) {} } return false; } return hasTextInFrames(window); """, text, ) return found is True def _click_by_exact_text(browser: BrowserAdapter, label: str) -> None: clicked = browser.execute_script( """ const label = String(arguments[0] || ''); function clickIn(doc) { const nodes = Array.from(doc.querySelectorAll('div, a, span, li, p, em, td, th, button')); const leaf = nodes.find((el) => (el.textContent || '').trim() === label && el.children.length === 0); if (leaf) { leaf.click(); return true; } const parent = nodes.find((el) => (el.textContent || '').trim() === label); if (parent) { parent.click(); return true; } return false; } function clickInFrames(win) { try { if (clickIn(win.document)) return true; } catch (e) {} for (let i = 0; i < win.frames.length; i++) { try { if (clickInFrames(win.frames[i])) return true; } catch (e) {} } return false; } return clickInFrames(window) ? label : ''; """, label, ) if clicked != label: try: browser.click_first([f"text={label}"]) except Exception: pass def _sort_my_contracts_by_apply_time(browser: BrowserAdapter) -> None: _click_by_exact_text(browser, "申请时间") browser.execute_script( """ const latestByApplyTime = true; function textOf(el) { return (el.textContent || '').replace(/\\s+/g, ' ').trim(); } function sortIn(doc) { const desc = Array.from(doc.querySelectorAll('[title], a, span, i, em')).find((el) => { const t = (el.getAttribute('title') || el.textContent || '').trim(); return t === '降序' || t === '从晚到早' || t === '倒序'; }); if (desc) { desc.click(); return 'desc'; } const bar = Array.from(doc.querySelectorAll('div, span')).find((el) => { const t = textOf(el); return t.indexOf('更多排序') >= 0 && t.length < 80; }); if (bar) { const downs = Array.from(bar.querySelectorAll('i, em, span, a')).filter((el) => { const t = (el.textContent || '').trim(); const cls = String(el.className || ''); return t === '∨' || t === 'v' || t === '↓' || t === '▼' || cls.indexOf('down') >= 0 || cls.indexOf('desc') >= 0; }); if (downs.length) { downs[downs.length - 1].click(); return 'arrow'; } } return ''; } function walk(win) { try { const hit = sortIn(win.document); if (hit) return hit; } catch (e) {} for (let i = 0; i < win.frames.length; i++) { try { const hit = walk(win.frames[i]); if (hit) return hit; } catch (e) {} } return ''; } return walk(window) || (latestByApplyTime ? 'apply-time' : ''); """ ) _wait_text_in_frames(browser, "合同编号", timeout_seconds=10) def _scrape_latest_my_contract_number(browser: BrowserAdapter) -> str: scanned = browser.execute_script( """ const latestByApplyTime = true; const skip = ['提交时自动生成', '自动生成', '==请选择==']; function scrape(doc) { const rows = Array.from(doc.querySelectorAll('tr')); let headerIdx = null; for (const tr of rows) { const cells = Array.from(tr.querySelectorAll('th, td')).map((c) => (c.innerText || '').replace(/\\s+/g, ' ').trim() ); if (!headerIdx && cells.indexOf('合同编号') >= 0) { headerIdx = {}; cells.forEach((h, i) => { if (h && headerIdx[h] === undefined) headerIdx[h] = i; }); continue; } if (!headerIdx) continue; const no = (cells[headerIdx['合同编号']] || '').trim(); if (!no || skip.indexOf(no) >= 0) continue; return latestByApplyTime ? no : no; } return ''; } function walk(win) { try { const no = scrape(win.document); if (no) return no; } catch (e) {} for (let i = 0; i < win.frames.length; i++) { try { const no = walk(win.frames[i]); if (no) return no; } catch (e) {} } return ''; } return walk(window); """ ) if isinstance(scanned, str) and _usable_contract_no(scanned.strip()): return scanned.strip() return "" def _read_contract_number_from_my_list( browser: BrowserAdapter, ctx: ContractFlowContext, timeout_seconds: int = 15, ) -> str: print("S13：刷新主界面 → 我的合同 → 申请时间 → 第一行最新合同号", flush=True) _open_my_contracts(browser, ctx) _sort_my_contracts_by_apply_time(browser) deadline = time.time() + timeout_seconds last = "" while time.time() < deadline: last = _scrape_latest_my_contract_number(browser) if last: print("从我的合同读到最新合同编号：", last, flush=True) return last time.sleep(0.3) print("我的合同列表未读到合同编号", flush=True) return last def _usable_contract_no(value: str) -> bool: text = (value or "").strip() if not text or text in ("提交时自动生成", "自动生成", "==请选择==", "[]"): return False if text in ("合同标题", "合同编号", "客户编码", "客户名称", "申请时间", "合同（协议）全称"): return False if text.startswith("HT"): return True if text.isdigit() and len(text) >= 10: return True return any(ch.isdigit() for ch in text) and len(text) >= 8 and " " not in text def _try_read(browser: BrowserAdapter, selectors: list[str]) -> str: if not browser.element_exists(selectors): return "" try: return (browser.read_first(selectors, "") or "").strip() except Exception: return "" def contract_version_label(ctx: ContractFlowContext) -> str: return "非格式合同" if ctx.template_type == "NON_FORMAT" else "格式合同"