当前位置：首页 > news >正文

孔夫子旧书网开放平台接口实战：古籍图书检索与商铺数据集成方案

news 2025/9/18 8:11:09

孔夫子旧书网作为国内最大的旧书交易平台，其开放接口为古籍爱好者、图书经销商和学术研究机构提供了便捷的图书数据获取渠道。与其他电商平台相比，孔夫子平台接口专注于旧书、古籍、善本等特色商品，具有独特的元数据结构和检索方式。本文将系统讲解孔夫子开放平台核心接口的技术实现，重点解决接口认证、古籍信息解析、商铺数据集成和批量检索优化等关键问题，提供一套可直接应用的完整解决方案。

一、接口基础信息与应用场景

接口核心信息

孔夫子开放平台主要接口的关键技术参数：

接口域名：https://open.kongfz.com/api
认证方式：API Key + 数字签名
请求格式：HTTP POST，JSON 格式参数
响应格式：JSON
编码格式：UTF-8
调用限制：单应用 QPS=3，日调用上限 3000 次

核心接口列表

接口名称	接口地址	功能描述
图书检索	`/v2/books/search`	按关键词、作者、年代等检索图书
图书详情	`/v2/books/detail`	获取单本图书的详细信息
商铺检索	`/v2/shops/search`	检索符合条件的商铺
商铺详情	`/v2/shops/detail`	获取商铺的详细信息和在售商品
分类列表	`/v2/categories`	获取图书分类体系

典型应用场景

古籍数字化项目：批量获取古籍元数据用于数字化存档
学术研究工具：分析特定时期或作者的著作流传情况
旧书商管理系统：多平台商品同步与价格监控
私人藏书管理：个人藏书与孔网数据比对分析
图书交易平台：集成孔网商品丰富自身平台内容

接口调用流程

plaintext

API Key申请 → 签名生成 → 请求参数构造 → 接口调用 → 响应解析 → 数据存储 → 缓存更新

点击获取key和secret

二、接口认证与参数详解

认证方式说明

孔夫子开放平台采用 API Key + 签名的认证方式：

开发者在平台注册并获取 API Key
每次请求时，使用 API Key 和请求参数生成签名
服务器验证签名有效性后处理请求

签名生成算法

将所有请求参数（不包括 signature）按参数名 ASCII 码升序排序
拼接为 "key=value&key=value" 格式的字符串
在字符串末尾拼接 API Secret
对拼接后的字符串进行 MD5 加密，得到 32 位小写字符串作为签名

公共请求参数

参数名	类型	说明
appKey	String	应用唯一标识
timestamp	Long	时间戳（毫秒）
signature	String	签名结果
format	String	响应格式，固定为 json

图书检索核心参数

参数名	类型	说明	是否必须
keyword	String	搜索关键词	否
author	String	作者	否
publisher	String	出版社	否
categoryId	Integer	分类 ID	否
era	String	年代，如 "清代"、"民国"	否
minPrice	Float	最低价格	否
maxPrice	Float	最高价格	否
bookCondition	Integer	品相：1 - 全新 2 - 九五品 3 - 九品 ... 8 - 八五品以下	否
page	Integer	页码，默认 1	否
pageSize	Integer	每页条数，1-20	否
sort	String	排序方式：price_asc, price_desc, publish_time_desc	否

响应结果结构

图书检索接口的核心响应字段：

json

{"code": 200,"message": "success","data": {"total": 1250,"page": 1,"pageSize": 20,"totalPage": 63,"items": [{"id": "123456","title": "论语集注","author": "朱熹 注","publisher": "中华书局","publishTime": "1983-05","era": "现代","edition": "第1版","binding": "平装","pages": 320,"price": 58.00,"originalPrice": 68.00,"bookCondition": 2,"bookConditionDesc": "九五品","shopId": "s7890","shopName": "古籍书店","coverImg": "https://img.kongfz.com/books/123456.jpg","description": "本书为论语集注的重印本...","tags": ["儒家", "经典", "哲学"]}]}
}

三、核心技术实现

1. 孔夫子接口认证工具类

python

运行

