本文目录导读：
Python 脚本（最常用）
使用 requests + 批量处理
import requests
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
# 方法1：顺序批量调用
def batch_call_sequential(api_list):
results = []
for api in api_list:
try:
response = requests.get(api['url'], params=api.get('params'))
results.append({
'api': api['url'],
'status': response.status_code,
'data': response.json()
})
except Exception as e:
print(f"Error: {api['url']} - {str(e)}")
return results
# 方法2：并发批量调用（推荐）
def batch_call_concurrent(api_list, max_workers=5):
def call_single_api(api):
try:
response = requests.get(
api['url'],
params=api.get('params'),
timeout=10
)
return {
'api': api['url'],
'status': response.status_code,
'data': response.json()
}
except Exception as e:
return {'api': api['url'], 'error': str(e)}
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {executor.submit(call_single_api, api): api for api in api_list}
for future in as_completed(futures):
results.append(future.result())
return results
# 使用示例
apis = [
{'url': 'https://api.example.com/users', 'params': {'id': 1}},
{'url': 'https://api.example.com/posts', 'params': {'page': 1}},
{'url': 'https://api.example.com/comments', 'params': {'post_id': 1}}
]
results = batch_call_concurrent(apis)
print(json.dumps(results, indent=2))
Bash Shell 脚本
#!/bin/bash
# API 列表文件格式：URL,METHOD,BODY
apis_file="apis.txt"
# 批量调用函数
batch_call_apis() {
while IFS=',' read -r url method body; do
echo "Calling: $url"
# HTTP GET
if [ "$method" = "GET" ]; then
curl -s -o /dev/null -w "Status: %{http_code}\n" "$url"
# HTTP POST
elif [ "$method" = "POST" ]; then
curl -s -X POST -H "Content-Type: application/json" \
-d "$body" "$url" | jq .
fi
# 添加延迟
sleep 1
done < "$apis_file"
}
# 或者使用 xargs 并发
parallel_batch_call() {
cat "$apis_file" | xargs -I {} -P 5 sh -c '
url=$(echo {} | cut -d"," -f1)
method=$(echo {} | cut -d"," -f2)
echo "Calling: $url"
curl -s "$url" | jq .
'
}
# 执行
batch_call_apis
Node.js 脚本
const axios = require('axios');
const fs = require('fs').promises;
// 批量调用函数
const batchCallAPI = async (apis) => {
const results = [];
// 并发控制
const concurrency = 3;
for (let i = 0; i < apis.length; i += concurrency) {
const batch = apis.slice(i, i + concurrency);
const promises = batch.map(api =>
axios.get(api.url, { params: api.params })
.then(response => ({
url: api.url,
status: response.status,
data: response.data
}))
.catch(error => ({
url: api.url,
error: error.message
}))
);
const batchResults = await Promise.all(promises);
results.push(...batchResults);
}
return results;
};
// 使用示例
const apis = [
{ url: 'https://api.example.com/users', params: { id: 1 } },
{ url: 'https://api.example.com/posts', params: { page: 1 } }
];
batchCallAPI(apis).then(results => {
console.log(JSON.stringify(results, null, 2));
});
高级特性封装
import asyncio
import aiohttp
from typing import List, Dict, Any
import time
class BatchAPICaller:
def __init__(self, max_concurrent=5, retry_times=3):
self.max_concurrent = max_concurrent
self.retry_times = retry_times
self.semaphore = asyncio.Semaphore(max_concurrent)
async def call_single_api(self, session, api_info):
async with self.semaphore:
for attempt in range(self.retry_times):
try:
method = api_info.get('method', 'GET').lower()
async with getattr(session, method)(
api_info['url'],
params=api_info.get('params'),
json=api_info.get('data'),
headers=api_info.get('headers'),
timeout=aiohttp.ClientTimeout(total=10)
) as response:
return {
'url': api_info['url'],
'status': response.status,
'data': await response.json(),
'time': time.time()
}
except Exception as e:
if attempt == self.retry_times - 1:
return {
'url': api_info['url'],
'error': str(e),
'attempts': attempt + 1
}
await asyncio.sleep(1 * (attempt + 1))
async def batch_call(self, apis: List[Dict[str, Any]]):
async with aiohttp.ClientSession() as session:
tasks = [self.call_single_api(session, api) for api in apis]
return await asyncio.gather(*tasks)
def run(self, apis):
return asyncio.run(self.batch_call(apis))
# 使用
caller = BatchAPICaller(max_concurrent=10, retry_times=2)
results = caller.run(apis)
实用工具函数
import pandas as pd
from typing import List, Dict
def load_apis_from_csv(file_path: str) -> List[Dict]:
"""从CSV文件加载API配置"""
df = pd.read_csv(file_path)
return df.to_dict('records')
def save_results(results: List[Dict], file_path: str):
"""保存结果到文件"""
df = pd.DataFrame(results)
df.to_csv(file_path, index=False)
def call_with_rate_limit(api_list, requests_per_second=10):
"""带速率限制的调用"""
import time
from collections import deque
results = []
timestamps = deque()
for api in api_list:
# 速率控制
now = time.time()
while timestamps and now - timestamps[0] > 1:
timestamps.popleft()
if len(timestamps) >= requests_per_second:
sleep_time = 1 - (now - timestamps[0])
if sleep_time > 0:
time.sleep(sleep_time)
# 调用API
response = requests.get(api['url'])
results.append({
'url': api['url'],
'status': response.status_code
})
timestamps.append(time.time())
return results
选择建议
- 数据量小、顺序执行：使用 Bash 脚本
- 需要复杂逻辑处理：使用 Python
- 前端或Node.js环境：使用 JavaScript
- 需要高并发：使用 Python asyncio 或 Node.js
- 需要重试机制：使用封装好的类库
需要我详细解释某个方法或根据你的具体场景定制脚本吗？
