影刀RPA API接口调用实战带认证的请求作者林焱什么情况用你用影刀RPA对接某个系统的API但接口需要Token认证Token有过期时间需要定期刷新不同接口需要不同的认证方式——有的用Bearer Token、有的用API Key、有的需要签名API认证是调用第三方接口的必经之路。没有正确的认证接口会返回401未授权。影刀RPA在采集数据时很多现代网站的数据接口都需要认证才能访问。本文讲清楚四种常见API认证方式的实现、Token自动刷新机制、签名认证的原理。怎么做一、四种常见认证方式认证方式位置典型格式适用场景Bearer TokenAuthorization头Authorization: Bearer xxxJWT、OAuth2API Key请求头或参数X-API-Key: xxx开放API| Basic Auth | Authorization头 |Authorization: Basic base64| 简单认证 || 签名认证 | 请求头或参数 |signmd5(paramssecret)| 支付、金融 |二、Bearer Token认证拼多多店群自动化上架方案importrequestsclassBearerTokenClient:Bearer Token认证客户端def__init__(self,base_url,tokenNone):self.base_urlbase_url self.sessionrequests.Session()iftoken:self.session.headers[Authorization]fBearer{token}defset_token(self,token):设置Tokenself.session.headers[Authorization]fBearer{token}defget(self,path,paramsNone):responseself.session.get(f{self.base_url}{path},paramsparams)returnself._handle(response)defpost(self,path,json_dataNone):responseself.session.post(f{self.base_url}{path},jsonjson_data)returnself._handle(response)def_handle(self,response):ifresponse.status_code401:return{success:False,error:Token无效或过期}elifresponse.status_code200:return{success:True,data:response.json()}else:return{success:False,error:fHTTP{response.status_code},data:response.text}# 使用clientBearerTokenClient(https://api.example.com,tokeneyJhbGciOiJIUzI1NiI...)resultclient.get(/users/me)三、Token自动刷新importtimeimportrequestsclassAutoRefreshTokenClient:带Token自动刷新的客户端def__init__(self,base_url,client_id,client_secret):self.base_urlbase_url self.client_idclient_id self.client_secretclient_secret self.sessionrequests.Session()self.tokenNoneself.token_expire_time0# Token过期时间戳def_ensure_token(self):确保Token有效current_timetime.time()# Token不存在或即将过期提前60秒刷新ifnotself.tokenorcurrent_timeself.token_expire_time-60:self._refresh_token()def_refresh_token(self):刷新Tokenprint(刷新Token...)responserequests.post(f{self.base_url}/oauth/token,data{grant_type:client_credentials,client_id:self.client_id,client_secret:self.client_secret,})ifresponse.status_code200:dataresponse.json()self.tokendata[access_token]expires_indata.get(expires_in,3600)# 默认1小时self.token_expire_timetime.time()expires_in self.session.headers[Authorization]fBearer{self.token}print(fToken刷新成功有效期{expires_in}秒)else:raiseException(fToken刷新失败:{response.status_code}{response.text})defrequest(self,method,path,**kwargs):发送请求自动处理Token刷新self._ensure_token()urlf{self.base_url}{path}responseself.session.request(method,url,**kwargs)# 如果返回401强制刷新Token后重试一次ifresponse.status_code401:print(收到401尝试刷新Token后重试...)self._refresh_token()responseself.session.request(method,url,**kwargs)returnresponse# 使用clientAutoRefreshTokenClient(base_urlhttps://api.example.com,client_idyour_client_id,client_secretyour_client_secret)# 每次请求自动检查Token是否有效responseclient.request(GET,/data/list,params{page:1})print(response.json())四、API Key认证classAPIKeyClient:API Key认证客户端def__init__(self,base_url,api_key,key_headerX-API-Key):self.base_urlbase_url self.sessionrequests.Session()self.session.headers[key_header]api_key self.session.headers[Content-Type]application/jsondefget(self,path,paramsNone):responseself.session.get(f{self.base_url}{path},paramsparams)returnresponse.json()defpost(self,path,dataNone):responseself.session.post(f{self.base_url}{path},jsondata)returnresponse.json()# 不同的API Key传递方式# 方式1请求头client1APIKeyClient(https://api.example.com,key_xxx,X-API-Key)# 方式2Authorization头client2APIKeyClient(https://api.example.com,key_xxx,Authorization)# 方式3URL参数classAPIKeyURLClient:def__init__(self,base_url,api_key,key_paramapi_key):self.base_urlbase_url self.api_keyapi_key self.key_paramkey_paramdefget(self,path,paramsNone):paramsparamsor{}params[self.key_param]self.api_key responserequests.get(f{self.base_url}{path},paramsparams)returnresponse.json()client3APIKeyURLClient(https://api.example.com,key_xxx,api_key)五、Basic Auth认证fromrequests.authimportHTTPBasicAuthimportbase64# 方式1用requests的auth参数responserequests.get(https://api.example.com/data,authHTTPBasicAuth(username,password))# 方式2手动构造Authorization头credentialsbase64.b64encode(fusername:password.encode()).decode()headers{Authorization:fBasic{credentials}}responserequests.get(https://api.example.com/data,headersheaders)# 方式3简单写法requests直接支持responserequests.get(https://username:passwordapi.example.com/data)六、签名认证很多中国互联网公司的API使用签名认证核心思路是把请求参数密钥做MD5/SHA256哈希importhashlibimporttimeimportrequestsimporturllib.parseclassSignatureClient:签名认证客户端def__init__(self,base_url,app_id,app_secret):self.base_urlbase_url self.app_idapp_id self.app_secretapp_secretdef_generate_sign(self,params): 生成签名 签名规则常见模式 1. 参数按key排序 2. 拼接成 key1value1key2value2 格式 3. 末尾拼接 secret 4. MD5哈希转大写 # 过滤空值和sign本身filtered{k:vfork,vinparams.items()ifvisnotNoneandk!sign}# 按key排序sorted_keyssorted(filtered.keys())# 拼接sign_str.join(f{k}{filtered[k]}forkinsorted_keys)# 加上secretsign_strfkey{self.app_secret}# MD5哈希signhashlib.md5(sign_str.encode(utf-8)).hexdigest().upper()returnsigndefrequest(self,method,path,paramsNone,dataNone):发送签名请求paramsparamsor{}# 添加公共参数params[app_id]self.app_id params[timestamp]str(int(time.time()))params[nonce]str(int(time.time()*1000))# 简单用时间戳做随机数# 生成签名params[sign]self._generate_sign(params)urlf{self.base_url}{path}ifmethodGET:responserequests.get(url,paramsparams)else:responserequests.post(url,data{**params,**(dataor{})})returnresponse.json()# 使用clientSignatureClient(base_urlhttps://api.example.com,app_idyour_app_id,app_secretyour_app_secret)resultclient.request(GET,/data/list,params{page:1,size:20})print(result)七、实战对接电商开放平台importhashlibimporthmacimporttimeimportrequestsimportjsonclassEcommerceAPI:电商开放平台API客户端def__init__(self,app_key,app_secret,base_urlhttps://openapi.example.com):self.app_keyapp_key self.app_secretapp_secret self.base_urlbase_urldef_make_request(self,method,paramsNone,bodyNone):构造并发送API请求paramsparamsor{}# 系统参数sys_params{app_key:self.app_key,method:method,timestamp:time.strftime(%Y-%m-%d %H:%M:%S),format:json,v:2.0,sign_method:md5,}# 合并参数all_params{**sys_params,**params}ifbody:all_params[body]json.dumps(body,ensure_asciiFalse)# 生成签名sign_strself.app_secretforkinsorted(all_params.keys()):sign_strf{k}{all_params[k]}sign_strself.app_secret all_params[sign]hashlib.md5(sign_str.encode(utf-8)).hexdigest().upper()# 发送请求responserequests.post(self.base_url,dataall_params)returnresponse.json()defget_order_list(self,statuspaid,page1,size20):获取订单列表returnself._make_request(order.list.get,params{status:status,page:page,page_size:size})defget_product_detail(self,product_id):获取商品详情returnself._make_request(product.detail.get,params{product_id:product_id})defupdate_stock(self,product_id,stock):更新库存returnself._make_request(product.stock.update,params{product_id:product_id},body{stock:stock})# 使用apiEcommerceAPI(app_keyxxx,app_secretyyy)# 获取订单ordersapi.get_order_list(statuspaid,page1)print(f获取到{len(orders.get(items,[]))}个订单)# 更新库存resultapi.update_stock(product_idP12345,stock200)print(f更新结果:{result})有什么坑坑1Token过期但不知道现象Token用了一段时间后所有请求都返回401。原因Token有有效期通常1-24小时过期后需要重新获取。很多新手只在流程开始时获取一次Token没有刷新机制。解决实现Token自动刷新——记录Token的过期时间每次请求前检查是否即将过期。收到401时自动刷新Token并重试。参考上面的AutoRefreshTokenClient。坑2签名算法和文档不一致现象按照文档写的签名算法但服务器返回签名错误。原因不同平台的签名规则细节不同——参数排序方式、编码方式、拼接顺序、大小写、是否包含空值参数等一个细节不对就会签名失败。TEMU店群如何管理运营解决仔细阅读API文档的签名规则。用文档提供的示例参数和签名结果验证你的算法是否正确。常见的坑参数值需要URL编码、中文用UTF-8编码、空参数不参与签名、布尔值转成true/false字符串。坑3时间戳导致签名失败现象签名算法没问题但偶尔返回签名错误或请求过期。原因签名中包含时间戳如果客户端和服务器时间差超过允许范围通常5分钟签名就会失效。解决确保客户端时间准确NTP同步。如果服务器在国内客户端时区设为东八区。签名前检查时间戳格式是否符合文档要求秒级还是毫秒级格式是时间戳还是日期字符串。坑4并发请求时Token竞争现象多线程同时发请求时Token刷新逻辑出错部分请求失败。原因多个线程同时检测到Token过期同时发起刷新请求后面的刷新可能使前面的Token失效。解决Token刷新加锁确保同一时间只有一个线程在刷新。其他线程等待刷新完成后再使用新Token。在影刀中避免在多线程场景下同时触发Token刷新。坑5影刀Python代码块中的requests和影刀HTTP指令的Cookie不共享现象在影刀【HTTP请求】指令中登录了但在Python代码块中用requests发请求登录态丢失。原因影刀的【HTTP请求】指令和Python代码块中的requests是两套独立的HTTP客户端Cookie/Session不共享。解决统一使用一种方式发请求——要么全用影刀的【HTTP请求】指令要么全用Python代码块中的requests。如果必须混用在Python代码块中从影刀变量获取Cookie手动设置到requests的Session中。