Python实现OneDrive每日文件上传失败自动重试3次的方案咨询

你遇到的Expecting value: line 1 column 1 (char 0)报错为偶发网络波动导致OneDrive分片上传接口返回非JSON格式响应引发的，属于可通过重试解决的临时性故障，以下提供两种实现方案：

方案1：手动实现重试逻辑（无第三方依赖） #

直接对上传核心逻辑加重试控制，自带指数退避等待，无需额外安装库，修改后的完整代码如下：

import time
if __name__ == '__main__':
    reg_manager = RegistryManager()
    keys = reg_manager.get_keys()
    max_retries = 3
    upload_success = False
    # 前置逻辑：数据导出、打包仅执行一次，避免重复导出浪费资源
    try:
        tables = BCPManager.load_tables()
        dir_name = datetime.today().strftime('%Y-%m-%d')
        dir_path = f'./files/{dir_name}'
        final_file = 'tables.zip'
        final_file_path = f'{dir_path}/{final_file}'
        FileManager.create_folder(dir_path)
        for table in tables:
            table_columns = BCPManager.get_table_columns(table, dir_path)
            table_data = BCPManager.get_table_data(table, dir_path, 3)
            table_path = f'{dir_path}/{table}.csv'
            FileManager.merge_table_files(table_path, table_columns, table_data)
        dir_files = FileManager.get_directory_files(dir_path)
        with ZipFile(
                final_file_path,
                'w',
                compression=ZIP_DEFLATED,
                compresslevel=9
        ) as zip_file:
            for dir_file in dir_files:
                if dir_file.name == final_file:
                    continue
                print(f'zipping file {dir_file.path}')
                zip_file.write(dir_file.path, arcname=dir_file.name)
                FileManager.remove_file(dir_file.path)
        file_size = os.path.getsize(final_file_path)
    except Exception as preprocess_exception:
        print('数据导出或打包失败，无法执行上传')
        print(preprocess_exception)
        # 打包失败直接发告警，无需重试
        message = Mail(
            from_email='test@outlook.com',
            to_emails=['test@gmail.com'],
            subject='OneDrive upload failed',
            html_content='数据导出打包环节出错，无法执行上传')
        try:
            sg = SendGridAPIClient(keys['sendgrid_api_key'])
            response = sg.send(message)
            print(response.status_code)
            print(response.body)
            print(response.headers)
        except Exception as e:
            print(e)
        exit()
    # 上传逻辑加重试
    for retry_count in range(max_retries):
        try:
            print(f'开始第{retry_count+1}次上传尝试')
            uploader = OneDriveUploader(
                'drives/{DriveID}/items/{ChildreID}',
                {
                    'client_id': keys['client_id'],
                    'client_secret': keys['client_secret']
                }
            )
            # 先清理之前可能残留的上传残次文件/文件夹
            uploader.delete_folder(dir_name)
            uploader.create_folder(dir_name)
            print('uploading file')
            with open(final_file_path, 'rb') as open_file:
                print('getting file chunks')
                chunks = FileManager.get_file_chunks(open_file)
                index = 0
                offset = 0
                headers = dict()
                upload_url = uploader.get_upload_link(dir_name, final_file)
                for chunk in chunks:
                    offset = index + len(chunk)
                    headers['Content-Type'] = 'application/octet-stream'
                    headers['Content-Length'] = str(file_size)
                    headers['Content-Range'] = f'bytes {index}-{offset - 1}/{file_size}'
                    uploader.upload_file_chunk(upload_url, chunk, headers)
                    index = offset
            # 上传成功
            print('上传成功，清理本地文件')
            FileManager.remove_file(final_file_path)
            upload_success = True
            break
        except Exception as uploader_exception:
            print(f'第{retry_count+1}次上传失败：{uploader_exception}')
            # 最后一次重试失败才发告警
            if retry_count == max_retries -1:
                break
            # 重试前等待，指数退避，避免频繁请求触发限流
            wait_time = 2 ** (retry_count + 1)
            print(f'等待{wait_time}秒后重试')
            time.sleep(wait_time)
    if not upload_success:
        print("今日上传任务失败")
        message = Mail(
            from_email='test@outlook.com',
            to_emails=['test@gmail.com'],
            subject='OneDrive upload failed',
            html_content='3次重试均失败，今日上传任务终止')
        try:
            sg = SendGridAPIClient(keys['sendgrid_api_key'])
            response = sg.send(message)
            print(response.status_code)
            print(response.body)
            print(response.headers)
        except Exception as e:
            print(e)

注意事项 #

• 代码中调用了uploader.delete_folder(dir_name)，如果你的OneDriveUploader类没有实现该方法，可以替换为对应的删除逻辑，避免上一次失败残留的半传文件导致重传冲突
• 数据导出、打包逻辑仅执行一次，避免重试时重复导出数据库数据浪费资源
• 重试等待采用指数退避策略，降低接口限流风险

方案2：使用tenacity库实现（代码更简洁） #

如果可以安装第三方库，使用tenacity可以更便捷实现重试逻辑：

1. 先安装库：pip install tenacity
2. 将上传逻辑封装为独立函数，加上重试装饰器即可

from tenacity import retry, stop_after_attempt, wait_exponential
# 封装上传函数
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def run_upload(final_file_path, dir_name, final_file, file_size, keys):
    uploader = OneDriveUploader(
        'drives/{DriveID}/items/{ChildreID}',
        {
            'client_id': keys['client_id'],
            'client_secret': keys['client_secret']
        }
    )
    uploader.delete_folder(dir_name)
    uploader.create_folder(dir_name)
    with open(final_file_path, 'rb') as open_file:
        chunks = FileManager.get_file_chunks(open_file)
        index = 0
        offset = 0
        headers = dict()
        upload_url = uploader.get_upload_link(dir_name, final_file)
        for chunk in chunks:
            offset = index + len(chunk)
            headers['Content-Type'] = 'application/octet-stream'
            headers['Content-Length'] = str(file_size)
            headers['Content-Range'] = f'bytes {index}-{offset - 1}/{file_size}'
            uploader.upload_file_chunk(upload_url, chunk, headers)
            index = offset

然后在主逻辑中调用该函数，捕获重试耗尽的异常发送告警即可。

内容的提问来源于stack exchange，提问作者Cristian Blanco