# 返利细单节点：本段字段与 run_import_flow。登录/查号/勾行/弹层在 rebate_common。 from __future__ import annotations import json import os import re import shutil import time from pathlib import Path from accept import AcceptanceReport, artifact_dir, evidence_shot from errors import NodeError from rebate_common import ( IMPORT_FLOW, PAY_MODE_NEXT_MONTH_N, TAX_RATE_DEFAULT, PAY_MODE_CODE, PAY_PARAM_DEFAULT, SETTLE_CODE, SETTLE_MODE_WIRE, COMPENSATION_DEFAULT, BrowserAdapter, RebateFlowContext, RebateImportLine, RebateImportDraft, RebateImportResult, open_rebate_center, login_rebate_center, _switch_agreement_frame, enter_purchase_agreement, search_contract, _header_dom_js, today_iso, year_end_iso, _cell, _tax_rate_of, _untaxed_price, _click_text, select_header_row_result, _close_header_query, _pick_detail_option, _read_detail_toast, _close_header_card, _close_detail_card, ) _IMPORT_ACCEPT = AcceptanceReport(flow=IMPORT_FLOW) # 细单按表 3 货品行：货品ID、数量、含税单价、无税=含税/(1+税率)、备注=omsId。 def import_lines_from_table3(rows: list[dict] | tuple[dict, ...]) -> tuple[RebateImportLine, ...]: lines: list[RebateImportLine] = [] for row in rows: if not isinstance(row, dict): continue goods_id = _cell(row.get("goods_id")) if not goods_id: continue tax_price = _cell(row.get("unit_price") or row.get("tax_price")) tax_rate = _tax_rate_of(row.get("purchase_tax_rate") or row.get("tax_rate")) lines.append( RebateImportLine( goods_code=goods_id, qty=_cell(row.get("qty")), tax_price=tax_price, order_no=_cell(row.get("oms_id") or row.get("order_no")), source_line_id=_cell(row.get("goods_source_id") or row.get("source_line_id")), goods_id=goods_id, tax_rate=tax_rate, net_price=_untaxed_price(tax_price, tax_rate), ) ) if not lines: raise NodeError("表 3 没有货品行，不能加细单") return tuple(lines) # 优先表 3 货品行，空缺再退回环境变量。 def import_draft_for_contract(contract_no: str, *, today: str | None = None) -> RebateImportDraft: number = (contract_no or "").strip() rows: list[dict] = [] if number: try: from repo import OracleRebateRepo rows = OracleRebateRepo.from_env().load_contract_goods(number) except Exception: rows = [] if rows: return build_import_draft(lines=import_lines_from_table3(rows), today=today) return import_draft_from_env(today=today) IMPORT = { "menu": [ "text=导入导出", "button:contains(导入导出)", "span:contains(导入导出)", "a:contains(导入导出)", ], "download": [ "text=下载模板", "text=下载导入模板", "text=导出模板", "text=导入模板", "span:contains(下载模板)", "a:contains(下载模板)", "button:contains(下载模板)", ], "upload_item": [ "text=导入", "span:contains(导入)", "button:contains(导入)", "a:contains(导入)", ], "file": [ 'input[type="file"]', 'input[accept*="excel"]', 'input[accept*=".xls"]', 'input[accept*="sheet"]', ], "confirm": [ "text=确定", "button:contains(确定)", "text=导入", "button:contains(导入)", ], } IMPORT_HEADERS = ( "货品编码", "数量", "含税协议单价", "税率", "关税税率", "承付模式", "承付参数", "细单开始日期", "细单结束日期", "采购结算方式", "特定补偿预估值", "进货补偿预估值", "细单备注", "货源细单ID", ) _HEADER_ALIASES = { "货品编码": ("货品ID", "货品编码", "操作码", "商品编码"), "数量": ("数量",), "含税协议单价": ("含税协议单价", "单价", "协议单价"), "税率": ("税率",), "关税税率": ("关税税率",), "承付模式": ("承付模式", "承付方式"), "承付参数": ("承付参数", "参数"), "细单开始日期": ("细单开始日期", "细单开始时间", "开始日期"), "细单结束日期": ("细单结束日期", "失效日期", "结束日期"), "采购结算方式": ("采购结算方式", "结算方式"), "特定补偿预估值": ("特定补偿预估值", "特定补偿"), "进货补偿预估值": ("进货补偿预估值", "进货补偿"), "细单备注": ("细单备注", "备注"), "货源细单ID": ("货源细单ID", "货源细单号", "货源信息"), } _DOWNLOAD_TITLES = ("下载模板", "下载导入模板", "导出模板", "导入模板", "模板下载") _SUCCESS_RE = re.compile(r"成功[^0-9]{0,10}(\d+)") _FAIL_RE = re.compile(r"失败[^0-9]{0,10}(\d+)") _READ_IMPORT_RESULT = """ /* READ_IMPORT_RESULT */ const body = (document.body && document.body.innerText) || ''; function num(re) { const m = body.match(re); return m ? Number(m[1]) : 0; } const reasons = Array.from(document.querySelectorAll( '.el-message, .el-notification, .el-dialog, .vxe-body--row, tr, .ant-message' )).map((el) => (el.innerText || '').replace(/\\s+/g, ' ').trim()).filter((t) => { return t && (t.indexOf('失败') >= 0 || t.indexOf('重复') >= 0 || t.indexOf('错误') >= 0); }).slice(0, 10); return { success: num(/成功[^0-9]{0,10}(\\d+)/), fail: num(/失败[^0-9]{0,10}(\\d+)/), reasons, body: body.slice(0, 800), }; """ # 返利中心增加细单按货品ID搜品种，用接口 3 的 goodsId。 def goods_search_id(line: RebateImportLine) -> str: return _cell(line.goods_id) or _cell(line.goods_code) # 把细单行收成旧 Excel 列字典。班次不走这条，页面走新增。 def build_import_rows( lines: list[RebateImportLine] | tuple[RebateImportLine, ...], *, today: str | None = None, ) -> list[dict[str, str]]: day = today_iso(today) end = year_end_iso(day) rows: list[dict[str, str]] = [] for line in lines: rows.append( { "货品编码": goods_search_id(line), "数量": _cell(line.qty), "含税协议单价": _cell(line.tax_price), "税率": TAX_RATE_DEFAULT, "关税税率": TAX_RATE_DEFAULT, "承付模式": PAY_MODE_CODE, "承付参数": PAY_PARAM_DEFAULT, "细单开始日期": day, "细单结束日期": end, "采购结算方式": SETTLE_CODE, "特定补偿预估值": COMPENSATION_DEFAULT, "进货补偿预估值": COMPENSATION_DEFAULT, "细单备注": _cell(line.order_no), "货源细单ID": _cell(line.source_line_id), } ) return rows # 校验细单至少一行，货品ID、订单号、数量、含税单价都要有。 def build_import_draft( *, lines: list[RebateImportLine] | tuple[RebateImportLine, ...], today: str | None = None, ) -> RebateImportDraft: packed = tuple(lines) if not packed: raise NodeError("细单至少一行") for line in packed: if not goods_search_id(line) or not _cell(line.order_no): raise NodeError("细单货品ID和订单号必须用订单中心的") if not _cell(line.qty) or not _cell(line.tax_price): raise NodeError("细单数量和含税协议单价必须用订单中心的") return RebateImportDraft(lines=packed, today=today_iso(today)) # 把 camelCase / snake_case 字典收成 RebateImportLine。无税单价按含税和税率算。 def _line_from_mapping(item: dict) -> RebateImportLine: goods_id = _cell(item.get("goodsId") or item.get("goods_id")) goods_code = _cell(item.get("goodsCode") or item.get("goods_code")) tax_price = _cell(item.get("taxPrice") or item.get("tax_price") or item.get("price") or item.get("unit_price")) tax_rate = _tax_rate_of( item.get("taxRate") or item.get("tax_rate") or item.get("purchase_tax_rate") ) return RebateImportLine( goods_code=goods_code, qty=_cell(item.get("qty") or item.get("quantity") or item.get("saleQty")), tax_price=tax_price, order_no=_cell(item.get("orderNo") or item.get("order_no")), source_line_id=_cell( item.get("sourceLineId") or item.get("source_line_id") or item.get("goodsSourceId") ), goods_id=goods_id, tax_rate=tax_rate, net_price=_untaxed_price(tax_price, tax_rate), ) # 从 REBATE_IMPORT_LINES 或单条环境变量拼细单草稿。表 3 空缺时干跑用。 def import_draft_from_env(*, today: str | None = None) -> RebateImportDraft: raw = (os.getenv("REBATE_IMPORT_LINES") or "").strip() if raw: payload = json.loads(raw) if not isinstance(payload, list): payload = [payload] lines = [_line_from_mapping(item) for item in payload] else: lines = [ RebateImportLine( goods_code=os.getenv("REBATE_GOODS_CODE") or "", qty=os.getenv("REBATE_QTY") or "", tax_price=os.getenv("REBATE_TAX_PRICE") or "", order_no=os.getenv("REBATE_ORDER_NO") or "", source_line_id=os.getenv("REBATE_SOURCE_LINE_ID") or "", goods_id=os.getenv("REBATE_GOODS_ID") or "", ) ] return build_import_draft(lines=lines, today=today) # 写一份带表头的空 xlsx，给旧 Excel 路径的单测用。 def write_canonical_template(path: Path) -> Path: from openpyxl import Workbook dest = Path(path) dest.parent.mkdir(parents=True, exist_ok=True) wb = Workbook() ws = wb.active ws.title = "细单" for col, name in enumerate(IMPORT_HEADERS, 1): ws.cell(1, col, name) wb.save(dest) return dest # 读工作表一行的标题文本。 def _header_titles(row) -> list[str]: titles = [] for cell in row: text = _cell(cell.value).lstrip("*") if text: titles.append(text) return titles # 找到含「货品编码」的表头行号。 def _header_row_index(ws) -> int: aliases = {name for names in _HEADER_ALIASES.values() for name in names} for index, row in enumerate(ws.iter_rows(min_row=1, max_row=15), 1): titles = _header_titles(row) if any(title in aliases for title in titles): return index return 1 # 表头标题到列号的映射。 def _col_map(ws, header_row: int) -> dict[str, int]: mapping: dict[str, int] = {} for cell in ws[header_row]: title = _cell(cell.value).lstrip("*") if not title or not cell.column: continue for field, names in _HEADER_ALIASES.items(): if title in names and field not in mapping: mapping[field] = int(cell.column) return mapping # 按列名把细单写进模板。班次不调用。 def fill_import_excel( template_path: Path, rows: list[dict[str, str]], output_path: Path | None = None, ) -> Path: from openpyxl import load_workbook src = Path(template_path) dest = Path(output_path) if output_path else src.with_name(src.stem + "_filled.xlsx") dest.parent.mkdir(parents=True, exist_ok=True) if src.suffix.lower() != ".xlsx": write_canonical_template(dest) elif src.resolve() != dest.resolve(): shutil.copy2(src, dest) wb = load_workbook(dest) ws = wb.active header_row = _header_row_index(ws) colmap = _col_map(ws, header_row) used = [int(cell.column) for cell in ws[header_row] if cell.column] next_col = (max(used) if used else 0) + 1 for field in IMPORT_HEADERS: if field in colmap: continue ws.cell(header_row, next_col, field) colmap[field] = next_col next_col += 1 start = header_row + 1 for offset, row in enumerate(rows): for field, value in row.items(): col = colmap.get(field) if col: ws.cell(start + offset, col, value) wb.save(dest) return dest # 等到下载目录出现新的 xlsx，忽略 .crdownload。 def _wait_local_download(dest_dir: Path, before: set[str], timeout_seconds: int = 20) -> Path: deadline = time.time() + timeout_seconds while time.time() <= deadline: hits = [ p for p in dest_dir.iterdir() if p.name not in before and p.suffix.lower() == ".xlsx" and not str(p).endswith(".crdownload") and not Path(str(p) + ".crdownload").exists() ] if hits: return max(hits, key=lambda p: p.stat().st_mtime) time.sleep(0.3) raise RuntimeError("未下到细单模板") # 点导入导出下模板。班次不走 Excel，单测还留着。 def download_detail_template(browser: BrowserAdapter, dest_dir: Path) -> Path: dest = Path(dest_dir) dest.mkdir(parents=True, exist_ok=True) setter = getattr(browser, "set_download_dir", None) if callable(setter): setter(str(dest.resolve())) before = {p.name for p in dest.iterdir()} _switch_agreement_frame(browser) if not _click_text(browser, "导入导出"): try: browser.click_first(IMPORT["menu"]) except Exception: pass downloaded = False for title in _DOWNLOAD_TITLES: if _click_text(browser, title): downloaded = True break if not downloaded: browser.click_first(IMPORT["download"]) waiter = getattr(browser, "wait_for_download", None) if callable(waiter): try: return Path(waiter(".xlsx")) except Exception: pass return _wait_local_download(dest, before) # 点导入把 xlsx 塞进文件框。班次不走这条。 def upload_detail_file(browser: BrowserAdapter, file_path: Path) -> None: _switch_agreement_frame(browser) if not _click_text(browser, "导入"): _click_text(browser, "导入导出") if not _click_text(browser, "导入"): try: browser.click_first(IMPORT["upload_item"]) except Exception: pass path = str(Path(file_path).resolve()) browser.upload_file_first(IMPORT["file"], path) try: browser.click_first(IMPORT["confirm"]) except Exception: pass # 从页面读导入成功/失败条数。新增路径用货品表行数代替。 def _scan_import_result(browser: BrowserAdapter, *, expected_lines: int = 0) -> RebateImportResult: raw = None try: raw = browser.execute_script(_READ_IMPORT_RESULT) except Exception: raw = None body = "" success = 0 fail = 0 reasons: list[str] = [] if isinstance(raw, dict): try: success = int(raw.get("success") or 0) except (TypeError, ValueError): success = 0 try: fail = int(raw.get("fail") or 0) except (TypeError, ValueError): fail = 0 reasons = [_cell(x) for x in (raw.get("reasons") or []) if _cell(x)] body = str(raw.get("body") or "") if not body: body = browser.read_body_text() or "" if success == 0: matched = _SUCCESS_RE.search(body) if matched: success = int(matched.group(1)) if fail == 0: matched = _FAIL_RE.search(body) if matched: fail = int(matched.group(1)) if "存在重复的编码" in body: if "存在重复的编码" not in reasons: reasons.append("存在重复的编码") if fail == 0: fail = expected_lines or 1 if success == 0 and fail == 0 and "导入成功" in body: success = expected_lines or 1 parsed = success + fail > 0 or "导入成功" in body or bool(reasons) return RebateImportResult(success, fail, tuple(reasons), parsed) # 货品工具栏是否已出现「新增」。必须先勾中总单。 def _goods_add_visible(browser: BrowserAdapter) -> bool: try: hit = browser.execute_script( """ /* GOODS_ADD_VISIBLE */ function compact(el) { return ((el && (el.innerText || '')) || '').replace(/\\s+/g, '').trim(); } function goodsTable() { const header = document.querySelector('.table-show .single-table'); const all = Array.from(document.querySelectorAll('.single-table')).filter((el) => { return el !== header && (el.offsetWidth || el.offsetHeight); }); return all.find((el) => compact(el).indexOf('货品') >= 0) || all.find((el) => el.closest('.elPlus-tab-pane')) || all[0] || null; } const table = goodsTable(); if (!table) return false; const nodes = Array.from(table.querySelectorAll( '.table-toolbar button, .table-toolbar .vxe-button, .table-toolbar span, .table-toolbar a, .table-toolbar-btn__text' )); return nodes.some((el) => { if (!(el.offsetWidth || el.offsetHeight)) return false; const t = compact(el); return t === '新增' || t === '+新增' || t === '新增细单' || t === '添加细单'; }); """ ) except Exception: hit = False return hit is True # 等到货品新增按钮出现。超时返回 False。 def _wait_goods_add_button(browser: BrowserAdapter, timeout: float = 8) -> bool: deadline = time.time() + timeout while time.time() <= deadline: _switch_agreement_frame(browser) if _goods_add_visible(browser): return True _click_text(browser, "货品") time.sleep(0.4) return False # 点货品工具栏「新增」，打开一条细单卡片。 def _click_add_detail(browser: BrowserAdapter) -> bool: _switch_agreement_frame(browser) clicked = browser.execute_script( """ /* CLICK_GOODS_ADD */ function compact(el) { return ((el && (el.innerText || '')) || '').replace(/\\s+/g, '').trim(); } function goodsTable() { const header = document.querySelector('.table-show .single-table'); const all = Array.from(document.querySelectorAll('.single-table')).filter((el) => { return el !== header && (el.offsetWidth || el.offsetHeight); }); return all.find((el) => compact(el).indexOf('货品') >= 0) || all.find((el) => el.closest('.elPlus-tab-pane')) || all[0] || null; } const table = goodsTable(); if (!table) { return {ok: false, error: 'goods-table'}; } const nodes = Array.from(table.querySelectorAll( '.table-toolbar button, .table-toolbar .vxe-button, .table-toolbar span, .table-toolbar a, .table-toolbar-btn__text' )); const hit = nodes.find((el) => { if (!(el.offsetWidth || el.offsetHeight)) return false; const t = compact(el); return t === '新增' || t === '+新增' || t === '新增细单' || t === '添加细单'; }); if (!hit) { return {ok: false}; } hit.click(); return {ok: true, text: compact(hit)}; """ ) if isinstance(clicked, dict) and clicked.get("ok"): time.sleep(0.8) return True return False # 只点货品工具栏（新增旁边那颗），不要点总单工具栏。 def _click_goods_toolbar(browser: BrowserAdapter, title: str) -> bool: clicked = browser.execute_script( """ /* SAVE_DETAIL */ const title = String(arguments[0] || '').replace(/\\s+/g, '').trim(); function compact(el) { return ((el && (el.innerText || '')) || '').replace(/\\s+/g, '').trim(); } function vis(el) { return !!(el && (el.offsetWidth || el.offsetHeight)); } const bars = Array.from(document.querySelectorAll('.table-toolbar')).filter(vis); const bar = bars.find((el) => { const t = compact(el); return t.indexOf('新增') >= 0 && t.indexOf('保存') >= 0 && t.indexOf('送审') < 0; }) || bars.find((el) => compact(el).indexOf('新增') >= 0); if (!bar) { return {ok: false, error: 'no-detail-bar'}; } const hit = Array.from(bar.querySelectorAll( 'button, .table-toolbar-btn, .table-toolbar-btn__text, span' )).find((el) => vis(el) && compact(el) === title); if (!hit) { return {ok: false, error: 'no-btn'}; } const btn = hit.closest('button, .table-toolbar-btn') || hit; btn.click(); return {ok: true, via: compact(hit)}; """, title, ) return bool(isinstance(clicked, dict) and clicked.get("ok")) # 用 ActionChains 点货品工具栏按钮，避开总单工具栏同名保存。 def _native_click_goods_toolbar(browser: BrowserAdapter, title: str) -> bool: driver = getattr(browser, "_driver", None) if driver is None: return False wanted = (title or "").replace(" ", "") try: from selenium.webdriver.common.action_chains import ActionChains for bar in driver.find_elements("css selector", ".table-toolbar"): text = (bar.text or "").replace("\n", " ") if "新增" not in text or "保存" not in text or "送审" in text: continue for el in bar.find_elements("css selector", "button, .table-toolbar-btn, .table-toolbar-btn__text"): raw = (el.text or "").replace("\n", "").replace(" ", "") if raw != wanted: continue driver.execute_script("arguments[0].scrollIntoView({block:'center'});", el) ActionChains(driver).move_to_element(el).pause(0.15).click().perform() return True except Exception: return False return False # 货品保存时若弹出确认，再点确认。 def _confirm_save_popup(browser: BrowserAdapter) -> bool: try: clicked = browser.execute_script( """ /* CONFIRM_SAVE_POPUP */ function compact(el) { return ((el && (el.innerText || '')) || '').replace(/\\s+/g, '').trim(); } function vis(el) { return !!(el && (el.offsetWidth || el.offsetHeight)); } const box = Array.from(document.querySelectorAll( '.elPlus-message-box, .el-message-box, .elPlus-dialog, .el-dialog, .vxe-modal--wrapper, .ant-modal' )).find((el) => { if (!vis(el)) return false; const t = compact(el); return t.indexOf('确认') >= 0 || t.indexOf('确定') >= 0 || t.indexOf('是否') >= 0; }); if (!box) { return {ok: false, asked: false}; } const hit = Array.from(box.querySelectorAll('button, .elPlus-button, span')).find((el) => { if (!vis(el)) return false; const t = compact(el); return t === '确定' || t === '确 定' || t === '确认' || t === '确 认' || t === '是'; }); if (!hit) { return {ok: false, asked: true}; } hit.click(); return {ok: true, asked: true, via: compact(hit)}; """ ) except Exception: clicked = None return bool(isinstance(clicked, dict) and clicked.get("ok")) # 细单卡片上货品编码输入是否已出来。 def _goods_id_field_visible(browser: BrowserAdapter) -> bool: try: hit = browser.execute_script( """ /* GOODS_ID_FIELD */ function labelText(el) { const n = el.querySelector('.search-label__text'); return ((n && n.innerText) || '').replace(/\\s+/g, '').trim(); } return Array.from(document.querySelectorAll('.search-condition-item')).some((el) => { return (el.offsetWidth || el.offsetHeight) && labelText(el) === '货品ID'; }); """ ) except Exception: hit = False return hit is True # 点货品编码三个点，准备按货品ID搜。 def _open_goods_id_query(browser: BrowserAdapter) -> None: browser.execute_script( """ /* OPEN_GOODS_ID_QUERY */ function labelText(el) { const n = el.querySelector('.search-label__text'); return ((n && n.innerText) || '').replace(/\\s+/g, '').trim(); } const ready = Array.from(document.querySelectorAll('.search-condition-item')).some((el) => { return (el.offsetWidth || el.offsetHeight) && labelText(el) === '货品ID'; }); if (ready) { return {ok: true, already: true}; } const modal = Array.from(document.querySelectorAll( '.vxe-modal--wrapper, .elPlus-dialog, .el-dialog, .elPlus-drawer' )).find((el) => el.offsetWidth || el.offsetHeight); const root = modal || document; const btn = root.querySelector('.search-button-content, .search-button'); if (btn && (btn.offsetWidth || btn.offsetHeight) && !root.querySelector('.table-show')) { btn.click(); return {ok: true, via: 'query'}; } if (modal) { const inner = modal.querySelector('.search-button-content, .search-button'); if (inner) { inner.click(); return {ok: true, via: 'modal'}; } } return {ok: false}; """ ) # 等到细单卡片货品编码字段出现。 def _wait_goods_id_field(browser: BrowserAdapter, timeout: float = 8) -> bool: deadline = time.time() + timeout retried_add = False while time.time() <= deadline: _switch_agreement_frame(browser) if _goods_id_field_visible(browser): return True _open_goods_id_query(browser) if _goods_id_field_visible(browser): return True if not retried_add: _click_add_detail(browser) retried_add = True time.sleep(0.4) return _goods_id_field_visible(browser) # 打开货品选择弹窗。 def _open_goods_picker(browser: BrowserAdapter) -> bool: opened = browser.execute_script( """ /* OPEN_GOODS_PICKER */ """ + _header_dom_js() + """ const item = detailItem('货品编码') || formItem('货品编码'); if (!item) { return {ok: false, error: 'no-goods-code'}; } item.scrollIntoView({block: 'center', inline: 'nearest'}); const via = clickDots(item); if (via) { return {ok: true, via: via}; } const btn = item.querySelector('button.elPlus-button, button'); if (!btn) { return {ok: false}; } btn.click(); return {ok: true, via: 'button'}; """ ) return bool(isinstance(opened, dict) and opened.get("ok")) # 等到货品选择弹窗出现。 def _wait_goods_picker(browser: BrowserAdapter, timeout: float = 12) -> bool: deadline = time.time() + timeout while time.time() <= deadline: hit = browser.execute_script( """ /* WAIT_GOODS_PICKER */ const el = Array.from(document.querySelectorAll('.vxe-modal--wrapper')).find((n) => { const t = ((n.innerText || '') + ''); const vis = n.offsetWidth || n.offsetHeight || String(n.className || '').indexOf('is--visible') >= 0; return vis && t.indexOf('货品选择') >= 0; }); return !!(el && (el.offsetWidth || el.offsetHeight)); """ ) if hit is True: return True time.sleep(0.4) return False # 弹窗里用货品ID搜，先点行里单选再点确定。不要用操作码或货品编码搜。 def _pick_goods_id_in_dialog(browser: BrowserAdapter, goods_id: str) -> bool: keyword = (goods_id or "").strip() if not keyword: return False if not _open_goods_picker(browser): return False if not _wait_goods_picker(browser): return False filled = browser.execute_script( """ /* FILL_GOODS_ID */ const no = String(arguments[0] || ''); function compact(el) { return ((el && (el.innerText || '')) || '').replace(/\\s+/g, ' ').trim(); } function vis(el) { return !!(el && (el.offsetWidth || el.offsetHeight)); } const modal = Array.from(document.querySelectorAll('.vxe-modal--wrapper')).find((el) => { return vis(el) && compact(el).indexOf('货品选择') >= 0; }); if (!modal) { return {ok: false, error: 'no-modal'}; } const tab = Array.from(modal.querySelectorAll('span, label, li, button, div')).find((el) => { return compact(el) === '货品ID'; }); if (tab) { tab.click(); } const input = modal.querySelector('input.elPlus-input__inner, input'); if (!input) { return {ok: false, error: 'no-input'}; } input.click(); input.focus(); const setter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value').set; setter.call(input, no); input.dispatchEvent(new Event('input', {bubbles: true})); input.dispatchEvent(new Event('change', {bubbles: true})); const query = Array.from(modal.querySelectorAll('button')).find((el) => { const t = compact(el); return t === '查询' || t.indexOf('查询') === 0; }); if (query) { query.click(); } return {ok: true, label: '货品ID', value: input.value}; """, keyword, ) if not (isinstance(filled, dict) and filled.get("ok") and filled.get("value") == keyword): return False deadline = time.time() + 8 while time.time() <= deadline: picked = browser.execute_script( """ /* PICK_GOODS_ROW */ const keyword = String(arguments[0] || '').replace(/\\s+/g, ''); function compact(el) { return ((el && (el.innerText || '')) || '').replace(/\\s+/g, ''); } function vis(el) { return !!(el && (el.offsetWidth || el.offsetHeight)); } const modal = Array.from(document.querySelectorAll('.vxe-modal--wrapper')).find((el) => { return vis(el) && compact(el).indexOf('货品选择') >= 0; }); const root = modal || document; const rows = Array.from(root.querySelectorAll('.vxe-body--row, tr, [role="row"]')); for (const row of rows) { const text = compact(row); if (text.indexOf(keyword) < 0) continue; if (text.indexOf('CG-') >= 0) continue; const radio = row.querySelector( '.vxe-cell--radio, .vxe-radio--icon, .vxe-radio--label, input[type="radio"]' ); if (radio) { radio.click(); } else { row.click(); } const ok = Array.from(root.querySelectorAll('button')).find((el) => { if (!vis(el)) return false; const t = compact(el); return t === '确定' || t === '确 定'; }); if (ok) { ok.click(); } return {ok: true, via: radio ? 'radio' : 'row'}; } return {ok: false}; """, keyword, ) if isinstance(picked, dict) and picked.get("ok"): time.sleep(0.5) return True time.sleep(0.4) return False # 读细单卡片分页文案，用来判断确认有没有把行落到表。 def _detail_card_pager(browser: BrowserAdapter) -> str: try: text = browser.execute_script( """ /* DETAIL_CARD_PAGER */ const el = document.querySelector('.right-btns'); return ((el && el.innerText) || '').replace(/\\s+/g, ' ').trim(); """ ) except Exception: text = "" return str(text or "") # 读货品表各行文本，供核对 goodsId 是否已在表里。 def _goods_table_rows(browser: BrowserAdapter) -> list[str]: try: rows = browser.execute_script( """ /* GOODS_TABLE_ROWS */ function compact(el) { return ((el && (el.innerText || '')) || '').replace(/\\s+/g, ' ').trim(); } function vis(el) { return !!(el && (el.offsetWidth || el.offsetHeight)); } const table = Array.from(document.querySelectorAll('.single-table')).find((el) => { return vis(el) && compact(el).indexOf('协议细单ID') >= 0; }); if (!table) { return []; } return Array.from(table.querySelectorAll('.vxe-body--row')).filter(vis).map((el) => compact(el).slice(0, 240)); """ ) except Exception: rows = None if isinstance(rows, list): return [str(item) for item in rows] return [] # 等到货品表出现该货品ID。 def _wait_goods_row(browser: BrowserAdapter, gid: str, timeout: float = 8) -> bool: number = (gid or "").strip() if not number: return False deadline = time.time() + timeout while time.time() <= deadline: rows = _goods_table_rows(browser) if any(number in row for row in rows): return True time.sleep(0.3) print("goods_table_rows=", _goods_table_rows(browser), "need=", number, flush=True) return False # 点货品卡片标题栏「确认 Ctrl + Enter」，不要点总单卡片确认。 def _confirm_detail_card(browser: BrowserAdapter) -> bool: before = _detail_card_pager(browser) clicked = browser.execute_script( """ /* CONFIRM_DETAIL_CARD */ function compact(el) { return ((el && (el.innerText || '')) || '').replace(/\\s+/g, ' ').trim(); } function vis(el) { return !!(el && (el.offsetWidth || el.offsetHeight)); } const card = Array.from(document.querySelectorAll('.table-card')).find((el) => { return vis(el) && compact(el).indexOf('货品编码') >= 0; }); if (!card) { return {ok: false, error: 'no-goods-card'}; } const nodes = Array.from(card.querySelectorAll('.card-header-btn, button.elPlus-button')); const texts = nodes.map((el) => compact(el)).filter(Boolean); const hit = nodes.find((el) => { if (!vis(el)) return false; const t = compact(el).replace(/\\s+/g, ''); return t.indexOf('确认') === 0 || t.indexOf('确认Ctrl') === 0; }); if (!hit) { return {ok: false, error: 'no-confirm', n: nodes.length, texts: texts.slice(0, 8)}; } hit.scrollIntoView({block: 'center', inline: 'nearest'}); hit.click(); return {ok: true, via: compact(hit).slice(0, 40), w: hit.offsetWidth, h: hit.offsetHeight}; """ ) driver = getattr(browser, "_driver", None) if driver is not None: try: from selenium.webdriver.common.action_chains import ActionChains for el in driver.find_elements("css selector", ".table-card .card-header-btn"): info = driver.execute_script( """ const el = arguments[0]; const card = el.closest('.table-card'); return { text: ((el.innerText || '') + '').replace(/\\s+/g, ' ').trim(), goods: !!card && ((card.innerText || '').indexOf('货品编码') >= 0) }; """, el, ) if not (isinstance(info, dict) and info.get("goods")): continue raw = str(info.get("text") or "") if "确认" not in raw.replace(" ", "").replace("\n", ""): continue driver.execute_script( "arguments[0].scrollIntoView({block:'center', inline:'nearest'});", el, ) time.sleep(0.2) ActionChains(driver).move_to_element(el).pause(0.15).click().perform() clicked = {"ok": True, "via": "native:" + raw[:40]} break except Exception as exc: print("native_confirm_err=", exc, flush=True) print("confirm_click=", clicked, "pager_before=", before, flush=True) if not (isinstance(clicked, dict) and clicked.get("ok")): return False time.sleep(0.6) err = _read_detail_toast(browser) if err: raise NodeError(err) errors = [] try: errors = browser.execute_script( """ /* DETAIL_FORM_ERRORS */ function compact(el) { return ((el && (el.innerText || '')) || '').replace(/\\s+/g, ' ').trim(); } return Array.from(document.querySelectorAll( '.table-card .elPlus-form-item.is-error, .table-card .elPlus-form-item__error' )).map((el) => compact(el)).filter(Boolean).slice(0, 8); """ ) or [] except Exception: errors = [] print("pager_after=", _detail_card_pager(browser), "errors=", errors, flush=True) if errors: raise NodeError("细单确认未过：" + "；".join(str(x) for x in errors[:3])) return True if isinstance(clicked, dict) and clicked.get("ok"): time.sleep(0.5) err = _read_detail_toast(browser) if err: raise NodeError(err) return True driver = getattr(browser, "_driver", None) if driver is not None: from selenium.webdriver.common.action_chains import ActionChains from selenium.webdriver.common.keys import Keys ActionChains(driver).key_down(Keys.CONTROL).send_keys(Keys.ENTER).key_up(Keys.CONTROL).perform() time.sleep(0.5) return True return False # 卡片输入要用键盘，JS 赋值 Vue 收不到，确认时会报必填。 def _type_detail_field(browser: BrowserAdapter, title: str, value: str) -> bool: wanted = _cell(value) if not wanted: return False marked = browser.execute_script( """ /* TYPE_DETAIL_FIELD */ """ + _header_dom_js() + """ const item = detailItem(arguments[0]); const input = item && item.querySelector('input, textarea'); if (!input) { return false; } input.setAttribute('data-rpa-fill', '1'); input.click(); return true; """, title, ) if not marked: return False driver = getattr(browser, "_driver", None) if driver is None: return True try: from selenium.webdriver.common.action_chains import ActionChains from selenium.webdriver.common.keys import Keys el = driver.find_element("css selector", '.table-card [data-rpa-fill="1"]') driver.execute_script("arguments[0].removeAttribute('data-rpa-fill')", el) ActionChains(driver).move_to_element(el).click().pause(0.05).key_down(Keys.CONTROL).send_keys("a").key_up( Keys.CONTROL ).send_keys(Keys.BACKSPACE).send_keys(wanted).send_keys(Keys.TAB).perform() return True except Exception: return False # 填数量、含税单价、无税单价、备注=订单号、特定/进货补偿=0、承付模式和参数 25、结算电汇。 def _fill_last_detail_qty_price(browser: BrowserAdapter, line: RebateImportLine) -> None: qty = _cell(line.qty) price = _cell(line.tax_price) remark = _cell(line.order_no) tax = _tax_rate_of(line.tax_rate) net = _cell(line.net_price) or _untaxed_price(price, tax) filled = browser.execute_script( """ /* FILL_DETAIL_CELLS */ """ + _header_dom_js() + """ const qty = String(arguments[0] || ''); const price = String(arguments[1] || ''); const remark = String(arguments[2] || ''); const compensation = String(arguments[3] || ''); const tax = String(arguments[4] || ''); const payParam = String(arguments[5] || ''); const net = String(arguments[6] || ''); function setByLabel(title, value) { if (!value) return false; const item = detailItem(title); if (!item) return false; const input = item.querySelector('input, textarea, .vxe-input--inner'); if (!input) return false; input.click(); const desc = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value') || Object.getOwnPropertyDescriptor(window.HTMLTextAreaElement.prototype, 'value'); if (desc && desc.set) { desc.set.call(input, value); } else { input.value = value; } input.dispatchEvent(new Event('input', {bubbles: true})); input.dispatchEvent(new Event('change', {bubbles: true})); return true; } return { qty: setByLabel('数量', qty), price: setByLabel('含税协议单价', price), net: setByLabel('无税协议价', net) || setByLabel('无税单价', net), tax: setByLabel('税率', tax), payParam: setByLabel('承付参数', payParam), remark: setByLabel('细单备注', remark), compensation: setByLabel('特定补偿预估值', compensation) || setByLabel('特定补偿', compensation), inbound: setByLabel('进货补偿预估值', compensation), payMode: shownValue(detailItem('承付模式')), settle: shownValue(detailItem('采购结算方式')), }; """, qty, price, remark, COMPENSATION_DEFAULT, tax, PAY_PARAM_DEFAULT, net, ) print("fill_detail=", filled, flush=True) try: browser.execute_script( """ document.dispatchEvent(new KeyboardEvent('keydown', { key: 'Escape', code: 'Escape', keyCode: 27, which: 27, bubbles: true })); """ ) except Exception: pass _type_detail_field(browser, "数量", qty) _type_detail_field(browser, "含税协议单价", price) _type_detail_field(browser, "无税协议价", net) _type_detail_field(browser, "承付参数", PAY_PARAM_DEFAULT) _type_detail_field(browser, "细单备注", remark) _type_detail_field(browser, "特定补偿预估值", COMPENSATION_DEFAULT) _type_detail_field(browser, "进货补偿预估值", COMPENSATION_DEFAULT) if not _pick_detail_option(browser, "承付模式", PAY_MODE_NEXT_MONTH_N, in_detail=True): raise NodeError("细单承付未点到下月N号结") _pick_detail_option(browser, "承付参数", PAY_PARAM_DEFAULT, in_detail=True) if not _pick_detail_option(browser, "采购结算方式", SETTLE_MODE_WIRE, in_detail=True): raise NodeError("细单结算未精确点到电汇") # 货品表是否已有该货品ID。重跑时跳过已存在的行。 def _goods_id_already_in_table(browser: BrowserAdapter, gid: str) -> bool: number = (gid or "").strip() if not number: return False return any(number in row for row in _goods_table_rows(browser)) # 按货品ID逐条新增细单，全部确认完再点货品工具栏保存。 # # 已在表里的货品ID跳过。不要每条都点工具栏保存，不要点总单保存。返回加进表的条数（含跳过）。 def add_detail_lines_by_goods_id( browser: BrowserAdapter, lines: list[RebateImportLine] | tuple[RebateImportLine, ...], *, save: bool = True, ) -> int: added = 0 created = 0 _close_header_card(browser) _close_detail_card(browser) for line in lines: gid = goods_search_id(line) if not gid: raise NodeError("细单货品ID必须用订单中心的 goodsId") if _goods_id_already_in_table(browser, gid): print("skip_detail_existing gid=", gid, flush=True) added += 1 continue if not _click_add_detail(browser): raise NodeError("选中总单后未出现货品新增") if not _pick_goods_id_in_dialog(browser, gid): raise NodeError(f"新增后未按货品ID搜到 {gid}，停下确认搜框是不是货品ID") _fill_last_detail_qty_price(browser, line) try: print( "card_btns=", browser.execute_script( """ return Array.from(document.querySelectorAll('.table-card .card-header-btn, .card-header-btn')).map((el) => ((el.innerText || '') + '').replace(/\\s+/g, ' ').trim()).filter(Boolean).slice(0, 8); """ ), flush=True, ) except Exception: pass confirmed = _confirm_detail_card(browser) print("confirm_detail=", confirmed, "gid=", gid, flush=True) if not confirmed: raise NodeError(f"细单 {gid} 未点到货品卡片「确认 Ctrl + Enter」，行还没落到货品表") if not _wait_goods_row(browser, gid): raise NodeError(f"细单确认后货品表没有 {gid}") added += 1 created += 1 if created: _close_detail_card(browser) if created and save: if not _native_click_goods_toolbar(browser, "保存"): if not _click_goods_toolbar(browser, "保存"): raise NodeError("未点到货品工具栏保存（新增旁边那颗），不要点总单保存") time.sleep(0.5) _confirm_save_popup(browser) err = _read_detail_toast(browser) if err: raise NodeError(err) return added # 等到能解析出导入成功/失败条数。新增路径用货品表行数。 def read_import_result(browser: BrowserAdapter, *, expected_lines: int = 0) -> RebateImportResult: deadline = time.time() + 20 last = RebateImportResult(0, 0, (), False) while time.time() <= deadline: last = _scan_import_result(browser, expected_lines=expected_lines) if last.parsed: return last time.sleep(0.4) return last # 查号 → 勾总单 → 按货品ID新增细单 → 保存。 # # 不做 Excel。写 ACCEPT SUMMARY REBATE_IMPORT。 def run_import_flow( browser: BrowserAdapter, ctx: RebateFlowContext, contract_no: str, draft: RebateImportDraft | None = None, *, download_dir: Path | None = None, ) -> RebateImportResult: _IMPORT_ACCEPT.reset() payload = draft or import_draft_for_contract(contract_no) dest = Path(download_dir) if download_dir else artifact_dir(IMPORT_FLOW) / "downloads" result = RebateImportResult(0, 0, (), False) try: open_rebate_center(browser, ctx) _IMPORT_ACCEPT.note("S02", "打开返利中心", {"url": ctx.rebate_url}, ok=True) try: login_rebate_center(browser, ctx) _IMPORT_ACCEPT.note("S03", "登录", {"user": ctx.username}, ok=True) except Exception as exc: evidence_shot(browser, "rebate-import-login-failed", failed=True) _IMPORT_ACCEPT.note( "S03", "登录", {"user": ctx.username, "error": exc}, ok=False, raise_error=False, ) raise enter_purchase_agreement(browser) _IMPORT_ACCEPT.note("S04", "进入采购协议", {"menu": "采购协议管理"}, ok=True) found = search_contract(browser, contract_no) _IMPORT_ACCEPT.note( "S05", "查号", {"contractNo": contract_no, "found": "yes" if found else "no"}, ok=found, ) if not found: raise NodeError("查号未搜到合同，不能加细单") _close_header_query(browser) picked = select_header_row_result(browser, contract_no) if not (isinstance(picked, dict) and picked.get("ok")): raise NodeError("未按合同编号勾中总单行") if not _wait_goods_add_button(browser): raise NodeError("点中总单后未出现货品新增，停下确认货品区是否已出") _close_header_card(browser) _IMPORT_ACCEPT.note("S06", "选中总单", {"selected": "yes"}, ok=True) added = add_detail_lines_by_goods_id(browser, payload.lines) first = payload.lines[0] _IMPORT_ACCEPT.note( "S07", "新增细单", { "lines": added, "goodsId": goods_search_id(first), "remark": _cell(first.order_no), "compensation": COMPENSATION_DEFAULT, }, ok=added == len(payload.lines), ) if added != len(payload.lines): raise NodeError(f"细单未加齐 {added}/{len(payload.lines)}") table_rows = _goods_table_rows(browser) missing = [ goods_search_id(line) for line in payload.lines if not any(goods_search_id(line) in row for row in table_rows) ] if missing: raise NodeError("货品表缺少 " + ",".join(missing)) _IMPORT_ACCEPT.note( "S08", "保存细单", {"saved": len(table_rows), "msg": f"{len(table_rows)}条货品已保存"}, ok=True, ) result = RebateImportResult(len(table_rows), 0, (), True) _IMPORT_ACCEPT.note( "S10", "读导入结果", {"success": result.success_count, "fail": result.fail_count}, ok=True, ) return result finally: print(_IMPORT_ACCEPT.summary_text(), end="", flush=True) _IMPORT_ACCEPT.dump()