import hashlib
import time
import json
from urllib.parse import urlencodeclass KongfzAuthUtil:"""孔夫子开放平台认证工具类"""@staticmethoddef generate_sign(params, app_secret):"""生成签名:param params: 参数字典:param app_secret: 应用密钥:return: 签名字符串"""try:# 1. 过滤空值参数和signature字段valid_params = {k: v for k, v in params.items() if v is not None and v != "" and k != "signature"}# 2. 按参数名ASCII升序排序sorted_params = sorted(valid_params.items(), key=lambda x: x[0])# 3. 拼接为key=value&key=value格式param_str = urlencode(sorted_params)# 4. 拼接appSecret并进行MD5加密sign_str = f"{param_str}{app_secret}"sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest().lower()return signexcept Exception as e:print(f"生成签名失败: {str(e)}")return None@staticmethoddef get_timestamp():"""获取当前时间戳（毫秒）"""return int(time.time() * 1000)

2. 图书检索接口客户端

python

运行

import requests
import time
from threading import Lock
from datetime import datetimeclass KongfzBookClient:"""孔夫子图书接口客户端"""def __init__(self, app_key, app_secret):self.app_key = app_keyself.app_secret = app_secretself.base_url = "https://open.kongfz.com/api"self.timeout = 15  # 超时时间(秒)self.qps_limit = 3  # QPS限制self.last_request_time = 0self.request_lock = Lock()  # 线程锁控制QPSdef _get_common_params(self):"""生成公共请求参数"""return {"appKey": self.app_key,"timestamp": KongfzAuthUtil.get_timestamp(),"format": "json"}def _check_qps(self):"""检查并控制QPS"""with self.request_lock:current_time = time.time()interval = 1.0 / self.qps_limit  # 每次请求最小间隔elapsed = current_time - self.last_request_timeif elapsed < interval:# 需要等待的时间time.sleep(interval - elapsed)self.last_request_time = time.time()def search_books(self,** kwargs):"""搜索图书:param kwargs: 搜索参数:return: 搜索结果字典"""# 检查QPS限制self._check_qps()# 1. 构造请求URLurl = f"{self.base_url}/v2/books/search"# 2. 构建请求参数params = self._get_common_params()# 添加业务参数valid_params = ["keyword", "author", "publisher", "categoryId", "era","minPrice", "maxPrice", "bookCondition", "page", "pageSize", "sort"]for param in valid_params:if param in kwargs and kwargs[param] is not None:params[param] = kwargs[param]# 3. 生成签名params["signature"] = KongfzAuthUtil.generate_sign(params, self.app_secret)# 4. 发送请求try:response = requests.post(url,json=params,headers={"Content-Type": "application/json;charset=utf-8","User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36"},timeout=self.timeout)# 5. 解析响应result = response.json()# 6. 处理响应结果if result.get("code") == 200:return self._parse_search_result(result.get("data", {}))else:raise Exception(f"搜索图书失败: {result.get('message', '未知错误')} "f"(错误码: {result.get('code', '未知')})")except Exception as e:print(f"图书搜索接口调用异常: {str(e)}")return Nonedef get_book_detail(self, book_id):"""获取图书详情:param book_id: 图书ID:return: 图书详情字典"""# 检查QPS限制self._check_qps()# 1. 构造请求URLurl = f"{self.base_url}/v2/books/detail"# 2. 构建请求参数params = self._get_common_params()params["id"] = book_id# 3. 生成签名params["signature"] = KongfzAuthUtil.generate_sign(params, self.app_secret)# 4. 发送请求try:response = requests.post(url,json=params,headers={"Content-Type": "application/json;charset=utf-8"},timeout=self.timeout)# 5. 解析响应result = response.json()# 6. 处理响应结果if result.get("code") == 200:return self._parse_book_detail(result.get("data", {}))else:raise Exception(f"获取图书详情失败: {result.get('message', '未知错误')} "f"(错误码: {result.get('code', '未知')})")except Exception as e:print(f"图书详情接口调用异常: {str(e)}")return Nonedef _parse_search_result(self, raw_data):"""解析搜索结果"""if not raw_data or "items" not in raw_data:return None# 处理图书列表books = []for item in raw_data["items"]:books.append({"id": item.get("id", ""),"title": item.get("title", ""),"author": item.get("author", ""),"publisher": item.get("publisher", ""),"publish_time": item.get("publishTime", ""),"era": item.get("era", ""),"edition": item.get("edition", ""),"binding": item.get("binding", ""),"pages": item.get("pages", 0),"price": float(item.get("price", 0)),"original_price": float(item.get("originalPrice", 0)),"book_condition": item.get("bookCondition", 0),"book_condition_desc": item.get("bookConditionDesc", ""),"shop_id": item.get("shopId", ""),"shop_name": item.get("shopName", ""),"cover_img": self._complete_image_url(item.get("coverImg", "")),"description": item.get("description", ""),"tags": item.get("tags", []),"fetch_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})return {"search_info": {"total": raw_data.get("total", 0),"page": raw_data.get("page", 1),"page_size": raw_data.get("pageSize", 20),"total_page": raw_data.get("totalPage", 0)},"books": books}def _parse_book_detail(self, raw_data):"""解析图书详情"""if not raw_data:return None# 处理图书详情信息return {"id": raw_data.get("id", ""),"title": raw_data.get("title", ""),"subtitle": raw_data.get("subtitle", ""),"author": raw_data.get("author", ""),"translator": raw_data.get("translator", ""),"publisher": raw_data.get("publisher", ""),"publish_time": raw_data.get("publishTime", ""),"print_time": raw_data.get("printTime", ""),"era": raw_data.get("era", ""),"edition": raw_data.get("edition", ""),"print": raw_data.get("print", ""),"binding": raw_data.get("binding", ""),"pages": raw_data.get("pages", 0),"word_count": raw_data.get("wordCount", 0),"isbn": raw_data.get("isbn", ""),"price": float(raw_data.get("price", 0)),"original_price": float(raw_data.get("originalPrice", 0)),"book_condition": raw_data.get("bookCondition", 0),"book_condition_desc": raw_data.get("bookConditionDesc", ""),"description": raw_data.get("description", ""),"content_desc": raw_data.get("contentDesc", ""),"author_desc": raw_data.get("authorDesc", ""),"shop_info": {"id": raw_data.get("shopId", ""),"name": raw_data.get("shopName", ""),"score": float(raw_data.get("shopScore", 0)),"sales": raw_data.get("shopSales", 0),"location": raw_data.get("shopLocation", "")},"images": [self._complete_image_url(img) for img in raw_data.get("images", [])],"tags": raw_data.get("tags", []),"category": raw_data.get("category", {}),"fetch_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}def _complete_image_url(self, url):"""补全图片URL"""if url and not url.startswith(("http://", "https://")):return f"https:{url}" if url.startswith("//") else f"https://img.kongfz.com{url}"return url

