# build-aux/jlcpcb_api.py - Reverse-engineered JLCPCB API client
#
# Copyright (C) 2025  Luke T. Shumaker <lukeshu@lukeshu.com>
# SPDX-License-Identifier: AGPL-3.0-or-later

import email.message
import enum
import io
import json
import os.path
import time
import typing
import urllib.error
import urllib.parse
import urllib.request

import pydantic

# pylint: disable=unused-variable
__all__ = [
    "Client",
]

# Struct types #################################################################

type _Incomplete = typing.Any


class BOMFilePermission(pydantic.BaseModel):
    currentUser: bool
    bomFileAccessId: bool
    read: bool
    modify: bool
    history: bool
    currentModify: bool
    customerName: _Incomplete | None
    expire: _Incomplete | None
    matchUserCode: str
    permissions: _Incomplete | None
    expireDate: _Incomplete | None
    starDate: _Incomplete | None
    linkUrl: _Incomplete | None


class BOMFileDetail(pydantic.BaseModel):
    bomMatchFileRecordAccessId: str  # ID returned from .upload_bomfile()
    bomFileName: str  # name passed to .upload_bomfile()
    delete: bool
    fileAccessID: str  # ???
    patchNumber: int
    bomMatchRecordAccessID: str  # ???
    createTime: int  # milliseconds since Unix epoch
    manualFlag: _Incomplete | None
    operatorName: _Incomplete | None


class BOMFileDetailEdits(typing.TypedDict):
    editPatchNumber: typing.NotRequired[int]
    # TODO: Presumably there are more fields


class BOMFileEdits(typing.TypedDict):
    delete: typing.NotRequired[typing.Literal[True]]
    # TODO: Presumably there are more fields


class BOMFileMatchComponentPrice(pydantic.BaseModel):
    componentPriceKeyId: _Incomplete | None  # I've only ever seen `null`
    componentPriceAccessId: _Incomplete | None  # I've only ever seen `null`
    lcscComponentId: _Incomplete | None  # I've only ever seen `null`
    componentCode: str  # LCSC "C____" part number
    startNumber: int  # min order qty for this price range
    endNumber: int  # max order qty for this price range, `-1` for infinity
    productPrice: float  # unit-price for this price range, in USD?
    erpComponentPriceKeyId: int | str  # ???
    createTime: _Incomplete | None  # I've only ever seen `null`
    updateTime: _Incomplete | None  # I've only ever seen `null`
    deleted: bool  # I've only ever seen `false`
    lcscPriceRangeCoefficient: int  # I've only ever seen `1`


class BOMFileMatchType(enum.IntEnum):
    no_matches = 1
    select_by_system = 3
    select_by_customer = 4


class BOMFileMatchSource(enum.IntEnum):
    from_value_designator_footprint = 8
    from_partnum = 9


class BOMFileMatchLibraryType(enum.StrEnum):
    basic = "base"
    extended = "expand"


