download_excel.py

    import boto3
    import os
    import openpyxl
    import logging
    from datetime import datetime
    import unicodedata
    import re
    
    class DisasterReportDownloader:
        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)
            
            if not os.path.exists(local_dir):
                os.makedirs(local_dir)
                
        def normalize_text(self, text):
            """テキストを正規化"""
            if not text:
                return ""
                
            # 文字列に変換
            text = str(text)
            
            # 全角→半角変換
            text = unicodedata.normalize('NFKC', text)
            
            # スペース除去
            text = re.sub(r'\s+', '', text)
            
            # 大文字小文字を区別しない
            text = text.lower()
            
            return text
                
        def is_disaster_report(self, cell_value):
            """セルの値が災害報告書のタイトルかどうかを判定"""
            # 検索対象のパターン
            patterns = [
                "業務上災害報告書兼災害分析表",
                "業務上災害報告書兼災害分析",
                "業務上災害報告書災害分析表",
                "業務災害報告書兼災害分析表",
                "業務上災害報告書",
                "災害報告書兼災害分析表"
            ]
            
            # セルの値を正規化
            normalized_value = self.normalize_text(cell_value)
            
            # パターンとマッチするか確認
            return any(
                pattern in normalized_value 
                for pattern in map(self.normalize_text, patterns)
            )
                
        def check_disaster_report(self, excel_path):
            """エクセルファイル内の全セルをチェックして災害報告書かどうかを判定"""
            try:
                wb = openpyxl.load_workbook(excel_path, data_only=True, read_only=True)
                
                for sheet in wb.worksheets:
                    for row in sheet.rows:
                        for cell in row:
                            if cell.value and self.is_disaster_report(cell.value):
                                self.logger.info(f"災害報告書を検出: {excel_path}")
                                self.logger.debug(f"検出文字列: {cell.value}")
                                wb.close()
                                return True
                
                wb.close()
                return False
                
            except Exception as e:
                self.logger.error(f"エクセルファイルの読み込みエラー {excel_path}: {str(e)}")
                return False
                
        def download_from_s3(self):
            """指定されたS3ディレクトリから災害報告書をダウンロード"""
            try:
                paginator = self.s3.get_paginator('list_objects_v2')
                
                for page in paginator.paginate(
                    Bucket=self.bucket_name,
                    Prefix=self.prefix
                ):
                    if 'Contents' not in page:
                        continue
                        
                    for obj in page['Contents']:
                        if not obj['Key'].lower().endswith(('.xlsx', '.xls')):
                            continue
                            
                        # 一時ファイル名を生成
                        temp_filename = f"temp_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.xlsx"
                        temp_path = os.path.join(self.local_dir, temp_filename)
                        
                        self.s3.download_file(
                            self.bucket_name,
                            obj['Key'],
                            temp_path
                        )
                        
                        # 災害報告書かチェック
                        if self.check_disaster_report(temp_path):
                            # S3のパス構造を維持したローカルパスを作成
                            relative_path = obj['Key'][len(self.prefix):]
                            final_path = os.path.join(self.local_dir, relative_path)
                            
                            # 保存先のディレクトリを作成
                            os.makedirs(os.path.dirname(final_path), exist_ok=True)
                            
                            # 既存ファイルがある場合は削除
                            if os.path.exists(final_path):
                                os.remove(final_path)
                            
                            # 一時ファイルを正式な場所に移動
                            os.rename(temp_path, final_path)
                            self.logger.info(f"ダウンロード完了: {final_path}")
                            self.logger.info(f"元のS3パス: {obj['Key']}")
                        else:
                            os.remove(temp_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'
        )
        
        bucket_name = "your-bucket-name"
        prefix = "path/to/excel/files/"
        local_dir = "disaster_reports"
        
        downloader = DisasterReportDownloader(bucket_name, prefix, local_dir)
        downloader.download_from_s3()
    
    if __name__ == "__main__":
        main()