3. 商铺接口客户端

python

运行

class KongfzShopClient:"""孔夫子商铺接口客户端"""def __init__(self, app_key, app_secret):self.app_key = app_keyself.app_secret = app_secretself.base_url = "https://open.kongfz.com/api"self.timeout = 15  # 超时时间(秒)self.qps_limit = 3  # QPS限制self.last_request_time = 0self.request_lock = Lock()  # 线程锁控制QPSdef _get_common_params(self):"""生成公共请求参数"""return {"appKey": self.app_key,"timestamp": KongfzAuthUtil.get_timestamp(),"format": "json"}def _check_qps(self):"""检查并控制QPS"""with self.request_lock:current_time = time.time()interval = 1.0 / self.qps_limit  # 每次请求最小间隔elapsed = current_time - self.last_request_timeif elapsed < interval:# 需要等待的时间time.sleep(interval - elapsed)self.last_request_time = time.time()def search_shops(self, **kwargs):"""搜索商铺:param kwargs: 搜索参数:return: 搜索结果字典"""# 检查QPS限制self._check_qps()# 1. 构造请求URLurl = f"{self.base_url}/v2/shops/search"# 2. 构建请求参数params = self._get_common_params()# 添加业务参数valid_params = ["keyword", "categoryId", "location", "minScore", "minSales", "isVip", "page", "pageSize", "sort"]for param in valid_params:if param in kwargs and kwargs[param] is not None:params[param] = kwargs[param]# 3. 生成签名params["signature"] = KongfzAuthUtil.generate_sign(params, self.app_secret)# 4. 发送请求try:response = requests.post(url,json=params,headers={"Content-Type": "application/json;charset=utf-8"},timeout=self.timeout)# 5. 解析响应result = response.json()# 6. 处理响应结果if result.get("code") == 200:return self._parse_shop_search_result(result.get("data", {}))else:raise Exception(f"搜索商铺失败: {result.get('message', '未知错误')} "f"(错误码: {result.get('code', '未知')})")except Exception as e:print(f"商铺搜索接口调用异常: {str(e)}")return Nonedef get_shop_detail(self, shop_id,** kwargs):"""获取商铺详情:param shop_id: 商铺ID:param kwargs: 其他参数，如需要返回的商品数量:return: 商铺详情字典"""# 检查QPS限制self._check_qps()# 1. 构造请求URLurl = f"{self.base_url}/v2/shops/detail"# 2. 构建请求参数params = self._get_common_params()params["id"] = shop_id# 添加其他参数if "goodsCount" in kwargs:params["goodsCount"] = min(20, max(0, int(kwargs["goodsCount"])))# 3. 生成签名params["signature"] = KongfzAuthUtil.generate_sign(params, self.app_secret)# 4. 发送请求try:response = requests.post(url,json=params,headers={"Content-Type": "application/json;charset=utf-8"},timeout=self.timeout)# 5. 解析响应result = response.json()# 6. 处理响应结果if result.get("code") == 200:return self._parse_shop_detail(result.get("data", {}))else:raise Exception(f"获取商铺详情失败: {result.get('message', '未知错误')} "f"(错误码: {result.get('code', '未知')})")except Exception as e:print(f"商铺详情接口调用异常: {str(e)}")return Nonedef _parse_shop_search_result(self, raw_data):"""解析商铺搜索结果"""if not raw_data or "items" not in raw_data:return None# 处理商铺列表shops = []for item in raw_data["items"]:shops.append({"id": item.get("id", ""),"name": item.get("name", ""),"logo": self._complete_image_url(item.get("logo", "")),"location": item.get("location", ""),"score": float(item.get("score", 0)),"sales": item.get("sales", 0),"goods_count": item.get("goodsCount", 0),"is_vip": item.get("isVip", False),"vip_level": item.get("vipLevel", 0),"open_time": item.get("openTime", ""),"description": item.get("description", ""),"main_category": item.get("mainCategory", ""),"fetch_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})return {"search_info": {"total": raw_data.get("total", 0),"page": raw_data.get("page", 1),"page_size": raw_data.get("pageSize", 20),"total_page": raw_data.get("totalPage", 0)},"shops": shops}def _parse_shop_detail(self, raw_data):"""解析商铺详情"""if not raw_data:return None# 处理商铺商品products = []if "products" in raw_data and isinstance(raw_data["products"], list):for p in raw_data["products"]:products.append({"id": p.get("id", ""),"title": p.get("title", ""),"price": float(p.get("price", 0)),"book_condition": p.get("bookConditionDesc", ""),"cover_img": self._complete_image_url(p.get("coverImg", "")),"publish_time": p.get("publishTime", "")})return {"id": raw_data.get("id", ""),"name": raw_data.get("name", ""),"logo": self._complete_image_url(raw_data.get("logo", "")),"banner": self._complete_image_url(raw_data.get("banner", "")),"location": raw_data.get("location", ""),"score": float(raw_data.get("score", 0)),"score_detail": raw_data.get("scoreDetail", {}),"sales": raw_data.get("sales", 0),"goods_count": raw_data.get("goodsCount", 0),"month_sales": raw_data.get("monthSales", 0),"is_vip": raw_data.get("isVip", False),"vip_level": raw_data.get("vipLevel", 0),"open_time": raw_data.get("openTime", ""),"description": raw_data.get("description", ""),"business_scope": raw_data.get("businessScope", ""),"contact": {"phone": raw_data.get("phone", ""),"wechat": raw_data.get("wechat", ""),"qq": raw_data.get("qq", "")},"categories": raw_data.get("categories", []),"products": products,"fetch_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}def _complete_image_url(self, url):"""补全图片URL"""if url and not url.startswith(("http://", "https://")):return f"https:{url}" if url.startswith("//") else f"https://img.kongfz.com{url}"return url

