297 lines
10 KiB
Python
297 lines
10 KiB
Python
# 选品班次：领取表 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, field
|
||
from typing import Any, Callable
|
||
|
||
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 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:
|
||
claimed: int = 0
|
||
added: int = 0
|
||
skipped: int = 0
|
||
failed: int = 0
|
||
cart_added: int = 0
|
||
cart_failed: int = 0
|
||
|
||
|
||
# 把表 1 行收成规则层的采购行。缺价格或数量保持 None。
|
||
def mapping_to_purchase(row: MappingRow) -> PurchaseLine:
|
||
sale_price = float(row.sale_price) if row.sale_price is not None else None
|
||
qty = float(row.plan_purchase_qty) if row.plan_purchase_qty is not None else None
|
||
return PurchaseLine(
|
||
approval_no=row.approval_no or "",
|
||
sale_price=sale_price,
|
||
qty=qty,
|
||
row_no=row.fid,
|
||
generic_name=row.goods_name or "",
|
||
spec=row.goods_spec or "",
|
||
manufacturer=row.manufacturer or "",
|
||
oms_id=row.oms_id,
|
||
bar_code=(row.bar_code or "").strip(),
|
||
org_code=row.org_code or "",
|
||
)
|
||
|
||
|
||
def run_batch(
|
||
repo,
|
||
*,
|
||
platforms: list[str],
|
||
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))
|
||
for row in rows:
|
||
purchase = mapping_to_purchase(row)
|
||
|
||
def renew(_fid: int = row.fid) -> None:
|
||
"""这一行还在搜时，把租约续到现在。"""
|
||
touch = getattr(repo, "touch_claim", None)
|
||
if touch is not None:
|
||
touch(_fid)
|
||
|
||
selection, row_audits = _select_row(
|
||
purchase,
|
||
platforms,
|
||
searchers,
|
||
login_errors,
|
||
before_search=renew,
|
||
)
|
||
result = build_result(purchase, selection)
|
||
_count(summary, result.cart_rows)
|
||
_mark_cart_stamp(result, cart_finish)
|
||
saved = repo.save_results(
|
||
[
|
||
FinishWrite(
|
||
oms_id=result.oms_id,
|
||
fid=row.fid,
|
||
deal_flag=result.deal_flag,
|
||
deal_result=result.deal_result,
|
||
cart_rows=result.cart_rows,
|
||
)
|
||
]
|
||
)
|
||
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],
|
||
searchers: dict[str, Searcher],
|
||
login_errors: dict[str, str],
|
||
before_search: Callable[[], None] | None = None,
|
||
) -> tuple[SelectionResult, list[QueryAudit]]:
|
||
"""校验采购行并搜各平台，交给统一拆量。校验不过不搜。
|
||
|
||
before_search 在每个平台开搜前调用，用来续租。
|
||
"""
|
||
row_reason = validate_purchase(purchase)
|
||
if row_reason:
|
||
return (
|
||
SelectionResult(
|
||
kind=KIND_BUSINESS_FAIL,
|
||
reason=row_reason,
|
||
allocations=[],
|
||
rejected=[],
|
||
malls=list(platforms),
|
||
),
|
||
[],
|
||
)
|
||
outcomes: list[MallOutcome] = []
|
||
audits: list[QueryAudit] = []
|
||
for mall in platforms:
|
||
if before_search is not None:
|
||
before_search()
|
||
outcome, audit = _search_one(purchase, mall, searchers, login_errors)
|
||
outcomes.append(outcome)
|
||
if audit is not None:
|
||
audits.append(audit)
|
||
return run_selection(purchase, outcomes), audits
|
||
|
||
|
||
# 搜一个平台。登录失败写 LOGIN_FAILED，查询抛错写 QUERY_FAILED。
|
||
#
|
||
# 异常原文进审计行，不写进失败原因列。
|
||
def _search_one(
|
||
purchase: PurchaseLine,
|
||
mall: str,
|
||
searchers: dict[str, Searcher],
|
||
login_errors: dict[str, str],
|
||
) -> tuple[MallOutcome, QueryAudit | None]:
|
||
login_error = login_errors.get(mall)
|
||
if login_error:
|
||
logger.warning("选品登录失败 mall=%s: %s", mall, login_error)
|
||
return MallOutcome(mall, failed_reason="LOGIN_FAILED"), None
|
||
searcher = searchers.get(mall)
|
||
if searcher is None:
|
||
return MallOutcome(mall, failed_reason="LOGIN_FAILED"), None
|
||
try:
|
||
candidates = searcher(purchase.bar_code)
|
||
except Exception as exc:
|
||
logger.exception("选品查询失败 mall=%s bar_code=%s", mall, purchase.bar_code)
|
||
return MallOutcome(mall, failed_reason="QUERY_FAILED"), _query_audit(
|
||
purchase, mall, exc
|
||
)
|
||
return MallOutcome(mall, candidates=candidates), None
|
||
|
||
|
||
# 有 oms_id 和表 1 fid 时收成一条审计。缺主键则只留日志。
|
||
def _query_audit(
|
||
purchase: PurchaseLine, mall: str, exc: Exception
|
||
) -> QueryAudit | None:
|
||
if purchase.oms_id is None or purchase.row_no is None:
|
||
return None
|
||
return QueryAudit(
|
||
oms_id=purchase.oms_id,
|
||
mapping_fid=purchase.row_no,
|
||
mall_name=mall,
|
||
bar_code=purchase.bar_code,
|
||
error_text=str(exc),
|
||
)
|
||
|
||
|
||
# 按表 2 行状态累加成功、系统故障、业务跳过。
|
||
def _count(summary: JobSummary, cart_rows: list[dict]) -> None:
|
||
for row in cart_rows:
|
||
if row.get("status") == STATUS_OK:
|
||
summary.added += 1
|
||
continue
|
||
reason = str(row.get("fail_reason") or "")
|
||
if reason.startswith("系统故障"):
|
||
summary.failed += 1
|
||
else:
|
||
summary.skipped += 1
|