class BOMFileMatchComponent(pydantic.BaseModel):
    bomResultAccessId: str  # ???
    matchType: BOMFileMatchType
    componentCode: str | None  # LCSC "C____" part number

    # Fields from the uploaded BOM
    bomComponentModel: str  # "Comment" column from the uploaded BOM; eg "10kOhm"
    bomSpecification: str  # "Footprint" column from the uploaded BOM; eg "0603" or KiCad-style "Resistor_SMD:R_0603_1608Metric"
    designator: str  # "Designator" column from the uploaded BOM; eg "R1,R2"
    designatorNum: int  # len(designator.split(","))

    # Fields from the matched part
    componentModel: str | None  # mfr part number; eg "0402WGF1001TCE"
    componentSpecification: str | None  # footprint; eg short "0603"
    componentLibraryType: BOMFileMatchLibraryType | None
    componentDescription: str | None  # freeform human-readable part description
    componentBrand: str | None  # mfr name; eg "UNI-ROYAL(Uniroyal Elec)"
    componentName: (
        str | None
    )  # short mfr name with mfr part number; eg "UNI-ROYAL 0402WGF1001TCE"
    lcscComponentId: (
        int | None
    )  # not anything to do with LCSC actually, this is a JLCPCB-assigned ID for the JLCPCB API

    # I think these are for if you provide the parts
    # instead of using JLCPCB's parts library?
    provider: _Incomplete | None
    productCodeProvider: _Incomplete | None
    sourcePartId: _Incomplete | None
    productRegion: _Incomplete | None

    shopCartNumber: int | None  # how many are currently in your shopping cart
    presaleNumber: int | None  # ???
    overseasShopNumber: int | None  # ???

    recommendNumber: (
        int | None
    )  # max(count_from_uploaded_bom * patchNumber + .lossNumber, .minPurchaseNum, .leastPatchNumber)
    buyNumber: int | None  # IDK when recommendNumber and buyNumber are different
    minPurchaseNum: int | None  # MOQ
    canPresaleNumber: int | None  # ???
    preMinPurchaseNum: int | None  # ???
    shopMinPurchaseNum: _Incomplete | None  # ???
    shopMaxPurchaseNum: _Incomplete | None  # ???
    overseasStockCount: int | None  # this is what shows as "stock" in the webui
    prices: list[BOMFileMatchComponentPrice] | None
    componentImageUrl: _Incomplete | None
    productBigImageAccessId: _Incomplete | None
    unitPrice: float | None  # unit-price at the price range for ?.buyNumber?
    originType: int | None  # I've only ever seen `0`
    totalMoney: float | None  # extended price for ?.buyNumber?
    lossNumber: int | None  # minimum margin for attrition
    leastNumber: int | None  # I've only ever seen `0`
    leastPatchNumber: int | None  # may be a fixed fee if under this number
    encapsulationNumber: int | None  # ???
    checked: bool | None
    theRatio: _Incomplete | None  # ???
    secondTypeNameEn: str | None  # JLCPCB human-readable category name
    matchWay: int | None  # I've only ever seen `1`
    privateStockCount: _Incomplete | None
    overseasShopStockCount: _Incomplete | None
    postStockCount: _Incomplete | None
    idleStockCount: _Incomplete | None
    deliveryTimeWayDays: _Incomplete | None
    isBuyComponent: str | None  # I've only ever seen `"1"`
    noBuyReason: _Incomplete | None
    componentAlternativesCode: (
        str | None
    )  # ??? sometimes null, sometimes empty-string, sometimes an LCSC "C____" number
    fullReelPrice: _Incomplete | None
    specificationCheckMsg: _Incomplete | None
    manualFlag: _Incomplete | None
    dealStatus: _Incomplete | None
    unhandledReason: _Incomplete | None
    matchSource: BOMFileMatchSource | None
    multipleRowTipMessage: (
        _Incomplete | None
    )  # eg: "The comment (0603WAF1502T5E) of this part does not match the one (15KΩ) provided in your BOM. Please confirm"
    catalogBlackFlag: _Incomplete | None
    multipleBomRowTipMessage: _Incomplete | None
    multipleDesTipMessage: _Incomplete | None
    moistureSensitivityLevelEn: str | None  # null, empty-string, "MSL 1", "MSL 3", ...
    needAuditFlag: bool | None
    specialComponentFee: (
        float | None
    )  # in USD, "The processing of this component is difficult. A special component fee of ${.specialComponentFee} per piece will be charged."
    orderInstructionEnglish: str | None
    bomMatchChangeLog: _Incomplete | None


class BOMFileMatch(pydantic.BaseModel):
    # "JLCPCB Parts" (in-stock)
    #
    # Parts that JLC has in stock.
    jlcStock: list[BOMFileMatchComponent]

    # "JLCPCB Parts" (not in-stock)
    #
    # Parts that JLC will buy: "Pre-order Items: Price for pre-order
    # items are for reference only. The confirmed price will be quoted
    # within 48 hours after finish the payment."
    jlcBuy: list[BOMFileMatchComponent]

    # "Global Sourcing Parts"
    overseasShop: list[BOMFileMatchComponent]

    # "Not Placed Parts"
    notSelect: list[BOMFileMatchComponent]

    # "Not Matched Parts"
    notMatch: list[BOMFileMatchComponent]