4. 数据管理器（缓存与批量处理）

python

运行

import os
import json
import sqlite3
from datetime import datetime, timedelta
import timeclass KongfzDataManager:"""孔夫子数据管理器，支持缓存与批量处理"""def __init__(self, app_key, app_secret, cache_dir="./kongfz_cache"):self.book_client = KongfzBookClient(app_key, app_secret)self.shop_client = KongfzShopClient(app_key, app_secret)self.cache_dir = cache_dirself.db_path = os.path.join(cache_dir, "kongfz_cache.db")self._init_cache()def _init_cache(self):"""初始化缓存数据库"""if not os.path.exists(self.cache_dir):os.makedirs(self.cache_dir)# 连接数据库conn = sqlite3.connect(self.db_path)cursor = conn.cursor()# 创建图书缓存表cursor.execute('''CREATE TABLE IF NOT EXISTS book_cache (book_id TEXT PRIMARY KEY,data TEXT,fetch_time TEXT)''')# 创建搜索缓存表cursor.execute('''CREATE TABLE IF NOT EXISTS search_cache (cache_key TEXT PRIMARY KEY,data TEXT,fetch_time TEXT,keyword TEXT)''')# 创建商铺缓存表cursor.execute('''CREATE TABLE IF NOT EXISTS shop_cache (shop_id TEXT PRIMARY KEY,data TEXT,fetch_time TEXT)''')conn.commit()conn.close()def search_books(self, keyword, use_cache=True, cache_ttl=3600, **kwargs):"""搜索图书，支持缓存:param keyword: 搜索关键词:param use_cache: 是否使用缓存:param cache_ttl: 缓存有效期(秒):param kwargs: 其他搜索参数:return: 搜索结果"""# 生成缓存键cache_key = self._generate_cache_key("book", keyword,** kwargs)# 尝试从缓存获取if use_cache:cached_data = self._get_cached_data("search_cache", cache_key, cache_ttl)if cached_data:print(f"使用缓存数据，关键词: {keyword}，页码: {kwargs.get('page', 1)}")return cached_data# 从接口获取print(f"调用接口搜索，关键词: {keyword}，页码: {kwargs.get('page', 1)}")result = self.book_client.search_books(keyword=keyword, **kwargs)# 更新缓存if result:self._update_cache("search_cache", cache_key, result, keyword)return resultdef get_book_detail(self, book_id, use_cache=True, cache_ttl=86400):"""获取图书详情，支持缓存:param book_id: 图书ID:param use_cache: 是否使用缓存:param cache_ttl: 缓存有效期(秒):return: 图书详情"""# 尝试从缓存获取if use_cache:cached_data = self._get_cached_data("book_cache", book_id, cache_ttl)if cached_data:print(f"使用缓存数据，图书ID: {book_id}")return cached_data# 从接口获取print(f"调用接口获取图书详情，ID: {book_id}")result = self.book_client.get_book_detail(book_id)# 更新缓存if result:self._update_cache("book_cache", book_id, result)return resultdef search_shops(self, keyword, use_cache=True, cache_ttl=3600,** kwargs):"""搜索商铺，支持缓存"""# 生成缓存键cache_key = self._generate_cache_key("shop", keyword, **kwargs)# 尝试从缓存获取if use_cache:cached_data = self._get_cached_data("search_cache", cache_key, cache_ttl)if cached_data:print(f"使用缓存数据，商铺搜索关键词: {keyword}")return cached_data# 从接口获取print(f"调用接口搜索商铺，关键词: {keyword}")result = self.shop_client.search_shops(keyword=keyword,** kwargs)# 更新缓存if result:self._update_cache("search_cache", cache_key, result, keyword)return resultdef get_shop_detail(self, shop_id, use_cache=True, cache_ttl=86400, **kwargs):"""获取商铺详情，支持缓存"""# 尝试从缓存获取if use_cache:cached_data = self._get_cached_data("shop_cache", shop_id, cache_ttl)if cached_data:print(f"使用缓存数据，商铺ID: {shop_id}")return cached_data# 从接口获取print(f"调用接口获取商铺详情，ID: {shop_id}")result = self.shop_client.get_shop_detail(shop_id,** kwargs)# 更新缓存if result:self._update_cache("shop_cache", shop_id, result)return resultdef batch_get_book_details(self, book_ids, use_cache=True, cache_ttl=86400):"""批量获取图书详情"""details = []for book_id in book_ids:detail = self.get_book_detail(book_id, use_cache, cache_ttl)if detail:details.append(detail)# 避免触发QPS限制，主动添加延迟time.sleep(0.5)return detailsdef _generate_cache_key(self, type_, keyword, **params):"""生成缓存键"""sorted_params = sorted(params.items(), key=lambda x: x[0])params_str = "&".join([f"{k}={v}" for k, v in sorted_params])return hashlib.md5(f"{type_}_{keyword}_{params_str}".encode()).hexdigest()def _get_cached_data(self, table, key, ttl):"""从缓存获取数据"""conn = sqlite3.connect(self.db_path)cursor = conn.cursor()cursor.execute(f"SELECT data, fetch_time FROM {table} WHERE {table[:-6]}_id = ?",(key,))result = cursor.fetchone()conn.close()if result:data_str, fetch_time = result# 检查缓存是否过期fetch_time_obj = datetime.strptime(fetch_time, "%Y-%m-%d %H:%M:%S")if (datetime.now() - fetch_time_obj).total_seconds() <= ttl:try:return json.loads(data_str)except:return Nonereturn Nonedef _update_cache(self, table, key, data, keyword=None):"""更新缓存"""if not data:returnconn = sqlite3.connect(self.db_path)cursor = conn.cursor()data_str = json.dumps(data, ensure_ascii=False)fetch_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")if table == "search_cache":# 搜索缓存有额外的keyword字段cursor.execute('''INSERT OR REPLACE INTO search_cache (cache_key, data, fetch_time, keyword)VALUES (?, ?, ?, ?)''', (key, data_str, fetch_time, keyword or ""))else:# 图书和商铺缓存cursor.execute(f'''INSERT OR REPLACE INTO {table} ({table[:-6]}_id, data, fetch_time)VALUES (?, ?, ?)''',(key, data_str, fetch_time))conn.commit()conn.close()def clean_expired_cache(self, max_age=86400 * 7):"""清理过期缓存，默认保留7天内的数据"""conn = sqlite3.connect(self.db_path)cursor = conn.cursor()# 计算过期时间expire_time = (datetime.now() - timedelta(seconds=max_age)).strftime("%Y-%m-%d %H:%M:%S")# 清理图书缓存cursor.execute("DELETE FROM book_cache WHERE fetch_time < ?", (expire_time,))deleted_books = cursor.rowcount# 清理搜索缓存cursor.execute("DELETE FROM search_cache WHERE fetch_time < ?", (expire_time,))deleted_searches = cursor.rowcount# 清理商铺缓存cursor.execute("DELETE FROM shop_cache WHERE fetch_time < ?", (expire_time,))deleted_shops = cursor.rowcountconn.commit()conn.close()print(f"清理过期缓存完成，删除图书缓存 {deleted_books} 条，搜索缓存 {deleted_searches} 条，商铺缓存 {deleted_shops} 条")return {"deleted_books": deleted_books,"deleted_searches": deleted_searches,"deleted_shops": deleted_shops}

