求助：如何实现Python登录站点拉取私有文件并上传至S3

核心问题 #

原有脚本直接拉取公开URL的文件流，没有处理站点登录态。你要对接的LDC站点基于Spree用户体系搭建，文件访问强制要求登录，且登录接口带动态CSRF校验，直接硬发POST登录请求会被服务端拦截。

改造逻辑 #

• 复用同一个requests.Session实例完成所有站点请求，自动维护会话cookie，保证登录态在后续文件拉取时自动生效
• 每次执行脚本先请求一次登录页，提取当前有效的动态CSRF令牌，按照表单结构拼接全量必填参数完成登录
• 完全保留原有S3流式分片上传的性能配置，不改动上传逻辑

运行前置依赖 #

执行前安装所需依赖包：
pip install requests boto3 beautifulsoup4

改造后可直接运行的代码 #

import threading
import requests
import boto3
from boto3.s3.transfer import TransferConfig
from bs4 import BeautifulSoup

def multi_part_upload_with_s3():
    # 按实际信息替换以下配置项
    LOGIN_PAGE_URL = "https://catalog.ldc.upenn.edu/login"
    LOGIN_SUBMIT_API = "https://catalog.ldc.upenn.edu/login"
    LDC_ACCOUNT = "你的站点登录邮箱"
    LDC_PASSWORD = "你的站点登录密码"
    TARGET_FILE_URL = "需要拉取的非公开文件直链"
    S3_BUCKET_NAME = "你的S3桶名称"
    S3_OBJECT_KEY = "examplePath/file"

    # 初始化带会话管理的请求客户端
    site_session = requests.Session()
    # 第一步：拉取登录页获取动态CSRF令牌
    login_page_resp = site_session.get(LOGIN_PAGE_URL, timeout=10)
    login_page_resp.raise_for_status()
    page_soup = BeautifulSoup(login_page_resp.text, "html.parser")
    csrf_token = page_soup.find("input", {"name": "authenticity_token"})["value"]

    # 第二步：按表单要求拼接全量登录参数，不能缺字段
    login_form_payload = {
        "utf8": "✓",
        "authenticity_token": csrf_token,
        "spree_user[login]": LDC_ACCOUNT,
        "spree_user[password]": LDC_PASSWORD,
        "spree_user[remember_me]": "1",
        "commit": "Login"
    }
    login_resp = site_session.post(LOGIN_SUBMIT_API, data=login_form_payload, timeout=10, allow_redirects=True)
    # 校验登录结果：返回页仍存在密码输入框则说明登录失败
    if "spree_user_password" in login_resp.text:
        raise RuntimeError("站点登录失败，请检查账号密码是否匹配")

    # 第三步：用同一会话拉取目标文件流，保持stream模式不占用本地内存
    file_resp = site_session.get(TARGET_FILE_URL, stream=True, timeout=30)
    file_resp.raise_for_status()
    # 兼容gzip压缩返回流，避免上传时连接报错
    file_resp.raw.decode_content = True

    # 复用原有S3分片上传逻辑
    upload_config = TransferConfig(
        multipart_threshold=1024 * 25,
        max_concurrency=30,
        multipart_chunksize=1024 * 25,
        use_threads=True
    )
    s3_session = boto3.Session()
    s3_client = s3_session.resource('s3')
    target_bucket = s3_client.Bucket(S3_BUCKET_NAME)
    target_bucket.upload_fileobj(file_resp.raw, S3_OBJECT_KEY, Config=upload_config, Callback=None)

注意事项 #

• 登录请求和文件拉取必须使用同一个requests.Session实例，跨实例请求会丢失会话cookie，导致文件请求被重定向回登录页
• 不要硬编码登录页HTML里展示的示例authenticity_token，该值是服务端每次请求动态生成的临时值，硬编码会触发CSRF校验失败
• 保持文件请求的stream=True配置，大文件直接流式传输到S3，不占用本地磁盘和内存，维持原有上传性能

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