我最早接触 item_search 招标搜索接口,是因为要给公司做一套自动盯标系统。每天人工去各省公共资源交易平台翻公告,既慢又容易漏,尤其关键词一多,眼睛完全不够用。后来仔细研究了一下第三方聚合平台提供的 item_search 接口,发现它本质上就是把全网散落的招标公告整理成一份可查询的 JSON 数据源,你用一组参数发一次 HTTP 请求,它就把匹配的公告列表返回来。这篇就想把从入门到精通的完整对接过程写下来,包括接口定义、签名鉴权、第一行请求、分页抓取、限流重试、幂等去重和常见问题排查,适合刚开始接接口的开发者,也适合已经跑通但想让程序更稳的老手。
1. 对接前准备:先搞清楚“item_search”到底是什么
1.1 接口定义里必须看清楚的五个信息
虽然都叫 item_search,但不同数据服务商给出的接口定义差别很大。我见过有的平台把接口路径直接写成/open/item_search,有的则放在/api/v1/search/tender下面,所以拿到文档后第一件事不是写代码,而是把下面五样东西抄到自己的笔记里。
- 接口名称和版本号,确认你拿到的文档和实际环境一致;
- 请求地址,区分测试环境和生产环境,很多时候这俩域名不一样;
- HTTP 方法,一般是 GET,但也有平台要求 POST,因为查询条件复杂时 GET 的 URL 长度不够用;
- 必填参数与选填参数,逐个确认类型、是否参与签名、最大长度;
- 响应结构,最好找到一份真实样例 JSON,确认字段层级和类型。
这些信息一般会写在“接口定义”章节,或者“开发规范”的附录里。我习惯先把它们整理成一张表格,对标文档去核对,尤其是参数名大小写。比如PageNo和page_no,看起来是同一个意思,少一个下划线服务器就报“参数缺失”。这类问题排查起来很烦,但完全可以在准备阶段避免。
还有一个容易忽略的地方:接口文档里的时间格式。有的平台返回2025-04-01 10:00:00,有的返回时间戳,还有的返回2025-04-01T10:00:00+08:00。这些字段后续要入库、排序、去重,解析方式完全不同。最好在准备阶段就确认好,并写一个统一的格式化函数。
1.2 签名鉴权:app_key 和 sign 是怎么算出来的
几乎所有正规的招标数据接口都会要求鉴权,最常见的做法是app_key + app_secret生成签名sign。我在刚开始对接时以为这只是走个过场,结果第一步就卡在签名上,生成出来的 sign 一直被提示“无效签名”。
签名机制的基本逻辑:调用方用私钥app_secret把请求参数混合加密,服务端用同样的规则重算一遍,如果两边结果一致,就说明请求没有被篡改。具体算法各平台有差异,但通用的套路是这样:
- 选取所有请求参数(除去 sign 本身,有的平台还要除去空值)。
- 按参数名字典序升序排列。
- 把
keyvalue形式拼接成一个字符串。 - 在拼接串的首尾加上
app_secret(加头加尾)。 - 对整个字符串做 MD5 或者 HMAC-SHA256,转成大写或小写。
举个我实际用过的例子:
import hashlib import time import requests APP_KEY = "你的app_key" APP_SECRET = "你的app_secret" def build_sign(params: dict, secret: str) -> str: # 过滤空值和 sign 本身,剩下所有参与签名的参数 filtered = {k: v for k, v in params.items() if k != "sign" and v not in ("", None)} sorted_keys = sorted(filtered.keys()) raw = "".join(f"{k}{filtered[k]}" for k in sorted_keys) raw = secret + raw + secret return hashlib.md5(raw.encode("utf-8")).hexdigest().upper() params = { "app_key": APP_KEY, "keyword": "智慧城市 招标", "page_no": 1, "page_size": 20, "timestamp": str(int(time.time())), } params["sign"] = build_sign(params, APP_SECRET) resp = requests.get( "https://api.你的平台.com/open/item_search", params=params, timeout=10, ) print(resp.status_code) print(resp.json())这段代码的核心不在 MD5 本身,而在于动态参数的一致性。比如timestamp必须在签名时和服务端收到时保持一致,所以不能用同一个字符串去签名、又用另一个时间戳去发送。我当时踩的坑就是先time.time()拼进字符串签名,然后又转成整型传参,导致签名对不上。解决方法是先定义好参数,再从参数里取对应值去签名。
另外,MD5 结果的大小写也容易忽略。有的平台要求小写,我遇到的是大写,别想当然。更稳妥的做法是拿到文档后,先看里面的示例请求,把响应里的sign和按自己算法算出来的结果对一遍,确认规则后再写封装函数。
1.3 准备调试环境:curl、Postman 和 Python 三件套
我推荐的调试顺序是:先用 curl 验证链路,再用 Postman 做参数可视化,最后写 Python 封装。很多人一上来就打开 IDE 敲代码,结果报错时不知道是网络问题、权限问题还是参数问题,排查范围太大。
curl 的优点是干净利落,能清楚看到原始请求:
curl -G "https://api.你的平台.com/open/item_search" \ --data-urlencode "app_key=你的app_key" \ --data-urlencode "keyword=智慧城市 招标" \ --data-urlencode "page_no=1" \ --data-urlencode "page_size=20" \ --data-urlencode "timestamp=1744000000" \ --data-urlencode "sign=你算出来的sign"注意这里必须用--data-urlencode,因为 keyword 里有中文,如果没有正确编码,服务端很可能拿不到完整的查询词。如果平台文档只给了一个不含中文的示例,你很难发现这个问题。
Postman 的优势是能直观地看响应体和请求头。把它当成接口文档的“活测试页”来用,尤其是排查 sign 问题时,可以直接对比每个参数在 Postman 里的值和在代码里发出的值,很容易定位到是哪一个参数变了。
Python 环境就不多说了,requests库是标配。我的建议是不要每次请求都requests.get一把梭,而是把请求封装成一个函数,统一处理签名、超时、重试和日志,后面章节会给出代码。
1.4 测试数据与沙箱环境
不少平台会给你一个测试环境的 app_key,专门用来联调。但也有一些平台没有沙箱,只允许用少量真实请求测试。这时候不要一上来就去拉全量数据,而是用一条最简单的 keyword 比如“实验室设备”测一条,确认返回结构正常后再逐步加条件。
我通常会在测试阶段固定一个测试关键词,比如“物业服务采购”,然后连续请求三次,观察返回的total是否稳定,items里的字段是否有缺失。这个动作能提前暴露两类问题:一是接口统计口径可能与页面不一致,二是某些字段在特定记录里为空,前端代码要兼容。
在测试环境里,还要专门试一次非法参数,比如page_no传 0,看看平台返回的是空列表还是报错信息。这能帮你判断后续代码应该按“空数据处理”还是按“异常分支处理”。了解平台的容错习惯,对接起来会顺手很多。
2. 发起第一次请求:从一行 curl 到一套可复用函数
2.1 先用 curl 跑通一个最简单的查询
假设你已经把 app_key 和 app_secret 拿到了,也有了接口文档,那么第一步不是写代码,而是手动组织参数并发一次 curl。这一步的目的很单纯:确认网络能通、鉴权能过、响应能解析。
以下面的请求为例:
curl -G "https://api.你的平台.com/open/item_search" \ --data-urlencode "app_key=test_123" \ --data-urlencode "keyword=物业" \ --data-urlencode "page_no=1" \ --data-urlencode "page_size=10" \ --data-urlencode "timestamp=1744000000" \ --data-urlencode "sign=abc123def456..."如果返回的 JSON 里有code: 0或success: true,那恭喜,链路已经通了。如果返回的是msg: sign error,不要急着去检查 sign 算法,先确认一件事:请求里的参数名、参数值和参与签名的参数顺序是否与文档完全一致。比如timestamp是否带了引号、是否包含毫秒,都可能导致签名不一致。
这里分享一个我在 curl 调试时常用的技巧:在请求后面加-v参数,可以看到完整的 HTTP 请求头和响应头。很多服务端错误会把具体原因放在响应头的X-Error-Message字段里,而不仅限于 Body。如果只看 Body,容易漏掉关键信息。
2.2 用 Python 封装第一个可复用搜索函数
curl 验证通过后,就可以把这套逻辑搬到 Python 里了。我推荐直接把签名、请求、超时、异常全部放进一个函数里,避免每次调用都重复写一坨。
import hashlib import time import requests class BidSearchClient: def __init__(self, app_key: str, app_secret: str, base_url: str): self.app_key = app_key self.app_secret = app_secret self.base_url = base_url def _sign(self, params: dict) -> str: filtered = {k: v for k, v in params.items() if k != "sign" and v not in ("", None)} sorted_keys = sorted(filtered.keys()) raw = "".join(f"{k}{filtered[k]}" for k in sorted_keys) raw = self.app_secret + raw + self.app_secret return hashlib.md5(raw.encode("utf-8")).hexdigest().upper() def search(self, keyword: str, page_no: int = 1, page_size: int = 20): params = { "app_key": self.app_key, "keyword": keyword, "page_no": page_no, "page_size": page_size, "timestamp": str(int(time.time())), } params["sign"] = self._sign(params) resp = requests.get(self.base_url, params=params, timeout=10) resp.raise_for_status() payload = resp.json() if payload.get("code") != 0: raise RuntimeError(payload.get("msg")) return payload.get("data", {}) client = BidSearchClient( app_key="test_123", app_secret="test_secret", base_url="https://api.你的平台.com/open/item_search", ) data = client.search("智慧路灯") print(data.get("total")) for item in data.get("items", []): print(item["title"], item["publish_time"])这个封装有几点值得注意:
- 签名函数和请求函数解耦,后续新增参数只需要改
search方法。 - 每次都从参数字典里生成
timestamp,然后传给签名函数,确保签名和请求用的是同一个时间戳。 raise_for_status()用来快速暴露 HTTP 层错误,但真正的业务错误还要看code。
这样第一个可复用的请求函数就完成了。你可以把它放进一个模块文件里,后续所有脚本都基于它扩展。
2.3 字段解析:从原始响应到结构化数据
接口返回的 JSON 通常长这样:
{ "code": 0, "msg": "ok", "data": { "total": 128, "has_next": true, "items": [ { "id": "20250401001", "title": "某市智慧城市一期招标公告", "region": "浙江", "publish_time": "2025-04-01 10:00:00", "deadline": "2025-05-01 09:30:00", "budget": 1500000, "content_url": "https://example.gov.cn/notice/123" } ] } }字段看起来很直观,但实际接入时你会发现不少问题:
id是全站唯一,还是只在当天唯一?一定要确认,这关系到后续去重。title里可能带\n,也可能包含空格,入库前要清洗。publish_time和deadline是本地时间还是 UTC?如果差 8 小时,你按天增量抓取时会漏掉一批。budget可能是数字类型,也可能是字符串"1500000",类型判断写死容易炸。
所以我习惯定义一个normalize_item函数,把所有字段统一转换成自己系统的内部格式:
def normalize_item(raw: dict) -> dict: return { "biz_id": raw.get("id"), "title": " ".join(raw.get("title", "").split()), "region": raw.get("region", ""), "publish_time": format_time(raw.get("publish_time")), "deadline": format_time(raw.get("deadline")), "budget": float(raw["budget"]) if raw.get("budget") else 0.0, "url": raw.get("content_url", ""), }" ".join(...split())这个写法能把连续多个空格、换行统一折叠成一个空格,对公告标题这种人工录入的数据很管用。
2.4 分页抓取:不要一把梭拉全网
第一次跑通接口后,很多人下意识会写一个循环,从第 1 页一直抓到has_next=False。这个思路本身没错,但直接梭哈很容易触发限流,也可能抓到重复或乱序数据。
我建议抓取策略这样定:
- 设置最大页数保护,比如最多抓 50 页,防止死循环。
- 每页之间固定 sleep 一小段时间,比如
time.sleep(1),避免突发流量。 - 所有抓下来的数据先放到一个临时列表里,等全部翻页结束后再批量入库,而不是每页都去操作数据库。
更重要的一点是:不要依赖页码做断点续传。因为招标公告随时可能新增,你上一页刚抓完第 10 页,下一页位置已经变了,就会漏掉或重复。如果平台支持游标参数cursor,优先用游标;如果只支持页码,那就在本地记录已抓过的id集合,抓完一批后做差集去重。
3. 稳定对接的硬指标:限流、重试、幂等、增量
3.1 频率控制:从响应头里读懂限流
接口对接不崩的秘密,大部分在限流上。很多平台在响应头里会返回当前账号的配额情况,常见的有:
X-RateLimit-Limit: 100 X-RateLimit-Remaining: 97 X-RateLimit-Reset: 1744001200这三个值分别代表单位时间总配额、剩余配额、重置时间戳。我强烈建议每次请求后都把这三个值打日志,哪怕只是简单 print,因为限流触发往往不是立即报错,而是响应变慢。如果你有日志,就能看到剩余配额从 20 慢慢耗到 1,然后下一请求被拒绝,整个过程一目了然。
如果平台不提供这些头字段,也可以通过响应码判断。常见的限流表现不是 200 + 错误码,而是直接返回 429 Too Many Requests。代码里需要对 429 做单独处理,不能简单raise_for_status就完事。
控制频率最基本的方式就是 sleep:
def search_with_rate_limit(client, keyword, pages=20): results = [] for page in range(1, pages + 1): data = client.search(keyword, page=page) results.extend(data.get("items", [])) has_next = data.get("has_next", False) if not has_next: break time.sleep(1.5) # 给足间隔,避免触发限流 return results如果你有多账号并发需求,建议改用令牌桶算法,而不是单纯 sleep。令牌桶的优点是能应对短时突发流量,同时又限制平均速率。Python 里可以用ratelimit库,或者自己用queue实现一个简单的令牌桶,网上有很多现成代码可以改。
3.2 超时设置和退避重试
接口请求最常见的异常不是业务错误,而是网络超时。很多人代码里timeout不写,结果服务端卡住时,线程也卡住,整个爬虫任务挂在那里一动不动。所以必须显式设置连接超时和读取超时。
我一般用timeout=(3.05, 10),第一个数字是连接建立超时,第二个是读取响应超时。如果请求失败,再按指数退避重试:
import time import requests def request_with_retry(url, params, retries=3): for attempt in range(retries): try: resp = requests.get(url, params=params, timeout=(3.05, 10)) if resp.status_code == 429: raise RateLimitError resp.raise_for_status() return resp except requests.exceptions.Timeout: wait_time = 2 ** attempt + 1 print(f"第 {attempt + 1} 次超时,{wait_time}s 后重试") time.sleep(wait_time) except RateLimitError: reset_time = int(resp.headers.get("X-RateLimit-Reset", time.time() + 30)) sleep_time = max(1, reset_time - time.time() + 1) time.sleep(sleep_time) return None这里2 ** attempt + 1就是指数退避:第一次等 2 秒,第二次等 5 秒,第三次等 9 秒。不要重试太多次,一般三次就够。如果连续三次都失败,应该发告警通知人,而不是无限重试浪费配额。
3.3 幂等性:重复请求不会重复入库
接口幂等性这个话题,在对接任何数据接口时都必须考虑。所谓幂等,通俗说就是同一个请求重复执行很多次,结果跟执行一次是一样的。对于招标搜索接口来说,它本身是查询接口,天然幂等;但你把查询结果写入自己的数据库时,就未必幂等了。
最常见的坑是:分页时页面位置偏移,导致同一公告被抓到两次。或者重试机制触发,同一页请求发了两遍。如果直接按title判断是否存在,很可能因为时效性标题有细微差别而失效。
所以我建议在业务表里加一个唯一约束,直接用接口返回的id作为业务主键或唯一索引:
CREATE TABLE tenders ( id INTEGER PRIMARY KEY AUTOINCREMENT, biz_id VARCHAR(64) NOT NULL UNIQUE, title TEXT, region VARCHAR(64), publish_time DATETIME, deadline DATETIME, budget REAL, url TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP );写入时用INSERT OR IGNORE(SQLite)或ON CONFLICT DO NOTHING(PostgreSQL),数据库天然帮你把重复数据挡在门外。
如果同一个接口返回的id不是全局唯一,比如是“日期+序号”,那就要拼接一个自己的唯一键,比如${source_code}:${id},确保不同数据源之间也不会冲突。
3.4 增量更新:别每次都全量拉取
当关键词一多、数据量一大,每次都全量分页拉取非常浪费配额,还容易把接口限流打满。更合理的做法是只拉更新时间大于本地最新时间的公告。
增量更新的设计核心在于比较基准。你可以记录本地库里某条记录的最晚publish_time,每次查询时在参数里加一个时间过滤条件:
params["start_time"] = "2025-04-01 00:00:00" params["end_time"] = "2025-04-01 23:59:59"但是要注意,不同平台对时间过滤的支持不同。有的平台只支持按发布日期过滤,有的只支持按更新时间过滤。招标公告发布日期比评审日期早,如果你用发布日期过滤,更新公告详情后,可能不会重新出现在搜索结果里,从而漏掉变更信息。
所以更稳妥的做法是每天做一次小的增量,每周做一次全量对齐。增量负责时效性,全量负责修正。全量对齐时用唯一键做差集,把已删除的公告标记为失效。这套逻辑虽然啰嗦,但能长期稳定运行。
4. 实战:做一个自动盯标小工具
4.1 场景与整体设计
当接口调用稳定之后,就可以把这些代码真正用起来。我的需求很简单:每天早上九点,扫描几个关键词,把前一天新发布的招标公告汇总成一条消息,推送到企业微信群。整个工具分为三步:拉取接口数据、写入 SQLite、生成推送文本。
关键词列表放在一个 JSON 文件里:
{ "keywords": ["智慧城市", "智慧路灯", "物业管理服务"] }主程序做的事情很简单:逐个关键词调用client.search(),然后用本地数据库里的biz_id判断哪些是新公告,把新公告组成列表。
4.2 核心代码实现
这里给一个完整可运行的简化版脚本:
import json import sqlite3 import time from datetime import datetime from bid_search_client import BidSearchClient DB_PATH = "tenders.db" KEYWORDS_PATH = "keywords.json" def get_db_conn(): conn = sqlite3.connect(DB_PATH) conn.execute(""" CREATE TABLE IF NOT EXISTS tenders ( biz_id TEXT PRIMARY KEY, keyword TEXT, title TEXT, region TEXT, publish_time TEXT, deadline TEXT, budget REAL, url TEXT, created_at TEXT ) """) return conn def save_new_items(conn, keyword, items): new_count = 0 for item in items: try: conn.execute( "INSERT OR IGNORE INTO tenders (biz_id, keyword, title, region, publish_time, deadline, budget, url, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", (item["id"], keyword, item["title"], item["region"], item["publish_time"], item["deadline"], item["budget"], item["content_url"], datetime.now().isoformat()) ) if conn.total_changes > 0: new_count += 1 except Exception as e: print(e) conn.commit() return new_count def build_message(new_items): lines = ["【招标盯标日报】", f"时间:{datetime.now().strftime('%Y-%m-%d %H:%M')}", ""] if not new_items: lines.append("今日没有新增公告。") else: for item in new_items: lines.append(f"标题:{item['title']}") lines.append(f"地区:{item['region']} | 发布时间:{item['publish_time']}") lines.append(f"链接:{item['url']}") lines.append("---") return "\n".join(lines) client = BidSearchClient( app_key="app_key", app_secret="app_secret", base_url="https://api.你的平台.com/open/item_search", ) with open(KEYWORDS_PATH, encoding="utf-8") as f: config = json.load(f) conn = get_db_conn() all_new = [] for keyword in config["keywords"]: page = 1 while page <= 10: data = client.search(keyword, page_no=page, page_size=50) items = data.get("items", []) if not items: break new_count = save_new_items(conn, keyword, items) if new_count > 0: all_new.extend(items[:new_count]) if not data.get("has_next"): break page += 1 time.sleep(1) conn.close() message = build_message(all_new) print(message) # 这里可以调用企微机器人 webhook 发送 message代码里的INSERT OR IGNORE保证了即使同一个公告被两个关键词都匹配到,biz_id相同,只会写一次。conn.total_changes > 0用来判断这一行是否真的插入了,如果不为 0,说明这条公告是新入库的,值得放进推送列表。
4.3 部署为定时任务
这个脚本最简单的方式是丢在服务器上用 cron 每天执行:
0 9 * * * cd /opt/bid-watcher && /usr/bin/python3 main.py >> logs/run.log 2>&1如果你用的是 Windows,也可以用计划任务。这里有个经验:不要直接跑python main.py而不输出日志,否则脚本崩了你根本不知道。所以一定要加日志重定向,并在脚本里用logging而不是print。
更进阶的做法是使用 systemd timer,好处在于可以管理依赖和重启策略。但对于一个 100 行不到的脚本,cron 完全够了。
4.4 效果与后续调优
跑起来后,你会很快发现两个问题:一是某些关键词太宽泛,匹配回来的公告一半不相关;二是同样的公告可能在多个关键词下重复入库,浪费查询配额。
宽泛问题要靠关键词优化解决,比如“智慧城市”可以改成“智慧城市 招标 OR 采购”,甚至直接用平台的排除词参数exclude_keyword。重复问题则可以用一个本地已见过的biz_id集合,在发送推送前去重。
我实际用下来,每天一遍全量加增量,耗时不到三分钟,配额消耗非常小。这套工具帮我从盯标工作里省下了大量时间,而且漏标的概率比人工低很多。
5. 常见问题与排查技巧实录
5.1 签名一直失败怎么办
这是对接 item_search 接口最高频的问题。按照我的经验,90% 的原因集中在四个方面:
- 参与签名的参数没有按字典序排序,或者包含了一些不应该包含的参数。
- 空值处理不当,比如某个字段传了空字符串,文档要求空值不参与签名,你却没有过滤。
- 时间戳不一致,签名时用的
timestamp和发送请求时用的不是同一个值。 - 编码问题,尤其是中文参数,签名时使用原始中文还是 URL 编码后的字符串,文档必须明确。
排查时我建议打印出参与签名的参数集合和最终拼接串,对照文档手算一遍。你很快就会发现差别。
5.2 返回“无权限”或“余额不足”怎么排查
“无权限”通常不是业务问题,而是鉴权层面的问题。检查顺序:
- app_key 是否填错,注意前后空格和大小写。
- IP 白名单是否包含当前服务器 IP。很多平台要求把公网 IP 加入白名单,本地测试换网后 IP 变了,就会报无权限。
- 接口套餐是否生效,有的接口需要单独开通,光有 app_key 还不够。
- 账号是否欠费或过期,这个最容易忽略,尤其是年付套餐到期后接口不会立即报“过期”,而是报“无权限”。
如果是“余额不足”,那就要看自家账户的计费模式。按次计费接口要做本地配额保护,比如在代码里统计每日调用次数,超过阈值后自动停止,否则很容易在调试阶段消耗完余额。
5.3 字段返回不全或格式异常
平台接口也会“抽风”。常见情况是返回的items中某条记录缺少budget字段,结果代码执行float(raw["budget"])直接抛 KeyError。解决方案是在解析函数里全部使用raw.get(key),并对缺失值给默认值。
还有一种情况是响应字段的类型在变化。比如publish_time有时是字符串,有时是数字时间戳。这时候不要自己猜,而是定义一个兼容函数:
def format_time(value): if not value: return None if isinstance(value, (int, float)): return datetime.fromtimestamp(value).strftime("%Y-%m-%d %H:%M:%S") if isinstance(value, str): # 尝试多种格式 for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S", "%Y/%m/%d"): try: return datetime.strptime(value, fmt).strftime("%Y-%m-%d %H:%M:%S") except ValueError: continue return str(value)这样即使平台调整了格式,你的代码也不需要立刻跟着改。
5.4 抓取到重复数据,如何快速清理
如果已经入库了一批重复数据,最简单的方法是给biz_id建唯一索引,然后删除重复行。
SQLite 里去重可以用:
DELETE FROM tenders WHERE id NOT IN ( SELECT MIN(id) FROM tenders GROUP BY biz_id );如果业务上确实需要多条记录(比如同一个招标项目有多次更正公告),那就要明确唯一键是什么。有的接口会给一个parent_id表示项目包,id表示公告子条目,这时候唯一键应该是parent_id而不是id,否则会把同一个项目的多份公告误判为重复。
5.5 一份避坑清单
| 问题类型 | 典型表现 | 处理建议 |
|---|---|---|
| 参数名不一致 | 报“缺少参数”,但文档明明有 | 用接口文档的原始参数名,不要自己改下划线 |
| 中文编码 | 搜索不到中文关键词结果 | 使用--data-urlencode或 requests 的 params 自动编码 |
| 时间戳单位 | 签名一直失败 | 确认单位是秒还是毫秒,并建立统一常量 |
| 响应字段缺失 | 解析时报 KeyError | 统一用.get()并设默认值 |
| 限流触发 | 返回 429 或响应头配额为 0 | 重试前读取Retry-After或X-RateLimit-Reset |
| 重复入库 | 数据量比预期大很多 | 建唯一索引并INSERT OR IGNORE |
| 分页偏移 | 漏数据或重复数据 | 用游标、或用本地biz_id去重 |
| 日志缺失 | 出错无法定位 | 每个请求和响应都打日志,至少记录 status 和 total |
这张表是我自己在对接过程中沉淀下来的,符合大多数平台的习惯。如果你对接的平台行为不太一样,也建议按这个分类方式维护一份自己的清单,后续维护成本会大幅下降。
再多说一句经验:接口对接跑通从来不是终点,稳定运行才是。上线前一定要把日志、错误告警、配额监控补齐,否则真出现数据断层,发现问题时已经晚了。我个人体会最深的一点是,接口文档里最不起眼的那行“注意:调用频率请控制在每秒不超过5次”,价值抵得上整个鉴权流程。所有你觉得理所当然的东西,都值得反复确认一遍。