影刀RPA进阶教程_API调用的进阶实战RESTful鉴权分页与错误处理
影刀RPA进阶教程:API调用的进阶实战——RESTful接口鉴权分页与错误处理
上一篇讲了影刀RPA的HTTP请求基础(GET/POST/JSON解析)。但真正工程级的API调用,有三个绕不过去的东西:鉴权、分页、错误处理。
这三个搞不定,API调用就只停留在"玩具"级别。这一篇全讲透。
鉴权的三种主流方式
不同的API系统用不同的鉴权方式,搞错了直接401。
方式一:Token方式(最常见)
Token类似临时通行证,登录后返回,后续所有请求带上。典型场景:电商平台开放API、企业内部系统。
# 第一步:发送登录请求,获取Tokenimportjson# POST到登录接口login_body=json.dumps({"username":"your_account","password":"your_password"})[video(video-ua386cnT-1781422012570)(type-csdn)(url-https://live.csdn.net/v/embed/525000)(image-https://v-blog.csdnimg.cn/asset/23da3fe1f67a47106d725406cfde9a97/cover/Cover0.jpg)(title-拼多多店群自动化上架方案)]# 假设登录接口返回 {"code":0, "data":{"token":"eyJhbGciOi..."}}# Token放在响应里# 第二步:后续请求Header中带Token# 影刀HTTP请求指令配置:# Headers: Authorization: Bearer eyJhbGciOi...Token有过期时间(通常几小时到几天),过期后需要重新登录获取。影刀流程里要做成:每次调用前判断Token是否过期,过期则自动重新登录。
方式二:API Key + Secret签名(复杂但安全)
典型场景:抖音开放平台、拼多多开放平台。需要把参数按规则拼接后做MD5/SHA256签名。
importhashlibimporttimedefgenerate_sign(api_key,secret,params):"""生成API签名"""# 1. 把所有参数按key字母排序sorted_params=sorted(params.items())# 2. 拼接成 key1=value1&key2=value2&secretsign_str='&'.join([f'{k}={v}'fork,vinsorted_params])sign_str+=f'&secret={secret}'# 3. MD5加密并转大写returnhashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()# 调用示例(拼多多开放平台风格)params={'type':'pdd.ddk.goods.search','timestamp':int(time.time()),'keyword':'羽绒服','page':1}params['sign']=generate_sign(api_key,secret,params)方式三:Cookie/Session(模拟登录)
典型场景:没有开放API,需要模拟用户登录后爬取数据。
在影刀RPA中就是:浏览器先做登录流程,登录完成后浏览器的Cookie自动维护,后续HTTP请求直接复用。
影刀的"HTTP请求"指令默认复用浏览器的Cookie,登录后直接调接口就行。但要注意Cookie过期时间。
分页的三种处理方式
方式一:页码翻页
# 接口返回:{"total": 5000, "page": 1, "page_size": 50, "data": [...]}total=response['total']page_size=response['page_size']total_pages=(total+page_size-1)//page_size# 向上取整forpageinrange(1,total_pages+1):# 发送请求,参数 page=当前页# 处理数据# 间隔0.5~1秒方式二:游标分页(Cursor/Offset)
# 接口返回:{"data": [...], "has_more": true, "next_cursor": "abc123"}has_more=Truecursor=Nonewhilehas_more:# 请求接口,参数 cursor=当前游标# 处理数据has_more=response['has_more']cursor=response.get('next_cursor')ifnothas_more:break游标分页比页码分页更稳——即使有新数据插入也不会重复或漏数据。
方式三:时间范围批量拉取
importdatetime start_date=datetime.date(2026,1,1)end_date=datetime.date(2026,6,10)current=start_date[video(video-3siRYi7H-1781422019172)(type-csdn)(url-https://live.csdn.net/v/embed/524993)(image-https://v-blog.csdnimg.cn/asset/a547123d88ad712dccba346c9217e237/cover/Cover0.jpg)(title-TEMU店群如何管理运营?)]whilecurrent<=end_date:# 请求接口,参数 start_time=current, end_time=current+6天# 处理一周的数据current+=datetime.timedelta(days=7)适合按时间范围查询的接口。批量拉取比逐日拉高效得多。
错误处理的工程级封装
importtimedefapi_call_with_retry(url,headers,body,max_retries=3):"""带重试的API调用"""retry_count=0base_wait=1# 基础等待秒数whileretry_count<=max_retries:try:# 发送HTTP请求(影刀指令)response=http_post(url,headers,body)# 判断HTTP状态码ifresponse.status_code==200:data=response.json()# 判断业务状态码ifdata.get('code')==0:returndata# 成功elifdata.get('code')in(401,403):# Token过期,需要重新登录refresh_token()continueelifdata.get('code')==429:# 被限流,等待后重试wait=int(response.headers.get('Retry-After',60))time.sleep(wait)continueelifdata.get('code')>=500:# 服务端错误,重试wait=base_wait*(2**retry_count)# 指数退避time.sleep(wait)retry_count+=1continueelse:# 其他业务错误,记录日志后跳过log_error(f"业务错误:{data.get('message')}")returnNoneelifresponse.status_code>=500:# 服务器错误time.sleep(base_wait*(2**retry_count))retry_count+=1continueelse:log_error(f"HTTP错误:{response.status_code}")returnNoneexceptTimeoutError:time.sleep(base_wait*(2**retry_count))retry_count+=1continue# 重试全部失败log_error("API调用重试耗尽")returnNone核心设计思想:
区分HTTP错误和业务错误,不同错误不同策略
指数退避重试(等1秒→2秒→4秒),避免雪崩
Token过期自动刷新,不用人工干预
所有失败都记录日志,方便排查
常见HTTP错误码速查
| 状态码 | 含义 | 应对策略 |
|---|---|---|
| 200 | 成功 | 判断业务code |
| 301/302 | 重定向 | 检查URL是否过期,影刀会自动跟随 |
| 400 | 参数错误 | 检查请求体JSON格式和数据 |
| 401 | 未授权 | Token过期或鉴权信息错误 |
| 403 | 禁止访问 | 权限不足或IP受限 |
| 404 | 接口不存在 | 检查URL拼写 |
| 429 | 被限流 | 降低频率,加长等待 |
| 500+ | 服务器错误 | 重试 |
#影刀RPA #RPA自动化 #API调用 #HTTP请求 #接口开发
作者:林焱
本文为《影刀RPA学习手册》系列文章之一,内容源于实操经验的整理与分享。
