如何实现本地目录文件自动上传至Azure Storage Explorer Blob存储？

当然有可行的方案！针对你的需求，主要有两种主流实现思路，我会结合你现有的代码给出具体示例：

这种方法适合对实时性要求不高的场景，通过定期检查目标目录，发现新文件或修改过的文件就自动上传。核心是用定时任务+本地记录避免重复上传。

实现步骤 #

1. 先安装必要依赖（如果用schedule库简化定时逻辑）：

pip install schedule

2. 结合你现有代码的完整实现：

import os
import time
import schedule
from azure.storage.blob import BlockBlobService, ContentSettings

# 你的Azure配置
accountName = "accountName"
ContainerSAS = "SAS_Key"
containerName = "containerName"

# 本地要监控的目录
watch_dir = "/path/to/your/local/directory"
# 记录已上传文件的本地文件（避免重复上传）
uploaded_record = "uploaded_files.txt"

# 初始化Blob服务
block_blob_service = BlockBlobService(account_name=accountName, sas_token=ContainerSAS)

def load_uploaded_files():
    """加载已上传的文件记录"""
    if not os.path.exists(uploaded_record):
        return set()
    with open(uploaded_record, "r") as f:
        return set(f.read().splitlines())

def save_uploaded_file(file_path):
    """保存已上传的文件记录"""
    with open(uploaded_record, "a") as f:
        f.write(f"{file_path}\n")

def upload_file_to_blob(file_path):
    """单个文件上传到Blob容器"""
    try:
        file_name = os.path.basename(file_path)
        # 可根据文件类型调整内容类型，比如图片用image/jpeg
        content_settings = ContentSettings(content_type="application/octet-stream")
        block_blob_service.create_blob_from_path(
            containerName,
            file_name,
            file_path,
            content_settings=content_settings
        )
        print(f"成功上传文件: {file_path}")
        save_uploaded_file(file_path)
    except Exception as e:
        print(f"上传文件失败 {file_path}: {str(e)}")

def check_and_upload():
    """检查目录中的文件并上传新文件"""
    uploaded_files = load_uploaded_files()
    for root, dirs, files in os.walk(watch_dir):
        for file in files:
            file_path = os.path.join(root, file)
            # 跳过已上传的文件
            if file_path not in uploaded_files:
                upload_file_to_blob(file_path)

# 设置定时任务：比如每5分钟检查一次，可根据需求调整
schedule.every(5).minutes.do(check_and_upload)

# 启动定时任务循环
if __name__ == "__main__":
    print("开始监控目录，等待上传任务...")
    while True:
        schedule.run_pending()
        time.sleep(1)

方案特点 #

• 实现简单，无额外系统依赖
• 通过本地记录文件避免重复上传，也可扩展为记录文件修改时间，支持文件更新后重新上传
• 适合对实时性要求不高（比如间隔几分钟检查一次）的场景

如果需要文件一出现就立刻上传，推荐用watchdog库监听文件系统的创建、修改事件，实时触发上传动作，效率更高。

实现步骤 #

1. 安装依赖：

pip install watchdog

2. 完整实现代码：

import os
import time
from azure.storage.blob import BlockBlobService, ContentSettings
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 你的Azure配置
accountName = "accountName"
ContainerSAS = "SAS_Key"
containerName = "containerName"

# 本地要监控的目录
watch_dir = "/path/to/your/local/directory"
# 记录已上传文件的修改时间（处理更新场景）
uploaded_files = {}

# 初始化Blob服务
block_blob_service = BlockBlobService(account_name=accountName, sas_token=ContainerSAS)

class UploadHandler(FileSystemEventHandler):
    def upload_file(self, file_path):
        """上传/更新文件到Blob容器"""
        try:
            file_name = os.path.basename(file_path)
            mtime = os.path.getmtime(file_path)
            # 如果文件未上传或有修改，则执行上传/更新
            if file_path not in uploaded_files or mtime > uploaded_files[file_path]:
                content_settings = ContentSettings(content_type="application/octet-stream")
                block_blob_service.create_blob_from_path(
                    containerName,
                    file_name,
                    file_path,
                    content_settings=content_settings
                )
                uploaded_files[file_path] = mtime
                print(f"成功上传/更新文件: {file_path}")
        except Exception as e:
            print(f"处理文件失败 {file_path}: {str(e)}")

    def on_created(self, event):
        """监听文件创建事件"""
        if not event.is_directory:
            # 等待文件完全写入（避免上传不完整的临时文件）
            time.sleep(1)
            self.upload_file(event.src_path)

    def on_modified(self, event):
        """监听文件修改事件"""
        if not event.is_directory:
            self.upload_file(event.src_path)

if __name__ == "__main__":
    # 可选：重启脚本后加载已上传文件的记录，避免重复上传
    # 可参考方案1的逻辑从本地文件加载uploaded_files
    
    event_handler = UploadHandler()
    observer = Observer()
    # recursive=True表示监控子目录
    observer.schedule(event_handler, watch_dir, recursive=True)
    observer.start()
    print(f"开始实时监控目录: {watch_dir}")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

方案特点 #

• 实时性强，文件创建/修改后立刻触发上传
• 支持文件更新后重新上传，无需等待定时轮询
• 需要处理文件写入未完成的情况（比如加短暂延时），避免上传不完整的文件

• 文件类型适配：可以扩展ContentSettings的逻辑，根据文件后缀自动设置对应的MIME类型（比如.txt设为text/plain，.jpg设为image/jpeg）
• 异常重试：可以给上传函数添加重试逻辑，处理网络波动导致的上传失败
• 后台运行：如果需要长期运行脚本，可以把它部署成系统服务（Linux用systemd，Windows用任务计划或服务）
• 大文件优化：针对大文件，可使用Azure Blob的分块上传功能，或者改用异步上传库提升效率

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