478 lines
15 KiB
Python
478 lines
15 KiB
Python
# 选品班次：领取戳、按条形码搜、系统故障短码。
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime, timedelta
|
||
|
||
from repo import MappingRow, MemorySelectRepo
|
||
from rules import Candidate, normalize_vendor
|
||
from job import run_batch
|
||
|
||
|
||
HANGZHOU = "杭州小药药自营旗舰店"
|
||
GUANGDONG = "广东壹号药业有限公司"
|
||
|
||
|
||
# 构造一条待选品表 1 行。
|
||
def _row(**overrides) -> MappingRow:
|
||
data = {
|
||
"fid": 1,
|
||
"oms_id": 2026100201,
|
||
"bar_code": "06901234567890",
|
||
"sale_price": 100.0,
|
||
"plan_purchase_qty": 10,
|
||
"approval_no": "国药准字H20100001",
|
||
"goods_name": "阿莫西林",
|
||
"goods_spec": "0.25g",
|
||
"manufacturer": "示例制药",
|
||
"org_code": "907",
|
||
"deal_flag": 0,
|
||
}
|
||
data.update(overrides)
|
||
return MappingRow(**data)
|
||
|
||
|
||
# 构造一条能通过规则的候选。
|
||
def _ok_candidate(mall: str = "1号药城", **overrides) -> Candidate:
|
||
vendor = overrides.pop("vendor", GUANGDONG if mall == "1号药城" else HANGZHOU)
|
||
data = {
|
||
"mall_name": mall,
|
||
"sku_id": "S1",
|
||
"name": "阿莫西林",
|
||
"spec": "0.25g",
|
||
"price": 90.0,
|
||
"expiry_raw": "13",
|
||
"expiry_months": 13.0,
|
||
"vendor_raw": vendor,
|
||
"vendor_norm": normalize_vendor(vendor),
|
||
"stock": 20.0,
|
||
}
|
||
data.update(overrides)
|
||
if "vendor_raw" in overrides and "vendor_norm" not in overrides:
|
||
data["vendor_norm"] = normalize_vendor(overrides["vendor_raw"])
|
||
return Candidate(**data)
|
||
|
||
|
||
# deal_flag=0 领走后变成 3，已完成的 1 不领。
|
||
def test_claim_pending_sets_processing_flag():
|
||
repo = MemorySelectRepo([_row(), _row(fid=2, oms_id=2026100202, deal_flag=1)])
|
||
claimed = repo.claim_pending()
|
||
assert [row.oms_id for row in claimed] == [2026100201]
|
||
assert repo.mapping[2026100201].deal_flag == 3
|
||
assert repo.mapping[2026100202].deal_flag == 1
|
||
|
||
|
||
# deal_flag=3 且 deal_time 超过租约的行可以再领。
|
||
def test_claim_reclaims_expired_flag_3():
|
||
now = datetime(2026, 10, 22, 12, 0, 0)
|
||
expired = now - timedelta(minutes=31)
|
||
repo = MemorySelectRepo(
|
||
[_row(deal_flag=3, deal_time=expired)]
|
||
)
|
||
claimed = repo.claim_pending(now=now)
|
||
assert [row.oms_id for row in claimed] == [2026100201]
|
||
assert repo.mapping[2026100201].deal_flag == 3
|
||
|
||
|
||
# deal_flag=3 且未过租约的行不领。
|
||
def test_claim_skips_fresh_flag_3():
|
||
now = datetime(2026, 10, 22, 12, 0, 0)
|
||
fresh = now - timedelta(minutes=1)
|
||
repo = MemorySelectRepo([_row(deal_flag=3, deal_time=fresh)])
|
||
claimed = repo.claim_pending(now=now)
|
||
assert claimed == []
|
||
assert repo.mapping[2026100201].deal_flag == 3
|
||
|
||
|
||
# 班次按条形码搜，不拿批准文号去搜。
|
||
def test_job_searches_barcode_not_approval_no():
|
||
repo = MemorySelectRepo([_row()])
|
||
seen: list[str] = []
|
||
|
||
# 记下查询键并返回一条合格候选。
|
||
def search_yaoex(query_key: str):
|
||
seen.append(query_key)
|
||
return [_ok_candidate()]
|
||
|
||
summary = run_batch(
|
||
repo,
|
||
platforms=["1号药城"],
|
||
searchers={"1号药城": search_yaoex},
|
||
login_errors={},
|
||
)
|
||
assert seen == ["06901234567890"]
|
||
assert summary.added == 1
|
||
assert repo.mapping[2026100201].deal_flag == 1
|
||
assert len(repo.cart) == 1
|
||
assert repo.cart[0]["bar_code"] == "06901234567890"
|
||
assert repo.cart[0]["status"] == 1
|
||
assert repo.cart[0]["purchase_qty"] == 10
|
||
|
||
|
||
def test_job_saves_each_claimed_row_before_the_next():
|
||
"""一行搜完就落这一行，不攒到班次结束。"""
|
||
|
||
class _Repo(MemorySelectRepo):
|
||
"""记下每次 save_results 的行。"""
|
||
|
||
def __init__(self, rows):
|
||
super().__init__(rows)
|
||
self.batches: list[list] = []
|
||
|
||
def save_results(self, items):
|
||
"""记下这一次落库，再交给内存仓库。"""
|
||
self.batches.append(list(items))
|
||
return super().save_results(items)
|
||
|
||
repo = _Repo(
|
||
[
|
||
_row(fid=1, oms_id=1, bar_code="111"),
|
||
_row(fid=2, oms_id=2, bar_code="222"),
|
||
]
|
||
)
|
||
|
||
# 两行都给一条合格候选。
|
||
def search(_query_key: str):
|
||
return [_ok_candidate()]
|
||
|
||
summary = run_batch(
|
||
repo,
|
||
platforms=["1号药城"],
|
||
searchers={"1号药城": search},
|
||
login_errors={},
|
||
)
|
||
assert summary.claimed == 2
|
||
assert summary.added == 2
|
||
assert len(repo.batches) == 2
|
||
assert [item.fid for batch in repo.batches for item in batch] == [1, 2]
|
||
assert repo.mapping[1].deal_flag == 1
|
||
assert repo.mapping[2].deal_flag == 1
|
||
|
||
|
||
def test_expired_lease_second_process_does_not_insert_second_cart_set():
|
||
"""租约过期后第二个进程不再插入第二套表 2。"""
|
||
from repo import FinishWrite
|
||
|
||
now = datetime(2026, 10, 22, 12, 0, 0)
|
||
repo = MemorySelectRepo([_row(deal_flag=3, deal_time=now - timedelta(minutes=31))])
|
||
claimed = repo.claim_pending(now=now)
|
||
assert [row.fid for row in claimed] == [1]
|
||
repo.cart.append(
|
||
{
|
||
"oms_id": 2026100201,
|
||
"platform_type": 66,
|
||
"purchase_qty": 10,
|
||
"status": 1,
|
||
}
|
||
)
|
||
repo.save_results(
|
||
[
|
||
FinishWrite(
|
||
oms_id=2026100201,
|
||
fid=1,
|
||
deal_flag=1,
|
||
deal_result="选品成功：1号药城",
|
||
cart_rows=[
|
||
{
|
||
"oms_id": 2026100201,
|
||
"platform_type": 66,
|
||
"purchase_qty": 10,
|
||
"status": 1,
|
||
}
|
||
],
|
||
)
|
||
]
|
||
)
|
||
assert len(repo.cart) == 1
|
||
assert repo.cart[0]["purchase_qty"] == 10
|
||
assert repo.mapping[2026100201].deal_flag == 1
|
||
|
||
|
||
# 没有条形码时不搜平台，业务失败。
|
||
def test_job_skips_search_when_barcode_missing():
|
||
repo = MemorySelectRepo([_row(bar_code="")])
|
||
called = {"n": 0}
|
||
|
||
# 本用例不应被调用。
|
||
def search_yaoex(_query_key: str):
|
||
called["n"] += 1
|
||
return [_ok_candidate()]
|
||
|
||
summary = run_batch(
|
||
repo,
|
||
platforms=["1号药城"],
|
||
searchers={"1号药城": search_yaoex},
|
||
login_errors={},
|
||
)
|
||
assert called["n"] == 0
|
||
assert summary.skipped == 1
|
||
assert repo.mapping[2026100201].deal_flag == 2
|
||
assert repo.cart[0]["status"] == 2
|
||
assert "条形码" in (repo.cart[0]["fail_reason"] or "")
|
||
|
||
|
||
# limit 只领 fid 最小的前 N 条。
|
||
def test_job_limit_claims_first_n():
|
||
repo = MemorySelectRepo(
|
||
[
|
||
_row(fid=1, oms_id=1, bar_code="111"),
|
||
_row(fid=2, oms_id=2, bar_code="222"),
|
||
_row(fid=3, oms_id=3, bar_code="333"),
|
||
]
|
||
)
|
||
claimed = repo.claim_pending(limit=2)
|
||
assert [row.oms_id for row in claimed] == [1, 2]
|
||
assert repo.mapping[3].deal_flag == 0
|
||
|
||
|
||
def test_both_malls_login_failed_marks_system_fault():
|
||
"""两边登录失败且凑不齐时，deal_flag 改回 0，不写表 2。"""
|
||
repo = MemorySelectRepo([_row()])
|
||
summary = run_batch(
|
||
repo,
|
||
platforms=["1号药城", "药帮忙"],
|
||
searchers={},
|
||
login_errors={"1号药城": "chrome down", "药帮忙": "timeout"},
|
||
)
|
||
assert summary.added == 0
|
||
assert repo.mapping[2026100201].deal_flag == 0
|
||
assert repo.mapping[2026100201].deal_result.startswith("系统故障")
|
||
assert repo.cart == []
|
||
|
||
|
||
# 杭州 3 + 广东 7 落两行成功，不插多余行。
|
||
def test_split_across_malls_writes_two_success_rows():
|
||
repo = MemorySelectRepo([_row()])
|
||
|
||
# 广东仓库存 20。
|
||
def search_yaoex(_key: str):
|
||
return [_ok_candidate("1号药城", stock=20.0)]
|
||
|
||
# 杭州仓库存 3。
|
||
def search_ybm(_key: str):
|
||
return [_ok_candidate("药帮忙", vendor=HANGZHOU, sku_id="HZ", stock=3.0)]
|
||
|
||
summary = run_batch(
|
||
repo,
|
||
platforms=["1号药城", "药帮忙"],
|
||
searchers={"1号药城": search_yaoex, "药帮忙": search_ybm},
|
||
login_errors={},
|
||
)
|
||
assert summary.added == 2
|
||
assert repo.mapping[2026100201].deal_flag == 1
|
||
ybm = next(row for row in repo.cart if row["platform_type"] == 48)
|
||
yaoex = next(row for row in repo.cart if row["platform_type"] == 66)
|
||
assert ybm["purchase_qty"] == 3
|
||
assert yaoex["purchase_qty"] == 7
|
||
assert ybm["multi_platform_flag"] == 1
|
||
|
||
|
||
# 一边登录失败、另一边能凑齐计划量时按成功落。
|
||
def test_peer_login_failed_but_other_fills_is_success():
|
||
repo = MemorySelectRepo([_row()])
|
||
|
||
# 广东仓库存 20，能单独凑齐。
|
||
def search_yaoex(_key: str):
|
||
return [_ok_candidate(stock=20.0)]
|
||
|
||
summary = run_batch(
|
||
repo,
|
||
platforms=["1号药城", "药帮忙"],
|
||
searchers={"1号药城": search_yaoex},
|
||
login_errors={"药帮忙": "timeout"},
|
||
)
|
||
assert summary.added == 1
|
||
assert repo.mapping[2026100201].deal_flag == 1
|
||
assert len(repo.cart) == 1
|
||
assert repo.cart[0]["platform_type"] == 66
|
||
|
||
|
||
def test_peer_login_failed_and_short_stock_is_system_fault():
|
||
"""一边故障且剩下的凑不齐时改回待领，不写成库存不足，不写表 2。"""
|
||
repo = MemorySelectRepo([_row()])
|
||
|
||
def search_yaoex(_key: str):
|
||
"""广东仓库存 4，凑不齐计划 10。"""
|
||
return [_ok_candidate(stock=4.0)]
|
||
|
||
summary = run_batch(
|
||
repo,
|
||
platforms=["1号药城", "药帮忙"],
|
||
searchers={"1号药城": search_yaoex},
|
||
login_errors={"药帮忙": "timeout"},
|
||
)
|
||
assert summary.added == 0
|
||
assert repo.mapping[2026100201].deal_flag == 0
|
||
assert repo.mapping[2026100201].deal_result.startswith("系统故障")
|
||
assert "库存不足" not in repo.mapping[2026100201].deal_result
|
||
assert repo.cart == []
|
||
|
||
|
||
# 两边都搜到但合计不够是业务失败库存不足。
|
||
def test_combined_short_stock_is_business_fail():
|
||
repo = MemorySelectRepo([_row()])
|
||
|
||
# 广东仓库存 2。
|
||
def search_yaoex(_key: str):
|
||
return [_ok_candidate(stock=2.0)]
|
||
|
||
# 杭州仓库存 2。
|
||
def search_ybm(_key: str):
|
||
return [_ok_candidate("药帮忙", vendor=HANGZHOU, sku_id="HZ", stock=2.0)]
|
||
|
||
summary = run_batch(
|
||
repo,
|
||
platforms=["1号药城", "药帮忙"],
|
||
searchers={"1号药城": search_yaoex, "药帮忙": search_ybm},
|
||
login_errors={},
|
||
)
|
||
assert summary.added == 0
|
||
assert summary.skipped == 2
|
||
assert repo.mapping[2026100201].deal_flag == 2
|
||
assert "库存不足" in repo.mapping[2026100201].deal_result
|
||
assert all(row["status"] == 2 for row in repo.cart)
|
||
|
||
|
||
# 查询抛错时原因列只写 QUERY_FAILED，不含异常原文。
|
||
def test_job_query_failed_writes_short_code_not_exception():
|
||
repo = MemorySelectRepo([_row()])
|
||
|
||
# 抛带 ORA 原文的错，确认原因列不收录。
|
||
def boom(_key: str):
|
||
raise RuntimeError("ORA-03114 secret-traceback")
|
||
|
||
summary = run_batch(
|
||
repo,
|
||
platforms=["1号药城"],
|
||
searchers={"1号药城": boom},
|
||
login_errors={},
|
||
)
|
||
assert summary.added == 0
|
||
assert repo.mapping[2026100201].deal_flag == 0
|
||
assert repo.cart == []
|
||
deal_result = repo.mapping[2026100201].deal_result
|
||
assert deal_result.startswith("系统故障")
|
||
assert "QUERY_FAILED" in deal_result
|
||
assert "ORA-03114" not in deal_result
|
||
assert "secret-traceback" not in deal_result
|
||
assert len(repo.query_audits) == 1
|
||
audit = repo.query_audits[0]
|
||
assert audit.oms_id == 2026100201
|
||
assert audit.mapping_fid == 1
|
||
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
|