324 lines
10 KiB
Python
324 lines
10 KiB
Python
# 选品规则：自营仓全称匹配、毛利严格大于 4%、效期不少于 12 个月，按仓序贪心拆量。
|
||
#
|
||
# 凑齐计划量才给出拆量行。凑不齐时 allocations 为空，调用方不落部分成功。
|
||
# 湖南壹号、甘肃壹号云不在自营集合。
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from calendar import monthrange
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, datetime
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
# 仓序：级内由近到远。全称不在这张表里的商家不是自营仓。
|
||
WAREHOUSE_ORDER = [
|
||
"杭州小药药自营旗舰店",
|
||
"长沙小药药自营旗舰店",
|
||
"湖北小药药自营旗舰店",
|
||
"广东壹号药业有限公司",
|
||
"昆山亿方药业有限公司",
|
||
"湖北亿昊药业有限公司",
|
||
"济南小药药自营旗舰店",
|
||
"天津壹号药业有限公司",
|
||
"重庆亿昊药业有限公司",
|
||
"辽宁亿昊药业有限公司",
|
||
"陕西壹号药业有限公司",
|
||
"河南壹号医药有限公司",
|
||
"重庆壹号医药有限公司",
|
||
"重庆小药药自营旗舰店",
|
||
]
|
||
|
||
WAREHOUSE_RANK = {name: index for index, name in enumerate(WAREHOUSE_ORDER)}
|
||
WAREHOUSE_ALIASES: dict[str, str] = {}
|
||
SELF_OPERATED_PREFIX = "[自营]"
|
||
MIN_MARGIN = Decimal("0.04")
|
||
MIN_EXPIRY_MONTHS = 12.0
|
||
DAYS_PER_MONTH = 30.4375
|
||
|
||
REASON_NOT_SELF = "NOT_SELF_WAREHOUSE"
|
||
REASON_MARGIN = "MARGIN_LT_4PCT"
|
||
REASON_EXPIRY = "EXPIRY_LT_12M"
|
||
REASON_EXPIRY_UNPARSEABLE = "EXPIRY_UNPARSEABLE"
|
||
REASON_PRICE_UNPARSEABLE = "PRICE_UNPARSEABLE"
|
||
|
||
|
||
# 一条商城搜索结果：商家、价格、效期、库存。后面按仓序拆量。
|
||
@dataclass
|
||
class Candidate:
|
||
mall_name: str
|
||
sku_id: str
|
||
name: str
|
||
spec: str
|
||
price: float | None
|
||
expiry_raw: str
|
||
expiry_months: float | None
|
||
vendor_raw: str
|
||
vendor_norm: str
|
||
stock: float | None
|
||
extra: dict[str, Any] = field(default_factory=dict)
|
||
|
||
|
||
# 表 1 的一条待选品需求。按条形码搜，计划量是 qty。
|
||
@dataclass
|
||
class PurchaseLine:
|
||
approval_no: str
|
||
sale_price: float | None
|
||
qty: float | None
|
||
row_no: int | None = None
|
||
generic_name: str = ""
|
||
spec: str = ""
|
||
manufacturer: str = ""
|
||
extra: dict[str, Any] = field(default_factory=dict)
|
||
oms_id: int | None = None
|
||
bar_code: str = ""
|
||
org_code: str = ""
|
||
|
||
|
||
# 没通过自营、毛利或效期的候选，以及原因码。
|
||
@dataclass
|
||
class Rejected:
|
||
candidate: Candidate
|
||
reason: str
|
||
|
||
|
||
# 一个仓分到的数量。同仓多条 sku 都留，合计才是该仓吃掉的量。
|
||
@dataclass
|
||
class Allocation:
|
||
candidate: Candidate
|
||
qty: int
|
||
|
||
|
||
# 拆量结果。凑齐时 allocations 合计等于计划量；凑不齐时 allocations 为空。
|
||
@dataclass
|
||
class AllocationResult:
|
||
allocations: list[Allocation]
|
||
rejected: list[Rejected]
|
||
unmatched_vendors: list[str]
|
||
passed_before_fold: list[Candidate]
|
||
|
||
|
||
# 去掉「[自营]」前缀和多余空白，收成仓序表里的商家全称。
|
||
def normalize_vendor(raw: Any) -> str:
|
||
text = "" if raw is None else str(raw)
|
||
text = text.replace("\u3000", " ").strip()
|
||
if text.startswith(SELF_OPERATED_PREFIX):
|
||
text = text[len(SELF_OPERATED_PREFIX) :]
|
||
text = " ".join(text.split())
|
||
return WAREHOUSE_ALIASES.get(text, text)
|
||
|
||
|
||
# 商家全称是否在自营仓序里。
|
||
def is_self_warehouse(vendor_norm: str) -> bool:
|
||
return vendor_norm in WAREHOUSE_RANK
|
||
|
||
|
||
# (销售价 - 平台价) / 销售价。用十进制，避免 104 与 99.84 漂成不等于 4%。
|
||
def margin(sale_price: float, platform_price: float) -> Decimal:
|
||
sale = Decimal(str(sale_price))
|
||
return (sale - Decimal(str(platform_price))) / sale
|
||
|
||
|
||
# 毛利必须严格大于 4%。刚好 4% 不通过。
|
||
def margin_ok(sale_price: float, platform_price: float) -> bool:
|
||
return margin(sale_price, platform_price) > MIN_MARGIN
|
||
|
||
|
||
# 把效期原文收成剩余月数。日期按今天算；「长期」按 120 个月；认不出返回 None。
|
||
def parse_expiry_months(raw: Any, today: date | None = None) -> float | None:
|
||
text = "" if raw is None else str(raw).strip()
|
||
if not text or text in {"-", "—", "/", "无", "null", "None"}:
|
||
return None
|
||
text = text.split()[0].strip()
|
||
today = today or date.today()
|
||
if "长期" in text:
|
||
return 120.0
|
||
|
||
chinese = re.fullmatch(r"(\d{4})年(\d{1,2})月(?:(\d{1,2})日?)?", text)
|
||
if chinese:
|
||
year, month = int(chinese.group(1)), int(chinese.group(2))
|
||
day = int(chinese.group(3) or 0)
|
||
return remaining_months(today, _safe_date(year, month, day))
|
||
|
||
if "月" in text:
|
||
matched = re.search(r"(\d+(?:\.\d+)?)", text)
|
||
if matched is None:
|
||
return None
|
||
return float(matched.group(1))
|
||
|
||
dotted = re.fullmatch(r"(\d{4})[./-](\d{1,2})(?:[./-](\d{1,2}))?", text)
|
||
if dotted:
|
||
year, month = int(dotted.group(1)), int(dotted.group(2))
|
||
day = int(dotted.group(3) or 0)
|
||
return remaining_months(today, _safe_date(year, month, day))
|
||
|
||
if re.fullmatch(r"\d{13}", text):
|
||
expiry_date = datetime.fromtimestamp(int(text) / 1000).date()
|
||
return remaining_months(today, expiry_date)
|
||
if re.fullmatch(r"\d{8}", text):
|
||
parsed = datetime.strptime(text, "%Y%m%d").date()
|
||
return remaining_months(today, parsed)
|
||
if re.fullmatch(r"\d{6}", text):
|
||
parsed = datetime.strptime(text, "%Y%m")
|
||
last_day = monthrange(parsed.year, parsed.month)[1]
|
||
return remaining_months(today, date(parsed.year, parsed.month, last_day))
|
||
|
||
if re.fullmatch(r"\d+(?:\.\d+)?", text):
|
||
value = float(text)
|
||
if value > 240:
|
||
return None
|
||
return value
|
||
|
||
expiry_date = _parse_expiry_date(text)
|
||
if expiry_date is None:
|
||
return None
|
||
return remaining_months(today, expiry_date)
|
||
|
||
|
||
# 只有年月、或日超出当月时，落到该月最后一天。
|
||
def _safe_date(year: int, month: int, day: int) -> date:
|
||
last_day = monthrange(year, month)[1]
|
||
if day <= 0 or day > last_day:
|
||
day = last_day
|
||
return date(year, month, day)
|
||
|
||
|
||
# 从今天到失效日还剩几个月。日差按 30.4375 天折成月。
|
||
def remaining_months(today: date, expiry: date) -> float:
|
||
months = (expiry.year - today.year) * 12 + (expiry.month - today.month)
|
||
months += (expiry.day - today.day) / DAYS_PER_MONTH
|
||
return months
|
||
|
||
|
||
# 解析 YYYY-MM-DD 或 YYYY-MM。只有年月时用该月最后一天。
|
||
def _parse_expiry_date(text: str) -> date | None:
|
||
for fmt in ("%Y-%m-%d", "%Y/%m/%d"):
|
||
try:
|
||
return datetime.strptime(text, fmt).date()
|
||
except ValueError:
|
||
continue
|
||
for fmt in ("%Y-%m", "%Y/%m"):
|
||
try:
|
||
parsed = datetime.strptime(text, fmt)
|
||
last_day = monthrange(parsed.year, parsed.month)[1]
|
||
return date(parsed.year, parsed.month, last_day)
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
|
||
# 需求本身能不能选：要有销售价、计划量和条形码。过了返回 None。
|
||
def validate_purchase(purchase: PurchaseLine) -> str | None:
|
||
if purchase.sale_price is None or purchase.sale_price <= 0:
|
||
return "NO_SALE_PRICE"
|
||
if purchase.qty is None or purchase.qty <= 0:
|
||
return "NO_QTY"
|
||
if not purchase.bar_code:
|
||
return "NO_BAR_CODE"
|
||
return None
|
||
|
||
|
||
# 单条候选过不过：自营仓、价格能解析、毛利严格大于 4%、效期不少于 12 个月。过了返回 None。
|
||
def evaluate_candidate(
|
||
candidate: Candidate,
|
||
purchase: PurchaseLine,
|
||
today: date | None = None,
|
||
) -> str | None:
|
||
if not is_self_warehouse(candidate.vendor_norm):
|
||
return REASON_NOT_SELF
|
||
if candidate.price is None:
|
||
return REASON_PRICE_UNPARSEABLE
|
||
if purchase.sale_price is None or purchase.sale_price <= 0:
|
||
return "NO_SALE_PRICE"
|
||
if not margin_ok(purchase.sale_price, candidate.price):
|
||
return REASON_MARGIN
|
||
|
||
expiry_months = candidate.expiry_months
|
||
if expiry_months is None:
|
||
expiry_months = parse_expiry_months(candidate.expiry_raw, today=today)
|
||
if expiry_months is None:
|
||
return REASON_EXPIRY_UNPARSEABLE
|
||
if expiry_months < MIN_EXPIRY_MONTHS:
|
||
return REASON_EXPIRY
|
||
|
||
return None
|
||
|
||
|
||
# 库存向下取整。没有或非正数当 0，不参与拆量。
|
||
def stock_units(stock: float | None) -> int:
|
||
if stock is None or stock <= 0:
|
||
return 0
|
||
return int(stock)
|
||
|
||
|
||
# 计划采购量向下取整。没有或非正数当 0。
|
||
def plan_qty(purchase: PurchaseLine) -> int:
|
||
qty = purchase.qty
|
||
if qty is None or qty <= 0:
|
||
return 0
|
||
return int(qty)
|
||
|
||
|
||
# 按仓序贪心拆计划量。同仓多个 sku 都参与，低价先吃。
|
||
#
|
||
# 凑齐才返回拆量行，合计等于计划量。凑不齐清空 allocations，不落部分成功。
|
||
def allocate(
|
||
candidates: list[Candidate],
|
||
purchase: PurchaseLine,
|
||
today: date | None = None,
|
||
) -> AllocationResult:
|
||
rejected: list[Rejected] = []
|
||
passed: list[Candidate] = []
|
||
unmatched: list[str] = []
|
||
seen_unmatched: set[str] = set()
|
||
|
||
for candidate in candidates:
|
||
if not is_self_warehouse(candidate.vendor_norm):
|
||
raw = candidate.vendor_raw
|
||
if raw and raw not in seen_unmatched:
|
||
unmatched.append(raw)
|
||
seen_unmatched.add(raw)
|
||
reason = evaluate_candidate(candidate, purchase, today=today)
|
||
if reason is None:
|
||
passed.append(candidate)
|
||
else:
|
||
rejected.append(Rejected(candidate, reason))
|
||
|
||
folded = _fold_same_warehouse(passed)
|
||
folded.sort(key=_sort_key)
|
||
|
||
remaining = plan_qty(purchase)
|
||
lines: list[Allocation] = []
|
||
if remaining > 0:
|
||
for candidate in folded:
|
||
take = min(stock_units(candidate.stock), remaining)
|
||
if take <= 0:
|
||
continue
|
||
lines.append(Allocation(candidate, take))
|
||
remaining -= take
|
||
if remaining == 0:
|
||
break
|
||
|
||
if remaining != 0:
|
||
lines = []
|
||
|
||
return AllocationResult(
|
||
allocations=lines,
|
||
rejected=rejected,
|
||
unmatched_vendors=unmatched,
|
||
passed_before_fold=passed,
|
||
)
|
||
|
||
|
||
# 同仓多个 sku 都留下。低价先拆由后面的排序决定，这里不再丢掉贵的那条。
|
||
def _fold_same_warehouse(passed: list[Candidate]) -> list[Candidate]:
|
||
return list(passed)
|
||
|
||
|
||
# 仓序在前，同仓低价在前，同价库存多的在前。
|
||
def _sort_key(candidate: Candidate) -> tuple[int, float, int]:
|
||
rank = WAREHOUSE_RANK[candidate.vendor_norm]
|
||
price = candidate.price if candidate.price is not None else float("inf")
|
||
return (rank, price, -stock_units(candidate.stock))
|