孔夫子旧书网 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 技术提取关键评价 出版年代可能存在模糊表述（如 "民国年间"），需特殊处理

六、功能扩展方向

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

​