# 选品规则：自营仓全称匹配、毛利严格大于 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))