孔夫子旧书网作为国内知名的古籍、二手书交易平台，其商品数据对于图书收藏、学术研究及二手书电商系统具有重要价值。本文将详细介绍孔夫子平台接口的调用方法，涵盖认证机制、搜索参数配置、数据解析及反爬策略，并提供可直接使用的 Python 代码实现，帮助开发者合规获取古籍和二手书数据。

一、孔夫子平台接口基础信息

孔夫子旧书网提供的开放接口主要包括图书搜索、商品详情、店铺信息等功能，其中/api/v1/books/search是获取图书列表的核心接口，特别适用于古籍、珍本、二手书的检索。

接口特点：

  采用 API Key 认证机制，部分接口需要商业合作授权

  支持按书名、作者、出版社、年代、品相等级等多维度筛选

  包含古籍特有的版本信息、刻印年代、装帧形式等字段

  提供卖家信誉、交易记录等二手书交易关键数据

接口端点：https://api.kongfz.com/api/v1/books/search

二、认证机制与核心参数

1. 认证方式

孔夫子接口采用简单直接的 API Key 认证：

  在孔夫子开发者平台注册并申请应用，获取 API Key

  在所有请求的 Header 中携带X-API-Key参数

  商业用户可申请更高权限的 Secret Key 进行签名认证

2. 核心搜索参数

  keyword：搜索关键字（书名、作者、ISBN 等，必填）

  category：图书分类（古籍 / 二手书 / 期刊等，可选）

  year_min/year_max：出版年代范围（可选）

  condition：品相等级（1-10 级，10 为全新，可选）

  price_min/price_max：价格区间（可选）

  publisher：出版社（可选）

  sort：排序方式（price_asc/price_desc/time_desc/credit_desc）

  page：页码（默认 1）

  limit：每页条数（1-20，默认 10）

  rare：是否仅显示珍本（true/false，可选）

3. 响应数据结构

  total：总结果数

  page/limit：分页信息

  books：图书列表数组

  filters：可用筛选条件

三、完整代码实现

