248 lines
8.1 KiB
Python
248 lines
8.1 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 = "系统故障：对端平台不可用，合计库存不足"
|
||
|
||
|
||
# 一行选品要写入表 1 / 表 2 的结果。
|
||
@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:
|
||
"""把判定收成表 2 行和表 1 的 deal_flag / deal_result。缺 oms_id 则失败。
|
||
|
||
LOGIN_FAILED / QUERY_FAILED 且凑不齐时 deal_flag 改回 0，不写表 2。
|
||
库存不足、毛利、效期仍写 deal_flag=2。
|
||
"""
|
||
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:
|
||
detail = "; ".join(selection.mall_errors.values()) or "未知错误"
|
||
return SelectResult(
|
||
oms_id=purchase.oms_id,
|
||
deal_flag=0,
|
||
deal_result=_clip_deal_result(f"系统故障：{detail}"),
|
||
cart_rows=[],
|
||
)
|
||
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,
|
||
)
|
||
|
||
|
||
# 成功行出现两个 platform_type 时 multi_platform_flag=1。
|
||
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
|
||
|
||
|
||
# 从表 2 行还原平台名，药帮忙在前。
|
||
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
|
||
|
||
|
||
# 按拆出仓构造一条 status=1 的表 2 行。
|
||
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
|
||
|
||
|
||
# 构造一条 status=2 的表 2 行，供应商用占位名。
|
||
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
|
||
|
||
|
||
# 表 2 公共字段。未知商城名直接失败。
|
||
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,
|
||
}
|
||
|
||
|
||
# 失败行采购量：计划量，缺或非正则 1。
|
||
def _qty(purchase: PurchaseLine) -> int:
|
||
qty = purchase.qty
|
||
if qty is None or qty <= 0:
|
||
return 1
|
||
return int(qty)
|
||
|
||
|
||
# 失败行单价用销售价；没有则 0.0100。
|
||
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")
|
||
|
||
|
||
_BUSINESS_PREFIX = "无符合规则商品，"
|
||
|
||
|
||
# 把规则短码收成表 1/表 2 可读的业务失败原因。多个短码都写上。
|
||
def _business_fail_reason(code: str) -> str:
|
||
parts = [part for part in code.split(";") if part]
|
||
if len(parts) <= 1:
|
||
return _one_business_fail_reason(code)
|
||
tails = [_reason_tail(_one_business_fail_reason(part)) for part in parts]
|
||
return _BUSINESS_PREFIX + "；".join(tails)
|
||
|
||
|
||
# 单个短码对应的文案。
|
||
def _one_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"{_BUSINESS_PREFIX}{code}"
|
||
return "无符合规则商品"
|
||
|
||
|
||
# 拼多条原因时去掉重复的前缀。
|
||
def _reason_tail(text: str) -> str:
|
||
if text.startswith(_BUSINESS_PREFIX):
|
||
return text[len(_BUSINESS_PREFIX) :]
|
||
return text
|
||
|
||
|
||
# 本平台有错误短码则带上；没有则写对端不可用。
|
||
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] + "…"
|
||
|
||
|
||
# 表 1 deal_result 是 VARCHAR2(200 BYTE) + ZHS16GBK，按字节截。
|
||
def _clip_deal_result(text: str) -> str:
|
||
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 ""
|