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()