深度分析电子元件API接口，用Python脚本实现

import requestsimport jsonimport loggingfrom typing import Dict, Optional, Listfrom requests.exceptions import RequestException# 配置日志（记录请求状态和错误）logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s")class LCSCAPI:
    def __init__(self, api_key: str):
        """
        初始化立创商城API客户端
        :param api_key: 立创开放平台获取的api_key
        """
        self.api_key = api_key
        self.base_url = "https://open.lcsc.com/api/v1"
        self.session = requests.Session()
        # 设置默认请求头（携带api_key）
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json;charset=UTF-8"
        })

    def _handle_response(self, response: requests.Response) -> Optional[Dict]:
        """
        统一处理接口响应（错误判断、JSON解析）
        :param response: 接口响应对象
        :return: 业务数据（无错误时）或None
        """
        try:
            response.raise_for_status()  # 抛出HTTP错误（如404、500）
            result = response.json()

            # 立创API错误码：code=200为成功，其他为错误
            if result.get("code") != 200:
                error_msg = result.get("msg", "未知错误")
                logging.error(f"API错误：{error_msg}（错误码：{result.get('code')}）")
                return None

            return result.get("data", {})  # 返回业务数据

        except RequestException as e:
            logging.error(f"请求异常：{str(e)}")
            return None
        except json.JSONDecodeError:
            logging.error(f"响应解析失败：{response.text[:200]}...")
            return None

    def search_part(self, keyword: str, page: int = 1, page_size: int = 20) -> Optional[Dict]:
        """
        搜索电子元件（按型号/关键词）
        :param keyword: 搜索关键词（如"STM32F103C8T6"、"0805电容"）
        :param page: 页码（默认第1页）
        :param page_size: 每页数量（最大50）
        :return: 搜索结果（含总条数、元件列表）
        """
        url = f"{self.base_url}/part/search"
        params = {
            "keyword": keyword,
            "page": page,
            "pageSize": page_size        }

        try:
            response = self.session.get(url, params=params, timeout=10)
            data = self._handle_response(response)
            if not data:
                return None

            # 格式化元件列表（提取核心字段）
            formatted_parts = []
            for part in data.get("parts", []):
                formatted_parts.append({
                    "part_number": part.get("partNumber"),  # 元件型号
                    "manufacturer": part.get("manufacturer"),  # 原厂
                    "category": part.get("categoryName"),  # 品类（如"MCU"）
                    "stock_quantity": part.get("stockQuantity"),  # 库存数量
                    "min_price": float(part.get("minPrice", 0)),  # 最低单价
                    "moq": part.get("moq"),  # 最小起订量
                    "datasheet_url": part.get("datasheetUrl"),  # datasheet链接
                    "lcsc_part_number": part.get("lcscPartNumber")  # 立创内部型号
                })

            return {
                "total_count": data.get("totalCount", 0),  # 总搜索结果数
                "current_page": page,
                "page_size": page_size,
                "parts": formatted_parts            }

        except Exception as e:
            logging.error(f"元件搜索失败：{str(e)}")
            return None

    def get_part_stock(self, part_number: str) -> Optional[Dict]:
        """
        查询指定型号的实时库存
        :param part_number: 元件型号（如"STM32F103C8T6"）
        :return: 库存信息（数量、位置、交货周期）
        """
        url = f"{self.base_url}/part/stock"
        params = {"partNumber": part_number}

        try:
            response = self.session.get(url, params=params, timeout=10)
            data = self._handle_response(response)
            if not data:
                return None

            # 格式化库存数据
            return {
                "part_number": part_number,
                "stock_quantity": data.get("stockQuantity"),  # 总库存
                "stock_detail": [  # 分仓库库存详情
                    {
                        "location": warehouse.get("warehouseName"),  # 仓库位置（如"深圳仓"）
                        "quantity": warehouse.get("quantity"),  # 该仓库库存
                        "lead_time": warehouse.get("leadTime")  # 交货周期
                    } for warehouse in data.get("stockDetails", [])
                ],
                "stock_status": "有货" if data.get("stockQuantity", 0) > 0 else "无货"
            }

        except Exception as e:
            logging.error(f"库存查询失败：{str(e)}")
            return None

    def get_part_price(self, part_number: str) -> Optional[Dict]:
        """
        查询指定型号的阶梯价格
        :param part_number: 元件型号
        :return: 价格信息（最小起订量、阶梯价）
        """
        url = f"{self.base_url}/part/price"
        params = {"partNumber": part_number}

        try:
            response = self.session.get(url, params=params, timeout=10)
            data = self._handle_response(response)
            if not data:
                return None

            # 格式化阶梯价数据
            price_breaks = []
            for pb in data.get("priceBreaks", []):
                price_breaks.append({
                    "min_quantity": pb.get("minQuantity"),  # 该阶梯最小数量
                    "max_quantity": pb.get("maxQuantity"),  # 该阶梯最大数量（None表示无上限）
                    "unit_price": float(pb.get("unitPrice", 0))  # 该阶梯单价
                })

            return {
                "part_number": part_number,
                "moq": data.get("moq"),  # 最小起订量
                "currency": data.get("currency", "CNY"),  # 货币类型（默认人民币）
                "price_breaks": price_breaks            }

        except Exception as e:
            logging.error(f"价格查询失败：{str(e)}")
            return None# 示例调用if __name__ == "__main__":
    # 1. 替换为你的立创API Key（从开放平台获取）
    LCSC_API_KEY = "your_lcsc_api_key"

    # 2. 初始化API客户端
    lcsc_api = LCSCAPI(api_key=LCSC_API_KEY)

    # 3. 搜索元件（以"STM32F103C8T6"为例）
    search_keyword = "STM32F103C8T6"
    search_result = lcsc_api.search_part(keyword=search_keyword, page=1, page_size=5)
    if search_result:
        print(f"=== 元件搜索结果（关键词：{search_keyword}）===")
        print(f"总找到 {search_result['total_count']} 个元件，当前第 {search_result['current_page']} 页\n")
        for i, part in enumerate(search_result["parts"], 1):
            print(f"{i}. 型号：{part['part_number']}")
            print(f"   原厂：{part['manufacturer']} | 品类：{part['category']}")
            print(f"   库存：{part['stock_quantity']} 件 | 最小单价：{part['min_price']} 元")
            print(f"   起订量：{part['moq']} 件 | datasheet：{part['datasheet_url'][:50]}...")
            print("-" * 100)

    # 4. 查询第一个元件的库存和价格
    if search_result and search_result["parts"]:
        target_part = search_result["parts"][0]["part_number"]
        print(f"\n=== 元件 {target_part} 详情 ===")

        # 4.1 查询库存
        stock_info = lcsc_api.get_part_stock(part_number=target_part)
        if stock_info:
            print(f"库存状态：{stock_info['stock_status']}（总库存：{stock_info['stock_quantity']} 件）")
            print("分仓库库存：")
            for warehouse in stock_info["stock_detail"]:
                print(f"   - {warehouse['location']}：{warehouse['quantity']} 件，交货周期：{warehouse['lead_time']}")

        # 4.2 查询价格
        price_info = lcsc_api.get_part_price(part_number=target_part)
        if price_info:
            print(f"\n价格信息（货币：{price_info['currency']}）：")
            print(f"最小起订量：{price_info['moq']} 件")
            print("阶梯价：")
            for pb in price_info["price_breaks"]:
                max_qty = pb["max_quantity"] if pb["max_quantity"] else "不限"
                print(f"   - {pb['min_quantity']}-{max_qty} 件：{pb['unit_price']} 元/件")