# 选品班次:领取表 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