upload_excel.py

    import boto3
    import os
    import logging
    from datetime import datetime
    import glob
    
    class DisasterReportUploader:
        def __init__(self, bucket_name, prefix, local_dir):
            """
            初期化
            Args:
                bucket_name (str): S3バケット名
                prefix (str): S3のディレクトリパス（プレフィックス）
                local_dir (str): アップロード元のローカルディレクトリ
            """
            self.bucket_name = bucket_name
            self.prefix = prefix.rstrip('/') + '/'
            self.local_dir = local_dir
            self.s3 = boto3.client('s3')
            
            self.logger = logging.getLogger(__name__)
            self.logger.setLevel(logging.INFO)
            
        def upload_to_s3(self, local_path):
            """
            ファイルをS3にアップロード
            Args:
                local_path (str): アップロードするローカルファイルのパス
            """
            try:
                # ローカルディレクトリからの相対パスを取得（フォルダ構造を維持）
                relative_path = os.path.relpath(local_path, self.local_dir)
                
                # S3のキーを生成（フォルダ階層を維持）
                s3_key = os.path.join(self.prefix, relative_path).replace('\\', '/')
                
                # アップロード前にファイルの存在確認
                try:
                    self.s3.head_object(Bucket=self.bucket_name, Key=s3_key)
                    self.logger.info(f"既存ファイルを上書き: {s3_key}")
                except:
                    self.logger.info(f"新規ファイルをアップロード: {s3_key}")
                
                # S3にアップロード
                self.s3.upload_file(
                    local_path,
                    self.bucket_name,
                    s3_key,
                    ExtraArgs={'ServerSideEncryption': 'AES256'}
                )
                
                self.logger.info(f"アップロード完了: {s3_key}")
                self.logger.info(f"ローカルパス: {local_path}")
                
                # アップロード日時をログに記録
                response = self.s3.head_object(Bucket=self.bucket_name, Key=s3_key)
                self.logger.info(f"最終更新日時: {response['LastModified']}")
                
            except Exception as e:
                self.logger.error(f"アップロードエラー {local_path}: {str(e)}")
                raise
                
        def upload_all_files(self):
            """ローカルディレクトリ内の全エクセルファイルをアップロード"""
            try:
                # エクセルファイルを検索
                excel_files = []
                for ext in ('*.xlsx', '*.xls'):
                    excel_files.extend(
                        glob.glob(os.path.join(self.local_dir, '**', ext), recursive=True)
                    )
                
                if not excel_files:
                    self.logger.warning(f"アップロード対象ファイルが見つかりません: {self.local_dir}")
                    return
                    
                # 各ファイルをアップロード
                for file_path in excel_files:
                    self.upload_to_s3(file_path)
                    
            except Exception as e:
                self.logger.error(f"アップロード処理エラー: {str(e)}")
                raise
                
        def upload_single_file(self, file_path):
            """
            指定された単一のファイルをアップロード
            Args:
                file_path (str): アップロードするファイルのパス
            """
            if not os.path.exists(file_path):
                self.logger.error(f"ファイルが存在しません: {file_path}")
                return
                
            if not file_path.lower().endswith(('.xlsx', '.xls')):
                self.logger.error(f"エクセルファイルではありません: {file_path}")
                return
                
            try:
                self.upload_to_s3(file_path)
            except Exception as e:
                self.logger.error(f"アップロード処理エラー: {str(e)}")
                raise
    
    def main():
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        
        # S3の設定
        bucket_name = "your-bucket-name"
        prefix = "path/to/excel/files/"
        local_dir = "disaster_reports"
        
        uploader = DisasterReportUploader(bucket_name, prefix, local_dir)
        
        # 全ファイルをアップロード
        uploader.upload_all_files()
        
        # または特定のファイルのみアップロード
        # file_path = "disaster_reports/example.xlsx"
        # uploader.upload_single_file(file_path)
    
    if __name__ == "__main__":
        main()