以下是 Python 实现的孔夫子旧书网图书搜索功能，包含 API 调用、数据解析和反爬策略：

  import requests

  import time

  import random

  from typing import Dict, List, Optional, Any

  from user_agent import generate_user_agent

  import logging

   

  # 配置日志

  logging.basicConfig(

      level=logging.INFO,

      format='%(asctime)s - %(levelname)s - %(message)s'

  )

  logger = logging.getLogger('kongfz_api')

   

  class KongfzBookAPI:

      def __init__(self, api_key: str, use_proxy: bool = False, proxy_pool: List[str] = None):

          """

          初始化孔夫子旧书网API客户端

         

          :param api_key: 平台申请的API Key

          :param use_proxy: 是否使用代理

          :param proxy_pool: 代理IP池列表

          """

          self.api_key = api_key

          self.base_url = "https://api.kongfz.com"

          self.search_endpoint = "/api/v1/books/search"

          self.detail_endpoint = "/api/v1/books/detail"

          self.max_limit = 20  # 最大每页条数

          self.use_proxy = use_proxy

          self.proxy_pool = proxy_pool or []

          self.session = self._init_session()

         

      def _init_session(self) -> requests.Session:

          """初始化请求会话，配置持久连接"""

          session = requests.Session()

          session.headers.update({

              "Accept": "application/json",

              "Content-Type": "application/json",

              "X-API-Key": self.api_key,

              "Connection": "keep-alive"

          })

          return session

     

      def _get_random_headers(self) -> Dict[str, str]:

          """生成随机请求头，降低反爬风险"""

          return {

              "User-Agent": generate_user_agent(),

              "Accept-Language": random.choice(["zh-CN,zh;q=0.9", "zh-TW,zh;q=0.9,en;q=0.8"]),

              "Referer": "https://www.kongfz.com/"

          }

     

      def _get_proxy(self) -> Optional[Dict[str, str]]:

          """从代理池获取随机代理"""

          if self.use_proxy and self.proxy_pool:

              proxy = random.choice(self.proxy_pool)

              return {"http": proxy, "https": proxy}

          return None

     

      def search_books(self,

                      keyword: str,

                      category: Optional[str] = None,

                      year_min: Optional[int] = None,

                      year_max: Optional[int] = None,

                      condition: Optional[int] = None,

                      price_min: Optional[float] = None,

                      price_max: Optional[float] = None,

                      publisher: Optional[str] = None,

                      sort: str = "time_desc",

                      page: int = 1,

                      limit: int = 10) -> Dict[str, Any]:

          """

          搜索孔夫子旧书网图书

         

          :param keyword: 搜索关键字

          :param category: 图书分类

          :param year_min: 最小出版年份

          :param year_max: 最大出版年份

          :param condition: 品相等级(1-10)

          :param price_min: 最低价格

          :param price_max: 最高价格

          :param publisher: 出版社

          :param sort: 排序方式

          :param page: 页码

          :param limit: 每页条数

          :return: 搜索结果

          """

          # 限制每页最大条数

          limit = min(limit, self.max_limit)

         

          # 构建查询参数

          params: Dict[str, Any] = {

              "keyword": keyword,

              "sort": sort,

              "page": page,

              "limit": limit

          }

         

          # 添加可选参数

          if category:

              params["category"] = category

          if year_min is not None:

              params["year_min"] = year_min

          if year_max is not None:

              params["year_max"] = year_max

          if condition is not None:

              params["condition"] = condition

          if price_min is not None:

              params["price_min"] = price_min

          if price_max is not None:

              params["price_max"] = price_max

          if publisher:

              params["publisher"] = publisher

         

          # 准备请求配置

          headers = self._get_random_headers()

          proxy = self._get_proxy()

         

          try:

              # 随机延迟，模拟人类行为

              time.sleep(random.uniform(0.8, 1.5))

             

              # 发送请求

              response = self.session.get(

                  f"{self.base_url}{self.search_endpoint}",

                  params=params,

                  headers=headers,

                  proxies=proxy,

                  timeout=15

              )

              response.raise_for_status()

             

              # 解析响应

              result = response.json()

             

              # 处理API错误

              if result.get("code") != 0:

                  logger.error(f"API错误: {result.get('msg')}")

                  return {

                      "success": False,

                      "error_code": result.get("code"),

                      "error_msg": result.get("msg")

                  }

                 

              # 解析搜索结果

              return self._parse_search_result(result.get("data", {}))

             

          except requests.exceptions.RequestException as e:

              logger.error(f"请求异常: {str(e)}")

              return {

                  "success": False,

                  "error_msg": f"请求异常: {str(e)}"

              }

          except Exception as e:

              logger.error(f"处理响应失败: {str(e)}")

              return {

                  "success": False,

                  "error_msg": f"处理响应失败: {str(e)}"

              }

     

      def _parse_search_result(self, raw_data: Dict[str, Any]) -> Dict[str, Any]:

          """解析搜索结果为结构化数据"""

          # 分页信息

          pagination = {

              "total": raw_data.get("total", 0),

              "page": raw_data.get("page", 1),

              "limit": raw_data.get("limit", 10),

              "pages": (raw_data.get("total", 0) + raw_data.get("limit", 10) - 1) //

                       raw_data.get("limit", 10)

          }

         

          # 解析图书列表

          books = []

          for item in raw_data.get("books", []):

              # 处理古籍特有的版本信息

              ancient_info = None

              if item.get("is_ancient"):

                  ancient_info = {

                      "edition": item.get("ancient_edition"),  # 版本

                      "engraving_year": item.get("engraving_year"),  # 刻印年代

                      "binding": item.get("binding"),  # 装帧

                      "seal_info": item.get("seal_info")  # 钤印信息

                  }

             

              books.append({

                  "book_id": item.get("id"),

                  "title": item.get("title"),

                  "author": item.get("author"),

                  "publisher": item.get("publisher"),

                  "publish_year": item.get("publish_year"),

                  "category": item.get("category"),

                  "is_ancient": item.get("is_ancient", False),  # 是否古籍

                  "ancient_info": ancient_info,

                  "condition": {

                      "level": item.get("condition_level"),  # 品相等级

                      "description": item.get("condition_desc")  # 品相描述

                  },

                  "price": {

                      "current": item.get("price"),

                      "original": item.get("original_price"),

                      "currency": "CNY"

                  },

                  "seller": {

                      "id": item.get("seller_id"),

                      "name": item.get("seller_name"),

                      "credit": item.get("seller_credit"),  # 信誉等级

                      "score": item.get("seller_score")  # 好评率

                  },

                  "images": {

                      "main": item.get("main_image"),

                      "thumbnail": item.get("thumbnail")

                  },

                  "url": item.get("url"),

                  "tags": item.get("tags", [])

              })

         

          # 解析可用筛选条件

          filters = self._parse_filters(raw_data.get("filters", {}))

         

          return {

              "success": True,

              "pagination": pagination,

              "books": books,

              "filters": filters

          }

     

      def _parse_filters(self, raw_filters: Dict[str, Any]) -> Dict[str, Any]:

          """解析筛选条件"""

          filters = {}

         

          # 分类筛选

          if "categories" in raw_filters:

              filters["categories"] = [

                  {"id": item.get("id"), "name": item.get("name"), "count": item.get("count")}

                  for item in raw_filters["categories"]

              ]

         

          # 品相筛选

          if "conditions" in raw_filters:

              filters["conditions"] = [

                  {"level": item.get("level"), "name": item.get("name"), "count": item.get("count")}

                  for item in raw_filters["conditions"]

              ]

             

          # 年代筛选

          if "years" in raw_filters:

              filters["years"] = raw_filters["years"]

             

          return filters

     

      def batch_search(self,

                      keyword: str,

                      max_pages: int = 3,

                      **kwargs) -> Dict[str, Any]:

          """

          批量获取多页搜索结果

         

          :param keyword: 搜索关键字

          :param max_pages: 最大获取页数

          :param**kwargs: 其他搜索参数

          :return: 合并的搜索结果

          """

          all_books = []

          current_page = 1

          total_pages = 1

         

          while current_page <= max_pages and current_page <= total_pages:

              logger.info(f"搜索第 {current_page} 页，关键字: {keyword}")

             

              # 搜索当前页

              result = self.search_books(

                  keyword=keyword,

                  page=current_page,

                  **kwargs

              )

             

              if not result.get("success"):

                  return result

                 

              # 收集图书数据

              all_books.extend(result.get("books", []))

             

              # 更新分页信息

              pagination = result.get("pagination", {})

              total_pages = pagination.get("pages", 1)

             

              # 准备下一页

              current_page += 1

             

              # 增加页数间隔，降低反爬风险

              if current_page <= max_pages:

                  time.sleep(random.uniform(1.5, 2.5))

         

          return {

              "success": True,

              "total_books": len(all_books),

              "books": all_books,

              "summary": {

                  "total_available": pagination.get("total", 0),

                  "fetched_pages": current_page - 1

              }

          }

   

  # 使用示例

  if __name__ == "__main__":

      # 替换为你的API Key

      API_KEY = "your_api_key"

     

      # 代理配置（可选）

      PROXY_POOL = [

          # "http://ip1:port",

          # "http://ip2:port"

      ]

     

      # 初始化API客户端

      kongfz_api = KongfzBookAPI(

          api_key=API_KEY,

          use_proxy=False,  # 根据需要开启

          proxy_pool=PROXY_POOL

      )

     

      # 示例1：搜索古籍

      ancient_result = kongfz_api.search_books(

          keyword="论语",

          category="ancient",  # 古籍分类

          year_min=1949,

          year_max=2023,

          condition=8,  # 8级及以上品相

          sort="price_asc",

          page=1,

          limit=10

      )

     

      if ancient_result["success"]:

          print(f"古籍搜索: 找到 {ancient_result['pagination']['total']} 本相关图书")

          if ancient_result["books"]:

              book = ancient_result["books"][0]

              print(f"书名: {book['title']}")

              print(f"作者: {book['author']}")

              print(f"价格: {book['price']['current']}元")

              print(f"品相: {book['condition']['level']}级 - {book['condition']['description']}")

              if book["is_ancient"]:

                  print(f"版本: {book['ancient_info']['edition']}")

     

      # 示例2：批量搜索二手书

      # batch_result = kongfz_api.batch_search(

      #     keyword="鲁迅全集",

      #     category="secondhand",  # 二手书分类

      #     price_min=50,

      #     price_max=500,

      #     max_pages=2

      # )

      #

      # if batch_result["success"]:

      #     print(f"\n批量搜索: 共获取 {batch_result['total_books']} 本图书")