# Base API machinery ###########################################################

_T = typing.TypeVar("_T")


class _ReqBody(typing.NamedTuple):
    mimetype: str
    content: bytes


def _json_req(obj: typing.Any) -> _ReqBody:
    return _ReqBody("application/json", json.dumps(obj).encode("utf-8"))


class _Resp(typing.NamedTuple):
    req: urllib.request.Request

    head: email.message.Message
    body: bytes


class _ResponseContainerNoData(pydantic.BaseModel):
    success: bool
    code: int
    message: str | None


class _ResponseContainer(pydantic.BaseModel, typing.Generic[_T]):
    success: bool
    code: int
    message: str | None
    data: _T


def _json_resp(ret_typ: type[_T], resp: _Resp) -> _T:
    body_nodata = _ResponseContainerNoData.model_validate_json(resp.body)
    if body_nodata.code != 200 or not body_nodata.success:
        raise urllib.error.HTTPError(
            resp.req.full_url,
            body_nodata.code,
            body_nodata.message or "",
            resp.head,
            io.BytesIO(resp.body),
        )

    container_typ: type[_ResponseContainer[_T]]
    if typing.TYPE_CHECKING:
        container_typ = _ResponseContainer[_T]
    else:
        container_typ = _ResponseContainer[ret_typ]
    body = container_typ.model_validate_json(resp.body)
    return body.data


class _PCBResponseContainerNoData(pydantic.BaseModel):
    code: int
    message: str | None


class _PCBResponseContainer(pydantic.BaseModel, typing.Generic[_T]):
    code: int
    message: str | None
    data: _T


def _pcb_json_resp(ret_typ: type[_T], resp: _Resp) -> _T:
    body_nodata = _PCBResponseContainerNoData.model_validate_json(resp.body)
    if body_nodata.code != 200:
        raise urllib.error.HTTPError(
            resp.req.full_url,
            body_nodata.code,
            body_nodata.message or "",
            resp.head,
            io.BytesIO(resp.body),
        )

    container_typ: type[_PCBResponseContainer[_T]]
    if typing.TYPE_CHECKING:
        container_typ = _PCBResponseContainer[_T]
    else:
        container_typ = _PCBResponseContainer[ret_typ]
    body = container_typ.model_validate_json(resp.body)
    return body.data


class _BaseClient:
    _parent: "_BaseClient | None"
    _pathprefix: str

    @property
    def _root(self) -> "_BaseClient":
        if not self._parent:
            return self
        return self._parent._root  # pylint: disable=protected-access

    @property
    def _fullpathprefix(self) -> str:
        if not self._parent:
            return self._pathprefix
        # pylint: disable=protected-access
        return self._parent._fullpathprefix + self._pathprefix

    @property
    def _session_id(self) -> str:
        return self._root._session_id  # pylint: disable=protected-access

    @property
    def _ua_string(self) -> str:
        return self._root._ua_string  # pylint: disable=protected-access

    def __init__(self, parent: "_BaseClient") -> None:
        self._parent = parent

    def _do_req(self, req: urllib.request.Request) -> _Resp:
        with urllib.request.urlopen(req) as resp:
            return _Resp(
                req=req,
                head=resp.msg,
                body=resp.read(),
            )

    def _do_post(self, pathsuffix: str, body: _ReqBody) -> _Resp:
        return self._do_req(
            urllib.request.Request(
                method="POST",
                url=self._fullpathprefix + pathsuffix,
                headers={
                    "User-Agent": self._ua_string,
                    "Content-Type": body.mimetype,
                    "Accept": "application/json",
                    "Cookie": f"JLCPCB_SESSION_ID={self._session_id}",
                },
                data=body.content,
            ),
        )

    def _do_get(self, pathsuffix: str, params: dict[str, str]) -> _Resp:
        return self._do_req(
            urllib.request.Request(
                method="GET",
                url=self._fullpathprefix
                + pathsuffix
                + "?"
                + urllib.parse.urlencode(
                    {
                        **params,
                        "_t": time.time_ns() // 1000000,
                    }
                ),
                headers={
                    "User-Agent": self._ua_string,
                    "Accept": "application/json",
                    "Cookie": f"JLCPCB_SESSION_ID={self._session_id}",
                },
            ),
        )


