影刀RPA进阶教程:API与HTTP请求的4种鉴权方式与分页处理

做RPA久了会发现:能直接调API拿数据,就不要去爬网页。

网页采集要处理登录态、等待加载、捕获元素、翻页、反爬……一套流程写下来至少200行指令。而API调用,发一个请求、收一份JSON、解析一下,数据就到手了,效率完全不在一个量级。

picture.image

我之前帮一个团队做飞书多维表格的读写,一开始想的是"模拟人工操作网页",写了一半发现复杂得离谱。后来换成飞书API,4条HTTP请求就搞定了

这篇文章把影刀RPA里调用API的完整流程讲清楚——GET和POST怎么发、4种鉴权方式分别怎么配、分页数据怎么处理。读完你可以把"爬网页采集"升级成"调API采集"。

picture.image


HTTP请求指令的核心配置

picture.image 在影刀指令面板搜索**"发送HTTP请求"**,这是所有API调用的入口。

配置项作用示例值
请求方式决定操作类型GET / POST / PUT / DELETE

picture.image | URL | API的地址 | https://api.example.com/v1/products | | 请求头 | 告诉服务器你的"身份"和"期望格式" | Authorization: Bearer xxx | | 请求体 | POST/PUT时携带的数据 | {"keyword": "耳机"} | | 超时时间 | 等待响应的最长时间 | 30秒 |

picture.image 最常用的两种请求方式:

GET请求(获取数据):

# 指令:发送HTTP请求

