318 lines
9.3 KiB
Python
318 lines
9.3 KiB
Python
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))
|