四、完整使用示例

1. 古籍图书检索与详情获取

python

运行

def ancient_book_search_demo():# 替换为实际的应用信息APP_KEY = "your_app_key"APP_SECRET = "your_app_secret"# 初始化数据管理器data_manager = KongfzDataManager(APP_KEY, APP_SECRET)# 搜索参数：查找清代古籍search_params = {"keyword": "古籍","era": "清代","bookCondition": 3,  # 九品"minPrice": 100,"maxPrice": 5000,"page": 1,"pageSize": 10,"sort": "price_asc"  # 价格从低到高}# 执行搜索result = data_manager.search_books(**search_params,use_cache=True,cache_ttl=3600  # 缓存1小时)if result:print(f"===== 古籍搜索结果: {search_params['keyword']} =====")print(f"共搜索到 {result['search_info']['total']} 本图书，第 {result['search_info']['page']}/{result['search_info']['total_page']} 页")# 打印搜索结果for i, book in enumerate(result["books"], 1):print(f"\n{i}. {book['title']}")print(f"作者: {book['author']}")print(f"年代: {book['era']}")print(f"品相: {book['book_condition_desc']}")print(f"价格: ¥{book['price']}")print(f"商家: {book['shop_name']}")# 获取第一本图书的详细信息if result["books"]:first_book_id = result["books"][0]["id"]book_detail = data_manager.get_book_detail(first_book_id)if book_detail:print(f"\n===== 图书详情: {book_detail['title']} =====")print(f"详细描述: {book_detail['description'][:200]}...")print(f"出版信息: {book_detail['publisher']} {book_detail['publish_time']}")print(f"页数: {book_detail['pages']}页")print(f"商家信息: {book_detail['shop_info']['name']} (评分: {book_detail['shop_info']['score']})")print(f"图书图片: {book_detail['images'][0] if book_detail['images'] else '无'}")# 清理过期缓存data_manager.clean_expired_cache()if __name__ == "__main__":ancient_book_search_demo()

