209 lines
6.6 KiB
Python
209 lines
6.6 KiB
Python
"""选品判定 → 表 2 行 + 表 1 deal_flag。
|
||
|
||
成功按拆出行落 status=1。凑不齐不落成功行。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from decimal import Decimal, ROUND_HALF_UP
|
||
from typing import Any
|
||
|
||
from mall_config import MALL_PLATFORM_TYPE, YAOEX_MALL_NAME, YBM_MALL_NAME
|
||
from pipeline import KIND_BUSINESS_FAIL, KIND_SUCCESS, KIND_SYSTEM_FAIL, SelectionResult
|
||
from rules import Allocation, PurchaseLine
|
||
|
||
OPERATE_TYPE_ADD_CART = 1
|
||
STATUS_OK = 1
|
||
STATUS_FAIL = 2
|
||
PUSH_NOT_SENT = 0
|
||
DEAL_OK = 1
|
||
DEAL_FAIL = 2
|
||
PLACEHOLDER_SUPPLIER = "无符合规则供应商"
|
||
DEAL_RESULT_MAX = 200
|
||
FAIL_REASON_MAX = 500
|
||
MALL_REPORT_ORDER = (YBM_MALL_NAME, YAOEX_MALL_NAME)
|
||
PEER_UNAVAILABLE = "系统故障：对端平台不可用，合计库存不足"
|
||
|
||
|
||
@dataclass
|
||
class SelectResult:
|
||
oms_id: int
|
||
deal_flag: int
|
||
deal_result: str
|
||
cart_rows: list[dict[str, Any]]
|
||
|
||
|
||
def money4(value: object) -> Decimal:
|
||
return Decimal(str(value)).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)
|
||
|
||
|
||
def build_result(purchase: PurchaseLine, selection: SelectionResult) -> SelectResult:
|
||
if purchase.oms_id is None:
|
||
raise ValueError("选品落表需要 oms_id")
|
||
|
||
if selection.kind == KIND_SUCCESS:
|
||
rows = [_success_row(purchase, item) for item in selection.allocations]
|
||
_set_multi_flag(rows)
|
||
malls = _malls_in_rows(rows)
|
||
return SelectResult(
|
||
oms_id=purchase.oms_id,
|
||
deal_flag=DEAL_OK,
|
||
deal_result=_clip_deal_result(f"选品成功：{'、'.join(malls)}"),
|
||
cart_rows=rows,
|
||
)
|
||
|
||
malls = selection.malls or list(MALL_PLATFORM_TYPE)
|
||
if selection.kind == KIND_SYSTEM_FAIL:
|
||
rows = [
|
||
_fail_row(purchase, mall, _system_fail_reason(mall, selection.mall_errors))
|
||
for mall in malls
|
||
]
|
||
detail = "; ".join(selection.mall_errors.values()) or "未知错误"
|
||
deal_result = _clip_deal_result(f"系统故障：{detail}")
|
||
else:
|
||
reason = _business_fail_reason(selection.reason)
|
||
rows = [_fail_row(purchase, mall, reason) for mall in malls]
|
||
deal_result = _clip_deal_result(reason)
|
||
|
||
_set_multi_flag(rows)
|
||
return SelectResult(
|
||
oms_id=purchase.oms_id,
|
||
deal_flag=DEAL_FAIL,
|
||
deal_result=deal_result,
|
||
cart_rows=rows,
|
||
)
|
||
|
||
|
||
def _set_multi_flag(rows: list[dict[str, Any]]) -> None:
|
||
types = {row["platform_type"] for row in rows}
|
||
flag = 1 if len(types) > 1 else 0
|
||
for row in rows:
|
||
row["multi_platform_flag"] = flag
|
||
|
||
|
||
def _malls_in_rows(rows: list[dict[str, Any]]) -> list[str]:
|
||
types = {row["platform_type"] for row in rows}
|
||
names: list[str] = []
|
||
for mall in MALL_REPORT_ORDER:
|
||
if MALL_PLATFORM_TYPE[mall] in types:
|
||
names.append(mall)
|
||
for mall, platform_type in MALL_PLATFORM_TYPE.items():
|
||
if platform_type in types and mall not in names:
|
||
names.append(mall)
|
||
return names
|
||
|
||
|
||
def _success_row(purchase: PurchaseLine, item: Allocation) -> dict[str, Any]:
|
||
price = money4(item.candidate.price)
|
||
qty = item.qty
|
||
row = _base_row(purchase, item.candidate.mall_name)
|
||
row.update(
|
||
{
|
||
"supplier_name": item.candidate.vendor_raw or PLACEHOLDER_SUPPLIER,
|
||
"purchase_price": price,
|
||
"purchase_qty": qty,
|
||
"purchase_total": money4(price * qty),
|
||
"status": STATUS_OK,
|
||
"fail_reason": None,
|
||
}
|
||
)
|
||
return row
|
||
|
||
|
||
def _fail_row(purchase: PurchaseLine, mall_name: str, reason: str) -> dict[str, Any]:
|
||
price = _fail_price(purchase)
|
||
qty = _qty(purchase)
|
||
row = _base_row(purchase, mall_name)
|
||
row.update(
|
||
{
|
||
"supplier_name": PLACEHOLDER_SUPPLIER,
|
||
"purchase_price": price,
|
||
"purchase_qty": qty,
|
||
"purchase_total": money4(price * qty),
|
||
"status": STATUS_FAIL,
|
||
"fail_reason": _clip(reason, FAIL_REASON_MAX),
|
||
}
|
||
)
|
||
return row
|
||
|
||
|
||
def _base_row(purchase: PurchaseLine, mall_name: str) -> dict[str, Any]:
|
||
if mall_name not in MALL_PLATFORM_TYPE:
|
||
raise ValueError(f"未知商城：{mall_name}")
|
||
return {
|
||
"operate_type": OPERATE_TYPE_ADD_CART,
|
||
"oms_id": purchase.oms_id,
|
||
"platform_type": MALL_PLATFORM_TYPE[mall_name],
|
||
"bar_code": purchase.bar_code,
|
||
"org_code": purchase.org_code or None,
|
||
"push_status": PUSH_NOT_SENT,
|
||
}
|
||
|
||
|
||
def _qty(purchase: PurchaseLine) -> int:
|
||
qty = purchase.qty
|
||
if qty is None or qty <= 0:
|
||
return 1
|
||
return int(qty)
|
||
|
||
|
||
def _fail_price(purchase: PurchaseLine) -> Decimal:
|
||
if purchase.sale_price is not None and purchase.sale_price > 0:
|
||
return money4(purchase.sale_price)
|
||
return money4("0.0100")
|
||
|
||
|
||
def _business_fail_reason(code: str) -> str:
|
||
mapping = {
|
||
"NO_HIT": "无符合规则商品，平台无命中",
|
||
"NO_BAR_CODE": "缺少条形码，无法检索平台",
|
||
"NO_SALE_PRICE": "缺少销售价格",
|
||
"NO_QTY": "缺少计划采购量",
|
||
"STOCK_INSUFFICIENT": "无符合规则商品，库存不足",
|
||
"MARGIN_LT_4PCT": "无符合规则商品，毛利不足 4%",
|
||
"EXPIRY_LT_12M": "无符合规则商品，效期不足 12 个月",
|
||
"EXPIRY_UNPARSEABLE": "无符合规则商品，效期不足 12 个月",
|
||
"NOT_SELF_WAREHOUSE": "无符合规则商品，非自营仓",
|
||
"NO_CANDIDATE_AFTER_RULES": "无符合规则商品",
|
||
}
|
||
if code in mapping:
|
||
return mapping[code]
|
||
if code:
|
||
return f"无符合规则商品，{code}"
|
||
return "无符合规则商品"
|
||
|
||
|
||
def _system_fail_reason(mall: str, mall_errors: dict[str, str]) -> str:
|
||
if mall in mall_errors:
|
||
return _clip(f"系统故障：{mall_errors[mall]}", FAIL_REASON_MAX)
|
||
return PEER_UNAVAILABLE
|
||
|
||
|
||
def _clip(text: str, max_len: int) -> str:
|
||
text = text.strip()
|
||
if len(text) <= max_len:
|
||
return text
|
||
return text[: max_len - 1] + "…"
|
||
|
||
|
||
def _clip_deal_result(text: str) -> str:
|
||
"""表 1 deal_result 是 VARCHAR2(200 BYTE) + ZHS16GBK，按字节截。"""
|
||
return _clip_encoded(text, DEAL_RESULT_MAX, "gbk")
|
||
|
||
|
||
def _clip_encoded(text: str, max_bytes: int, encoding: str) -> str:
|
||
text = text.strip()
|
||
raw = text.encode(encoding, errors="replace")
|
||
if len(raw) <= max_bytes:
|
||
return text
|
||
ellipsis = "…"
|
||
ell = ellipsis.encode(encoding, errors="replace")
|
||
budget = max(0, max_bytes - len(ell))
|
||
cut = raw[:budget]
|
||
while cut:
|
||
try:
|
||
return cut.decode(encoding) + ellipsis
|
||
except UnicodeDecodeError:
|
||
cut = cut[:-1]
|
||
return ellipsis if len(ell) <= max_bytes else ""
|