388 lines
16 KiB
Python
388 lines
16 KiB
Python
"""Flowchart-sized CONTRACT_CREATE browser steps for YiSaiQi.
|
||
|
||
Each function is intentionally small enough to map to one YiSaiQi flowchart
|
||
node or one script block behind a flowchart node.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from typing import Protocol
|
||
|
||
from .errors import (
|
||
ContractCreateError,
|
||
LoginFailed,
|
||
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 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 select_first(self, selectors: list[str], value: str) -> None: ...
|
||
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: 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
|
||
order_no_list: list[str] = field(default_factory=list)
|
||
|
||
|
||
def run_contract_create_flow(browser: BrowserAdapter, ctx: ContractFlowContext) -> str:
|
||
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):
|
||
return ""
|
||
return read_contract_number(browser)
|
||
|
||
|
||
# 合同起草页（模板选择）直达 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"
|
||
)
|
||
|
||
# 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)"]',
|
||
}
|
||
|
||
# 货品明细行的字段后缀；行索引从 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(添加行)",
|
||
]
|
||
|
||
# 客商选择窗口（双击客户名称后弹出）选择器。若真实页面改版，仅需调整此处。
|
||
CUSTOMER_PICKER = {
|
||
"search_input": [
|
||
'input[placeholder*="客商"]',
|
||
'input[placeholder*="编码"]',
|
||
'input[placeholder*="搜索"]',
|
||
'input[name*="search"]',
|
||
'input[class*="search"]',
|
||
],
|
||
"search_button": [
|
||
"text=搜索",
|
||
"text=查询",
|
||
"button:contains(搜索)",
|
||
"a:contains(搜索)",
|
||
],
|
||
"result_rows": [
|
||
"//table//tr[contains(.,'{code}')]",
|
||
"//*[contains(@class,'list')]//*[contains(text(),'{code}')]",
|
||
"text={code}",
|
||
],
|
||
"confirm_button": [
|
||
"text=确定",
|
||
"text=确认",
|
||
"button:contains(确定)",
|
||
"button:contains(确认)",
|
||
],
|
||
}
|
||
|
||
|
||
def open_contract_system(browser: BrowserAdapter, ctx: ContractFlowContext) -> None:
|
||
browser.open_url(ctx.contract_url)
|
||
|
||
|
||
def login_contract_system(browser: BrowserAdapter, ctx: ContractFlowContext) -> None:
|
||
try:
|
||
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"]'])
|
||
browser.wait_for_text("合同", timeout_seconds=20)
|
||
except Exception as exc:
|
||
browser.screenshot("contract-login-failed")
|
||
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)
|
||
|
||
|
||
def select_contract_template(browser: BrowserAdapter, ctx: ContractFlowContext) -> None:
|
||
# 已通过 URL 直接指定模板，本步骤仅兜底：若尚未进入表单则再次导航
|
||
if "客户编码" not in browser.read_body_text():
|
||
browser.open_url(FORM_PAGE_URL)
|
||
browser.wait_for_text("客户编码", timeout_seconds=20)
|
||
|
||
|
||
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)为空，无法加载客商")
|
||
|
||
# 双击客户名称字段打开客商选择窗口，按编码搜索并带回客商
|
||
browser.double_click_first([F["customer_name"]])
|
||
_pick_customer_from_dialog(browser, ctx.customer_code)
|
||
loaded_name = _wait_customer_name(browser, timeout=8)
|
||
|
||
# 兜底：双击未弹出选择窗时，直接填客商操作码触发自动加载
|
||
if not loaded_name:
|
||
browser.fill_first([F["customer_code"]], ctx.customer_code)
|
||
loaded_name = _wait_customer_name(browser, timeout=8)
|
||
|
||
if not loaded_name:
|
||
browser.screenshot("contract-customer-load-failed")
|
||
raise SupplierContractDisabled(
|
||
f"客商加载失败，请确认输入的客商操作码是否有误：{ctx.customer_code}"
|
||
)
|
||
|
||
if expected_name and loaded_name != expected_name:
|
||
browser.screenshot("contract-customer-name-mismatch")
|
||
raise SupplierContractDisabled(
|
||
f"客商名称校验不通过：期望「{expected_name}」，实际「{loaded_name}」"
|
||
)
|
||
|
||
browser.screenshot("contract-signatory-info-filled")
|
||
|
||
|
||
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:
|
||
browser.wait_for_element(CUSTOMER_PICKER["search_input"], timeout_seconds=10)
|
||
browser.fill_first(CUSTOMER_PICKER["search_input"], code)
|
||
browser.click_first(CUSTOMER_PICKER["search_button"])
|
||
time.sleep(1)
|
||
rows = [sel.format(code=code) for sel in CUSTOMER_PICKER["result_rows"]]
|
||
browser.click_first(rows)
|
||
browser.click_first(CUSTOMER_PICKER["confirm_button"])
|
||
|
||
|
||
def fill_contract_basic_info(browser: BrowserAdapter, ctx: ContractFlowContext) -> None:
|
||
if ctx.contract_category:
|
||
browser.fill_first([F["contract_category"]], ctx.contract_category)
|
||
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"]'])
|
||
browser.screenshot("contract-basic-info-filled")
|
||
|
||
|
||
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 ContractCreateError(f"结算方式下拉未选中，仍停留在「==请选择==」，期望「{value}」")
|
||
|
||
|
||
def fill_contract_amount_and_dates(browser: BrowserAdapter, ctx: ContractFlowContext) -> None:
|
||
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)
|
||
browser.screenshot("contract-amount-and-dates-filled")
|
||
|
||
|
||
def fill_contract_goods_lines(browser: BrowserAdapter, ctx: ContractFlowContext) -> None:
|
||
for index, line in enumerate(ctx.goods_lines):
|
||
row = index # 第 1 行 row=0，第 2 行 row=1，依此类推
|
||
if index > 0:
|
||
browser.click_first(ADD_ROW_SELECTORS)
|
||
browser.wait_for_element([_goods_selector("goods_code", row)], timeout_seconds=15)
|
||
|
||
browser.fill_first([_goods_selector("goods_code", row)], line.goods_code)
|
||
browser.fill_first([_goods_selector("goods_name", row)], line.goods_name)
|
||
if line.mdm_code:
|
||
browser.fill_first([_goods_selector("goods_mdm", row)], line.mdm_code)
|
||
if line.approval_no:
|
||
browser.fill_first([_goods_selector("goods_approval", row)], line.approval_no)
|
||
if line.spec:
|
||
browser.fill_first([_goods_selector("goods_spec", row)], line.spec)
|
||
if line.unit:
|
||
browser.fill_first([_goods_selector("goods_unit", row)], line.unit)
|
||
browser.fill_first([_goods_selector("goods_qty", row)], str(line.quantity))
|
||
if line.unit_price:
|
||
browser.fill_first([_goods_selector("goods_price", row)], str(line.unit_price))
|
||
browser.screenshot("contract-goods-lines-filled")
|
||
|
||
|
||
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.fill_first([F["is_major_contract"]], ctx.is_major_contract)
|
||
browser.fill_first([F["is_prepayment"]], ctx.is_prepayment)
|
||
if ctx.belong_year:
|
||
browser.fill_first([F["belong_year"]], ctx.belong_year)
|
||
if ctx.payment_method:
|
||
browser.fill_first([F["payment_method"]], ctx.payment_method)
|
||
if ctx.payment_days:
|
||
browser.fill_first([F["payment_days"]], ctx.payment_days)
|
||
if ctx.contract_status:
|
||
browser.fill_first([F["contract_status"]], ctx.contract_status)
|
||
if ctx.contains_blood_products:
|
||
browser.fill_first([F["contains_blood"]], ctx.contains_blood_products)
|
||
browser.screenshot("contract-payment-filled")
|
||
|
||
|
||
def upload_contract_document_if_needed(browser: BrowserAdapter, ctx: ContractFlowContext) -> None:
|
||
if ctx.source_type != "YYC":
|
||
# XYY / OFFLINE：跳过合同证文上传
|
||
return
|
||
if ctx.submit_mode == "dry_run":
|
||
# dry_run 只跑到提交前，不真正上传附件
|
||
browser.screenshot("contract-pdf-upload-skipped")
|
||
return
|
||
if not ctx.merged_pdf_path:
|
||
raise PdfMergeUnsupported("YYC 合同起草缺少合并后的 PDF 路径，无法上传")
|
||
browser.upload_file_first(['input[type="file"]', "label=合同证文上传"], ctx.merged_pdf_path)
|
||
browser.screenshot("contract-pdf-uploaded")
|
||
|
||
|
||
def submit_contract_for_approval(browser: BrowserAdapter, ctx: ContractFlowContext) -> bool:
|
||
if ctx.submit_mode == "dry_run":
|
||
browser.screenshot("contract-dry-run-ready")
|
||
return False
|
||
browser.click_first(["text=提交", "text=提交送审", "text=送审"])
|
||
browser.wait_for_text("合同编号", timeout_seconds=30)
|
||
browser.screenshot("contract-submitted")
|
||
return True
|
||
|
||
|
||
def read_contract_number(browser: BrowserAdapter) -> str:
|
||
return browser.read_first(
|
||
['input[name="contractNo"]', 'input[name="contractSystemNo"]', "label=合同编号", "text=合同编号"],
|
||
"contract_number",
|
||
)
|
||
|
||
|
||
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 "格式合同"
|