119 lines
3.5 KiB
Python
119 lines
3.5 KiB
Python
"""选品班次：领取表 1 → 按条形码搜平台 → 统一拆量写表 2。"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Callable
|
|
|
|
from persist import STATUS_OK, build_result
|
|
from pipeline import KIND_BUSINESS_FAIL, MallOutcome, SelectionResult, run_selection
|
|
from repo import MappingRow
|
|
from rules import Candidate, PurchaseLine, validate_purchase
|
|
|
|
Searcher = Callable[[str], list[Candidate]]
|
|
|
|
|
|
@dataclass
|
|
class JobSummary:
|
|
claimed: int = 0
|
|
added: int = 0
|
|
skipped: int = 0
|
|
failed: int = 0
|
|
|
|
|
|
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,
|
|
) -> JobSummary:
|
|
rows = repo.claim_pending(limit=limit)
|
|
summary = JobSummary(claimed=len(rows))
|
|
for row in rows:
|
|
purchase = mapping_to_purchase(row)
|
|
selection = _select_row(purchase, platforms, searchers, login_errors)
|
|
result = build_result(purchase, selection)
|
|
_count(summary, result.cart_rows)
|
|
repo.save_result(
|
|
result.oms_id, result.deal_flag, result.deal_result, result.cart_rows
|
|
)
|
|
return summary
|
|
|
|
|
|
def _select_row(
|
|
purchase: PurchaseLine,
|
|
platforms: list[str],
|
|
searchers: dict[str, Searcher],
|
|
login_errors: dict[str, str],
|
|
) -> SelectionResult:
|
|
row_reason = validate_purchase(purchase)
|
|
if row_reason:
|
|
return SelectionResult(
|
|
kind=KIND_BUSINESS_FAIL,
|
|
reason=row_reason,
|
|
allocations=[],
|
|
rejected=[],
|
|
malls=list(platforms),
|
|
)
|
|
if not purchase.bar_code:
|
|
return SelectionResult(
|
|
kind=KIND_BUSINESS_FAIL,
|
|
reason="NO_BAR_CODE",
|
|
allocations=[],
|
|
rejected=[],
|
|
malls=list(platforms),
|
|
)
|
|
outcomes = [
|
|
_search_one(purchase, mall, searchers, login_errors) for mall in platforms
|
|
]
|
|
return run_selection(purchase, outcomes)
|
|
|
|
|
|
def _search_one(
|
|
purchase: PurchaseLine,
|
|
mall: str,
|
|
searchers: dict[str, Searcher],
|
|
login_errors: dict[str, str],
|
|
) -> MallOutcome:
|
|
login_error = login_errors.get(mall)
|
|
if login_error:
|
|
return MallOutcome(mall, failed_reason=f"LOGIN_FAILED:{login_error}")
|
|
searcher = searchers.get(mall)
|
|
if searcher is None:
|
|
return MallOutcome(mall, failed_reason="LOGIN_FAILED:未登录")
|
|
try:
|
|
candidates = searcher(purchase.bar_code)
|
|
except Exception as exc:
|
|
return MallOutcome(mall, failed_reason=f"QUERY_FAILED:{exc}")
|
|
return MallOutcome(mall, candidates=candidates)
|
|
|
|
|
|
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
|