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