![picture.image](https://p3-volc-community-sign.byteimg.com/tos-cn-i-tlddhu82om/733823cb4a2d47569d71c32136e4ccc9~tplv-tlddhu82om-image.image?=&rk3s=8031ce6d&x-expires=1784181313&x-signature=VwjPqnoapjSaqEPcPfxMIk397T0%3D)
# 请求方式:GET
# URL:https://api.example.com/products?page=1&size=20
# 请求头:{"Content-Type": "application/json"}
# 输出:响应内容(JSON字符串)

picture.image

POST请求(提交数据):

# 指令:发送HTTP请求
# 请求方式:POST

![picture.image](https://p3-volc-community-sign.byteimg.com/tos-cn-i-tlddhu82om/e4cee9171f5b48189134762e8985c6ae~tplv-tlddhu82om-image.image?=&rk3s=8031ce6d&x-expires=1784181313&x-signature=Tm7OUN6XM%2FzcfZmTVZAPQOCh6Ww%3D)
# URL:https://api.example.com/search
# 请求头:{"Content-Type": "application/json", "Authorization": "Bearer xxx"}
# 请求体:{"keyword": "蓝牙耳机", "page": 1}
# 输出:响应内容(JSON字符串)

四种API鉴权方式详解

API不是谁都能调用的。大多数API都需要"验明正身",下面四种是最常见的鉴权方式。

方式一:API Key(最简单,最常用)

原理: 在URL或请求头里带一个密钥字符串,服务器校验密钥是否有效。

# ===== 方式1:在URL里带API Key =====
# 指令:发送HTTP请求
# URL:https://api.example.com/products?api_key=your_api_key_12345

# ===== 方式2:在请求头里带API Key(更规范) =====
# 请求头设置:
#   Content-Type: application/json
#   X-API-Key: your_api_key_12345

影刀里的配置步骤:

picture.image

  1. 在"发送HTTP请求"指令的详情面板中找到**"请求头"**区域
  2. 点击"添加",键填X-API-Key,值填你的API Key
  3. 如果API Key是放在URL里的,直接在URL后面加?api_key=xxx

适用场景: 大部分公开API(天气、汇率、物流查询等)。

方式二:Bearer Token(标准OAuth2方式)

原理: 先通过账号密码获取一个"临时通行证"(Token),后续所有请求都带着这个Token。Token有有效期(通常1-2小时),过期后需要重新获取。

# ===== 步骤1:获取Token =====
# POST https://api.example.com/auth/login
# 请求体:{"username": "admin", "password": "123456"}
# 响应:{"access_token": "eyJhbGciOiJIUzI1NiIs..."}

# ===== 步骤2:调用业务接口时带Token =====
# 请求头设置:
#   Authorization: Bearer eyJhbGciOiJIUzI1NiIs...
#   Content-Type: application/json

影刀里的配置步骤(两步走):

第一步,先发一个POST请求获取Token:

# 指令:发送HTTP请求
# 请求方式:POST

![picture.image](https://p3-volc-community-sign.byteimg.com/tos-cn-i-tlddhu82om/4b4e28d1aa0a47c38269d328bb98f9bc~tplv-tlddhu82om-image.image?=&rk3s=8031ce6d&x-expires=1784181313&x-signature=A8CygKaQT1LhD1I7tT4a5lQuYeY%3D)
# URL:https://api.example.com/auth/login
# 请求体:{"username": "xxx", "password": "xxx"}
# 输出变量:login_response(JSON字符串)

第二步,用"执行Python代码"解析Token并保存:

import json
response = json.loads(login_response)
token = response.get("access_token")
# token可以存成全局变量供后续请求使用

第三步,后续请求的请求头里带上Token:

# 请求头:
#   Authorization: Bearer {token}
#   Content-Type: application/json

Token过期的处理: 在调用接口的Try-Catch里判断,如果返回"401 Unauthorized",重新执行"获取Token"步骤,然后重试原请求。

方式三:Basic Auth(用户名+密码编码)

原理: 用户名和密码用Base64编码后放在请求头里。安全性较低,建议只在HTTPS环境下使用。

# 用户名:admin,密码:123456
# 拼接字符串 "admin:123456" → Base64编码 → "YWRtaW46MTIzNDU2"
# 请求头:Authorization: Basic YWRtaW46MTIzNDU2

影刀里的配置方式: 在"发送HTTP请求"指令的详情面板中,有专门的**"认证方式"**下拉选项,选"Basic Auth",然后填入用户名和密码,影刀会自动做Base64编码。

方式四:动态签名(最复杂,少用)

原理: 把请求参数按一定规则拼接,加上密钥后做哈希计算,生成一个签名串。服务器收到后按同样的规则验签。

场景: 淘宝开放平台、拼多多开放平台等电商API基本都是这种。

# Python代码:生成签名(示例)
import hashlib
import time

def generate_sign(params, secret):
    # 参数按字典序排序
    sorted_keys = sorted(params.keys())
    sign_str = secret
    for key in sorted_keys:
        sign_str += key + str(params[key])
    sign_str += secret
    # MD5加密
    return hashlib.md5(sign_str.encode()).hexdigest()

# 使用
params = {"method": "taobao.item.get", "timestamp": int(time.time())}
sign = generate_sign(params, "your_app_secret")
params["sign"] = sign

这种鉴权方式在影刀里没有"一键配置",需要用"执行Python代码"指令单独处理签名逻辑。


分页数据处理:四种方式

API返回的数据通常不会一次性全给你,而是分页返回。四种分页方式,处理逻辑各不相同。

方式一:页码分页(最常用)

特征: 参数里有pagesize,响应里有total总条数。

# ============================================
# 执行Python代码:循环获取所有页数据
# 输入:base_url(文本),page_size(数字)
# 输出:all_data(二维列表)
# ============================================

import requests
import json

def fetch_all_pages(base_url, page_size=100):
    all_data = []
    page = 1
    total = None
    
    while True:
        # 构造分页URL
        url = f"{base_url}?page={page}&size={page_size}"
        
        # 发送请求(这里假设在影刀里已拿到响应,直接用)
        response = json.loads(response_text)  # response_text是从影刀传进来的
        
        # 解析数据
        data = response.get("data", [])
        total = response.get("total", 0)
        
        if not data:
            break
        
        all_data.extend(data)
        
        # 判断是否最后一页
        if len(all_data) >= total:
            break
        
        page += 1
    
    return all_data

方式二:游标分页(Cursor)

特征: 响应里有一个next_cursorlast_id,下一次请求带上它才能拿到下一页。

# ============================================
# 执行Python代码:游标分页循环
# 输入:base_url(文本)
# 输出:all_data(二维列表)
# ============================================

import json

def fetch_with_cursor(base_url):
    all_data = []
    cursor = None
    has_more = True
    
    while has_more:
        # 构造URL,带上游标
        url = base_url
        if cursor:
            url += f"?cursor={cursor}"
        
        # 解析响应
        response = json.loads(response_text)
        
        data = response.get("data", [])
        all_data.extend(data)
        
        # 读取下一页游标
        cursor = response.get("next_cursor")
        has_more = response.get("has_more", False)
        
        # 如果没有更多数据,退出
        if not has_more or not cursor:
            break
    
    return all_data

方式三:时间范围分页

特征: 按时间段分批拉取,比如每次拉取1天的数据,循环处理30天。

# 按日期循环,从2026-07-01到2026-07-30
from datetime import datetime, timedelta

start_date = datetime(2026, 7, 1)
end_date = datetime(2026, 7, 30)
current_date = start_date

all_data = []

while current_date <= end_date:
    date_str = current_date.strftime("%Y-%m-%d")
    url = f"{base_url}?date={date_str}"
    
    # 拉取该日数据
    response = json.loads(response_text)
    all_data.extend(response.get("data", []))
    
    current_date += timedelta(days=1)

print(f"共获取 {len(all_data)} 条数据")

方式四:连续请求直到数据为空

特征: 没有总量、没有下一页标识,一直请求直到返回的数据为空

all_data = []
page = 1

while True:
    url = f"{base_url}?page={page}"
    response = json.loads(response_text)
    data = response.get("data", [])
    
    if not data:
        break
    
    all_data.extend(data)
    page += 1

API响应解析与错误处理

API返回的通常是JSON格式,需要用"执行Python代码"解析。

# ============================================
# 执行Python代码:解析API响应
# 输入:api_response(文本),JSON字符串
# 输出:parsed_data(字典/列表),error_msg(文本)
# ============================================

import json

def parse_api_response(api_response):
    try:
        data = json.loads(api_response)
        
        # 检查是否成功(不同API格式不同)
        code = data.get("code")
        if code == 0 or code == 200:
            # 成功
            result = data.get("data", {})
            error_msg = ""
            return result, error_msg
        else:
            # 业务错误
            error_msg = data.get("msg", "未知错误")
            return None, error_msg
            
    except json.JSONDecodeError:
        return None, "JSON解析失败,响应不是合法JSON"

parsed_data, error_msg = parse_api_response(api_response)

if error_msg:
    # 影刀里判断error_msg是否为空
    输出日志("API调用失败:" + error_msg)
else:
    输出日志("API调用成功,数据:" + parsed_data)

在影刀里的完整流程:

Try:
    # 1. 发送HTTP请求
    发送HTTP请求(...) → 存入「api_response」
    
    # 2. 解析响应
    执行Python代码(解析逻辑, 输入=api_response) → 输出「parsed_data」和「error_msg」
    
    # 3. 判断是否成功
    如果 error_msg != "":
        输出日志("【API错误】" + error_msg)
        发送告警通知("API调用失败")
        终止流程()
    否则:
        输出日志("API调用成功,获取到" + 列表长度(parsed_data) + "条数据")
        写入Excel(parsed_data)

Catch:
    输出日志("【HTTP异常】" + 错误信息)
    截图保存("api_error.png")

完整案例:调用API获取商品数据并入库

# ============================================
# 完整流程:调用电商API获取商品数据,存入SQLite
# ============================================

# 步骤1:获取Token(假设需要Bearer Token认证)
发送HTTP请求(
    方式=POST,
    URL="https://api.shop.com/auth/login",
    请求体={"username": "admin", "password": "pwd123"}
) → 存入「login_response」

执行Python代码(解析Token, 输入=login_response) → 输出「access_token」

# 步骤2:循环分页获取商品数据
所有商品 = []
当前页 = 1
每页数量 = 50
是否还有下一页 = True

循环 当 是否还有下一页 == True:
    # 发送带Token的请求
    发送HTTP请求(
        方式=GET,
        URL=f"https://api.shop.com/products?page={当前页}&size={每页数量}",
        请求头={"Authorization": "Bearer " + access_token}
    ) → 存入「page_response」
    
    # 解析数据
    执行Python代码(解析商品列表, 输入=page_response) → 输出「商品列表」和「总条数」
    
    # 追加到总数据
    所有商品.extend(商品列表)
    
    # 判断是否最后一页
    如果 列表长度(所有商品) >= 总条数:
        是否还有下一页 = False
    否则:
        当前页 = 当前页 + 1

# 步骤3:写入SQLite
调用子流程(写入SQLite, 数据=所有商品)

# 步骤4:发送通知
发送飞书通知(f"采集完成,共{len(所有商品)}条商品数据")

API调用参数配置清单

配置项推荐值原因
超时时间30秒API响应可能受网络影响,设长一点容错
重试次数2次网络抖动时自动重试
请求间隔0.5秒避免触发API限流
单次分页大小50-100条太大容易超时,太小请求次数多
Token刷新在Catch里检测401并刷新Token过期自动续期
响应大小限制10MB以内太大可能导致解析卡顿

容易踩的坑速查表

现象排查方向修复方法
返回401 UnauthorizedToken过期或无效重新获取Token后重试
返回403 Forbidden权限不足或IP被拉黑检查API Key是否有对应权限
返回429 Too Many Requests触发限流降低请求频率,加间隔等待
JSON解析报错响应不是JSON(可能是HTML错误页)先打印响应内容,看是不是返回了HTML
分页只有第一页的数据没有正确处理分页循环检查是否在循环里更新了页码或游标
请求成功但数据为空日期或参数不对检查请求参数格式是否正确
中文参数乱码编码问题URL里的中文需要用URL编码
Token获取成功但业务接口报错Token没有对应的资源权限检查Token的scope/权限范围

一条核心建议

能用API就不要爬网页。 API采集比网页采集稳定10倍,速度也快10倍。

判断一个平台能不能用API的标准:

  1. 去搜"XXX开放平台"或"XXX开发者文档"
  2. 看有没有"API文档"和"App Key申请"入口
  3. 如果有 → 优先用API
  4. 如果没有 → 再考虑网页采集

API调用的维护成本极低——页面改版、class变动、XPath失效这些事在API世界里完全不存在。只需要关心参数格式和鉴权方式,其他都稳定。

如果平台API需要企业认证才能申请,也不要放弃。很多平台的API申请流程比想象中简单(拼多多、淘宝的开放平台,只要有店铺就能申请),花一周申请通过,省一年的爬虫维护时间


推荐资源

  • Postman(API调试工具):调试API接口必备
  • 影刀社区示例:"API数据采集模板"(搜索"API"可找到相关案例)
  • JSON格式化工具:在线JSON解析、格式化、校验

#影刀RPA #RPA自动化 #API调用 #HTTP请求 #数据采集

作者:林焱

本文为《影刀RPA学习手册》系列文章之一,内容源于实操经验的整理与分享。

0
0
0
0
评论
未登录
暂无评论