Python脚本中Azure Token无法续期引发401错误的技术咨询

核心问题分析 #

你的脚本当前遇到401错误时直接跳过请求，完全未触发renew_token续期逻辑，这是Token无法自动续期的根本原因。另外还有几个潜在问题：

• 用递归重试续Token，处理大量数据时可能触发栈溢出
• 重复复用csv_file_path变量，导致文件路径被覆盖
• 300万条数据循环请求未做节流，容易触发API限流

修改后的完整代码 #

import requests
import csv
import os
import time

# 配置参数
API_BASE_URL = 'https://<WEBSITE HERE>'
GROUP_DETAILS_URL = f"{API_BASE_URL}/api/v2/groups/{{}}/details"
TOKEN_RENEW_URL = f"{API_BASE_URL}/api/v1/tokens/<TOKEN ID HERE>/renew/"
INITIAL_TOKEN = '<AZURE TOKEN HERE>'
INPUT_CSV = os.path.expanduser("C:/Users/FILE.csv")
OUTPUT_CSV = os.path.expanduser("C:/Users/FILE_OUTPUT.CSV")

# 初始化请求头
my_headers = {'Authorization': f'Bearer {INITIAL_TOKEN}'}

def renew_token(max_retries=3):
    """循环重试续期Token，避免递归栈溢出"""
    retry_count = 0
    while retry_count < max_retries:
        try:
            response = requests.post(TOKEN_RENEW_URL, headers=my_headers)
            response.raise_for_status()
            response_json = response.json()
            new_token = response_json['<TOKEN HERE>']
            my_headers['Authorization'] = f'Bearer {new_token}'
            print("Token续期成功")
            return True
        except requests.exceptions.HTTPError as e:
            print(f"续期Token失败 {e.response.status_code}: {e}")
            retry_count += 1
            time.sleep(5)
        except ValueError as e:
            print(f"解析Token续期响应失败: {e}")
            retry_count += 1
            time.sleep(5)
    print("达到最大重试次数，Token续期失败")
    return False

def get_group_details(group_name, max_retries=2):
    """获取群组详情，支持Token过期后重试"""
    retry_count = 0
    while retry_count < max_retries:
        try:
            response = requests.get(GROUP_DETAILS_URL.format(group_name), headers=my_headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 401:
                # 401时尝试续期Token，成功后重试请求
                if renew_token():
                    retry_count += 1
                    continue
                else:
                    print(f"群组 {group_name} 获取失败：无法续期Token")
                    return None
            else:
                print(f"群组 {group_name} 获取失败 {e.response.status_code}: {e}")
                return None
        except ValueError as e:
            print(f"解析群组 {group_name} 响应失败: {e}")
            return None
    print(f"群组 {group_name} 达到最大重试次数，获取失败")
    return None

if __name__ == "__main__":
    # 打开输出CSV文件
    with open(OUTPUT_CSV, 'w', newline='', encoding='utf-8') as csv_file:
        writer = csv.writer(csv_file)
        writer.writerow(['Group Name', 'Owner'])

        # 读取输入群组列表
        with open(INPUT_CSV, 'r', encoding='utf-8') as read_obj:
            csv_reader = csv.reader(read_obj)
            next(csv_reader)  # 跳过表头

            for idx, row in enumerate(csv_reader, 1):
                group_name = row[0].strip()
                if not group_name:
                    continue

                # 每处理100条请求暂停1秒，避免API限流
                if idx % 100 == 0:
                    print(f"已处理 {idx} 个群组，暂停1秒")
                    time.sleep(1)

                group_details = get_group_details(group_name)
                if not group_details:
                    writer.writerow([group_name, ''])
                    continue

                # 提取所有者信息
                owner = group_details.get('owner', {}).get('sAMAccountName', '') if group_details.get('owner') else ''
                writer.writerow([group_name, owner])

    print("所有群组处理完成")

关键修改说明 #

1. 401错误触发续期逻辑：在get_group_details中捕获401错误后，自动调用renew_token续期，成功后重试请求
2. 替换递归为循环重试：renew_token改用while循环，避免处理百万级数据时出现栈溢出
3. 分离输入输出路径：拆分原重复的csv_file_path变量，避免文件路径被覆盖
4. 添加请求节流：每处理100条数据暂停1秒，降低API限流风险
5. 模块化函数：将Token续期、群组详情获取拆分为独立函数，提升代码可维护性
6. 指定文件编码：打开CSV时指定utf-8编码，避免中文乱码问题

额外建议 #

• 处理300万条数据时，建议分批次保存进度，比如每10万条写入一次临时文件，防止程序崩溃导致数据丢失
• 新增失败日志文件，将获取失败的群组单独记录，方便后续补采
• 如果API支持批量查询接口，优先使用批量请求减少调用次数，大幅提升处理效率

内容的提问来源于stack exchange，提问作者Мирослав Бадански