92 lines
2.6 KiB
Python
92 lines
2.6 KiB
Python
"""再搜条形码，保留原始商品字段给加车用。"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
from cart_match import CartOffer
|
|
from yaoex_login import SELECTION_ROOT
|
|
|
|
if str(SELECTION_ROOT) not in sys.path:
|
|
sys.path.append(str(SELECTION_ROOT))
|
|
|
|
from search_yaoex import map_yaoex_products, query_yaoex_products # noqa: E402
|
|
from search_ybm import map_ybm_products, query_ybm_products # noqa: E402
|
|
|
|
|
|
def search_yaoex_offers(
|
|
auth,
|
|
query_key: str,
|
|
session: requests.Session | None = None,
|
|
cache: dict[str, list[CartOffer]] | None = None,
|
|
) -> list[CartOffer]:
|
|
store = cache if cache is not None else {}
|
|
if query_key in store:
|
|
return store[query_key]
|
|
own = session is None
|
|
client = session or requests.Session()
|
|
try:
|
|
products, user_id = query_yaoex_products(client, auth, query_key)
|
|
candidates = map_yaoex_products(products, user_id)
|
|
offers = [
|
|
_from_candidate(candidate, raw)
|
|
for candidate, raw in zip(candidates, products)
|
|
]
|
|
finally:
|
|
if own:
|
|
client.close()
|
|
store[query_key] = offers
|
|
return offers
|
|
|
|
|
|
def search_ybm_offers(
|
|
cookies: dict[str, str],
|
|
query_key: str,
|
|
session: requests.Session | None = None,
|
|
cache: dict[str, list[CartOffer]] | None = None,
|
|
) -> list[CartOffer]:
|
|
store = cache if cache is not None else {}
|
|
if query_key in store:
|
|
return store[query_key]
|
|
own = session is None
|
|
client = session or requests.Session()
|
|
try:
|
|
if own:
|
|
for name, value in cookies.items():
|
|
client.cookies.set(name, value)
|
|
rows = query_ybm_products(client, query_key)
|
|
candidates = map_ybm_products(rows)
|
|
offers = [
|
|
_from_candidate(candidate, _ybm_raw(row))
|
|
for candidate, row in zip(candidates, rows)
|
|
]
|
|
finally:
|
|
if own:
|
|
client.close()
|
|
store[query_key] = offers
|
|
return offers
|
|
|
|
|
|
def _ybm_raw(row: dict[str, Any]) -> dict[str, Any]:
|
|
product = row.get("productInfo")
|
|
raw = dict(product) if isinstance(product, dict) else dict(row)
|
|
if not raw.get("skuId"):
|
|
raw["skuId"] = row.get("skuId")
|
|
return raw
|
|
|
|
|
|
def _from_candidate(candidate, raw: dict[str, Any]) -> CartOffer:
|
|
extra = dict(candidate.extra)
|
|
extra["raw"] = raw
|
|
return CartOffer(
|
|
sku_id=candidate.sku_id,
|
|
vendor_raw=candidate.vendor_raw,
|
|
vendor_norm=candidate.vendor_norm,
|
|
stock=candidate.stock,
|
|
price=candidate.price,
|
|
extra=extra,
|
|
)
|