如何检测存储桶新增文件 用Python将CEF转JSON存储并记录流位置

新增文件检测实现 #

你可以根据自己的技术栈选以下两种方案：

• 事件通知方案（适配实时流场景优先选）：目前主流云厂商的对象存储服务（AWS S3、阿里云OSS、腾讯云COS等）都支持配置对象上传事件通知，你可以将新增文件的事件通知推送到配套的消息队列服务中，用Python编写消费端监听消息队列即可实时获取新增文件的路径、大小、修改时间等元数据，无需自行遍历存储桶，延迟最低也不会出现漏检。拿到新增文件的Key后，你就可以通过SDK拉取文件内容，完成CEF到JSON的格式转换，再写入目标存储桶即可。
• 定时轮询比对方案：如果无法使用事件通知能力，你可以自行实现轮询逻辑：固定间隔遍历存储桶指定前缀下的所有文件，将文件的Key（文件路径）、LastModified（最后修改时间）、ETag（文件内容哈希标识）三个字段和上一次的记录做比对，新出现的Key、或者三个字段有更新的文件即为待处理的新增/更新文件。注意文件量较大时要调用SDK的分页遍历接口，避免一次性拉取全量文件导致内存溢出。

处理位点存储实现 #

没有通用的内置流位置键，你可以自行实现轻量的位点存储能力，常用的存储方式有两种：

• 对象存储存位点文件：你可以在源桶或者目标桶中单独创建一个隐藏的位点文件，比如命名为.last_processed_site.json，每次处理完一批文件后，将最后处理的文件的修改时间、文件Key等信息更新到该文件中，下次轮询或者服务重启时，先读取该位点文件的内容，仅拉取修改时间晚于位点记录时间的文件即可，无需重复遍历历史文件。
  Python实现示例（以AWS S3为例，其他云厂商替换为对应SDK即可）：

import json
import boto3

s3_client = boto3.client("s3")
# 位点文件存储的桶名，也可以用你的源桶或者目标桶
SITE_STORAGE_BUCKET = "your-bucket-name"
# 位点文件的Key，加.前缀默认会在控制台文件列表中隐藏，避免误删
SITE_FILE_KEY = ".last_processed_site.json"

# 读取上次处理位点
def load_process_site():
    try:
        resp = s3_client.get_object(Bucket=SITE_STORAGE_BUCKET, Key=SITE_FILE_KEY)
        return json.loads(resp["Body"].read().decode("utf-8"))
    except s3_client.exceptions.NoSuchKey:
        # 首次运行无位点记录，返回初始值
        return {"last_processed_time": "1970-01-01T00:00:00Z", "last_processed_key": ""}

# 更新处理位点
def save_process_site(site_data: dict):
    s3_client.put_object(
        Bucket=SITE_STORAGE_BUCKET,
        Key=SITE_FILE_KEY,
        Body=json.dumps(site_data, ensure_ascii=False).encode("utf-8"),
        ContentType="application/json"
    )

• 本地/轻量数据库存位点：如果是单节点运行的转换程序，也可以直接将位点存在本地JSON文件、SQLite数据库中，实现逻辑更简单，多节点部署时要注意加分布式锁避免位点写入冲突。

额外建议：处理文件前先判断目标桶是否已经存在转换完成的JSON文件，保证处理逻辑幂等，避免重复处理导致重复写入。

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