孔夫子旧书网 API 实战：古籍与二手书数据获取及接口调用方案

​

 孔夫子旧书网作为国内知名的古籍、二手书交易平台，其商品数据对于图书收藏、学术研究及二手书电商系统具有重要价值。本文将详细介绍孔夫子平台接口的调用方法，涵盖认证机制、搜索参数配置、数据解析及反爬策略，并提供可直接使用的 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 技术提取关键评价
   出版年代可能存在模糊表述（如 "民国年间"），需特殊处理

六、功能扩展方向

   开发古籍版本比对工具，基于多本同书数据进行版本差异分析
   构建卖家信誉评估系统，结合历史交易和评价数据
   实现图书价格趋势分析，追踪古籍市场价格波动
   开发古籍修复需求识别功能，基于品相描述自动判断修复需求

​
审核编辑 黄宇