excel_to_csv_converter.py

import pandas as pd
import os
import logging
import json
import boto3
from botocore.config import Config
import subprocess
from pdf2image import convert_from_path
import tempfile
import base64
from PIL import Image
import io

class ExcelStructuredConverter:
    def __init__(self, input_bucket, input_prefix, output_bucket, output_prefix):
        """
        初期化
        Args:
            input_bucket (str): 入力S3バケット名
            input_prefix (str): 入力S3プレフィックス
            output_bucket (str): 出力S3バケット名
            output_prefix (str): 出力S3プレフィックス
        """
        self.input_bucket = input_bucket
        self.input_prefix = input_prefix
        self.output_bucket = output_bucket
        self.output_prefix = output_prefix
        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(logging.INFO)
        self.s3 = boto3.client('s3')
        
        config = Config(
            retries=dict(max_attempts=10),
            read_timeout=300,
            connect_timeout=300
        )
        self.bedrock = boto3.client(
            service_name='bedrock-runtime',
            config=config,
            region_name='us-east-1'
        )
        
        self.temp_dir = tempfile.mkdtemp()
            
    def download_from_s3(self, key, local_path):
        """S3からファイルをダウンロード"""
        try:
            self.s3.download_file(self.input_bucket, key, local_path)
            self.logger.info(f"S3からダウンロード完了: {key}")
        except Exception as e:
            self.logger.error(f"S3ダウンロードエラー: {str(e)}")
            raise
            
    def upload_to_s3(self, local_path, key):
        """S3にファイルをアップロード"""
        try:
            self.s3.upload_file(local_path, self.output_bucket, key)
            self.logger.info(f"S3へアップロード完了: {key}")
        except Exception as e:
            self.logger.error(f"S3アップロードエラー: {str(e)}")
            raise

    def convert_excel_to_pdf(self, excel_path):
        """ExcelファイルをPDFに変換"""
        try:
            pdf_path = os.path.splitext(excel_path)[0] + '.pdf'
            
            subprocess.run([
                'soffice',
                '--headless',
                '--convert-to', 'pdf',
                '--outdir', os.path.dirname(pdf_path),
                excel_path
            ], check=True)
            
            return pdf_path
        except Exception as e:
            self.logger.error(f"PDF変換エラー: {str(e)}")
            raise

    def convert_pdf_to_png(self, pdf_path):
        """PDFをPNG画像に変換"""
        try:
            images = convert_from_path(pdf_path)
            png_paths = []
            
            for i, image in enumerate(images):
                png_path = os.path.splitext(pdf_path)[0] + f'_page_{i+1}.png'
                image.save(png_path, 'PNG')
                png_paths.append(png_path)
                
            return png_paths
        except Exception as e:
            self.logger.error(f"PNG変換エラー: {str(e)}")
            raise

    def encode_image_base64(self, image_path):
        """画像をbase64エンコード"""
        with open(image_path, 'rb') as image_file:
            return base64.b64encode(image_file.read()).decode('utf-8')

    def process_with_claude(self, image_path):
        """ClaudeでPNG画像を解析してJSON形式で出力"""
        try:
            base64_image = self.encode_image_base64(image_path)
            
            prompt = """
            この画像はエクセルデータを画像にしたものです。
            すべての本文を以下の形式でJSONとして構造化してください：
            
            {
                "headers": ["列名1", "列名2", ...],
                "data": [
                    {"列名1": "値1", "列名2": "値2", ...},
                    ...
                ]
            }
            
            正確にエクセルのデータを抽出してください。
            """
            
            body = json.dumps({
                "prompt": prompt,
                "max_tokens_to_sample": 4096,
                "temperature": 0.0,
                "top_p": 1,
                "anthropic_version": "bedrock-2023-05-31",
                "image": base64_image
            })
            
            response = self.bedrock.invoke_model(
                modelId="anthropic.claude-3-5-sonnet-20240620-v1:0",
                body=body
            )
            
            response_body = json.loads(response.get('body').read())
            return json.loads(response_body.get('completion'))
            
        except Exception as e:
            self.logger.error(f"Claude APIエラー: {str(e)}")
            return None
            
    def save_to_csv(self, data, output_path):
        """JSON形式のデータをCSVとして保存"""
        try:
            df = pd.DataFrame(data['data'])
            df.to_csv(output_path, index=False, encoding='utf-8')
            self.logger.info(f"CSV保存完了: {output_path}")
        except Exception as e:
            self.logger.error(f"CSV保存エラー: {str(e)}")
            raise

    def convert_all_excel_files(self):
        """S3のExcelファイルを全て処理"""
        try:
            response = self.s3.list_objects_v2(
                Bucket=self.input_bucket,
                Prefix=self.input_prefix
            )
            
            if 'Contents' not in response:
                self.logger.warning(f"Excelファイルが見つかりません: {self.input_prefix}")
                return
                
            for obj in response['Contents']:
                if obj['Key'].endswith(('.xlsx', '.xls')):
                    self.process_single_file(obj['Key'])
                    
        except Exception as e:
            self.logger.error(f"変換処理中のエラー: {str(e)}")
            raise
            
    def process_single_file(self, s3_key):
        """S3の単一のExcelファイルを処理"""
        try:
            filename = os.path.basename(s3_key)
            base_name = os.path.splitext(filename)[0]
            excel_path = os.path.join(self.temp_dir, filename)
            
            self.download_from_s3(s3_key, excel_path)

            pdf_path = self.convert_excel_to_pdf(excel_path)
            
            png_paths = self.convert_pdf_to_png(pdf_path)
            
            for i, png_path in enumerate(png_paths):
                structured_data = self.process_with_claude(png_path)
                
                if structured_data:
                    csv_path = os.path.join(self.temp_dir, f"{base_name}_page_{i+1}.csv")
                    df = pd.DataFrame(structured_data['data'])
                    df.to_csv(csv_path, index=False, encoding='utf-8')
                    
                    s3_csv_key = f"{self.output_prefix}/{base_name}_page_{i+1}.csv"
                    self.upload_to_s3(csv_path, s3_csv_key)
                    
                    os.remove(csv_path)
                
                os.remove(png_path)
            
            os.remove(pdf_path)
            os.remove(excel_path)
            
        except Exception as e:
            self.logger.error(f"ファイル {s3_key} の処理中にエラー: {str(e)}")
            raise

def main():
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    
    input_bucket = "input-bucket-name"
    input_prefix = "excel-files/"
    output_bucket = "output-bucket-name"
    output_prefix = "csv-files/"
    
    converter = ExcelStructuredConverter(
        input_bucket, input_prefix,
        output_bucket, output_prefix
    )
    converter.convert_all_excel_files()

if __name__ == "__main__":
    main()