linyuhang
/
AutoRpa
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: 支付落表取消查详情，细单ID自生成，按订单号判重 

Excel 没有平台细单号。落库时生成 pay_order_dtl_id 给订单中心防重传，本表按 pay_order_no 跳过已落订单。
This commit is contained in:
Yuuuu 2026-10-04 17:28:52 +08:00
parent bab6954584
commit 3257081523
11 changed files with 76 additions and 255 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
rpa_cart/README.md
View File

@ -21,9 +21,9 @@ uv run python main.py --days 0 # 仅当天（默认）
uv run python main.py --day 2026-09-30 --days 0
```
过滤：只落 待发货 / 待收货 / 已完成。单价用折后单价，总价 `round(折后单价×数量, 4)`。`pay_order_no` 用商城订单号，`pay_order_dtl_id` 用详情接口 `orderDetailId`。
过滤：只落 待发货 / 待收货 / 已完成。单价用折后单价，总价 `round(折后单价×数量, 4)`。`pay_order_no` 用商城订单号。`pay_order_dtl_id` 落库时自生成（UUID hex），不是平台数据，给订单中心防重传。
幂等：同一 `pay_order_no + pay_order_dtl_id` 不重复插入。细单号拉不到的留下，不猜。
幂等：同一 `pay_order_no` 不重复插入。Excel 没有细单 ID，不再打详情接口。
```bash
uv run pytest -v
@ -31,7 +31,7 @@ uv run pytest -v
## Windows 真库
Mac 打不到 `192.168.12.244`，**插入表 2 在 Windows 上测**。导出/解析在 Mac 已经跑过，Win 上要测的是登录、详情、落表。
Mac 打不到 `192.168.12.244`，**插入表 2 在 Windows 上测**。导出/解析在 Mac 已经跑过，Win 上要测的是登录、导出、落表。
Python 3.12 + Instant Client（位数一致）+ Chrome。密码不要写进仓库。把本包和 `rpa_selection/.env` 一起带过去。
@ -55,7 +55,7 @@ uv run python main.py --dry-run --day 2026-09-30 --days 0
uv run python main.py --day 2026-09-30 --days 0
```
看控制台 `inserted / unmatched`。细单号拉不到是 `NO_DTL_ID`。
看控制台 `inserted / skipped_dup / unmatched`。同一订单号再跑应 `skipped_dup` 上升。
查库：

7
rpa_cart/job.py
View File

@ -2,9 +2,8 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from settle import MallLine, attach_detail_ids, match_pay_lines
from settle import MallLine, match_pay_lines
@dataclass
@ -24,12 +23,10 @@ class PaySummary:
def run_yaoex_pay(
repo,
lines: list[MallLine],
details: dict[str, dict[str, Any]],
*,
dry_run: bool = False,
) -> PaySummary:
attached = attach_detail_ids(lines, details)
result = match_pay_lines(attached, repo.load_existing_pay_keys())
result = match_pay_lines(lines, repo.load_existing_pay_order_nos())
if result.rows and not dry_run:
repo.save_pay_rows(result.rows)
reasons: dict[str, int] = {}

24
rpa_cart/main.py
View File

@ -10,8 +10,7 @@ from db import RepoUnavailable
from job import run_yaoex_pay
from parse_xls import parse_order_xls
from repo import EmptyPayRepo, OraclePayRepo
from settle import PAID_STATUSES
from yaoex_client import day_range, export_orders, fetch_order_details
from yaoex_client import day_range, export_orders
from yaoex_login import load_cart_dotenv, resolve_yaoex_auth
PROJECT_ROOT = Path(__file__).resolve().parent
@ -39,9 +38,8 @@ def main(argv: list[str] | None = None, repo=None) -> None:
load_cart_dotenv()
xls_path = args.xls
tmp_xls: Path | None = None
token = ""
if xls_path is None or not args.dry_run:
if xls_path is None:
try:
auth = resolve_yaoex_auth(
secrets_path=args.secrets,
@ -55,10 +53,6 @@ def main(argv: list[str] | None = None, repo=None) -> None:
except Exception as exc:
print(f"1号药城登录失败：{exc}")
raise SystemExit(2) from exc
if xls_path is None:
if not token:
raise SystemExit("导出需要登录。")
end = date.fromisoformat(args.day) if args.day else date.today()
begin, end_ts = day_range(end, args.days)
print(f"导出 {begin} ~ {end_ts}")
@ -71,18 +65,6 @@ def main(argv: list[str] | None = None, repo=None) -> None:
lines = parse_order_xls(xls_path)
print(f"导出行 {len(lines)}")
order_nos = sorted({line.order_no for line in lines if line.status in PAID_STATUSES})
details: dict = {}
if order_nos:
if not token:
auth = resolve_yaoex_auth(
secrets_path=args.secrets,
auth_path=args.yaoex_auth,
fresh=args.fresh_login,
)
token = auth.token
details = fetch_order_details(token, order_nos)
print(f"拉取详情 {len(details)} 单")
if repo is None:
try:
@ -94,7 +76,7 @@ def main(argv: list[str] | None = None, repo=None) -> None:
print(f"无库，dry-run 只解析：{exc}")
repo = EmptyPayRepo()
summary = run_yaoex_pay(repo, lines, details, dry_run=args.dry_run)
summary = run_yaoex_pay(repo, lines, dry_run=args.dry_run)
print(
"落表 "
f"inserted={summary.inserted} "

17
rpa_cart/repo.py
View File

@ -7,13 +7,12 @@ from typing import Any
from db import CREATE_USER_NAME, RepoUnavailable, connect_from_env, oracle_dsn
from settle import YAOEX_PLATFORM_TYPE
_SELECT_PAY_KEYS = """
SELECT pay_order_no, pay_order_dtl_id
_SELECT_PAY_ORDER_NOS = """
SELECT DISTINCT pay_order_no
FROM rpa_cart_settlement_goods_data
WHERE operate_type = 2
AND platform_type = :platform_type
AND pay_order_no IS NOT NULL
AND pay_order_dtl_id IS NOT NULL
"""
_INSERT_PAY = """
@ -50,7 +49,7 @@ _PAY_BIND_KEYS = (
class EmptyPayRepo:
def load_existing_pay_keys(self) -> set[tuple[str, str]]:
def load_existing_pay_order_nos(self) -> set[str]:
return set()
def save_pay_rows(self, rows: list[dict[str, Any]]) -> None:
@ -66,16 +65,12 @@ class OraclePayRepo:
oracle_dsn()
return cls(connect=connect_from_env)
def load_existing_pay_keys(self) -> set[tuple[str, str]]:
def load_existing_pay_order_nos(self) -> set[str]:
conn = self._connect()
try:
with conn.cursor() as cur:
cur.execute(_SELECT_PAY_KEYS, {"platform_type": YAOEX_PLATFORM_TYPE})
return {
(str(order_no), str(dtl_id))
for order_no, dtl_id in cur.fetchall()
if order_no and dtl_id
}
cur.execute(_SELECT_PAY_ORDER_NOS, {"platform_type": YAOEX_PLATFORM_TYPE})
return {str(order_no) for (order_no,) in cur.fetchall() if order_no}
except Exception as exc:
raise RepoUnavailable(str(exc)) from exc
finally:

68
rpa_cart/settle.py
View File

@ -1,7 +1,8 @@
"""1 号药城导出行 → 表 2 operate_type=2。不对加购行。"""
from __future__ import annotations
from dataclasses import dataclass, replace
import uuid
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
from typing import Any
@ -16,6 +17,10 @@ def money4(value: object) -> Decimal:
return Decimal(str(value)).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)
def new_pay_dtl_id() -> str:
return uuid.uuid4().hex
@dataclass(frozen=True)
class MallLine:
order_no: str
@ -30,7 +35,6 @@ class MallLine:
qty: int
goods_amount: Decimal
order_paid: Decimal
pay_order_dtl_id: str = ""
@dataclass
@ -41,47 +45,21 @@ class MatchResult:
skipped_status: list[MallLine]
def attach_detail_ids(
lines: list[MallLine], details: dict[str, dict[str, Any]]
) -> list[MallLine]:
remaining: dict[str, list[dict[str, Any]]] = {}
for order_no, payload in details.items():
remaining[order_no] = list(payload.get("productList") or [])
attached: list[MallLine] = []
for line in lines:
pool = remaining.get(line.order_no) or []
idx = _find_product(pool, line)
if idx is None:
attached.append(line)
continue
product = pool.pop(idx)
dtl_id = product.get("orderDetailId")
attached.append(replace(line, pay_order_dtl_id="" if dtl_id is None else str(dtl_id)))
return attached
def match_pay_lines(
lines: list[MallLine],
existing_keys: set[tuple[str, str]],
existing_order_nos: set[str],
) -> MatchResult:
result = MatchResult(rows=[], unmatched=[], skipped_dup=[], skipped_status=[])
seen = set(existing_keys)
for line in lines:
if line.status not in PAID_STATUSES:
result.skipped_status.append(line)
continue
if not line.pay_order_dtl_id:
result.unmatched.append((line, "NO_DTL_ID"))
if line.order_no in existing_order_nos:
result.skipped_dup.append(line)
continue
if len(line.bar_code) > 40:
result.unmatched.append((line, "BAR_CODE_TOO_LONG"))
continue
key = (line.order_no, line.pay_order_dtl_id)
if key in seen:
result.skipped_dup.append(line)
continue
seen.add(key)
result.rows.append(build_pay_row(line))
return result
@ -101,34 +79,8 @@ def build_pay_row(line: MallLine) -> dict[str, Any]:
"status": None,
"fail_reason": None,
"pay_order_no": line.order_no,
"pay_order_dtl_id": line.pay_order_dtl_id,
"pay_order_dtl_id": new_pay_dtl_id(),
"push_status": PUSH_NOT_SENT,
"org_code": None,
"multi_platform_flag": 0,
}
def _find_product(pool: list[dict[str, Any]], line: MallLine) -> int | None:
for idx, product in enumerate(pool):
qty = int(Decimal(str(product.get("quantity") or 0)))
pay_price = money4(product.get("payPrice4") or 0)
if qty != line.qty:
continue
if pay_price != money4(line.pay_price):
continue
if not _goods_name_matches(product, line.goods_name):
continue
return idx
return None
def _goods_name_matches(product: dict[str, Any], goods_name: str) -> bool:
if not goods_name:
return True
product_name = str(product.get("productName") or "")
short_name = str(product.get("shortName") or "")
if not product_name and not short_name:
return True
if goods_name in {product_name, short_name}:
return True
return goods_name in product_name

44
rpa_cart/tests/test_job.py
View File

@ -9,7 +9,7 @@ class MemoryPayRepo:
self.existing = existing or set()
self.saved: list[dict] = []
def load_existing_pay_keys(self):
def load_existing_pay_order_nos(self):
return set(self.existing)
def save_pay_rows(self, rows):
@ -30,48 +30,32 @@ def _line(**overrides) -> MallLine:
qty=10,
goods_amount=Decimal("100.00"),
order_paid=Decimal("98.50"),
pay_order_dtl_id="",
)
data.update(overrides)
return MallLine(**data)
def test_job_attaches_dtl_and_inserts_without_cart():
def test_job_inserts_generated_dtl_without_details():
repo = MemoryPayRepo()
details = {
"ZQD1": {
"productList": [
{
"orderDetailId": 881,
"productName": "阿莫西林",
"quantity": 10,
"payPrice4": 9.85,
}
]
}
}
summary = run_yaoex_pay(repo, [_line()], details)
summary = run_yaoex_pay(repo, [_line()])
assert summary.inserted == 1
assert summary.unmatched == 0
assert repo.saved[0]["oms_id"] == 0
assert repo.saved[0]["pay_order_dtl_id"] == "881"
assert repo.saved[0]["pay_order_dtl_id"]
assert len(repo.saved[0]["pay_order_dtl_id"]) <= 40
assert repo.saved[0]["purchase_total"] == Decimal("98.5000")
def test_job_dry_run_does_not_insert():
repo = MemoryPayRepo()
details = {
"ZQD1": {
"productList": [
{
"orderDetailId": 881,
"productName": "阿莫西林",
"quantity": 10,
"payPrice4": 9.85,
}
]
}
}
summary = run_yaoex_pay(repo, [_line()], details, dry_run=True)
summary = run_yaoex_pay(repo, [_line()], dry_run=True)
assert summary.inserted == 1
assert repo.saved == []
def test_job_skips_existing_order_no():
repo = MemoryPayRepo(existing={"ZQD1"})
summary = run_yaoex_pay(repo, [_line()])
assert summary.inserted == 0
assert summary.skipped_dup == 1
assert repo.saved == []

1
rpa_cart/tests/test_parse_xls.py
View File

@ -65,7 +65,6 @@ def test_parse_order_xls_reads_settlement_columns(tmp_path: Path):
assert line.qty == 20
assert line.pay_price == Decimal("65.49")
assert line.unit_price == Decimal("65.49")
assert line.pay_order_dtl_id == ""
def test_parse_order_xls_takes_first_barcode_when_cell_has_many(tmp_path: Path):

13
rpa_cart/tests/test_repo.py
View File

@ -49,7 +49,7 @@ def test_save_pay_rows_inserts_operate_type_2():
"status": None,
"fail_reason": None,
"pay_order_no": "ZQD1",
"pay_order_dtl_id": "881",
"pay_order_dtl_id": "a1b2c3d4e5f6789012345678901234ab",
"push_status": 0,
"org_code": None,
"multi_platform_flag": 0,
@ -61,10 +61,10 @@ def test_save_pay_rows_inserts_operate_type_2():
assert executed[0][1]["operate_type"] == 2
assert executed[0][1]["oms_id"] == 0
assert executed[0][1]["pay_order_no"] == "ZQD1"
assert executed[0][1]["pay_order_dtl_id"] == "881"
assert executed[0][1]["pay_order_dtl_id"] == "a1b2c3d4e5f6789012345678901234ab"
def test_load_existing_pay_keys():
def test_load_existing_pay_order_nos():
class FakeCursor:
def __enter__(self):
return self
@ -74,9 +74,10 @@ def test_load_existing_pay_keys():
def execute(self, sql, params=None):
self.sql = sql
self.params = params
def fetchall(self):
return [("ZQD1", "881"), ("ZQD2", None)]
return [("ZQD1",), ("ZQD1",), (None,)]
class FakeConn:
def cursor(self):
@ -86,5 +87,5 @@ def test_load_existing_pay_keys():
pass
repo = OraclePayRepo(connect=lambda: FakeConn())
keys = repo.load_existing_pay_keys()
assert keys == {("ZQD1", "881")}
keys = repo.load_existing_pay_order_nos()
assert keys == {"ZQD1"}

99
rpa_cart/tests/test_settle.py
View File

@ -3,7 +3,6 @@ from decimal import Decimal
from settle import (
PAID_STATUSES,
MallLine,
attach_detail_ids,
build_pay_row,
match_pay_lines,
money4,
@ -24,7 +23,6 @@ def _line(**overrides) -> MallLine:
qty=200,
goods_amount=Decimal("2000.00"),
order_paid=Decimal("1970.00"),
pay_order_dtl_id="881",
)
data.update(overrides)
return MallLine(**data)
@ -43,7 +41,7 @@ def test_drops_cancelled_and_keeps_paid_statuses():
_line(status="已完成", order_no="P3"),
_line(status="退款/退换货", order_no="R1"),
]
result = match_pay_lines(lines, existing_keys=set())
result = match_pay_lines(lines, existing_order_nos=set())
kept = {row["pay_order_no"] for row in result.rows}
assert kept == {"P1", "P2", "P3"}
assert {line.order_no for line in result.skipped_status} == {"C1", "R1"}
@ -51,7 +49,7 @@ def test_drops_cancelled_and_keeps_paid_statuses():
def test_purchase_total_uses_discounted_price_times_qty():
result = match_pay_lines([_line()], existing_keys=set())
result = match_pay_lines([_line()], existing_order_nos=set())
assert len(result.rows) == 1
row = result.rows[0]
assert row["purchase_price"] == Decimal("9.8500")
@ -61,104 +59,61 @@ def test_purchase_total_uses_discounted_price_times_qty():
assert row["platform_type"] == 66
assert row["status"] is None
assert row["pay_order_no"] == "ZQD1"
assert row["pay_order_dtl_id"] == "881"
assert row["pay_order_dtl_id"]
assert len(row["pay_order_dtl_id"]) <= 40
assert row["oms_id"] == 0
assert row["org_code"] is None
assert row["multi_platform_flag"] == 0
def test_inserts_excel_line_without_cart_match():
result = match_pay_lines([_line(bar_code="999")], existing_keys=set())
result = match_pay_lines([_line(bar_code="999")], existing_order_nos=set())
assert len(result.rows) == 1
assert result.rows[0]["bar_code"] == "999"
assert result.rows[0]["oms_id"] == 0
assert result.unmatched == []
def test_same_barcode_two_lines_both_insert():
def test_same_order_two_lines_both_insert_with_distinct_dtl_ids():
lines = [
_line(order_no="A", qty=10, pay_price=Decimal("53.40"), pay_order_dtl_id="1"),
_line(order_no="A", qty=50, pay_price=Decimal("55.21"), pay_order_dtl_id="2"),
_line(order_no="A", qty=10, pay_price=Decimal("53.40")),
_line(order_no="A", qty=50, pay_price=Decimal("55.21")),
]
result = match_pay_lines(lines, existing_keys=set())
assert [row["pay_order_dtl_id"] for row in result.rows] == ["1", "2"]
result = match_pay_lines(lines, existing_order_nos=set())
assert len(result.rows) == 2
assert [row["pay_order_no"] for row in result.rows] == ["A", "A"]
ids = [row["pay_order_dtl_id"] for row in result.rows]
assert all(ids)
assert ids[0] != ids[1]
assert all(row["oms_id"] == 0 for row in result.rows)
def test_skips_existing_pay_order_and_dtl():
result = match_pay_lines([_line()], existing_keys={("ZQD1", "881")})
def test_skips_existing_order_no():
result = match_pay_lines([_line()], existing_order_nos={"ZQD1"})
assert result.rows == []
assert len(result.skipped_dup) == 1
def test_unmatched_without_dtl_is_not_inserted():
result = match_pay_lines([_line(pay_order_dtl_id="")], existing_keys=set())
assert result.rows == []
assert result.unmatched[0][1] == "NO_DTL_ID"
def test_existing_order_no_skips_all_its_lines_other_orders_insert():
lines = [
_line(order_no="OLD", qty=10),
_line(order_no="OLD", qty=20),
_line(order_no="NEW", qty=5),
]
result = match_pay_lines(lines, existing_order_nos={"OLD"})
assert [row["pay_order_no"] for row in result.rows] == ["NEW"]
assert [line.order_no for line in result.skipped_dup] == ["OLD", "OLD"]
def test_unmatched_barcode_over_40_is_not_inserted():
result = match_pay_lines(
[_line(bar_code="1" * 41)],
existing_keys=set(),
existing_order_nos=set(),
)
assert result.rows == []
assert result.unmatched[0][1] == "BAR_CODE_TOO_LONG"
def test_attach_detail_ids_matches_qty_and_pay_price():
lines = [
_line(pay_order_dtl_id="", qty=10, pay_price=Decimal("5.29"), goods_name="乳酶生片"),
_line(pay_order_dtl_id="", qty=50, pay_price=Decimal("55.21"), goods_name="阿德福韦酯胶囊"),
]
details = {
"ZQD1": {
"productList": [
{
"orderDetailId": 1001,
"productName": "乳酶生片",
"quantity": 10,
"payPrice4": 5.29,
},
{
"orderDetailId": 1002,
"productName": "阿德福韦酯胶囊",
"quantity": 50,
"payPrice4": 55.21,
},
]
}
}
attached = attach_detail_ids(lines, details)
assert [line.pay_order_dtl_id for line in attached] == ["1001", "1002"]
def test_attach_detail_ids_matches_short_name_when_product_name_has_brand():
lines = [
_line(
pay_order_dtl_id="",
qty=120,
pay_price=Decimal("35.26"),
goods_name="艾司奥美拉唑镁肠溶胶囊",
)
]
details = {
"ZQD1": {
"productList": [
{
"orderDetailId": 366807077,
"productName": "艾速平 艾司奥美拉唑镁肠溶胶囊",
"shortName": "艾司奥美拉唑镁肠溶胶囊",
"quantity": 120,
"payPrice4": 35.26,
}
]
}
}
attached = attach_detail_ids(lines, details)
assert attached[0].pay_order_dtl_id == "366807077"
def test_build_pay_row_from_excel_fields():
row = build_pay_row(_line())
assert row["oms_id"] == 0
@ -167,3 +122,5 @@ def test_build_pay_row_from_excel_fields():
assert row["supplier_name"] == "广东壹号药业有限公司"
assert row["bar_code"] == "6921665004933"
assert row["push_status"] == 0
assert row["pay_order_dtl_id"]
assert len(row["pay_order_dtl_id"]) <= 40

14
rpa_cart/tests/test_yaoex_client.py
View File

@ -1,6 +1,6 @@
from datetime import date
from yaoex_client import day_range, export_orders, fetch_order_details
from yaoex_client import day_range, export_orders
class FakeResponse:
@ -46,15 +46,3 @@ def test_export_orders_posts_json_params():
assert payload["token"] == "tok"
assert "ZQD" not in payload["jsonParams"]
assert "2026-10-03 00:00:00" in payload["jsonParams"]
def test_fetch_order_details_uses_order_id():
session = FakeSession(
FakeResponse(
json_data={"data": {"productList": [{"orderDetailId": 1}]}},
content_type="application/json",
)
)
details = fetch_order_details("tok", ["ZQD1"], session=session)
assert details["ZQD1"]["productList"][0]["orderDetailId"] == 1
assert session.calls[0][1]["jsonParams"] == '{"orderId": "ZQD1"}'

36
rpa_cart/yaoex_client.py
View File

@ -1,4 +1,4 @@
"""1 号药城订单导出与详情。"""
"""1 号药城订单导出。"""
from __future__ import annotations
import json
@ -13,7 +13,6 @@ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
GATEWAY = "https://gateway-b2b.fangkuaiyi.com"
EXPORT_URL = f"{GATEWAY}/api/order/export"
DETAIL_URL = f"{GATEWAY}/api/order/detail"
def day_range(day: date | None = None, days: int = 0) -> tuple[str, str]:
@ -63,39 +62,6 @@ def export_orders(
client.close()
def fetch_order_details(
token: str,
order_nos: list[str],
*,
session: requests.Session | None = None,
) -> dict[str, dict[str, Any]]:
own = session is None
client = session or requests.Session()
details: dict[str, dict[str, Any]] = {}
try:
for order_no in order_nos:
response = client.post(
DETAIL_URL,
data=_form(token, {"jsonParams": json.dumps({"orderId": order_no})}),
headers={
**_headers(),
"clientInfo": json.dumps({"clientAppVersionCode": 6996}),
},
timeout=60,
verify=False,
)
response.raise_for_status()
payload = response.json()
data = payload.get("data") if isinstance(payload, dict) else None
if not isinstance(data, dict):
data = payload if isinstance(payload, dict) else {}
details[order_no] = data
return details
finally:
if own:
client.close()
def _form(token: str, extra: dict[str, Any]) -> dict[str, Any]:
payload = {
"traderName": "yaoex_pc",