instant-bom/bom_assistant/suppliers/lcsc/lcsc.py

from __future__ import annotations

import json
import urllib.request
from typing import Any

from bom_assistant.session.models import NormalizedParams
from bom_assistant.suppliers.base import SearchResult, SupplierAdapter

_FACET_URL = "https://wmsc.lcsc.com/ftps/wm/product/query/param/group"
_SEARCH_URL = "https://wmsc.lcsc.com/ftps/wm/product/query/list"
_UA = "Mozilla/5.0 (X11; Linux x86_64; rv:120.0) Gecko/20100101 Firefox/120.0"
_PAGE_SIZE = 25

_FILTER_DEFAULTS: dict[str, Any] = {
    "keyword": "",
    "brandIdList": [],
    "encapValueList": [],
    "isStock": False,
    "isOtherSuppliers": False,
    "isAsianBrand": False,
    "isDeals": False,
    "isEnvironment": False,
    "paramNameValueMap": {},
}


def _build_filter_payload(category_id: int, filters: dict[str, Any] | None) -> dict[str, Any]:
    payload = dict(_FILTER_DEFAULTS)
    payload["catalogIdList"] = [category_id]
    if filters:
        payload.update(filters)
    return payload


def _post(url: str, payload: dict[str, Any]) -> dict[str, Any]:
    data = json.dumps(payload).encode()
    headers = {
        "User-Agent": _UA,
        "Accept": "application/json, text/plain, */*",
        "Content-Type": "application/json;charset=utf-8",
        "Origin": "https://www.lcsc.com",
        "Referer": "https://www.lcsc.com/",
    }
    req = urllib.request.Request(url, data=data, headers=headers, method="POST")
    with urllib.request.urlopen(req, timeout=15) as resp:
        return json.loads(resp.read().decode())


class LcscAdapter(SupplierAdapter):
    """
    LCSC supplier adapter.

    Modes:
      api_token=None (default) — talks to LCSC's real internal JSON API
        (wmsc.lcsc.com/ftps/wm/product/query/*), reverse-engineered from live
        browser traffic. No CSRF/session dance needed — plain POST with
        User-Agent/Origin/Referer is sufficient.
      api_token=<key>          — official API mode (not yet implemented, raises
        NotImplementedError until LCSC credentials are available).
    """

    def search(
        self,
        category_id: int,
        params: NormalizedParams,
        filters: dict[str, Any] | None = None,
        page: int = 1,
    ) -> SearchResult:
        if self.api_token:
            return self._search_official(category_id, params, filters, page)
        return self._search_live(category_id, filters, page)

    def query_facets(
        self,
        category_id: int,
        filters: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """
        Returns LCSC's own facet groups for this category (+ any already-applied
        filters): {"Package": [...], "Manufacturer": [...], "Packaging": [...],
        "paramNameValueMap": {...}}, or {"error": "..."} on failure.
        """
        payload = _build_filter_payload(category_id, filters)
        try:
            body = _post(_FACET_URL, payload)
        except Exception as exc:
            return {"error": str(exc)}
        if not body.get("ok"):
            return {"error": body.get("msg") or "facet query failed"}
        return body.get("result", {})

    # ------------------------------------------------------------------
    # Live API mode
    # ------------------------------------------------------------------

    def _search_live(
        self,
        category_id: int,
        filters: dict[str, Any] | None,
        page: int,
    ) -> SearchResult:
        payload = _build_filter_payload(category_id, filters)
        payload["currentPage"] = page
        payload["pageSize"] = _PAGE_SIZE

        try:
            body = _post(_SEARCH_URL, payload)
        except Exception as exc:
            return SearchResult(count=0, page=page, page_size=_PAGE_SIZE, items=[], error=str(exc))

        if not body.get("ok"):
            error = body.get("msg") or "search failed"
            return SearchResult(count=0, page=page, page_size=_PAGE_SIZE, items=[], error=error)

        result = body.get("result", {})
        return SearchResult(
            count=result.get("totalRow", 0),
            page=result.get("currPage", page),
            page_size=result.get("pageRow", _PAGE_SIZE),
            items=result.get("dataList", []),
        )

    # ------------------------------------------------------------------
    # Official API mode (placeholder — needs LCSC API key + HMAC impl)
    # ------------------------------------------------------------------

    def _search_official(
        self,
        category_id: int,
        params: NormalizedParams,
        filters: dict[str, Any] | None,
        page: int,
    ) -> SearchResult:
        raise NotImplementedError(
            "Official LCSC API not yet implemented. "
            "Use LcscAdapter() without api_token to use the live API mode."
        )