孔夫子旧书网作为国内知名的古籍、二手书交易平台,其商品数据对于图书收藏、学术研究及二手书电商系统具有重要价值。本文将详细介绍孔夫子平台接口的调用方法,涵盖认证机制、搜索参数配置、数据解析及反爬策略,并提供可直接使用的 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_keyself.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_proxyself.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 sessiondef _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 Nonedef 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"] = categoryif year_min is not None:params["year_min"] = year_minif year_max is not None:params["year_max"] = year_maxif condition is not None:params["condition"] = conditionif price_min is not None:params["price_min"] = price_minif price_max is not None:params["price_max"] = price_maxif 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 = Noneif 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 filtersdef 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 = 1total_pages = 1while 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 KeyAPI_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 技术提取关键评价
- 出版年代可能存在模糊表述(如 "民国年间"),需特殊处理
六、功能扩展方向
- 开发古籍版本比对工具,基于多本同书数据进行版本差异分析
- 构建卖家信誉评估系统,结合历史交易和评价数据
- 实现图书价格趋势分析,追踪古籍市场价格波动
- 开发古籍修复需求识别功能,基于品相描述自动判断修复需求