128 lines
4.5 KiB
Python
128 lines
4.5 KiB
Python
# 药帮忙按条形码搜索，收成选品候选。无货的价格记 0，不参与拆量。
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
import requests
|
||
|
||
from http_util import post_json_with_retry
|
||
from mall_config import YBM_MALL_NAME, YBM_SEARCH_API_URL, YBM_SORT_STRATEGY
|
||
from rules import Candidate, normalize_vendor, parse_expiry_months
|
||
from util import normalize_text, parse_float_or_none, parse_int_or_none
|
||
|
||
|
||
# 拼药帮忙搜索体。关键字是条形码，只要效期超过一年的。
|
||
def build_ybm_search_payload(query_key: str) -> dict[str, Any]:
|
||
return {
|
||
"queryWord": query_key,
|
||
"searchScene": 1,
|
||
"nearEffect": "",
|
||
"minPrice": "",
|
||
"maxPrice": "",
|
||
"drugClassificationsStr": "",
|
||
"isThirdCompany": 0,
|
||
"isEffectPeriodOverOneYear": 1,
|
||
"sortStrategy": YBM_SORT_STRATEGY,
|
||
}
|
||
|
||
|
||
# 取平台价。price 空了再用 fob。
|
||
def pick_ybm_platform_price(product: dict[str, Any]) -> Any:
|
||
price = product.get("price")
|
||
if normalize_text(price) != "":
|
||
return price
|
||
fob = product.get("fob")
|
||
if normalize_text(fob) != "":
|
||
return fob
|
||
return ""
|
||
|
||
|
||
# 按条形码查药帮忙，返回商品行。接口声明失败就报错，不当成没搜到。
|
||
def query_ybm_products(
|
||
session: requests.Session,
|
||
query_key: str,
|
||
) -> list[dict[str, Any]]:
|
||
headers = {
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Content-Type": "application/json",
|
||
"Origin": "https://www.ybm100.com",
|
||
"Referer": "https://www.ybm100.com/new/",
|
||
"X-Requested-With": "XMLHttpRequest",
|
||
"User-Agent": (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/152.0.0.0 Safari/537.36"
|
||
),
|
||
}
|
||
payload = build_ybm_search_payload(query_key)
|
||
payload_json = post_json_with_retry(
|
||
session,
|
||
YBM_SEARCH_API_URL,
|
||
platform_name=YBM_MALL_NAME,
|
||
query_key=query_key,
|
||
json=payload,
|
||
headers=headers,
|
||
)
|
||
if not payload_json.get("success"):
|
||
error_msg = normalize_text(payload_json.get("msg")) or "未知错误"
|
||
raise RuntimeError(f"药帮忙查询失败：{error_msg}")
|
||
|
||
data_section = payload_json.get("data") or {}
|
||
rows = data_section.get("rows") or []
|
||
if not isinstance(rows, list):
|
||
raise RuntimeError(f"药帮忙接口返回 data.rows 结构异常：{type(rows)}")
|
||
return rows
|
||
|
||
|
||
# 把药帮忙商品收成候选。库存为 0 时价格记 0，后面拆量不会吃它。
|
||
def map_ybm_products(rows: list[dict[str, Any]]) -> list[Candidate]:
|
||
candidates: list[Candidate] = []
|
||
for row in rows:
|
||
product = row.get("productInfo") or {}
|
||
available_qty = product.get("availableQty", "")
|
||
available_num = parse_int_or_none(available_qty)
|
||
no_stock = available_num == 0 or normalize_text(available_qty) == "0"
|
||
raw_price = 0 if no_stock else pick_ybm_platform_price(product)
|
||
vendor_raw = normalize_text(product.get("shopName"))
|
||
expiry_raw = normalize_text(product.get("nearEffect"))
|
||
candidates.append(
|
||
Candidate(
|
||
mall_name=YBM_MALL_NAME,
|
||
sku_id=normalize_text(product.get("skuId", row.get("skuId", ""))),
|
||
name=normalize_text(product.get("showName")),
|
||
spec=normalize_text(product.get("spec")),
|
||
price=parse_float_or_none(raw_price),
|
||
expiry_raw=expiry_raw,
|
||
expiry_months=parse_expiry_months(expiry_raw),
|
||
vendor_raw=vendor_raw,
|
||
vendor_norm=normalize_vendor(vendor_raw),
|
||
stock=parse_float_or_none(available_qty),
|
||
extra={"库存说明": "无货" if no_stock else "有货"},
|
||
)
|
||
)
|
||
return candidates
|
||
|
||
|
||
# 按条形码搜药帮忙。同一查询键本班只打一次接口。
|
||
def search_candidates(
|
||
cookies: dict[str, str],
|
||
query_key: str,
|
||
session: requests.Session | None = None,
|
||
cache: dict[str, list[Candidate]] | None = None,
|
||
) -> list[Candidate]:
|
||
store = cache if cache is not None else {}
|
||
if query_key in store:
|
||
return store[query_key]
|
||
own_session = session or requests.Session()
|
||
try:
|
||
if session is None:
|
||
for name, value in cookies.items():
|
||
own_session.cookies.set(name, value)
|
||
rows = query_ybm_products(own_session, query_key)
|
||
mapped = map_ybm_products(rows)
|
||
finally:
|
||
if session is None:
|
||
own_session.close()
|
||
store[query_key] = mapped
|
||
return mapped
|