368 lines
14 KiB
Python
368 lines
14 KiB
Python
# 返利总单：表 3 MDM、计划员廖建琦、复用已有总单号。
|
||
|
||
from accept import accept_log_path
|
||
from apa_browser_adapter import chrome_debug_port, chrome_session_flow, resolve_chrome_profile_dir
|
||
from rebate_flow_steps import (
|
||
CONTACT_DEFAULT,
|
||
LOGIN,
|
||
PAY_MODE_NEXT_MONTH_N,
|
||
PAY_PARAM_DEFAULT,
|
||
PLANNER_DEFAULT,
|
||
RebateFlowContext,
|
||
_pick_person,
|
||
build_header_draft,
|
||
fill_rebate_header,
|
||
header_draft_from_table3,
|
||
import_lines_from_table3,
|
||
run_header_flow,
|
||
)
|
||
|
||
|
||
# 总单流程假浏览器，按脚本关键字回勾选和协议总单ID。
|
||
class FakeBrowser:
|
||
# 记下页面脚本回传的勾选、保存和协议号。
|
||
def __init__(
|
||
self,
|
||
*,
|
||
table_text="",
|
||
body_text="",
|
||
login_body="",
|
||
order_no="",
|
||
existing_order_no="",
|
||
):
|
||
self.table_text = table_text
|
||
self.body_text = body_text
|
||
self.login_body = login_body or body_text
|
||
self.order_no = order_no
|
||
self.existing_order_no = existing_order_no
|
||
self.saved = False
|
||
self.opened: list[str] = []
|
||
self.filled: list[tuple[list[str], str]] = []
|
||
self.selected: list[tuple[list[str], str]] = []
|
||
self.clicked: list[list[str]] = []
|
||
self.script_count = 0
|
||
|
||
# 记下打开过的 URL。
|
||
def open_url(self, url: str) -> None:
|
||
self.opened.append(url)
|
||
|
||
# 记下填过的选择器和值。
|
||
def fill_first(self, selectors: list[str], value: str) -> None:
|
||
self.filled.append((list(selectors), value))
|
||
|
||
# 记下点过的选择器。
|
||
def click_first(self, selectors: list[str]) -> None:
|
||
self.clicked.append(list(selectors))
|
||
|
||
# 记下原生点击，行为同 click_first。
|
||
def click_native(self, selectors: list[str]) -> None:
|
||
self.click_first(selectors)
|
||
|
||
# 假正文里没有这段字就抛 TimeoutError。
|
||
def wait_for_text(self, text: str, timeout_seconds: int = 15) -> None:
|
||
haystack = f"{self.body_text}\n{self.table_text}\n{self.login_body}"
|
||
if text not in haystack:
|
||
raise TimeoutError(text)
|
||
|
||
# 假页面一律当作元素已在。
|
||
def wait_for_element(self, selectors: list[str], timeout_seconds: int = 15) -> None:
|
||
return None
|
||
|
||
# 假页面一律当作元素已在。
|
||
def element_exists(self, selectors: list[str]) -> bool:
|
||
return True
|
||
|
||
# 返回构造时塞进的正文。
|
||
def read_body_text(self) -> str:
|
||
return self.body_text
|
||
|
||
# 按脚本标记回勾选、保存、协议号。
|
||
def execute_script(self, script: str, *args):
|
||
self.script_count += 1
|
||
text = script or ""
|
||
if "OPEN_HEADER_ROW" in text:
|
||
return {"ok": True, "via": "row"}
|
||
if "SELECT_HEADER_ROW" in text:
|
||
return {"ok": True, "text": str(args[0] if args else "")}
|
||
if "HEADER_SELECTION" in text:
|
||
no = str(args[0] if args else "")
|
||
return {"ok": True, "cells": [no], "contract": no, "checkedIdx": 0, "pager": "共1条"}
|
||
if "CLOSE_HEADER_QUERY" in text or "CLOSE_DETAIL_CARD" in text:
|
||
return {"ok": True}
|
||
if "CLICK_HEADER_TOOLBAR" in text:
|
||
title = str(args[0] if args else "")
|
||
self.clicked.append([title])
|
||
if title == "保存":
|
||
self.saved = True
|
||
return {"ok": True}
|
||
if "DETAIL_TOAST" in text:
|
||
return {"texts": []}
|
||
if "HEADER_CARD" in text:
|
||
return False
|
||
if "OPEN_LABELED_PICKER" in text:
|
||
return {"ok": True}
|
||
if "FILL_HEADER_LOOKUP" in text or "PICK_HEADER_LOOKUP" in text:
|
||
return {"ok": True}
|
||
if "CONFIRM_HEADER_CARD" in text:
|
||
return {"ok": True, "via": "确认"}
|
||
if "SET_HEADER_CELL" in text:
|
||
return {"ok": False}
|
||
if "SET_LABELED" in text:
|
||
label = args[0] if args else ""
|
||
value = args[1] if len(args) > 1 else ""
|
||
self.filled.append(([str(label)], str(value)))
|
||
return True
|
||
if "READ_LABELED_INPUT" in text:
|
||
title = str(args[0] if args else "")
|
||
if title == "承付参数":
|
||
return {"value": PAY_PARAM_DEFAULT}
|
||
if title == "计划员":
|
||
return {"value": "李四"}
|
||
return {"value": ""}
|
||
if "READ_REBATE_ORDER_NO" in text:
|
||
return self.existing_order_no or (self.order_no if self.saved else "")
|
||
return self.table_text
|
||
|
||
# 假截图，返回空路径。
|
||
def screenshot(self, name: str) -> str:
|
||
return name
|
||
|
||
# 假读取，返回空字符串。
|
||
def read_first(self, selectors: list[str], value_name: str = "") -> str:
|
||
return self.order_no
|
||
|
||
# 假切入 iframe，什么都不做。
|
||
def switch_to_frame(self, selectors: list[str]) -> None:
|
||
return None
|
||
|
||
# 假切回主文档，什么都不做。
|
||
def switch_to_default_content(self) -> None:
|
||
return None
|
||
|
||
# 记下选过的选择器和值。
|
||
def select_first(self, selectors: list[str], value: str) -> None:
|
||
self.selected.append((list(selectors), value))
|
||
|
||
# 记下 xform 选择。
|
||
def select_xform(self, name: str, value: str) -> None:
|
||
return None
|
||
|
||
# 记下上传路径。
|
||
def upload_file_first(self, selectors: list[str], file_path: str) -> None:
|
||
return None
|
||
|
||
|
||
# 构造一段假返利会话入参。
|
||
def _ctx() -> RebateFlowContext:
|
||
return RebateFlowContext(rebate_url="http://example", username="u", password="p")
|
||
|
||
|
||
# 构造一份带测试客商编码的总单草稿。
|
||
def _draft():
|
||
return build_header_draft(
|
||
planner_name="李四",
|
||
supplier_code="DW0100046361",
|
||
rebate_unit_code="901",
|
||
today="2026-10-09",
|
||
)
|
||
|
||
|
||
# 总单供应商用表 3 MDM，计划员仍是廖建琦，细单货品ID和备注来自表 3。
|
||
def test_header_draft_from_table3_uses_mdm_and_planner():
|
||
draft = header_draft_from_table3(
|
||
[
|
||
{
|
||
"supplier_id": "SUP001",
|
||
"supplier_mdm_code": "DW0100046361",
|
||
"planner_name": "李四",
|
||
"purchaser_name": "张三",
|
||
"org_code": "901",
|
||
"goods_id": "4000018689",
|
||
"qty": "10",
|
||
"unit_price": "10",
|
||
"oms_id": "20260928000003",
|
||
}
|
||
],
|
||
today="2026-10-12",
|
||
)
|
||
assert draft.supplier_code == "DW0100046361"
|
||
assert draft.planner_name == PLANNER_DEFAULT
|
||
assert draft.rebate_unit_code == "901"
|
||
assert draft.contact_name == CONTACT_DEFAULT
|
||
lines = import_lines_from_table3(
|
||
[
|
||
{
|
||
"goods_id": "4000018689",
|
||
"qty": "10",
|
||
"unit_price": "10",
|
||
"oms_id": "20260928000003",
|
||
"goods_source_id": "SRC1",
|
||
}
|
||
]
|
||
)
|
||
assert lines[0].goods_id == "4000018689"
|
||
assert lines[0].order_no == "20260928000003"
|
||
assert lines[0].tax_rate == "0.13"
|
||
assert lines[0].net_price == "8.8496"
|
||
|
||
|
||
# 总单默认走现行字段：下月N号结、协议起止当天、返利合约否。
|
||
def test_header_defaults_follow_mapping_not_scheme_55():
|
||
draft = _draft()
|
||
assert draft.planner_name == "李四"
|
||
assert draft.pay_mode == PAY_MODE_NEXT_MONTH_N
|
||
assert draft.protocol_start == "2026-10-09"
|
||
assert draft.protocol_end == "2026-10-09"
|
||
assert draft.rebate_contract == "否"
|
||
assert draft.settle_mode == "否"
|
||
|
||
|
||
# 计划员名为空时回退廖建琦，联系人苏文辉。
|
||
def test_header_draft_defaults_planner():
|
||
draft = build_header_draft(planner_name="", supplier_code="S1", rebate_unit_code="901")
|
||
assert draft.planner_name == PLANNER_DEFAULT
|
||
assert draft.contact_name == CONTACT_DEFAULT
|
||
|
||
|
||
# 填总单时用草稿里的计划员、结算和承付模式。
|
||
def test_fill_header_uses_mapping_values():
|
||
browser = FakeBrowser(
|
||
table_text="HT1 重庆亿昊",
|
||
body_text="返利合约 计划员 承付模式 保存",
|
||
)
|
||
fill_rebate_header(browser, _draft())
|
||
filled_values = [value for _, value in browser.filled]
|
||
selected_values = [value for _, value in browser.selected]
|
||
assert "李四" in filled_values
|
||
assert "2026-10-09" in filled_values
|
||
assert PAY_MODE_NEXT_MONTH_N in selected_values
|
||
assert "否" in selected_values
|
||
assert "廖建峰" not in filled_values
|
||
assert "账期" not in selected_values
|
||
assert "2026-12-31" not in filled_values
|
||
|
||
|
||
# 总单流程写 ACCEPT，带上协议总单ID。
|
||
def test_header_flow_writes_accept_with_order_no(tmp_path, monkeypatch):
|
||
monkeypatch.delenv("RPA_ACCEPT_LOG", raising=False)
|
||
monkeypatch.setattr("accept.SCREENSHOT_ROOT", tmp_path)
|
||
browser = FakeBrowser(
|
||
table_text="HT1 重庆亿昊",
|
||
body_text="采购协议管理 合同编号 查询 返利合约 承付模式 计划员 保存 总单号 FL20261009001",
|
||
login_body="采购协议管理",
|
||
order_no="FL20261009001",
|
||
)
|
||
order_no = run_header_flow(browser, _ctx(), "HT1", _draft())
|
||
assert order_no == "FL20261009001"
|
||
text = accept_log_path("REBATE_HEADER").read_text(encoding="utf-8")
|
||
assert "ACCEPT SUMMARY REBATE_HEADER 8/8 ok" in text
|
||
assert "ACCEPT S09 ok 读总单号" in text
|
||
assert "rebateOrderNo=FL20261009001" in text
|
||
assert "planner=李四" in text
|
||
assert f"payMode={PAY_MODE_NEXT_MONTH_N}" in text
|
||
assert "protocolEnd=2026-10-09" in text
|
||
|
||
|
||
# 表上已有协议总单ID则复用，不再新建。
|
||
def test_header_reuses_existing_order_no_without_create(tmp_path, monkeypatch):
|
||
monkeypatch.delenv("RPA_ACCEPT_LOG", raising=False)
|
||
monkeypatch.setattr("accept.SCREENSHOT_ROOT", tmp_path)
|
||
browser = FakeBrowser(
|
||
table_text="HT1 重庆亿昊 FL_EXISTING",
|
||
body_text="采购协议管理 合同编号 查询 返利合约 承付模式 计划员 保存 总单号 FL_EXISTING",
|
||
login_body="采购协议管理",
|
||
existing_order_no="FL_EXISTING",
|
||
)
|
||
order_no = run_header_flow(browser, _ctx(), "HT1", _draft())
|
||
assert order_no == "FL_EXISTING"
|
||
assert ["新增"] not in browser.clicked
|
||
assert ["保存"] in browser.clicked
|
||
text = accept_log_path("REBATE_HEADER").read_text(encoding="utf-8")
|
||
assert "rebateOrderNo=FL_EXISTING" in text
|
||
assert "reused=yes" in text
|
||
assert "ACCEPT S07 ok 填总单" in text
|
||
|
||
|
||
# 已登录会话不再填账密。
|
||
def test_header_reuses_logged_in_session(tmp_path, monkeypatch):
|
||
monkeypatch.delenv("RPA_ACCEPT_LOG", raising=False)
|
||
monkeypatch.setattr("accept.SCREENSHOT_ROOT", tmp_path)
|
||
browser = FakeBrowser(
|
||
table_text="HT1 重庆亿昊",
|
||
body_text="采购协议管理 待审批 合同编号 查询 返利合约 承付模式 计划员 保存",
|
||
login_body="采购协议管理",
|
||
order_no="FL1",
|
||
)
|
||
run_header_flow(browser, _ctx(), "HT1", _draft())
|
||
login_fills = [value for selectors, value in browser.filled if selectors == LOGIN["user"]]
|
||
assert login_fills == []
|
||
assert "u" not in [value for _, value in browser.filled]
|
||
|
||
|
||
# 总单和查号共用同一 Chrome 会话和端口。
|
||
def test_header_chrome_session_reuses_find():
|
||
assert chrome_session_flow("REBATE_HEADER") == "REBATE_FIND"
|
||
assert resolve_chrome_profile_dir("REBATE_HEADER") == resolve_chrome_profile_dir("REBATE_FIND")
|
||
assert chrome_debug_port("REBATE_HEADER") == chrome_debug_port("REBATE_FIND")
|
||
|
||
|
||
# 计划员三个点打开自己的选择层，不拿货源选择当层。
|
||
class PlannerPickBrowser:
|
||
# 记下计划员选择层是否打开、选中的姓名和 ID。
|
||
def __init__(self):
|
||
self.planner_name = ""
|
||
self.planner_id = ""
|
||
self.layer_open = False
|
||
self.scripts: list[str] = []
|
||
|
||
# 按脚本标记模拟计划员三个点弹层。
|
||
def execute_script(self, script: str, *args):
|
||
text = script or ""
|
||
self.scripts.append(text)
|
||
if "READ_LABELED_INPUT" in text:
|
||
title = str(args[0] if args else "")
|
||
if title == "计划员ID":
|
||
return {"value": self.planner_id}
|
||
if title == "计划员":
|
||
return {"value": self.planner_name}
|
||
return {"value": ""}
|
||
if "CLOSE_HEADER_QUERY" in text:
|
||
return {"ok": True}
|
||
if "OPEN_LABELED_PICKER" in text:
|
||
self.layer_open = True
|
||
return {"ok": True, "via": "more-outline"}
|
||
if "PERSON_PICKER_VISIBLE" in text:
|
||
return self.layer_open
|
||
if "FOCUS_PERSON_FIELD" in text:
|
||
return {"ok": True, "ph": "请输入计划员", "title": "计划员选择"}
|
||
if "CLICK_PERSON_QUERY" in text:
|
||
return {"ok": True, "via": "query"}
|
||
if "PERSON_PICKER_ROWS" in text:
|
||
return {"ok": True, "match": True, "count": 1, "row": "廖建琦"}
|
||
if "PICK_PERSON_ROW" in text:
|
||
self.planner_name = "廖建琦"
|
||
self.planner_id = "U1001"
|
||
self.layer_open = False
|
||
return {"ok": True, "text": "廖建琦"}
|
||
if "CLOSE_PERSON_PICKER" in text:
|
||
self.layer_open = False
|
||
return {"ok": True, "via": "close"}
|
||
if "SCAN_LOOKUP_OVERLAYS" in text:
|
||
return ["计划员选择"] if self.layer_open else []
|
||
if "SOURCE_CHOICE" in text or "CLOSE_SOURCE_CHOICE" in text:
|
||
return False
|
||
return None
|
||
|
||
|
||
# 计划员三个点按姓名选廖建琦并绑上 ID，不走货源细单。
|
||
def test_pick_person_selects_liao_and_binds_id():
|
||
browser = PlannerPickBrowser()
|
||
_pick_person(browser, "计划员", PLANNER_DEFAULT)
|
||
assert browser.planner_name == "廖建琦"
|
||
assert browser.planner_id == "U1001"
|
||
joined = "\n".join(browser.scripts)
|
||
assert "FOCUS_PERSON_FIELD" in joined
|
||
assert "PICK_PERSON_ROW" in joined
|
||
assert "FOCUS_SOURCE_FIELD" not in joined
|
||
assert "PICK_SOURCE_ROW" not in joined
|