如何每60分钟暂停while循环刷新Bearer Token以持续API调用？

针对你遇到的Token过期问题，既可以定时刷新Token，也可以结合「定时预刷新+请求失败自动重试」的方案，后者更可靠，能应对Token提前失效、计时偏差等情况。下面是具体优化后的代码和关键点说明：

核心优化点 #

• 记录Token生成时间，每55分钟自动刷新（提前5分钟，避免刚好到点失效）
• 捕获API返回的401 Unauthorized错误，触发Token刷新并重试当前请求
• 增加请求重试机制，处理网络波动导致的临时失败
• 分批保存数据，避免程序崩溃丢失所有已获取的信息

优化后的完整代码 #

import requests
import time
import json
import pandas as pd
from datetime import datetime

# -------------------- 生成Bearer Token ----------------------#
def get_access_token():   
    a = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
    b = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
    payload = {
        'key_id': a,
        'key_secret': b
    }
    response = requests.post(
        'https://api.zdxcloud.net/v1/oauth/token',
        headers={'Content-Type': 'application/json', 'accept': 'application/json'},
        data=json.dumps(payload)
    )
    response.raise_for_status()  # 捕获Token请求失败的情况
    token = response.json()['token']
    return "Bearer " + token

# -------------------- 初始化变量 ----------------------#
accesstoken = get_access_token()
token_create_time = datetime.now()  # 记录Token生成时间

# 读取DeviceID列表
xl = pd.read_json("D:\\Users.json")
deviceid = xl['id'].tolist()
max_id_count = min(20000, len(deviceid))  # 最多处理20000条，避免超出列表长度

data = {'mydata': []}
save_batch_size = 1000  # 每处理1000条保存一次数据

# -------------------- 获取设备详情主逻辑 ----------------------#
print(f"开始时间: {datetime.now()}")

for i in range(max_id_count):
    id = deviceid[i]
    url = f"https://api.zdxcloud.net/v1/users/{id}"
    headers = {
        'Accept': 'application/json',
        'Authorization': accesstoken
    }

    # 检查是否需要刷新Token（每55分钟刷新一次）
    time_diff = (datetime.now() - token_create_time).total_seconds()
    if time_diff > 55 * 60:
        print("Token即将过期，刷新中...")
        accesstoken = get_access_token()
        token_create_time = datetime.now()
        headers['Authorization'] = accesstoken

    # 带重试的请求逻辑
    retry_count = 3
    while retry_count > 0:
        try:
            response = requests.get(url, headers=headers, proxies=proxies)
            response.raise_for_status()  # 抛出HTTP错误
            results = response.json()
            data['mydata'].append(results)
            break
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                # Token失效，刷新后重试
                print("Token已失效，刷新后重试当前请求...")
                accesstoken = get_access_token()
                token_create_time = datetime.now()
                headers['Authorization'] = accesstoken
                retry_count -= 1
            else:
                # 其他HTTP错误，记录并跳过
                print(f"获取ID {id} 失败，错误: {e}")
                break
        except requests.exceptions.RequestException as e:
            # 网络错误，重试
            print(f"网络错误，重试获取ID {id}: {e}")
            retry_count -= 1
            time.sleep(2)  # 重试前等待2秒
    else:
        # 重试3次都失败，记录
        print(f"ID {id} 经过3次重试仍失败，跳过")

    # 分批保存数据
    if (i + 1) % save_batch_size == 0:
        print(f"已处理 {i+1} 条，保存数据中...")
        with open("D:\\SysUserDetails.json", "w") as outfile:
            json.dump(data, outfile, indent=2)

# 最后保存所有数据
print("处理完成，最终保存数据...")
with open("D:\\SysUserDetails.json", "w") as outfile:
    json.dump(data, outfile, indent=2)

print(f"结束时间: {datetime.now()}")

关键说明 #

1. 定时预刷新Token：设置55分钟刷新一次，比Token有效期提前5分钟，避免刚好在请求时Token过期，导致请求失败。
2. 401错误重试：如果API返回401（Token失效），自动刷新Token并重试当前请求，应对Token意外失效的情况。
3. 请求重试机制：针对网络波动等临时错误，最多重试3次，提高程序稳定性。
4. 分批保存数据：每处理1000条就保存一次，避免程序崩溃导致之前的所有数据丢失。
5. 替换while循环为for循环：更直观控制处理的ID数量，避免原代码中i == 20000的逻辑漏洞（比如deviceid长度不足20000时会无限循环）。

内容的提问来源于stack exchange，提问作者Vijay Kumar