# API endpoints ################################################################


class Client(_BaseClient):
    __session_id: str
    __ua_string: str

    @property
    def _session_id(self) -> str:
        return self.__session_id

    @property
    def _ua_string(self) -> str:
        return self.__ua_string

    def __init__(
        self,
        *,
        session_id: str,
        baseurl: str = "https://jlcpcb.com/api",
        ua_string: str = "jlcpcb_api.py (https://git.lukeshu.com/sbc-harness-hardware/tree/build-aux/jlcpcb_api.py)",
    ) -> None:
        # pylint: disable=super-init-not-called
        self._parent = None
        self._pathprefix = baseurl
        self.__session_id = session_id
        self.__ua_string = ua_string

        self.core_platform = self.__class__.CorePlatform(self)
        self.smt_component_order_platform = self.__class__.SMTComponentOrderPlatform(
            self
        )

    class CorePlatform(_BaseClient):
        _pathprefix = "/overseas-core-platform"

        def __init__(self, parent: _BaseClient) -> None:
            super().__init__(parent)

            self.bom_match_record_controller = self.__class__.BOMMatchRecordController(
                self
            )
            self.shopping_cart = self.__class__.ShoppingCart(self)

        class BOMMatchRecordController(_BaseClient):
            _pathprefix = "/bomMatchRecordController"

            def check_whitelist(
                self,
                topic: typing.Literal[
                    "bomShareOpenList", "bomMatchOpenList", "smt_manual_white_list"
                ],
            ) -> bool:
                return _json_resp(bool, self._do_get("/checkWhiteList/" + topic, {}))

            def upload_bomfile(self, name: str, content: bytes) -> str:
                """Return the newly created bomfile ID."""
                assert "/" not in name
                _, ext = os.path.splitext(name)
                typ = {
                    ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
                    ".xls": "application/vnd.ms-excel",
                    ".csv": "text/csv",
                }[ext.lower()]
                boundary = "----geckoformboundary22e1a6f299b864383bdfbddbc67c22ab"
                return _json_resp(
                    str,
                    self._do_post(
                        "/uploadBomFile",
                        _ReqBody(
                            f"multipart/form-data; boundary={boundary}",
                            f"--{boundary}\r\n"
                            f'Content-Disposition: form-data; name="file"; filename="{name}"\r\n'
                            f"Content-Type: {typ}\r\n"
                            f"\r\n".encode("utf-8")
                            + content
                            + b"\r\n"
                            + f"--{boundary}--\r\n".encode("utf-8"),
                        ),
                    ),
                )

            def bomfile_get_parse_status(self, bomfile_id: str) -> int:
                return _json_resp(
                    int,
                    self._do_get(
                        "/getParseStatus",
                        {"bomMatchFileRecordAccessId": bomfile_id},
                    ),
                )

            def bomfile_get_permission(self, bomfile_id: str) -> BOMFilePermission:
                return _json_resp(
                    BOMFilePermission,
                    self._do_post(
                        "/checkPermission",
                        _json_req({"bomFileAccessId": bomfile_id}),
                    ),
                )

            def bomfile_update_details(
                self,
                bomfile_id: str,
                edits: BOMFileDetailEdits,
                *,
                replace_id: bool = False,
                update_strategy: int = 3,
            ) -> None:
                assert len(edits) > 0
                return _json_resp(
                    None.__class__,
                    self._do_post(
                        "/updateBomMatchDetail",
                        _json_req(
                            {
                                **edits,
                                "fileRecordAccessId": bomfile_id,
                                "noReplaceId": not replace_id,
                                "updateStrategy": update_strategy,
                            }
                        ),
                    ),
                )

            def bomfile_get_details(self, bomfile_id: str) -> BOMFileDetail:
                return _json_resp(
                    BOMFileDetail,
                    self._do_get(
                        "/getBomFileDetail",
                        {"bomAccessId": bomfile_id},
                    ),
                )

            def bomfile_update(
                self, bomfile_id: str, bomfile_name: str, edits: BOMFileEdits
            ) -> None:
                assert len(edits) > 0
                _json_resp(
                    None.__class__,
                    self._do_post(
                        "/updateBomMatchFile",
                        _json_req(
                            {
                                "bomFileName": bomfile_name,
                                "bomMatchFileRecordAccessId": bomfile_id,
                                **edits,
                            }
                        ),
                    ),
                )

            def bomfile_get_match(
                self, bomfile_id: str, count: int, show_manual_only_flag: bool = False
            ) -> BOMFileMatch:
                return _json_resp(
                    BOMFileMatch,
                    self._do_post(
                        "/getMatchResult",
                        _json_req(
                            {
                                "bomUuid": bomfile_id,
                                "patchNumber": count,
                                "showManualOnlyFlag": show_manual_only_flag,
                            }
                        ),
                    ),
                )

            def bomfile_export_match(
                self, bomfile_id: str, count: int, show_manual_only_flag: bool = False
            ) -> bytes:
                """Like .bomfile_get_match(), but the result is returned as an XLS (not XLSX!) spreadsheet."""
                return self._do_post(
                    "/exportBomMatchResult",
                    _json_req(
                        {
                            "bomUuid": bomfile_id,
                            "patchNumber": count,
                            "showManualOnlyFlag": show_manual_only_flag,
                        }
                    ),
                ).body

        class ShoppingCart(_BaseClient):
            _pathprefix = "/shoppingCart"

            def __init__(self, parent: _BaseClient) -> None:
                super().__init__(parent)

                self.smt_good = self.__class__.SMTGood(self)

            class SMTGood(_BaseClient):
                _pathprefix = "/smtGood"

                def get_component_detail(self, lcsc_cnumber: str) -> _Incomplete:
                    return _json_resp(
                        typing.Any,
                        self._do_get(
                            "/getComponentDetail", {"componentCode": lcsc_cnumber}
                        ),
                    )

    class PCBOrder(_BaseClient):
        _pathprefix = "/overseas-pcb-order/v1"

        def __init__(self, parent: _BaseClient) -> None:
            super().__init__(parent)

            self.order = self.__class__.ShoppingCart(self)

        class ShoppingCart(_BaseClient):
            _pathprefix = "/shoppingCart"

            def __init__(self, parent: _BaseClient) -> None:
                super().__init__(parent)

                self.smt_good = self.__class__.SMTGood(self)

            class SMTGood(_BaseClient):
                _pathprefix = "/smtGood"

                def get_component_detail(self, jlcpcb_lcsc_id: int) -> _Incomplete:
                    """Like core_platform get_component_detail(), but
                    uses JLCPCB's internal "LcscId" instead of the
                    LCSC "C_____" number.  Return type is the same,
                    except that in core_platform
                    .overseasComponentCatalogBizKey is a string of an
                    integer, while here it is just an integer.

                    """
                    return _pcb_json_resp(
                        typing.Any,
                        self._do_get(
                            "/getComponentDetail",
                            {"componentCode": str(jlcpcb_lcsc_id)},
                        ),
                    )

    class SMTComponentOrderPlatform(_BaseClient):
        _pathprefix = "/overseas-smt-component-order-platform/v1"

        def __init__(self, parent: _BaseClient) -> None:
            super().__init__(parent)

            self.order = self.__class__.Order(self)

        class Order(_BaseClient):
            _pathprefix = "/overseasSmtComponentOrder"

            def __init__(self, parent: _BaseClient) -> None:
                super().__init__(parent)

                self.shop_cart = self.__class__.ShopCart(self)

            class ShopCart(_BaseClient):
                _pathprefix = "/smtComponentShopCart"

                def get_search_type(self) -> str:
                    """IDK what the string means."""
                    return _json_resp(
                        str, self._do_get("/getOverseasShopSearchType", {})
                    )