from __future__ import annotations import re from calendar import monthrange from dataclasses import dataclass, field from datetime import date, datetime 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 = 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" REASON_SAME_WAREHOUSE = "SAME_WAREHOUSE_NOT_BEST" @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) @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 @dataclass class Allocation: candidate: Candidate qty: int @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 def margin(sale_price: float, platform_price: float) -> float: return (sale_price - platform_price) / sale_price def margin_ok(sale_price: float, platform_price: float) -> bool: return margin(sale_price, platform_price) >= MIN_MARGIN 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) 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 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 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" return 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 def stock_units(stock: float | None) -> int: if stock is None or stock <= 0: return 0 return int(stock) def plan_qty(purchase: PurchaseLine) -> int: qty = purchase.qty if qty is None or qty <= 0: return 0 return int(qty) 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, rejected) 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, ) def _fold_same_warehouse( passed: list[Candidate], rejected: list[Rejected], ) -> list[Candidate]: best: dict[tuple[str, str], Candidate] = {} order: list[tuple[str, str]] = [] for candidate in passed: key = (candidate.mall_name, candidate.vendor_norm) current = best.get(key) if current is None: best[key] = candidate order.append(key) continue if _sku_better(candidate, current): rejected.append(Rejected(current, REASON_SAME_WAREHOUSE)) best[key] = candidate else: rejected.append(Rejected(candidate, REASON_SAME_WAREHOUSE)) return [best[key] for key in order] def _sku_better(left: Candidate, right: Candidate) -> bool: left_price = left.price if left.price is not None else float("inf") right_price = right.price if right.price is not None else float("inf") if left_price != right_price: return left_price < right_price return stock_units(left.stock) > stock_units(right.stock) 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))