2. 商铺搜索与数据集成

python

运行

def shop_search_demo():# 替换为实际的应用信息APP_KEY = "your_app_key"APP_SECRET = "your_app_secret"# 初始化数据管理器data_manager = KongfzDataManager(APP_KEY, APP_SECRET)# 搜索古籍商铺shop_search_params = {"keyword": "古籍","location": "北京","minScore": 4.5,"minSales": 1000,"isVip": 1,"page": 1,"pageSize": 5,"sort": "sales_desc"  # 按销量排序}# 执行商铺搜索shop_result = data_manager.search_shops(** shop_search_params)if shop_result:print(f"===== 商铺搜索结果: {shop_search_params['keyword']} =====")print(f"共搜索到 {shop_result['search_info']['total']} 家商铺")# 打印商铺信息for i, shop in enumerate(shop_result["shops"], 1):print(f"\n{i}. {shop['name']}")print(f"位置: {shop['location']}")print(f"评分: {shop['score']} 分")print(f"销量: {shop['sales']} 单")print(f"在售图书: {shop['goods_count']} 本")print(f"开店时间: {shop['open_time']}")# 获取第一个商铺的详细信息if shop_result["shops"]:first_shop_id = shop_result["shops"][0]["id"]shop_detail = data_manager.get_shop_detail(first_shop_id,goodsCount=5  # 获取5件商品)if shop_detail:print(f"\n===== 商铺详情: {shop_detail['name']} =====")print(f"经营范围: {shop_detail['business_scope']}")print(f"联系方式: {shop_detail['contact']['phone']}")print(f"本月销量: {shop_detail['month_sales']} 单")# 打印商铺的热门商品if shop_detail["products"]:print("\n热门商品:")for i, product in enumerate(shop_detail["products"], 1):print(f"{i}. {product['title']} - ¥{product['price']} ({product['book_condition']})")if __name__ == "__main__":shop_search_demo()

五、常见问题与优化建议

1. 常见错误码及解决方案

错误码	说明	解决方案
200	成功	-
400	参数错误	检查必填参数是否齐全，参数格式是否正确
401	认证失败	检查 appKey 是否正确，签名是否有效，时间戳是否过期
403	权限不足	检查应用是否已申请该接口权限
429	频率超限	降低调用频率，确保不超过 QPS 限制
500	服务器错误	记录错误信息，稍后重试
503	服务维护	等待服务恢复后再调用