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

from __future__ import annotations

import json
import re
import urllib.request
from typing import Any

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

_SEARCH_URL = "https://lcsc.com/api/products/search"
_SESSION_SEED_URL = "https://lcsc.com/products/Capacitors_11.html"
_UA = "Mozilla/5.0 (X11; Linux x86_64; rv:120.0) Gecko/20100101 Firefox/120.0"


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

    Modes:
      api_token=None (default) — scraper mode: acquires CSRF session from a
        category page, then POSTs to lcsc.com/api/products/search.
      api_token=<key>          — official API mode (not yet implemented, raises
        NotImplementedError until LCSC credentials are available).
    """

    def __init__(self, api_token: str | None = None) -> None:
        super().__init__(api_token)
        self._cookies: str | None = None
        self._csrf: str | None = None

    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_scraper(category_id, params, filters, page)

    # ------------------------------------------------------------------
    # Scraper mode
    # ------------------------------------------------------------------

    def _ensure_session(self) -> None:
        if self._csrf is not None:
            return
        req = urllib.request.Request(_SESSION_SEED_URL, headers={"User-Agent": _UA})
        with urllib.request.urlopen(req, timeout=15) as resp:
            self._cookies = resp.headers.get("Set-Cookie", "")
            html = resp.read().decode("utf-8", errors="replace")
        m = re.search(r'csrfToken["\s:=]+(["\'])([A-Za-z0-9_\-]+)\1', html)
        self._csrf = m.group(2) if m else ""

    def _search_scraper(
        self,
        category_id: int,
        params: NormalizedParams,
        filters: dict[str, Any] | None,
        page: int,
    ) -> SearchResult:
        try:
            self._ensure_session()
        except Exception as exc:
            return SearchResult(count=0, page=page, page_size=25, items=[], error=f"session: {exc}")

        payload: dict[str, Any] = {
            "current_page": page,
            "page_size": 25,
            "catalog_id": category_id,
            "in_stock": False,
            "is_RoHS": False,
            "show_icon": False,
        }
        if filters:
            payload.update(filters)

        data = json.dumps(payload).encode()
        headers: dict[str, str] = {
            "User-Agent": _UA,
            "Content-Type": "application/json",
            "X-CSRF-Token": self._csrf or "",
            "Referer": f"https://lcsc.com/category/{category_id}.html",
        }
        if self._cookies:
            headers["Cookie"] = self._cookies

        req = urllib.request.Request(_SEARCH_URL, data=data, headers=headers, method="POST")
        try:
            with urllib.request.urlopen(req, timeout=15) as resp:
                body = json.loads(resp.read().decode())
        except Exception as exc:
            return SearchResult(count=0, page=page, page_size=25, items=[], error=str(exc))

        vo = body.get("data", {}).get("productSearchResultVO", {})
        return SearchResult(
            count=vo.get("totalCount", 0),
            page=vo.get("currentPage", page),
            page_size=vo.get("pageSize", 25),
            items=vo.get("productList", []),
        )

    # ------------------------------------------------------------------
    # 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:
        # TODO: HMAC-sign request → POST https://ips.lcsc.com/rest/wmsc2agent/category/product/{category_id}
        raise NotImplementedError(
            "Official LCSC API not yet implemented. "
            "Use LcscAdapter() without api_token to use scraper mode."
        )