请求编写本地新增CSV文件自动上传至AWS S3 Bucket的脚本方案

核心思路 #

基于你已有的手动上传代码，我们可以通过本地记录文件追踪已处理文件的方式，实现脚本运行时检测新增CSV并自动上传的功能，无需7×24小时驻留运行。

模块实现细节 #

• 文件状态记录：用本地文本文件存储已上传的CSV文件名，每次运行脚本时对比当前文件夹文件与记录，筛选新增文件。
• 新增文件检测：遍历目标文件夹，筛选.csv后缀文件，再与记录文件内容做差集，得到待上传的新增文件列表。
• S3批量上传：复用你现有的上传逻辑，调整为批量处理模式，保留原文件名作为S3对象的Key。

完整代码实现 #

import os
import pandas as pd
from io import StringIO
import boto3

# 配置参数
TARGET_FOLDER = r"你的目标文件夹路径"
S3_BUCKET_NAME = "BucketName"
RECORD_FILE = "uploaded_files.txt"

# 推荐通过环境变量/AWS配置文件加载凭证，避免硬编码密钥
s3 = boto3.client('s3')

def get_uploaded_files():
    """读取已上传的文件记录"""
    if os.path.exists(RECORD_FILE):
        with open(RECORD_FILE, 'r', encoding='utf-8') as f:
            return set(f.read().splitlines())
    return set()

def save_uploaded_files(uploaded_files):
    """更新已上传文件记录"""
    with open(RECORD_FILE, 'w', encoding='utf-8') as f:
        f.write('\n'.join(uploaded_files))

def upload_single_csv(file_path, bucket_name):
    """上传单个CSV到S3，保留原文件名"""
    try:
        df = pd.read_csv(file_path)
        csv_buf = StringIO()
        df.to_csv(csv_buf, header=True, index=False)
        csv_buf.seek(0)
        s3_key = os.path.basename(file_path)
        s3.put_object(Bucket=bucket_name, Body=csv_buf.getvalue(), Key=s3_key)
        print(f"上传成功: {s3_key}")
        return True
    except Exception as e:
        print(f"上传失败 {file_path}: {str(e)}")
        return False

def main():
    uploaded_files = get_uploaded_files()
    # 遍历目标文件夹筛选CSV文件
    current_csv = [os.path.join(TARGET_FOLDER, f) for f in os.listdir(TARGET_FOLDER) 
                   if f.lower().endswith('.csv')]
    # 筛选新增文件
    new_files = [f for f in current_csv if os.path.basename(f) not in uploaded_files]
    
    if not new_files:
        print("无新增CSV文件")
        return
    
    # 批量上传并记录成功项
    success_list = []
    for file in new_files:
        if upload_single_csv(file, S3_BUCKET_NAME):
            success_list.append(os.path.basename(file))
    
    # 更新记录文件
    uploaded_files.update(success_list)
    save_uploaded_files(uploaded_files)
    print(f"本次完成 {len(success_list)} 个文件上传")

if __name__ == "__main__":
    main()

关键注意事项 #

• 凭证安全：切勿在代码中硬编码AWS密钥，建议通过系统环境变量（AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY）或本地~/.aws/credentials配置文件加载，boto3会自动读取这些配置。
• 精准去重：若需避免重名文件误判，可将记录内容改为文件的修改时间或哈希值，提升识别准确性。
• 定期运行：可通过Windows任务计划、Linux cron等系统定时工具，设置脚本周期性运行，实现自动检测上传。

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