四、代码核心功能解析

1. 反爬策略实现

  随机生成 User-Agent 和请求头，模拟不同浏览器行为

  加入随机请求延迟，避免固定访问频率被识别

  支持代理 IP 池配置，分散请求来源

  使用持久化 Session，模拟正常用户浏览行为

2. 古籍数据特色处理

  专门解析古籍特有的版本、刻印年代、装帧等信息

  区分古籍与普通二手书的数据结构

  提取钤印信息等古籍收藏关键维度

3. 搜索功能设计

  支持完整的图书筛选参数，满足古籍和二手书的搜索需求

  提供单页搜索和多页批量搜索两种模式

  批量搜索时动态调整间隔时间，平衡效率与安全性

4. 数据结构化

  按图书类型组织数据，区分普通二手书和古籍

  提取卖家信誉、品相描述等二手交易关键信息

  解析可用筛选条件，便于前端实现高级筛选功能

五、实战注意事项

1. 接口权限与申请

  孔夫子 API 分为免费版和商业版，免费版有调用频率限制（通常 QPS≤2）

  古籍珍本等敏感数据需要申请商业授权

  个人开发者需提供身份证明，企业开发者需提供营业执照

2. 反爬与合规

  免费版接口请勿进行高频次调用，建议单 IP 日调用不超过 1000 次

  数据使用需遵守孔夫子平台的版权协议，不得用于商业竞品

  尊重古籍数据的知识产权，引用时需注明来源

3. 搜索策略优化

  古籍搜索建议结合年代和版本筛选，提高精准度

  批量获取数据时，合理设置max_pages参数，避免触发限制

  对稀缺古籍建立缓存机制，缓存周期建议 7-30 天

4. 数据处理建议

  书名和作者可能存在异体字、通假字，需进行文字规范化处理

  品相描述为文本信息，可通过 NLP 技术提取关键评价

  出版年代可能存在模糊表述（如 "民国年间"），需特殊处理

六、功能扩展方向

  开发古籍版本比对工具，基于多本同书数据进行版本差异分析

  构建卖家信誉评估系统，结合历史交易和评价数据

  实现图书价格趋势分析，追踪古籍市场价格波动

  开发古籍修复需求识别功能，基于品相描述自动判断修复需求