Compare commits
1 Commits
ops/oracle
...
main
| Author | SHA1 | Date |
|---|---|---|
 Yuuuu
|
bcf5e03d07 |
16
AGENTS.md
16
AGENTS.md
|
|
@ -1,6 +1,6 @@
|
|||
# AutoRpa
|
||||
|
||||
洁达 RPA。当前主线是**订单中心入站**（`rpa_oms`）、**选品班次**（`rpa_selection/`）、**加购班次**（`rpa_cart/`）、**合同班次**（`rpa_contract/`）和**返利班次**（`rpa_rebate/`）：收件落 Oracle，请求里不跑选品、不录合同；选品另起进程捞 `deal_flag=0`；加购另起进程领表 2 成功行；合同另起进程领表 3；返利另起进程领表 4 起草行 / 表 3 只录协议行。根上不再有 `选品流程/`、`state_center/`、`rpa_yisaiqi/`。
|
||||
洁达 RPA。当前主线是**订单中心入站**（`rpa_oms`）、**选品班次**（`rpa_selection/`）、**加购班次**（`rpa_cart/`）、**合同班次**（`rpa_contract/`）和**返利班次**（`rpa_rebate/`）：收件落 Oracle，请求里不跑选品、不录合同；选品另起进程捞 `deal_flag=0`，凑齐后同一进程按本次拆定的 sku 加车；加购进程只补表 2 里仍是「系统管理员」或过期「加购中」的行；合同另起进程领表 3；返利另起进程领表 4 起草行 / 表 3 只录协议行。根上不再有 `选品流程/`、`state_center/`、`rpa_yisaiqi/`。
|
||||
|
||||
智力用在**定落点、想清楚对错**；速度和 token 靠下面这张地图定点打开，禁止用扫仓代替思考。改对应了，反工和消耗都会少。
|
||||
|
||||
|
|
@ -28,7 +28,7 @@
|
|||
| 接口3 收件契约 | `rpa_oms/app/inbound/purchase_contract.py` | `rpa_oms/tests/test_purchase_contract_receive.py::点名` |
|
||||
| 接口3 Oracle 落表 | `rpa_oms/app/inbound/oracle_contract_repo.py` | `rpa_oms/tests/test_oracle_contract_repo.py::点名` |
|
||||
| 选品规则 / 拆量 | `rpa_selection/rules.py`、`pipeline.py` | `rpa_selection/tests/test_pipeline.py::test_split_across_malls_is_success` |
|
||||
| 选品班次落表 | `rpa_selection/job.py`、`persist.py`、`repo.py` | `rpa_selection/tests/test_job.py::test_job_searches_barcode_not_approval_no` |
|
||||
| 选品班次落表 / 凑齐后加车 | `rpa_selection/job.py`、`persist.py`、`repo.py`；加车 `rpa_cart/cart_job.py` `add_known_lines` | `rpa_selection/tests/test_job.py::test_success_adds_each_split_sku_without_rereading_table2` |
|
||||
| 合同班合单落表 | `rpa_contract/job.py`、`group.py`、`persist.py`、`repo.py` | `rpa_contract/tests/test_job.py::test_job_merges_same_supplier_and_writes_table4` |
|
||||
| 合同页面步骤 | `rpa_contract/contract_login.py`（S02–S05）/ `contract_form.py`（S06–S08、S10）/ `contract_goods.py`（S09）/ `contract_upload.py`（S11）/ `contract_submit.py`（S12）/ `contract_number.py`（S13）；共用 `contract_common.py`；入口 `contract_flow_steps.py` | `ACCEPT SUMMARY` 或 `rpa_contract/screenshots/` 下 `ACCEPT.txt` |
|
||||
| 合同入参 / 客商草稿 | `rpa_contract/contract_draft.py` | `rpa_contract` 点名用例 |
|
||||
|
|
@ -81,7 +81,7 @@
|
|||
|
||||
## 选品落点
|
||||
|
||||
班次作业在 `rpa_selection/`，不在接口 1 请求里跑。从表 1 领 `deal_flag=0`，按**条形码**搜 1 号药城 / 药帮忙，两平台合并后按四级仓序贪心拆量，写入表 2。凑齐才落成功行；凑不齐不落部分成功。
|
||||
班次作业在 `rpa_selection/`，不在接口 1 请求里跑。从表 1 领 `deal_flag=0`，按**条形码**搜 1 号药城 / 药帮忙，两平台合并后按四级仓序贪心拆量，写入表 2。凑齐在内存里定，才落成功行；凑不齐不落部分成功。凑齐的新行同一班按这次拆定的 sku 加车，不再去表 2 重捞来判断齐没齐。表 2 已有该 `omsId` 时不加第二份。
|
||||
|
||||
口径：`docs/superpowers/specs/2026-10-02-selection-split-allocation-design.md`。
|
||||
|
||||
|
|
@ -89,17 +89,17 @@
|
|||
| --- | --- | --- |
|
||||
| 领取 | `claim_pending` | `deal_flag` 0→3；`FOR UPDATE SKIP LOCKED`；中途崩溃停在 3，超过租约（`deal_time` 30 分钟）可再领 |
|
||||
| 规则 | `allocate` / `run_selection` | 自营仓全称匹配、毛利严格大于 4%、效期 ≥ 12 月；同仓多个 sku 都参与拆量，低价先吃；按仓序拆量 |
|
||||
| 落表 | `build_result` + `save_result` | 成功每条拆量一行 `status=1`，合计=计划量；`push_status=0` |
|
||||
| 落表 | `build_result` + `save_result` | 成功每条拆量一行 `status=1`，合计=计划量；`push_status=0`。要马上加车的新行更新人写成「加购中」，并 RETURNING 回 fid |
|
||||
| 库存不足 | `KIND_BUSINESS_FAIL` | 两边都搜到但合计不够：`deal_flag=2`，该 `omsId` 不重跑 |
|
||||
| 对端故障 | `KIND_SYSTEM_FAIL` | 一边故障且剩下的凑不齐：可把 `deal_flag` 改回 0；禁止写成库存不足 |
|
||||
|
||||
湖南壹号、甘肃壹号云不在自营集合。不加购，不推接口 2。
|
||||
湖南壹号、甘肃壹号云不在自营集合。凑齐后按本次 sku 加车，不推接口 2。加购失败不回滚表 1 的选品成功。登录或查询失败把表 2 戳放回「系统管理员」。同单有成功又有失败时，失败原因带「未加齐」。
|
||||
|
||||
验收：`cd rpa_selection` → `uv run pytest tests/test_pipeline.py::test_split_across_malls_is_success -q`（杭州 3 + 广东 7 = 计划 10）。
|
||||
|
||||
## 加购落点
|
||||
|
||||
班次作业在 `rpa_cart/`，不在选品请求里跑。领表 2 `operate_type=1 AND status=1 AND push_status=0`，戳为 `系统管理员` 或过期的 `加购中`。登录 1 号药城 / 药帮忙，按条形码再搜，对上 `supplier_name`，按行加车。
|
||||
补偿班在 `rpa_cart/`。正常凑齐的加车发生在选品班里。本班领表 2 `operate_type=1 AND status=1 AND push_status=0`，戳为 `系统管理员` 或过期的 `加购中`。这些行没有 sku，登录 1 号药城 / 药帮忙后按条形码再搜，对上 `supplier_name`，按行加车。
|
||||
|
||||
| 步 | 函数 | 要点 |
|
||||
| --- | --- | --- |
|
||||
|
|
@ -171,8 +171,8 @@
|
|||
- 接口1：点名 `rpa_oms/tests/test_sales_goods_receive.py` 或 `test_oracle_repo.py`；禁止为「显得全面」跑整包再去连真库。
|
||||
- 接口2：点名 `rpa_oms/tests/test_cart_settlement_outbound.py::test_assembles_unpushed_rows_as_document_envelope`；SQL 截组 `tests/test_oracle_cart_repo.py::test_list_unpushed_keeps_whole_oms_group_when_limit_is_3`。
|
||||
- 接口3：点名 `rpa_oms/tests/test_purchase_contract_receive.py` 或 `test_oracle_contract_repo.py`。
|
||||
- 选品：点名 `rpa_selection/tests/test_pipeline.py::test_split_across_malls_is_success` 或 `test_rules.py::test_margin_exactly_4_percent_passes`。
|
||||
- 加购：点名 `rpa_cart/tests/test_cart_job.py::test_job_adds_split_rows_without_insert` 或 `test_job_marks_out_of_stock`。
|
||||
- 选品：点名 `rpa_selection/tests/test_pipeline.py::test_split_across_malls_is_success` 或 `test_rules.py::test_margin_exactly_4_percent_passes`。凑齐后加车点名 `rpa_selection/tests/test_job.py::test_success_adds_each_split_sku_without_rereading_table2`。
|
||||
- 加购：点名 `rpa_cart/tests/test_cart_job.py::test_job_adds_split_rows_without_insert` 或 `test_job_marks_out_of_stock`。选品交来的 sku 点名 `test_known_lines_keep_each_sku_and_flag_incomplete`。
|
||||
- 合同班：点名 `rpa_contract/tests/test_job.py::test_job_merges_same_supplier_and_writes_table4` 或 `test_job_marks_diligence_fail`。
|
||||
- 返利班：点名 `rpa_rebate/tests/test_job.py::test_job_finds_contract_and_writes_agreement` 或 `test_job_leaves_not_found_for_next_shift`。
|
||||
- 返利查号页：只读控制台 `ACCEPT SUMMARY REBATE_FIND` 或 `rpa_rebate/screenshots/REBATE_FIND/ACCEPT.txt`。
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@
|
|||
|
||||
## 加购
|
||||
|
||||
选品已经把「哪个仓、加几件」写成表 2：`operate_type=1`、`status=1`、`push_status=0`、`update_user_name=系统管理员`。本班读这些行，登录 1 号药城 / 药帮忙，按条形码再搜一次，对上 `supplier_name`，改数量，加入购物车。同一 `omsId` 拆两仓、两平台要加齐。
|
||||
选品凑齐的行在选品班里按本次拆定的 sku 直接加车，不经过这里。本命令只补还停在 `update_user_name=系统管理员`、或租约过期的「加购中」的行。那些行表 2 里没有 sku，仍按条形码再搜，对上 `supplier_name` 再加。同一 `omsId` 拆两仓、两平台要加齐。
|
||||
|
||||
成功不插新行，只把 `update_user_name` 改成 `加购班`。加不进去（登录挂、库存没了、中包装加不进、对不上仓）改该行 `status=2`，写失败原因。领取时先盖 `加购中`；中途崩溃停在这一戳。`update_time` 超过 30 分钟的「加购中」下一班会再领；未过期的需改回 `系统管理员` 再跑。
|
||||
|
||||
|
|
|
|||
|
|
@ -37,6 +37,23 @@ class CartSummary:
|
|||
failed: int = 0
|
||||
|
||||
|
||||
# 选品已经拆定的一行。加这行时不再搜、不再按商家名重对。
|
||||
@dataclass
|
||||
class KnownLine:
|
||||
fid: int
|
||||
oms_id: int
|
||||
platform_type: int
|
||||
supplier_name: str
|
||||
bar_code: str
|
||||
qty: int
|
||||
sku_id: str
|
||||
vendor_raw: str
|
||||
vendor_norm: str
|
||||
stock: float | None
|
||||
price: float | None
|
||||
raw: dict
|
||||
|
||||
|
||||
def _session(repo):
|
||||
"""有 session 就借一条，没有就空上下文。商城调用必须在这个块外面。"""
|
||||
session = getattr(repo, "session", None)
|
||||
|
|
@ -71,26 +88,120 @@ def run_cart_batch(
|
|||
(row, *_add_one(row, searchers, adders, login_errors, readers))
|
||||
for row in group
|
||||
]
|
||||
_apply_outcomes(repo, oms_id, outcomes, summary, audits)
|
||||
repo.save_shift_errors(audits)
|
||||
return summary
|
||||
|
||||
|
||||
def add_known_lines(
|
||||
repo,
|
||||
lines: list[KnownLine],
|
||||
*,
|
||||
adders: dict[int, Adder],
|
||||
login_errors: dict[int, str],
|
||||
cart_readers: dict[int, CartReader] | None = None,
|
||||
) -> CartSummary:
|
||||
"""按调用方已经拆定的 sku 加车并盖章。不再搜条形码，不再按商家名重对。
|
||||
|
||||
同一 oms_id 先全部加完再盖章。成功只改戳。业务失败改 status=2。
|
||||
同组有成功又有失败时，失败原因带「未加齐」。登录失败把戳放回系统管理员。
|
||||
行必须已经是「加购中」，否则盖章 SQL 不会改到它。
|
||||
"""
|
||||
audits: list[ShiftError] = []
|
||||
readers = cart_readers or {}
|
||||
summary = CartSummary(claimed=len(lines))
|
||||
for oms_id, group in _by_known(lines).items():
|
||||
outcomes = [
|
||||
(line, *_add_known(line, adders, login_errors, readers)) for line in group
|
||||
]
|
||||
_apply_outcomes(repo, oms_id, outcomes, summary, audits)
|
||||
repo.save_shift_errors(audits)
|
||||
return summary
|
||||
|
||||
|
||||
# 按盖章结果改表 2。加车必须发生在这个 session 外面。
|
||||
def _apply_outcomes(repo, oms_id, outcomes, summary: CartSummary, audits: list) -> None:
|
||||
incomplete = _is_incomplete(outcomes)
|
||||
if incomplete:
|
||||
logger.warning("加购未加齐 oms_id=%s", oms_id)
|
||||
with _session(repo):
|
||||
for row, reason, code, text in outcomes:
|
||||
for item, reason, code, text in outcomes:
|
||||
if reason is None:
|
||||
repo.mark_success(row.fid)
|
||||
repo.mark_success(item.fid)
|
||||
summary.added += 1
|
||||
continue
|
||||
if incomplete:
|
||||
reason = _incomplete_reason(reason)
|
||||
if _retryable(code, reason):
|
||||
repo.mark_release(row.fid, reason)
|
||||
repo.mark_release(item.fid, reason)
|
||||
else:
|
||||
repo.mark_fail(row.fid, reason)
|
||||
repo.mark_fail(item.fid, reason)
|
||||
summary.failed += 1
|
||||
if code and text:
|
||||
audits.append(ShiftError("加购", str(row.fid), code, text))
|
||||
repo.save_shift_errors(audits)
|
||||
return summary
|
||||
audits.append(ShiftError("加购", str(item.fid), code, text))
|
||||
|
||||
|
||||
# 按 oms_id 保序分组。同组先加完再盖章，才能看出有没有加齐。
|
||||
def _by_known(lines: list[KnownLine]) -> OrderedDict[int, list[KnownLine]]:
|
||||
groups: OrderedDict[int, list[KnownLine]] = OrderedDict()
|
||||
for line in lines:
|
||||
groups.setdefault(int(line.oms_id), []).append(line)
|
||||
return groups
|
||||
|
||||
|
||||
def _add_known(
|
||||
line: KnownLine,
|
||||
adders: dict[int, Adder],
|
||||
login_errors: dict[int, str],
|
||||
cart_readers: dict[int, CartReader],
|
||||
) -> tuple[str | None, str | None, str | None]:
|
||||
"""对一条已经拆定的行加车。返回 (失败原因, 审计短码, 原文)。成功三项都是 None。
|
||||
|
||||
不看选品当时的库存。库存以商城加车回包为准。购物车里已经不少于本行时直接成功。
|
||||
没有加车函数时放回待领，让补偿班再处理。
|
||||
"""
|
||||
platform = line.platform_type
|
||||
if platform in login_errors:
|
||||
text = (login_errors[platform] or "").strip()
|
||||
logger.warning("加购登录失败 platform=%s: %s", platform, login_errors[platform])
|
||||
return _system("LOGIN_FAILED"), ("LOGIN_FAILED" if text else None), (text or None)
|
||||
adder = adders.get(platform)
|
||||
if adder is None:
|
||||
return _system("LOGIN_FAILED"), "LOGIN_FAILED", "未配置加车"
|
||||
bar_code = (line.bar_code or "").strip()
|
||||
if not bar_code:
|
||||
return "缺少条形码", None, None
|
||||
qty = int(line.qty)
|
||||
if qty <= 0:
|
||||
return "采购数量无效", None, None
|
||||
reader = cart_readers.get(platform)
|
||||
if reader is not None:
|
||||
try:
|
||||
already = int(reader(bar_code, line.supplier_name) or 0)
|
||||
except Exception as exc:
|
||||
logger.exception("加购查购物车失败 platform=%s bar_code=%s", platform, bar_code)
|
||||
return _system("QUERY_FAILED"), "QUERY_FAILED", _text(exc)
|
||||
if already >= qty:
|
||||
return None, None, None
|
||||
offer = CartOffer(
|
||||
sku_id=line.sku_id,
|
||||
vendor_raw=line.vendor_raw,
|
||||
vendor_norm=line.vendor_norm,
|
||||
stock=line.stock,
|
||||
price=line.price,
|
||||
extra={"raw": line.raw} if isinstance(line.raw, dict) else {},
|
||||
)
|
||||
try:
|
||||
outcome = adder(offer, qty)
|
||||
except Exception as exc:
|
||||
logger.exception("加购加车失败 platform=%s fid=%s", platform, line.fid)
|
||||
return _system("ADD_FAILED"), "ADD_FAILED", _text(exc)
|
||||
if outcome.kind == KIND_OK:
|
||||
return None, None, None
|
||||
reason = (outcome.reason or "加购失败").strip()
|
||||
if outcome.kind == KIND_SYSTEM:
|
||||
return _system(reason), None, None
|
||||
return clip_reason(reason), None, None
|
||||
|
||||
|
||||
# 按 oms_id 保序分组。同组先加完再盖章，才能看出有没有加齐。
|
||||
|
|
|
|||
|
|
@ -3,7 +3,15 @@
|
|||
from contextlib import contextmanager
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from cart_job import KIND_BUSINESS, KIND_OK, KIND_SYSTEM, AddOutcome, run_cart_batch
|
||||
from cart_job import (
|
||||
KIND_BUSINESS,
|
||||
KIND_OK,
|
||||
KIND_SYSTEM,
|
||||
AddOutcome,
|
||||
KnownLine,
|
||||
add_known_lines,
|
||||
run_cart_batch,
|
||||
)
|
||||
from cart_match import CartOffer
|
||||
from cart_repo import STAMP_CLAIMING, STAMP_DONE, STAMP_SELECTION, MemoryCartRepo
|
||||
|
||||
|
|
@ -449,3 +457,93 @@ def test_rerun_skips_add_when_cart_already_has_the_line():
|
|||
assert repo.rows[1]["status"] == 1
|
||||
assert repo.rows[1]["update_user_name"] == STAMP_DONE
|
||||
assert repo.rows[1]["fail_reason"] is None
|
||||
|
||||
|
||||
def _known(fid: int, sku: str, **overrides) -> KnownLine:
|
||||
"""一条已经拆定的加购行。raw 里的 skuId 必须跟 sku 一起交到加车。"""
|
||||
data = dict(
|
||||
fid=fid,
|
||||
oms_id=2026100201,
|
||||
platform_type=YBM,
|
||||
supplier_name=HANGZHOU,
|
||||
bar_code="06901234567890",
|
||||
qty=5,
|
||||
sku_id=sku,
|
||||
vendor_raw=HANGZHOU,
|
||||
vendor_norm=HANGZHOU,
|
||||
stock=5.0,
|
||||
price=90.0,
|
||||
raw={"skuId": sku},
|
||||
)
|
||||
data.update(overrides)
|
||||
return KnownLine(**data)
|
||||
|
||||
|
||||
# 同仓两条 sku 各加各的，不按商家名并到低价那条。一条失败时只给失败行标未加齐。
|
||||
def test_known_lines_keep_each_sku_and_flag_incomplete():
|
||||
repo = MemoryCartRepo(
|
||||
[
|
||||
_row(
|
||||
fid=11,
|
||||
platform_type=YBM,
|
||||
supplier_name=HANGZHOU,
|
||||
purchase_qty=6,
|
||||
update_user_name=STAMP_CLAIMING,
|
||||
),
|
||||
_row(
|
||||
fid=12,
|
||||
platform_type=YBM,
|
||||
supplier_name=HANGZHOU,
|
||||
purchase_qty=4,
|
||||
update_user_name=STAMP_CLAIMING,
|
||||
),
|
||||
]
|
||||
)
|
||||
added: list[tuple[str, int]] = []
|
||||
|
||||
# 高价那条被商城拒绝。
|
||||
def add_ybm(offer: CartOffer, qty: int):
|
||||
added.append((offer.sku_id, qty))
|
||||
assert offer.extra["raw"]["skuId"] == offer.sku_id
|
||||
if offer.sku_id == "DEAR":
|
||||
return AddOutcome(kind=KIND_BUSINESS, reason="库存不足")
|
||||
return AddOutcome(kind=KIND_OK)
|
||||
|
||||
summary = add_known_lines(
|
||||
repo,
|
||||
[_known(11, "CHEAP", qty=6), _known(12, "DEAR", qty=4, price=92.0)],
|
||||
adders={YBM: add_ybm},
|
||||
login_errors={},
|
||||
)
|
||||
assert added == [("CHEAP", 6), ("DEAR", 4)]
|
||||
assert summary.added == 1
|
||||
assert summary.failed == 1
|
||||
assert repo.rows[11]["status"] == 1
|
||||
assert repo.rows[11]["update_user_name"] == STAMP_DONE
|
||||
assert repo.rows[12]["status"] == 2
|
||||
assert repo.rows[12]["fail_reason"] == "未加齐：库存不足"
|
||||
|
||||
|
||||
# 没配加车函数时放回系统管理员，status 仍是 1。
|
||||
def test_known_line_without_adder_releases_for_retry():
|
||||
repo = MemoryCartRepo([_row(fid=1, update_user_name=STAMP_CLAIMING)])
|
||||
summary = add_known_lines(
|
||||
repo,
|
||||
[
|
||||
_known(
|
||||
1,
|
||||
"S1",
|
||||
platform_type=YAOEX,
|
||||
supplier_name=GUANGDONG,
|
||||
vendor_raw=GUANGDONG,
|
||||
vendor_norm=GUANGDONG,
|
||||
)
|
||||
],
|
||||
adders={},
|
||||
login_errors={},
|
||||
)
|
||||
assert summary.added == 0
|
||||
assert summary.failed == 1
|
||||
assert repo.rows[1]["status"] == 1
|
||||
assert repo.rows[1]["update_user_name"] == STAMP_SELECTION
|
||||
assert "LOGIN_FAILED" in (repo.rows[1]["fail_reason"] or "")
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
# 洁达商城选品
|
||||
|
||||
班次作业：从表 1 捞 `deal_flag=0`，按**条形码**搜 1 号药城 / 药帮忙，规则选出仓，结果写入表 2。不加购，不推订单中心。
|
||||
班次作业：从表 1 捞 `deal_flag=0`，按**条形码**搜 1 号药城 / 药帮忙，规则选出仓，结果写入表 2。凑齐的行在本班按这次拆定的 sku 加车。不推订单中心。
|
||||
|
||||
口径：`docs/RPA订单中心接口开发口径_20260930.md` 第 3 节。
|
||||
|
||||
|
|
@ -13,6 +13,7 @@ rpa_sales_goods_mapping_data (deal_flag=0)
|
|||
→ 按条形码搜索（批准文号只落表，不拿去搜）
|
||||
→ 选品规则（两平台合并，按仓序拆量）
|
||||
→ rpa_cart_settlement_goods_data（operate_type=1, push_status=0）
|
||||
→ 凑齐的新行标成「加购中」，按本次 sku 加车，成功改「加购班」
|
||||
→ 表 1 deal_flag=1 成功 / 2 失败
|
||||
```
|
||||
|
||||
|
|
|
|||
|
|
@ -1,23 +1,49 @@
|
|||
# 选品班次：领取表 1 → 按条形码搜平台 → 统一拆量写表 2。
|
||||
#
|
||||
# 凑齐在内存里定。传入 cart_finish 时，新插入的成功行标成「加购中」，
|
||||
# 落库后用这次拆定的 sku 加车，不再去表 2 重捞。
|
||||
# 崩在 deal_flag=3：超过租约可再领，未过期需把 3 改回 0。
|
||||
# 查询失败原因列只写 QUERY_FAILED，原文进 rpa_selection_query_audit。
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Callable
|
||||
|
||||
from persist import STATUS_OK, build_result
|
||||
from mall_config import MALL_PLATFORM_TYPE
|
||||
from persist import DEAL_OK, STATUS_OK, build_result
|
||||
from pipeline import KIND_BUSINESS_FAIL, MallOutcome, SelectionResult, run_selection
|
||||
from repo import FinishWrite, MappingRow, QueryAudit
|
||||
from rules import Candidate, PurchaseLine, validate_purchase
|
||||
from repo import STAMP_SELECTION, FinishWrite, MappingRow, QueryAudit
|
||||
from rules import Allocation, Candidate, PurchaseLine, validate_purchase
|
||||
|
||||
STAMP_CLAIMING = "加购中"
|
||||
|
||||
Searcher = Callable[[str], list[Candidate]]
|
||||
logger = logging.getLogger("rpa_selection")
|
||||
|
||||
|
||||
# 凑齐后交给加车的一行。sku 和 raw 来自这次搜索，不从表 2 反查。
|
||||
@dataclass
|
||||
class CartLine:
|
||||
fid: int
|
||||
oms_id: int
|
||||
platform_type: int
|
||||
supplier_name: str
|
||||
bar_code: str
|
||||
qty: int
|
||||
sku_id: str
|
||||
vendor_raw: str
|
||||
vendor_norm: str
|
||||
stock: float | None
|
||||
price: float | None
|
||||
raw: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
# cart_finish(lines) -> (加购成功行数, 失败行数)。未传入则只落表。
|
||||
CartFinish = Callable[[list[CartLine]], tuple[int, int]]
|
||||
|
||||
|
||||
# 一个班次的领取和落表计数。
|
||||
@dataclass
|
||||
class JobSummary:
|
||||
|
|
@ -25,6 +51,8 @@ class JobSummary:
|
|||
added: int = 0
|
||||
skipped: int = 0
|
||||
failed: int = 0
|
||||
cart_added: int = 0
|
||||
cart_failed: int = 0
|
||||
|
||||
|
||||
# 把表 1 行收成规则层的采购行。缺价格或数量保持 None。
|
||||
|
|
@ -52,11 +80,14 @@ def run_batch(
|
|||
searchers: dict[str, Searcher],
|
||||
login_errors: dict[str, str],
|
||||
limit: int | None = None,
|
||||
cart_finish: CartFinish | None = None,
|
||||
) -> JobSummary:
|
||||
"""领取待选品行，按行搜平台、拆量，一行搜完就落这一行。
|
||||
|
||||
每个平台开搜前刷新 deal_time。落库前由仓库再确认仍是 deal_flag=3，
|
||||
且表 2 还没有该 oms_id。limit 交给领取层。查询失败原文在该行落表后另写。
|
||||
cart_finish 只在这一行新插入了凑齐的成功行之后调用，参数是这次拆定的 sku。
|
||||
表 2 已有该 oms_id 时不调用，避免再加一份。
|
||||
"""
|
||||
rows = repo.claim_pending(limit=limit)
|
||||
summary = JobSummary(claimed=len(rows))
|
||||
|
|
@ -78,7 +109,8 @@ def run_batch(
|
|||
)
|
||||
result = build_result(purchase, selection)
|
||||
_count(summary, result.cart_rows)
|
||||
repo.save_results(
|
||||
_mark_cart_stamp(result, cart_finish)
|
||||
saved = repo.save_results(
|
||||
[
|
||||
FinishWrite(
|
||||
oms_id=result.oms_id,
|
||||
|
|
@ -89,11 +121,92 @@ def run_batch(
|
|||
)
|
||||
]
|
||||
)
|
||||
fids = list(saved[0]) if saved else []
|
||||
if cart_finish is not None and result.deal_flag == DEAL_OK:
|
||||
_finish_cart(summary, cart_finish, purchase, selection, result.cart_rows, fids)
|
||||
if row_audits:
|
||||
repo.save_query_audits(row_audits)
|
||||
return summary
|
||||
|
||||
|
||||
# 要马上加车的成功行先标成加购中，补偿班才不会在加车完成前领走。其余行保持系统管理员。
|
||||
def _mark_cart_stamp(result, cart_finish: CartFinish | None) -> None:
|
||||
claim = cart_finish is not None and result.deal_flag == DEAL_OK and bool(result.cart_rows)
|
||||
for row in result.cart_rows:
|
||||
if claim and row.get("status") == STATUS_OK:
|
||||
row["update_user_name"] = STAMP_CLAIMING
|
||||
else:
|
||||
row["update_user_name"] = STAMP_SELECTION
|
||||
|
||||
|
||||
# 新插入的凑齐行交给加车。行数对不上就留下加购中，不拿错 sku 去加。
|
||||
def _finish_cart(
|
||||
summary: JobSummary,
|
||||
cart_finish: CartFinish | None,
|
||||
purchase: PurchaseLine,
|
||||
selection: SelectionResult,
|
||||
cart_rows: list[dict],
|
||||
fids: list[int],
|
||||
) -> None:
|
||||
if cart_finish is None or not fids:
|
||||
return
|
||||
if purchase.oms_id is None:
|
||||
return
|
||||
lines = _cart_lines(purchase, selection.allocations, cart_rows, fids)
|
||||
if not lines:
|
||||
logger.error(
|
||||
"选品加购行数对不上 oms_id=%s allocations=%s fids=%s",
|
||||
purchase.oms_id,
|
||||
len(selection.allocations),
|
||||
len(fids),
|
||||
)
|
||||
return
|
||||
try:
|
||||
added, failed = cart_finish(lines)
|
||||
except Exception:
|
||||
logger.exception("选品后加购未盖章 oms_id=%s", purchase.oms_id)
|
||||
return
|
||||
summary.cart_added += int(added)
|
||||
summary.cart_failed += int(failed)
|
||||
|
||||
|
||||
# 按拆量顺序配上刚插入的 fid。顺序或状态不一致就返回空，调用方不加车。
|
||||
def _cart_lines(
|
||||
purchase: PurchaseLine,
|
||||
allocations: list[Allocation],
|
||||
cart_rows: list[dict],
|
||||
fids: list[int],
|
||||
) -> list[CartLine]:
|
||||
if len(allocations) != len(fids) or len(cart_rows) != len(fids):
|
||||
return []
|
||||
if any(row.get("status") != STATUS_OK for row in cart_rows):
|
||||
return []
|
||||
lines: list[CartLine] = []
|
||||
for item, fid in zip(allocations, fids):
|
||||
candidate = item.candidate
|
||||
platform = MALL_PLATFORM_TYPE.get(candidate.mall_name)
|
||||
if platform is None:
|
||||
return []
|
||||
raw = candidate.extra.get("raw") if isinstance(candidate.extra, dict) else None
|
||||
lines.append(
|
||||
CartLine(
|
||||
fid=int(fid),
|
||||
oms_id=int(purchase.oms_id or 0),
|
||||
platform_type=int(platform),
|
||||
supplier_name=candidate.vendor_raw,
|
||||
bar_code=purchase.bar_code,
|
||||
qty=int(item.qty),
|
||||
sku_id=candidate.sku_id,
|
||||
vendor_raw=candidate.vendor_raw,
|
||||
vendor_norm=candidate.vendor_norm,
|
||||
stock=candidate.stock,
|
||||
price=candidate.price,
|
||||
raw=raw if isinstance(raw, dict) else {},
|
||||
)
|
||||
)
|
||||
return lines
|
||||
|
||||
|
||||
def _select_row(
|
||||
purchase: PurchaseLine,
|
||||
platforms: list[str],
|
||||
|
|
|
|||
|
|
@ -1,8 +1,9 @@
|
|||
# 选品班次入口：登录 1 号药城 / 药帮忙，按条形码搜，结果写入表 2。不加购，不推接口 2。
|
||||
# 选品班次入口：登录 1 号药城 / 药帮忙，按条形码搜，凑齐后按本次 sku 加车。不推接口 2。
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
|
|
@ -21,7 +22,12 @@ from auth_login import (
|
|||
)
|
||||
from rpa_oracle import RepoUnavailable, load_job_env
|
||||
from job import run_batch
|
||||
from mall_config import YAOEX_MALL_NAME, YBM_MALL_NAME
|
||||
from mall_config import (
|
||||
YAOEX_MALL_NAME,
|
||||
YAOEX_PLATFORM_TYPE,
|
||||
YBM_MALL_NAME,
|
||||
YBM_PLATFORM_TYPE,
|
||||
)
|
||||
from repo import OracleSelectRepo
|
||||
from rules import Candidate
|
||||
from search_yaoex import YaoexAuth
|
||||
|
|
@ -29,6 +35,7 @@ from search_yaoex import search_candidates as search_yaoex_candidates
|
|||
from search_ybm import search_candidates as search_ybm_candidates
|
||||
|
||||
DEFAULT_CLAIM_LIMIT = 50
|
||||
_CART_ROOT = Path(__file__).resolve().parent.parent / "rpa_cart"
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
|
|
@ -178,6 +185,15 @@ def main(argv: list[str] | None = None, repo=None) -> None:
|
|||
|
||||
searchers[YBM_MALL_NAME] = search_ybm
|
||||
|
||||
adders, login_by_type = _cart_adders(
|
||||
platforms,
|
||||
yaoex_auth=yaoex_auth,
|
||||
ybm_cookies=ybm_cookies,
|
||||
yaoex_session=yaoex_session,
|
||||
ybm_session=ybm_session,
|
||||
login_errors=login_errors,
|
||||
)
|
||||
cart_finish = _cart_finish(repo, adders, login_by_type)
|
||||
try:
|
||||
summary = run_batch(
|
||||
repo,
|
||||
|
|
@ -185,6 +201,7 @@ def main(argv: list[str] | None = None, repo=None) -> None:
|
|||
searchers=searchers,
|
||||
login_errors=login_errors,
|
||||
limit=args.limit,
|
||||
cart_finish=cart_finish,
|
||||
)
|
||||
finally:
|
||||
yaoex_session.close()
|
||||
|
|
@ -194,12 +211,122 @@ def main(argv: list[str] | None = None, repo=None) -> None:
|
|||
print(
|
||||
f"领取={summary.claimed} ADDED={summary.added} "
|
||||
f"SKIPPED={summary.skipped} FAILED={summary.failed} "
|
||||
f"加购成功={summary.cart_added} 加购失败={summary.cart_failed} "
|
||||
f"用时 {elapsed:.2f} 秒"
|
||||
)
|
||||
print("已写入 rpa_cart_settlement_goods_data，push_status=0，未推订单中心，未加购。")
|
||||
print("已写入 rpa_cart_settlement_goods_data，push_status=0，未推订单中心。")
|
||||
finally:
|
||||
runtime.close()
|
||||
|
||||
|
||||
# 把加购包放进导入路径。选品目录仍排在前面，避免拿错同名模块。
|
||||
def _ensure_cart_path() -> None:
|
||||
if _CART_ROOT.is_dir() and str(_CART_ROOT) not in sys.path:
|
||||
sys.path.append(str(_CART_ROOT))
|
||||
|
||||
|
||||
# 用本班已登录的会话加车。登录失败的平台不配加车函数。
|
||||
def _cart_adders(
|
||||
platforms: list[str],
|
||||
*,
|
||||
yaoex_auth,
|
||||
ybm_cookies,
|
||||
yaoex_session: requests.Session,
|
||||
ybm_session: requests.Session,
|
||||
login_errors: dict[str, str],
|
||||
) -> tuple[dict, dict[int, str]]:
|
||||
_ensure_cart_path()
|
||||
from cart_yaoex import add_yaoex_cart
|
||||
from cart_ybm import add_ybm_cart
|
||||
|
||||
adders: dict = {}
|
||||
if yaoex_auth is not None and YAOEX_MALL_NAME in platforms:
|
||||
|
||||
# 按选品拆定的 sku 加入 1 号药城购物车。
|
||||
def add_yaoex(offer, qty: int):
|
||||
return add_yaoex_cart(yaoex_auth, offer, qty, session=yaoex_session)
|
||||
|
||||
adders[YAOEX_PLATFORM_TYPE] = add_yaoex
|
||||
if ybm_cookies is not None and YBM_MALL_NAME in platforms:
|
||||
|
||||
# 按选品拆定的 sku 加入药帮忙购物车。
|
||||
def add_ybm(offer, qty: int):
|
||||
return add_ybm_cart(ybm_cookies, offer, qty, session=ybm_session)
|
||||
|
||||
adders[YBM_PLATFORM_TYPE] = add_ybm
|
||||
login_by_type: dict[int, str] = {}
|
||||
if YAOEX_MALL_NAME in login_errors:
|
||||
login_by_type[YAOEX_PLATFORM_TYPE] = login_errors[YAOEX_MALL_NAME]
|
||||
if YBM_MALL_NAME in login_errors:
|
||||
login_by_type[YBM_PLATFORM_TYPE] = login_errors[YBM_MALL_NAME]
|
||||
return adders, login_by_type
|
||||
|
||||
|
||||
# 内存选品仓库上的盖章。字段和加购班的成功、失败、放回一致。
|
||||
class _MemoryCartStamp:
|
||||
# 盖章写回选品仓库里刚插入的表 2 行。
|
||||
def __init__(self, select_repo):
|
||||
self._repo = select_repo
|
||||
self.shift_errors: list = []
|
||||
|
||||
# 成功只改戳，status 保持 1。
|
||||
def mark_success(self, fid: int) -> None:
|
||||
row = self._row(fid)
|
||||
row["update_user_name"] = "加购班"
|
||||
row["fail_reason"] = None
|
||||
|
||||
# 失败改 status=2，戳改成加购班。
|
||||
def mark_fail(self, fid: int, reason: str) -> None:
|
||||
row = self._row(fid)
|
||||
row["status"] = 2
|
||||
row["fail_reason"] = reason
|
||||
row["update_user_name"] = "加购班"
|
||||
|
||||
# 登录或查询失败放回待领。status 不动。
|
||||
def mark_release(self, fid: int, reason: str) -> None:
|
||||
row = self._row(fid)
|
||||
row["update_user_name"] = "系统管理员"
|
||||
row["fail_reason"] = reason
|
||||
|
||||
# 记下未捕获异常原文。不改表 2。
|
||||
def save_shift_errors(self, items) -> None:
|
||||
self.shift_errors.extend(items)
|
||||
|
||||
# 按刚插入的 fid 找表 2 行。找不到则 KeyError。
|
||||
def _row(self, fid: int) -> dict:
|
||||
for row in self._repo.cart:
|
||||
if int(row.get("fid") or 0) == int(fid):
|
||||
return row
|
||||
raise KeyError(fid)
|
||||
|
||||
|
||||
# 真库和内存选品仓库都能盖加购章。真库复用选品这条池，避免再开一条连接池。
|
||||
def _stamp_repo(select_repo):
|
||||
if isinstance(select_repo, OracleSelectRepo):
|
||||
_ensure_cart_path()
|
||||
from cart_repo import OracleCartRepo
|
||||
|
||||
return OracleCartRepo(select_repo._pool)
|
||||
return _MemoryCartStamp(select_repo)
|
||||
|
||||
|
||||
# 落库之后按拆定的行加车。返回加购成功行数和失败行数。
|
||||
def _cart_finish(select_repo, adders: dict, login_by_type: dict[int, str]):
|
||||
_ensure_cart_path()
|
||||
from cart_job import add_known_lines
|
||||
|
||||
# 同一单的拆量行一次加完。
|
||||
def finish(lines):
|
||||
summary = add_known_lines(
|
||||
_stamp_repo(select_repo),
|
||||
lines,
|
||||
adders=adders,
|
||||
login_errors=login_by_type,
|
||||
)
|
||||
return summary.added, summary.failed
|
||||
|
||||
return finish
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@
|
|||
#
|
||||
# 从表 1 领取 deal_flag=0 的行改成 3，选品成功后写表 2、表 1 改成 1。
|
||||
# 进程中途退出时行停在 3，超过租约（deal_time）才能再领。
|
||||
# 不处理加购，不推接口 2。
|
||||
# 表 2 的更新人由落库行带入。凑齐后的加车在班次里做，不在这个仓库里。
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -51,7 +51,8 @@ SET deal_flag = 3,
|
|||
WHERE fid IN ({placeholders})
|
||||
"""
|
||||
|
||||
# 选品无登录用户：创建人/更新人固定 0 / 系统管理员，时间用库时钟。
|
||||
# 选品无登录用户：创建人固定 0 / 系统管理员。更新人由调用方传入。
|
||||
# 凑齐后马上加车的行传入「加购中」，避免补偿班在加车完成前领走。fid 由触发器生成，RETURNING 取回。
|
||||
_INSERT_CART_SQL = """
|
||||
INSERT INTO rpa_cart_settlement_goods_data (
|
||||
operate_type, oms_id, platform_type, supplier_name, bar_code,
|
||||
|
|
@ -64,10 +65,13 @@ INSERT INTO rpa_cart_settlement_goods_data (
|
|||
:purchase_price, :purchase_qty, :purchase_total, :status, :fail_reason,
|
||||
:push_status, :org_code, :multi_platform_flag,
|
||||
SYSDATE, 0, '系统管理员',
|
||||
SYSDATE, 0, '系统管理员'
|
||||
SYSDATE, 0, :update_user_name
|
||||
)
|
||||
RETURNING fid INTO :out_fid
|
||||
"""
|
||||
|
||||
STAMP_SELECTION = "系统管理员"
|
||||
|
||||
_COMPLETE_SQL = """
|
||||
UPDATE rpa_sales_goods_mapping_data
|
||||
SET deal_flag = :deal_flag,
|
||||
|
|
@ -193,6 +197,7 @@ class MemorySelectRepo:
|
|||
self.mapping: dict[int, MappingRow] = {row.oms_id: row for row in rows}
|
||||
self.cart: list[dict[str, Any]] = []
|
||||
self.query_audits: list[QueryAudit] = []
|
||||
self._cart_fid = 0
|
||||
|
||||
# 领取待选品任务，把 deal_flag 从 0 或过期 3 改成 3。
|
||||
#
|
||||
|
|
@ -241,23 +246,40 @@ class MemorySelectRepo:
|
|||
return
|
||||
row.deal_time = now or datetime.now()
|
||||
|
||||
def save_results(self, items: list[FinishWrite]) -> None:
|
||||
def save_results(self, items: list[FinishWrite]) -> list[list[int]]:
|
||||
"""把选品结果写入内存表 2，并按 fid 改表 1。
|
||||
|
||||
落库前再读：仍是 deal_flag=3，并且表 2 还没有该 oms_id，才插入。
|
||||
已经不是本班领的，整单跳过。已有加购行时只改表 1，不插第二套。
|
||||
返回值与 items 对齐：每一单新插入的表 2 fid，顺序与 cart_rows 一致。未插入则为空。
|
||||
"""
|
||||
inserted: list[list[int]] = []
|
||||
for item in items:
|
||||
row = self._row_by_fid(item.fid)
|
||||
if row.oms_id != item.oms_id:
|
||||
raise KeyError(item.oms_id)
|
||||
if row.deal_flag != 3:
|
||||
inserted.append([])
|
||||
continue
|
||||
exists = any(int(cart.get("oms_id") or 0) == int(item.oms_id) for cart in self.cart)
|
||||
fids: list[int] = []
|
||||
if item.cart_rows and not exists:
|
||||
self.cart.extend(dict(cart_row) for cart_row in item.cart_rows)
|
||||
for cart_row in item.cart_rows:
|
||||
stored = dict(cart_row)
|
||||
stored["fid"] = self._alloc_cart_fid()
|
||||
stored.setdefault("update_user_name", STAMP_SELECTION)
|
||||
self.cart.append(stored)
|
||||
fids.append(int(stored["fid"]))
|
||||
row.deal_flag = item.deal_flag
|
||||
row.deal_result = item.deal_result
|
||||
inserted.append(fids)
|
||||
return inserted
|
||||
|
||||
# 给新插入的内存表 2 行配一个 fid。已有行的 fid 不重用。
|
||||
def _alloc_cart_fid(self) -> int:
|
||||
used = [int(row["fid"]) for row in self.cart if row.get("fid") is not None]
|
||||
self._cart_fid = max([self._cart_fid, *used, 0]) + 1
|
||||
return self._cart_fid
|
||||
|
||||
# 记下查询失败原文。不改表 1 / 表 2。
|
||||
def save_query_audits(self, items: list[QueryAudit]) -> None:
|
||||
|
|
@ -369,29 +391,29 @@ class OracleSelectRepo:
|
|||
except Exception as exc:
|
||||
raise RepoUnavailable("数据库不可用") from exc
|
||||
|
||||
def save_results(self, items: list[FinishWrite]) -> None:
|
||||
def save_results(self, items: list[FinishWrite]) -> list[list[int]]:
|
||||
"""把这一调用里的表 2 行和表 1 完成更新落在同一个事务。
|
||||
|
||||
每一单先再读：仍是 deal_flag=3，并且表 2 还没有该 oms_id，才插入。
|
||||
已有加购行时只改表 1。空列表不借连接。任一行失败整笔回滚。
|
||||
库故障对外「数据库不可用」，不含 ORA 原文。
|
||||
返回值与 items 对齐：每一单新插入的表 2 fid，顺序与 cart_rows 一致。未插入则为空。
|
||||
"""
|
||||
if not items:
|
||||
return
|
||||
return []
|
||||
inserted: list[list[int]] = [[] for _ in items]
|
||||
try:
|
||||
with self._pool.acquire() as conn:
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cart_binds: list[dict] = []
|
||||
complete_binds: list[dict] = []
|
||||
for item in items:
|
||||
for index, item in enumerate(items):
|
||||
if not _claim_still_open(cur, item.fid):
|
||||
continue
|
||||
if item.cart_rows and not _cart_already_written(cur, item.oms_id):
|
||||
cart_binds.extend(
|
||||
{key: cart_row.get(key) for key in _CART_BIND_KEYS}
|
||||
for cart_row in item.cart_rows
|
||||
)
|
||||
inserted[index] = [
|
||||
_insert_cart_row(cur, cart_row) for cart_row in item.cart_rows
|
||||
]
|
||||
complete_binds.append(
|
||||
{
|
||||
"deal_flag": item.deal_flag,
|
||||
|
|
@ -400,11 +422,10 @@ class OracleSelectRepo:
|
|||
"fid": item.fid,
|
||||
}
|
||||
)
|
||||
if cart_binds:
|
||||
cur.executemany(_INSERT_CART_SQL, cart_binds)
|
||||
if complete_binds:
|
||||
cur.executemany(_COMPLETE_SQL, complete_binds)
|
||||
conn.commit()
|
||||
return inserted
|
||||
except Exception as exc:
|
||||
_rollback(conn)
|
||||
if isinstance(exc, RepoUnavailable):
|
||||
|
|
@ -451,6 +472,26 @@ def _cart_already_written(cur, oms_id: int) -> bool:
|
|||
return int(rows[0][0]) > 0
|
||||
|
||||
|
||||
# 插入一行表 2 并取回触发器生成的 fid。缺更新人时写成系统管理员。
|
||||
def _insert_cart_row(cur, cart_row: dict[str, Any]) -> int:
|
||||
params = {key: cart_row.get(key) for key in _CART_BIND_KEYS}
|
||||
params["update_user_name"] = cart_row.get("update_user_name") or STAMP_SELECTION
|
||||
out = cur.var(int)
|
||||
params["out_fid"] = out
|
||||
cur.execute(_INSERT_CART_SQL, params)
|
||||
return _fid_from_var(out)
|
||||
|
||||
|
||||
# RETURNING 的值可能是单元素列表，也可能是标量。空值当落库失败。
|
||||
def _fid_from_var(var) -> int:
|
||||
value = var.getvalue()
|
||||
if isinstance(value, list):
|
||||
value = value[0] if value else None
|
||||
if value is None:
|
||||
raise RepoUnavailable("数据库不可用")
|
||||
return int(value)
|
||||
|
||||
|
||||
# 返回截断原文后的审计行。
|
||||
def _clipped_audit(item: QueryAudit) -> QueryAudit:
|
||||
return QueryAudit(
|
||||
|
|
|
|||
|
|
@ -183,7 +183,10 @@ def map_yaoex_products(products: list[dict[str, Any]], user_id: Any) -> list[Can
|
|||
vendor_raw=vendor_raw,
|
||||
vendor_norm=normalize_vendor(vendor_raw),
|
||||
stock=parse_float_or_none(product.get("stockCount")),
|
||||
extra={"stockCountDesc": product.get("stockCountDesc", "")},
|
||||
extra={
|
||||
"stockCountDesc": product.get("stockCountDesc", ""),
|
||||
"raw": product,
|
||||
},
|
||||
)
|
||||
)
|
||||
return candidates
|
||||
|
|
|
|||
|
|
@ -85,6 +85,9 @@ def map_ybm_products(rows: list[dict[str, Any]]) -> list[Candidate]:
|
|||
raw_price = 0 if no_stock else pick_ybm_platform_price(product)
|
||||
vendor_raw = normalize_text(product.get("shopName"))
|
||||
expiry_raw = normalize_text(product.get("nearEffect"))
|
||||
raw = dict(product) if isinstance(product, dict) else {}
|
||||
if not raw.get("skuId") and row.get("skuId"):
|
||||
raw["skuId"] = row.get("skuId")
|
||||
candidates.append(
|
||||
Candidate(
|
||||
mall_name=YBM_MALL_NAME,
|
||||
|
|
@ -97,7 +100,10 @@ def map_ybm_products(rows: list[dict[str, Any]]) -> list[Candidate]:
|
|||
vendor_raw=vendor_raw,
|
||||
vendor_norm=normalize_vendor(vendor_raw),
|
||||
stock=parse_float_or_none(available_qty),
|
||||
extra={"库存说明": "无货" if no_stock else "有货"},
|
||||
extra={
|
||||
"库存说明": "无货" if no_stock else "有货",
|
||||
"raw": raw,
|
||||
},
|
||||
)
|
||||
)
|
||||
return candidates
|
||||
|
|
|
|||
|
|
@ -1,10 +1,19 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
import main
|
||||
_CART_ROOT = Path(__file__).resolve().parents[2] / "rpa_cart"
|
||||
if str(_CART_ROOT) not in sys.path:
|
||||
sys.path.append(str(_CART_ROOT))
|
||||
|
||||
import cart_ybm # noqa: E402
|
||||
import cart_yaoex # noqa: E402
|
||||
|
||||
import main # noqa: E402
|
||||
from job import run_batch
|
||||
from mall_config import YAOEX_MALL_NAME, YBM_MALL_NAME
|
||||
from repo import MappingRow, MemorySelectRepo
|
||||
|
|
@ -80,10 +89,20 @@ def test_main_writes_cart_rows_from_mapping(monkeypatch):
|
|||
)
|
||||
]
|
||||
|
||||
# 挡住真实加车。入口应在落表后按拆定的 sku 调到这里。
|
||||
def fake_add(_auth, offer, qty, **_kwargs):
|
||||
from cart_add import KIND_OK, AddOutcome
|
||||
|
||||
assert offer.sku_id == "S1"
|
||||
assert qty == 10
|
||||
return AddOutcome(kind=KIND_OK)
|
||||
|
||||
monkeypatch.setattr(main, "login_yaoex", fake_login_yaoex)
|
||||
monkeypatch.setattr(main, "login_ybm", fake_login_ybm)
|
||||
monkeypatch.setattr(main, "search_yaoex_candidates", fake_search_yaoex)
|
||||
monkeypatch.setattr(main, "search_ybm_candidates", fake_search_ybm)
|
||||
monkeypatch.setattr(cart_yaoex, "add_yaoex_cart", fake_add)
|
||||
monkeypatch.setattr(cart_ybm, "add_ybm_cart", fake_add)
|
||||
|
||||
main.main(["--fresh-login"], repo=repo)
|
||||
|
||||
|
|
@ -92,6 +111,7 @@ def test_main_writes_cart_rows_from_mapping(monkeypatch):
|
|||
assert repo.cart[0]["platform_type"] == 66
|
||||
assert repo.cart[0]["status"] == 1
|
||||
assert repo.cart[0]["bar_code"] == "06901234567890"
|
||||
assert repo.cart[0]["update_user_name"] == "加购班"
|
||||
|
||||
|
||||
def test_default_limit_caps_claim_at_50(monkeypatch):
|
||||
|
|
@ -120,7 +140,14 @@ def test_default_limit_caps_claim_at_50(monkeypatch):
|
|||
seen["limit"] = kwargs.get("limit")
|
||||
claimed = bound_repo.claim_pending(limit=kwargs.get("limit"))
|
||||
seen["count"] = len(claimed)
|
||||
return SimpleNamespace(claimed=len(claimed), added=0, skipped=0, failed=0)
|
||||
return SimpleNamespace(
|
||||
claimed=len(claimed),
|
||||
added=0,
|
||||
skipped=0,
|
||||
failed=0,
|
||||
cart_added=0,
|
||||
cart_failed=0,
|
||||
)
|
||||
|
||||
def no_browser(*_args, **_kwargs):
|
||||
"""挡住浏览器登录。"""
|
||||
|
|
|
|||
|
|
@ -122,7 +122,7 @@ def test_job_saves_each_claimed_row_before_the_next():
|
|||
def save_results(self, items):
|
||||
"""记下这一次落库，再交给内存仓库。"""
|
||||
self.batches.append(list(items))
|
||||
super().save_results(items)
|
||||
return super().save_results(items)
|
||||
|
||||
repo = _Repo(
|
||||
[
|
||||
|
|
@ -362,3 +362,116 @@ def test_job_query_failed_writes_short_code_not_exception():
|
|||
assert audit.mall_name == "1号药城"
|
||||
assert audit.bar_code == "06901234567890"
|
||||
assert audit.error_text == "ORA-03114 secret-traceback"
|
||||
|
||||
|
||||
# 凑齐后按这次拆定的 sku 加车，同仓两条都不并成低价那条。
|
||||
def test_success_adds_each_split_sku_without_rereading_table2():
|
||||
repo = MemorySelectRepo([_row()])
|
||||
seen: list[tuple[str, int, str]] = []
|
||||
|
||||
# 同仓低价 6、高价 10，计划 10，应拆成 6 + 4。
|
||||
def search_ybm(_key: str):
|
||||
return [
|
||||
_ok_candidate(
|
||||
"药帮忙",
|
||||
vendor=HANGZHOU,
|
||||
sku_id="CHEAP",
|
||||
price=90.0,
|
||||
stock=6.0,
|
||||
extra={"raw": {"skuId": "CHEAP"}},
|
||||
),
|
||||
_ok_candidate(
|
||||
"药帮忙",
|
||||
vendor=HANGZHOU,
|
||||
sku_id="DEAR",
|
||||
price=92.0,
|
||||
stock=10.0,
|
||||
extra={"raw": {"skuId": "DEAR"}},
|
||||
),
|
||||
]
|
||||
|
||||
# 记下加车入参。行必须已经落成加购中，fid 来自这次插入。
|
||||
def finish(lines):
|
||||
assert [row["update_user_name"] for row in repo.cart] == ["加购中", "加购中"]
|
||||
for line in lines:
|
||||
seen.append((line.sku_id, line.qty, line.raw.get("skuId")))
|
||||
matched = next(row for row in repo.cart if row["fid"] == line.fid)
|
||||
matched["update_user_name"] = "加购班"
|
||||
return len(lines), 0
|
||||
|
||||
summary = run_batch(
|
||||
repo,
|
||||
platforms=["药帮忙"],
|
||||
searchers={"药帮忙": search_ybm},
|
||||
login_errors={},
|
||||
cart_finish=finish,
|
||||
)
|
||||
assert summary.added == 2
|
||||
assert summary.cart_added == 2
|
||||
assert summary.cart_failed == 0
|
||||
assert seen == [("CHEAP", 6, "CHEAP"), ("DEAR", 4, "DEAR")]
|
||||
assert [row["update_user_name"] for row in repo.cart] == ["加购班", "加购班"]
|
||||
assert repo.mapping[2026100201].deal_flag == 1
|
||||
|
||||
|
||||
# 凑不齐不加车，失败行戳保持系统管理员。
|
||||
def test_business_fail_does_not_add_cart():
|
||||
repo = MemorySelectRepo([_row()])
|
||||
called = {"n": 0}
|
||||
|
||||
def search_ybm(_key: str):
|
||||
return [_ok_candidate("药帮忙", vendor=HANGZHOU, sku_id="HZ", stock=2.0)]
|
||||
|
||||
def finish(_lines):
|
||||
called["n"] += 1
|
||||
return 0, 0
|
||||
|
||||
summary = run_batch(
|
||||
repo,
|
||||
platforms=["药帮忙"],
|
||||
searchers={"药帮忙": search_ybm},
|
||||
login_errors={},
|
||||
cart_finish=finish,
|
||||
)
|
||||
assert called["n"] == 0
|
||||
assert summary.cart_added == 0
|
||||
assert summary.added == 0
|
||||
assert repo.mapping[2026100201].deal_flag == 2
|
||||
assert repo.cart
|
||||
assert all(row["update_user_name"] == "系统管理员" for row in repo.cart)
|
||||
|
||||
|
||||
# 表 2 已有这单时不再加一份。
|
||||
def test_existing_cart_rows_are_not_added_again():
|
||||
repo = MemorySelectRepo(
|
||||
[_row(deal_flag=3, deal_time=datetime.now() - timedelta(minutes=31))]
|
||||
)
|
||||
repo.cart.append(
|
||||
{
|
||||
"oms_id": 2026100201,
|
||||
"platform_type": 66,
|
||||
"purchase_qty": 10,
|
||||
"status": 1,
|
||||
"update_user_name": "系统管理员",
|
||||
}
|
||||
)
|
||||
called = {"n": 0}
|
||||
|
||||
def search_yaoex(_key: str):
|
||||
return [_ok_candidate()]
|
||||
|
||||
def finish(_lines):
|
||||
called["n"] += 1
|
||||
return 1, 0
|
||||
|
||||
summary = run_batch(
|
||||
repo,
|
||||
platforms=["1号药城"],
|
||||
searchers={"1号药城": search_yaoex},
|
||||
login_errors={},
|
||||
cart_finish=finish,
|
||||
)
|
||||
assert called["n"] == 0
|
||||
assert summary.cart_added == 0
|
||||
assert len(repo.cart) == 1
|
||||
assert repo.mapping[2026100201].deal_flag == 1
|
||||
|
|
|
|||
|
|
@ -27,6 +27,8 @@ def test_yaoex_field_mapping_and_price_passthrough():
|
|||
assert item.vendor_norm == "广东壹号药业有限公司"
|
||||
assert item.stock == 40.0
|
||||
assert item.mall_name == "1号药城"
|
||||
assert item.extra["raw"]["spuCode"] == "SPU1"
|
||||
assert item.extra["stockCountDesc"] == ""
|
||||
|
||||
|
||||
def test_ybm_field_mapping_zero_stock_price_is_zero():
|
||||
|
|
@ -54,6 +56,7 @@ def test_ybm_field_mapping_zero_stock_price_is_zero():
|
|||
assert item.stock == 0.0
|
||||
assert item.expiry_raw == "14个月"
|
||||
assert item.mall_name == "药帮忙"
|
||||
assert item.extra["raw"]["skuId"] == "SKU9"
|
||||
|
||||
|
||||
def test_ybm_uses_fob_when_price_empty():
|
||||
|
|
|
|||
|
|
@ -47,6 +47,15 @@ def test_from_env_without_user_raises(monkeypatch):
|
|||
OracleSelectRepo.from_env()
|
||||
|
||||
|
||||
# RETURNING 输出。仓库读 getvalue()[0]。
|
||||
class _OutBox:
|
||||
def __init__(self):
|
||||
self.value = None
|
||||
|
||||
def getvalue(self):
|
||||
return [self.value]
|
||||
|
||||
|
||||
# 录 SQL 的假游标，fetchall 返回预设行。
|
||||
class _FakeCursor:
|
||||
# 挂上共享的 executed 列表和预设查询结果。
|
||||
|
|
@ -66,6 +75,7 @@ class _FakeCursor:
|
|||
self._fail_on_many = fail_on_many
|
||||
self._fail_update_at = fail_update_at
|
||||
self._updates = 0
|
||||
self._next_fid = 1000
|
||||
self.description = description
|
||||
self.rowcount = 1
|
||||
|
||||
|
|
@ -77,16 +87,27 @@ class _FakeCursor:
|
|||
def __exit__(self, *args):
|
||||
return False
|
||||
|
||||
# 记下 RETURNING 的输出变量。getvalue 给成单元素列表。
|
||||
def var(self, *_args, **_kwargs):
|
||||
return _OutBox()
|
||||
|
||||
# 记下 SQL 和 bind。一条 IN 超过 1000 个 fid 时抛错，模拟 ORA-01795。
|
||||
def execute(self, sql, params=None):
|
||||
params = params or {}
|
||||
params = dict(params or {})
|
||||
out = params.pop("out_fid", None)
|
||||
_reject_oversized_in(params)
|
||||
if str(sql).lstrip().lower().startswith("update"):
|
||||
text = str(sql).lstrip().lower()
|
||||
if self._fail_on_many and "insert into rpa_cart_settlement_goods_data" in text:
|
||||
raise RuntimeError("batch failed")
|
||||
if text.startswith("update"):
|
||||
self._updates = getattr(self, "_updates", 0) + 1
|
||||
if self._fail_update_at is not None and self._updates == self._fail_update_at:
|
||||
raise RuntimeError("chunk failed")
|
||||
self._last_sql = sql
|
||||
self._executed.append((sql, params))
|
||||
if out is not None:
|
||||
self._next_fid += 1
|
||||
out.value = self._next_fid
|
||||
|
||||
# 记下批量 SQL。fail_on_many 时抛错，用来点名整批回滚。
|
||||
def executemany(self, sql, seq):
|
||||
|
|
@ -247,8 +268,9 @@ def test_save_result_inserts_cart_then_updates_mapping():
|
|||
assert conn.committed is True
|
||||
assert conn.rolled_back is False
|
||||
assert pool.acquires == 1
|
||||
assert len(conn.many) == 2
|
||||
sql = " ".join(conn.many[0][0].split())
|
||||
inserts = _cart_inserts(conn)
|
||||
assert len(inserts) == 1
|
||||
sql = " ".join(inserts[0][0].split())
|
||||
assert "rpa_cart_settlement_goods_data" in sql.lower()
|
||||
assert "create_user_id" in sql
|
||||
assert "create_user_name" in sql
|
||||
|
|
@ -256,14 +278,19 @@ def test_save_result_inserts_cart_then_updates_mapping():
|
|||
assert "update_user_id" in sql
|
||||
assert "update_user_name" in sql
|
||||
assert "SYSDATE, 0, '系统管理员'" in sql
|
||||
assert ":update_user_name" in sql
|
||||
assert "returning fid into :out_fid" in sql.lower()
|
||||
assert sql.count("SYSDATE") == 2
|
||||
assert conn.many[0][1][0]["oms_id"] == 2026100201
|
||||
assert "create_user_name" not in conn.many[0][1][0]
|
||||
update_sql = " ".join(conn.many[1][0].split()).lower()
|
||||
assert inserts[0][1]["oms_id"] == 2026100201
|
||||
assert inserts[0][1]["update_user_name"] == "系统管理员"
|
||||
assert "create_user_name" not in inserts[0][1]
|
||||
assert "out_fid" not in inserts[0][1]
|
||||
assert len(conn.many) == 1
|
||||
update_sql = " ".join(conn.many[0][0].split()).lower()
|
||||
assert "rpa_sales_goods_mapping_data" in update_sql
|
||||
assert "where fid = :fid" in update_sql
|
||||
assert conn.many[1][1][0]["fid"] == 42
|
||||
assert conn.many[1][1][0]["deal_flag"] == 1
|
||||
assert conn.many[0][1][0]["fid"] == 42
|
||||
assert conn.many[0][1][0]["deal_flag"] == 1
|
||||
|
||||
|
||||
# 一批两行表 2 只从池里借一次连接。
|
||||
|
|
@ -279,21 +306,21 @@ def test_save_result_acquires_once_for_two_cart_rows():
|
|||
fid=42,
|
||||
)
|
||||
assert pool.acquires == 1
|
||||
inserts = [item for item in conn.many if "rpa_cart_settlement_goods_data" in item[0].lower()]
|
||||
inserts = _cart_inserts(conn)
|
||||
updates = [item for item in conn.many if "rpa_sales_goods_mapping_data" in item[0].lower()]
|
||||
assert len(inserts) == 1
|
||||
assert len(inserts[0][1]) == 2
|
||||
assert len(inserts) == 2
|
||||
assert [item[1]["purchase_qty"] for item in inserts] == [7, 3]
|
||||
assert len(updates) == 1
|
||||
assert len(updates[0][1]) == 1
|
||||
assert conn.committed is True
|
||||
|
||||
|
||||
# 两单的表 2 和表 1 完成更新各一次 executemany，只借一条连接。
|
||||
# 两单的表 2 按行取回 fid，表 1 完成更新一次 executemany，只借一条连接。
|
||||
def test_save_results_batches_two_orders_in_one_acquire():
|
||||
conn = _FakeConn()
|
||||
pool = _FakePool(conn)
|
||||
repo = OracleSelectRepo(pool)
|
||||
repo.save_results(
|
||||
fids = repo.save_results(
|
||||
[
|
||||
FinishWrite(
|
||||
oms_id=2026100201,
|
||||
|
|
@ -301,8 +328,8 @@ def test_save_results_batches_two_orders_in_one_acquire():
|
|||
deal_flag=1,
|
||||
deal_result="选品成功：1号药城、药帮忙",
|
||||
cart_rows=[
|
||||
_cart_row(purchase_qty=7),
|
||||
_cart_row(platform_type=48, purchase_qty=3),
|
||||
_cart_row(purchase_qty=7, update_user_name="加购中"),
|
||||
_cart_row(platform_type=48, purchase_qty=3, update_user_name="加购中"),
|
||||
],
|
||||
),
|
||||
FinishWrite(
|
||||
|
|
@ -314,18 +341,23 @@ def test_save_results_batches_two_orders_in_one_acquire():
|
|||
),
|
||||
]
|
||||
)
|
||||
assert fids == [[1001, 1002], [1003]]
|
||||
assert pool.acquires == 1
|
||||
assert conn.committed is True
|
||||
assert conn.rolled_back is False
|
||||
assert len(conn.many) == 2
|
||||
insert_sql, insert_binds = conn.many[0]
|
||||
update_sql, update_binds = conn.many[1]
|
||||
assert "rpa_cart_settlement_goods_data" in insert_sql.lower()
|
||||
assert [row["oms_id"] for row in insert_binds] == [
|
||||
assert len(conn.many) == 1
|
||||
inserts = _cart_inserts(conn)
|
||||
update_sql, update_binds = conn.many[0]
|
||||
assert [row["oms_id"] for _sql, row in inserts] == [
|
||||
2026100201,
|
||||
2026100201,
|
||||
2026100202,
|
||||
]
|
||||
assert [row["update_user_name"] for _sql, row in inserts] == [
|
||||
"加购中",
|
||||
"加购中",
|
||||
"系统管理员",
|
||||
]
|
||||
assert "where fid = :fid" in " ".join(update_sql.split()).lower()
|
||||
assert [row["fid"] for row in update_binds] == [42, 43]
|
||||
assert [row["deal_flag"] for row in update_binds] == [1, 2]
|
||||
|
|
@ -402,6 +434,15 @@ def test_save_query_audits_swallows_write_failure():
|
|||
assert conn.committed is False
|
||||
|
||||
|
||||
def _cart_inserts(conn) -> list:
|
||||
"""表 2 插入语句。RETURNING 走 execute，不进 executemany。"""
|
||||
return [
|
||||
item
|
||||
for item in conn.executed
|
||||
if "insert into rpa_cart_settlement_goods_data" in item[0].lower()
|
||||
]
|
||||
|
||||
|
||||
def _fid_bind_count(params: dict) -> int:
|
||||
"""数出本段 bind 里的 fid 个数。"""
|
||||
return sum(
|
||||
|
|
|
|||
Loading…
Reference in New Issue