Python如何从给定URL的OneDrive仓库按指定文件名下载CSV文件

依赖准备 #

仅需要安装第三方库requests，内置库无需额外安装：

pip install requests

核心功能实现 #

共享链接转可用ID #

OneDrive的公开共享链接不能直接请求文件列表，需要先按微软规则编码为Graph API可识别的共享ID。

完整可运行代码 #

import requests
import base64
import os
import subprocess

def encode_onedrive_share_url(share_url: str) -> str:
    # 按照微软Graph API规则编码共享链接为共享ID
    base64_encoded = base64.b64encode(share_url.encode("utf-8")).decode("utf-8")
    # 修正编码格式
    formatted_id = base64_encoded.rstrip("=").replace("/", "_").replace("+", "-")
    return f"u!{formatted_id}"

def download_target_csv(onedrive_folder_url: str, target_filename: str, save_dir: str = "./") -> str:
    """
    从OneDrive共享文件夹下载指定文件名的CSV到本地
    :param onedrive_folder_url: OneDrive共享文件夹的公开访问链接
    :param target_filename: 要下载的CSV文件名（带后缀）
    :param save_dir: 本地保存路径，默认当前工作目录
    :return: 下载后的本地文件完整路径
    """
    share_id = encode_onedrive_share_url(onedrive_folder_url)
    # 调用Graph API获取共享文件夹下所有文件列表
    list_req_url = f"https://graph.microsoft.com/v1.0/shares/{share_id}/root/children"
    resp = requests.get(list_req_url)
    if resp.status_code != 200:
        raise ConnectionError(f"访问共享文件夹失败，状态码：{resp.status_code}")
    
    # 匹配目标文件
    file_list = resp.json()["value"]
    target_file = next((f for f in file_list if f["name"] == target_filename), None)
    if not target_file:
        raise FileNotFoundError(f"共享文件夹中未找到目标文件：{target_filename}")
    
    # 下载文件到本地
    download_url = target_file["@microsoft.graph.downloadUrl"]
    file_content = requests.get(download_url).content
    local_save_path = os.path.join(save_dir, target_filename)
    with open(local_save_path, "wb") as f:
        f.write(file_content)
    return local_save_path

def run_auto_analysis(onedrive_share_url: str, csv_file_list: list, main_script_path: str = "your_main_script.py"):
    """
    自动化执行下载->分析->删除的全流程
    :param onedrive_share_url: OneDrive共享文件夹链接
    :param csv_file_list: 所有需要处理的CSV文件名列表
    :param main_script_path: 你的主分析脚本的本地路径
    """
    for csv_name in csv_file_list:
        local_csv_path = ""
        try:
            # 第一步：下载CSV
            local_csv_path = download_target_csv(onedrive_share_url, csv_name)
            print(f"[完成] {csv_name} 已下载到本地")
            # 第二步：运行主分析脚本，根据你的主脚本参数调整命令
            # 若主脚本不需要传参，直接去掉后面的local_csv_path参数即可
            subprocess.run(["python", main_script_path, local_csv_path], check=True)
            print(f"[完成] {csv_name} 分析处理完成")
        except Exception as e:
            print(f"[错误] 处理{csv_name}时失败：{str(e)}")
        finally:
            # 第三步：删除本地CSV文件，避免残留
            if local_csv_path and os.path.exists(local_csv_path):
                os.remove(local_csv_path)
                print(f"[完成] {csv_name} 本地文件已清理\n")

# 调用示例
if __name__ == "__main__":
    # 替换为你自己的OneDrive共享文件夹链接
    YOUR_ONEDRIVE_FOLDER_URL = "https://1drv.ms/xxxxxxxxx"
    # 替换为你需要处理的所有CSV文件名
    NEED_PROCESS_CSV = ["title.csv", "data_2024.csv", "report.csv"]
    # 替换为你的主分析脚本路径
    YOUR_MAIN_SCRIPT = "main_analysis.py"

    run_auto_analysis(YOUR_ONEDRIVE_FOLDER_URL, NEED_PROCESS_CSV, YOUR_MAIN_SCRIPT)

注意事项 #

• 上述代码仅适用于公开访问的OneDrive共享文件夹，如果你的共享文件夹需要登录权限，需要先在Azure平台注册应用获取认证凭证，调用MSAL库获取访问令牌后，将令牌加入Graph API请求的Authorization请求头即可
• 如果共享文件夹包含多层子目录，调整Graph API的请求路径为https://graph.microsoft.com/v1.0/shares/{share_id}/root:/{子文件夹路径}:/children即可检索子目录内的文件
• 主分析脚本如果是读取固定路径的CSV文件，不需要传入参数，修改subprocess.run的调用参数为subprocess.run(["python", main_script_path], check=True)即可

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