198 lines
7.3 KiB
Python
198 lines
7.3 KiB
Python
# 加购班次：领表 2 选品成功行 → 再搜条形码对仓 → 加车。
|
||
#
|
||
# 成功盖章，失败改 status=2。崩在「加购中」：超过租约可再领，未过期需改回「系统管理员」。
|
||
# 查询失败原因列只写 QUERY_FAILED / LOGIN_FAILED / ADD_FAILED。原文进审计表，会话结束后另写。
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from collections import OrderedDict
|
||
from contextlib import nullcontext
|
||
from dataclasses import dataclass
|
||
from typing import Callable
|
||
|
||
from cart_add import ( # noqa: F401
|
||
KIND_BUSINESS,
|
||
KIND_OK,
|
||
KIND_SYSTEM,
|
||
AddOutcome,
|
||
)
|
||
from cart_match import CartOffer, match_offer
|
||
from cart_repo import CartRow, clip_reason
|
||
from rpa_oracle import ShiftError
|
||
|
||
Searcher = Callable[[str], list[CartOffer]]
|
||
Adder = Callable[[CartOffer, int], AddOutcome]
|
||
CartReader = Callable[[str, str], int]
|
||
_RETRY_CODES = ("LOGIN_FAILED", "QUERY_FAILED")
|
||
logger = logging.getLogger("rpa_cart")
|
||
INCOMPLETE_MARK = "未加齐"
|
||
|
||
|
||
# 一个加购班次的领取和盖章计数。
|
||
@dataclass
|
||
class CartSummary:
|
||
claimed: int = 0
|
||
added: int = 0
|
||
failed: int = 0
|
||
|
||
|
||
def _session(repo):
|
||
"""有 session 就借一条，没有就空上下文。商城调用必须在这个块外面。"""
|
||
session = getattr(repo, "session", None)
|
||
if session is None:
|
||
return nullcontext()
|
||
return session()
|
||
|
||
|
||
def run_cart_batch(
|
||
repo,
|
||
*,
|
||
searchers: dict[int, Searcher],
|
||
adders: dict[int, Adder],
|
||
login_errors: dict[int, str],
|
||
limit: int | None = None,
|
||
platform_types: list[int] | None = None,
|
||
cart_readers: dict[int, CartReader] | None = None,
|
||
) -> CartSummary:
|
||
"""领取待加购行，按行搜仓、加车、盖章。
|
||
|
||
session 只包领取和每一组的盖章。搜索和加车在连接归还之后跑。一组盖完再处理下一组。
|
||
登录失败和查询失败把戳改回系统管理员。购物车里已有不少于本行的数量则直接成功，不加第二份。
|
||
limit 交给领取层。
|
||
"""
|
||
audits: list[ShiftError] = []
|
||
readers = cart_readers or {}
|
||
with _session(repo):
|
||
rows = repo.claim_pending(limit=limit, platform_types=platform_types)
|
||
summary = CartSummary(claimed=len(rows))
|
||
for oms_id, group in _by_oms(rows).items():
|
||
outcomes = [
|
||
(row, *_add_one(row, searchers, adders, login_errors, readers))
|
||
for row in group
|
||
]
|
||
incomplete = _is_incomplete(outcomes)
|
||
if incomplete:
|
||
logger.warning("加购未加齐 oms_id=%s", oms_id)
|
||
with _session(repo):
|
||
for row, reason, code, text in outcomes:
|
||
if reason is None:
|
||
repo.mark_success(row.fid)
|
||
summary.added += 1
|
||
continue
|
||
if incomplete:
|
||
reason = _incomplete_reason(reason)
|
||
if _retryable(code, reason):
|
||
repo.mark_release(row.fid, reason)
|
||
else:
|
||
repo.mark_fail(row.fid, reason)
|
||
summary.failed += 1
|
||
if code and text:
|
||
audits.append(ShiftError("加购", str(row.fid), code, text))
|
||
repo.save_shift_errors(audits)
|
||
return summary
|
||
|
||
|
||
# 按 oms_id 保序分组。同组先加完再盖章，才能看出有没有加齐。
|
||
def _by_oms(rows: list[CartRow]) -> OrderedDict[int, list[CartRow]]:
|
||
groups: OrderedDict[int, list[CartRow]] = OrderedDict()
|
||
for row in rows:
|
||
groups.setdefault(int(row.oms_id), []).append(row)
|
||
return groups
|
||
|
||
|
||
# 同组里既有成功又有失败，才算这单没加齐。
|
||
def _is_incomplete(outcomes: list[tuple]) -> bool:
|
||
flags = [reason is None for _row, reason, _code, _text in outcomes]
|
||
return any(flags) and not all(flags)
|
||
|
||
|
||
# 失败原因前加上未加齐，已带前缀的不再套一层。
|
||
def _incomplete_reason(reason: str) -> str:
|
||
text = (reason or "").strip()
|
||
if text.startswith(INCOMPLETE_MARK):
|
||
return text
|
||
return f"{INCOMPLETE_MARK}：{text}" if text else INCOMPLETE_MARK
|
||
|
||
|
||
def _retryable(code: str | None, reason: str | None) -> bool:
|
||
"""登录失败和查询失败放回队列。加车失败、对不上仓、库存不足不放。"""
|
||
blob = f"{code or ''} {reason or ''}"
|
||
return any(token in blob for token in _RETRY_CODES)
|
||
|
||
|
||
def _add_one(
|
||
row: CartRow,
|
||
searchers: dict[int, Searcher],
|
||
adders: dict[int, Adder],
|
||
login_errors: dict[int, str],
|
||
cart_readers: dict[int, CartReader],
|
||
) -> tuple[str | None, str | None, str | None]:
|
||
"""对一行查购物车、搜仓并加车。
|
||
|
||
返回 (失败原因, 审计短码, 原文)。成功三项都是 None。
|
||
购物车里该条形码和供应商的数量已经不少于本行时直接成功，多出来的数量不加第二份。
|
||
业务原因（对不上仓、库存、商城回的业务失败）不进审计。没有原文也不进。
|
||
"""
|
||
platform = row.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)
|
||
searcher = searchers.get(platform)
|
||
if searcher is None:
|
||
return _system("LOGIN_FAILED"), None, None
|
||
adder = adders.get(platform)
|
||
if adder is None:
|
||
return _system("ADD_FAILED"), None, None
|
||
bar_code = (row.bar_code or "").strip()
|
||
if not bar_code:
|
||
return "缺少条形码", None, None
|
||
qty = int(row.purchase_qty)
|
||
if qty <= 0:
|
||
return "采购数量无效", None, None
|
||
reader = cart_readers.get(platform)
|
||
if reader is not None:
|
||
try:
|
||
already = int(reader(bar_code, row.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
|
||
try:
|
||
offers = searcher(bar_code)
|
||
except Exception as exc:
|
||
logger.exception("加购查询失败 platform=%s bar_code=%s", platform, bar_code)
|
||
return _system("QUERY_FAILED"), "QUERY_FAILED", _text(exc)
|
||
offer = match_offer(offers, row.supplier_name)
|
||
if offer is None:
|
||
return "未找到对应仓", None, None
|
||
if offer.stock is not None and offer.stock < qty:
|
||
return "库存不足", None, None
|
||
try:
|
||
outcome = adder(offer, qty)
|
||
except Exception as exc:
|
||
logger.exception("加购加车失败 platform=%s fid=%s", platform, row.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
|
||
|
||
|
||
# 取出异常原文。空串不当审计正文。
|
||
def _text(exc: BaseException) -> str | None:
|
||
text = str(exc).strip()
|
||
return text or None
|
||
|
||
|
||
# 把系统故障收成「系统故障：短码」并截断。已带前缀的不再套一层。
|
||
def _system(detail: str) -> str:
|
||
text = (detail or "未知错误").strip()
|
||
if text.startswith("系统故障"):
|
||
return clip_reason(text)
|
||
return clip_reason(f"系统故障：{text}")
|