Claude3を使ってパワポ資料を読み込む処理をLambda関数上で実行してみた

import base64
import json
import re
import subprocess
from concurrent.futures import Future, ThreadPoolExecutor
from pathlib import Path

import boto3
from pdf2image import convert_from_path

s3 = boto3.resource("s3")
bedrock_runtime = boto3.client("bedrock-runtime", region_name="us-east-1")

def convert_pptx_to_pdf(
    s3_bucket_name: str,  # bucket name of input. ex: "bucket-name"
    s3_key_pptx: str,  # object to pptx. ex: "input/setsumeikai_17.pptx"
    s3_key_prefix_output: str | None,  # folder to output in bucket. ex: "output"
):
    # check input
    if not s3_key_pptx.endswith(".pptx"):
        print("WARNING: The input file is not pptx file")
        return None

    # get pptx from S3 bucket
    bucket = s3.Bucket(s3_bucket_name)
    filepath_tmp_pptx = Path("/tmp") / s3_key_pptx

    filepath_tmp_pptx.parent.mkdir(parents=True, exist_ok=True)
    bucket.download_file(s3_key_pptx, filepath_tmp_pptx.as_posix())

    # convert pptx to pdf with libreoffice
    command = " ".join(
        [
            "/opt/libreoffice7.6/program/soffice",
            "--headless",
            "--norestore",
            "--invisible",
            "--nodefault",
            "--nofirststartwizard",
            "--nolockcheck",
            "--nologo",
            "--convert-to pdf:writer_pdf_Export",
            f"--outdir {Path(filepath_tmp_pptx).parent.as_posix()}",
            filepath_tmp_pptx.as_posix(),
        ]
    )
    proc = subprocess.run(
        command,
        shell=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    print(f"STDOUT: {proc.stdout}")
    print(f"STDERR: {proc.stderr}")

    filepath_pdf = filepath_tmp_pptx.with_suffix(".pdf")

    if not filepath_pdf.exists():
        print(f"ERROR: The PDF file not created: {filepath_pdf}")
        return None

    print(f"PDF: {filepath_pdf}")
    print(f"Size: {Path(filepath_pdf).stat().st_size}")

    # put pdf to S3 bucket
    s3_key_pdf = filepath_pdf.relative_to("/tmp")
    if s3_key_prefix_output is not None and s3_key_prefix_output != "":
        s3_key_pdf = Path(s3_key_prefix_output) / s3_key_pdf

    with open(filepath_pdf, "rb") as data:
        bucket.put_object(Key=s3_key_pdf.as_posix(), Body=data)

    # return pdf fielpath and s3 object key
    return filepath_pdf, s3_key_pdf

def convert_pdf_to_images(
    filepath_pdf: Path,  # temp file path of pdf
    s3_bucket_name: str,  # bucket name to output. ex: "bucket-name"
    s3_key_prefix_output: str | None,  # folder to output in bucket. if set None, same folder as input object. ex: "output"
):
    # check input
    if not filepath_pdf.exists() or not filepath_pdf.suffix == ".pdf":
        print("WARNING: The input file is not pdf file")
        return None

    # convert pdf to image with pdf2image
    folderpath_images = Path(filepath_pdf).parent / "images"

    folderpath_images.mkdir(parents=True, exist_ok=True)
    convert_from_path(filepath_pdf, output_folder=folderpath_images, fmt="jpeg")

    filepaths_image = list(folderpath_images.glob("*.jpg"))
    print(filepaths_image)

    # put object to S3 bucket
    bucket = s3.Bucket(s3_bucket_name)
    s3_keys_image = []
    for filepath_img in filepaths_image:

        s3_key_image = filepath_img.relative_to("/tmp")
        if s3_key_prefix_output is not None and s3_key_prefix_output != "":
            s3_key_image = Path(s3_key_prefix_output) / s3_key_image

        with open(filepath_img, "rb") as data:
            bucket.put_object(Key=s3_key_image.as_posix(), Body=data)

        s3_keys_image.append(s3_key_image)

    # return images fielpath and s3 object key
    return filepaths_image, s3_keys_image

def read_image_to_text(
    filepath_image: Path,
    s3_bucket_name: str,  # bucket name to output. ex: "bucket-name"
    s3_key_prefix_output: str | None,  # folder to output in bucket. if set None, same folder as input object. ex: "output"
):
    # check input
    if not filepath_image.exists() and not filepath_image.suffix in [".jpg", ".jpeg"]:
        print(f"WARNING: The input file is not image file: {filepath_image}")
        return None

    # prepare message for bedrock API
    prompt = "これはパワーポイントファイルのスクリーンショットです。文字起こしして、markdown形式で出力してください。出力は、'''などで囲わず、markdownのみ出力してください。人間が認識するような順番・親子関係としてheadingしてください。写真やフロー図・構成図などがスクリーンショット内部に含まれている場合は、内容を説明して[]で囲ってテキストとして出力してください。オブジェクトや図形の説明などは不要です。このスクリーンショットは画像ですが、スクリーンショット自体の説明は不要です"

    with open(filepath_image, "rb") as f:
        bytes_image = f.read()
    bytes_image_b64 = base64.b64encode(bytes_image).decode("utf-8")  # Base64変換

    content_image = {
        "type": "image",
        "source": {
            "type": "base64",
            "media_type": "image/jpeg",
            "data": bytes_image_b64,
        },
    }
    content_prompt = {
        "type": "text",
        "text": prompt,
    }
    content = [content_image, content_prompt]
    messages = [
        {"role": "user", "content": content},
    ]

    # call bedrock API
    body = json.dumps(
        {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 500,
            "temperature": 0,
            "messages": messages,
        }
    )
    response = bedrock_runtime.invoke_model(
        body=body,
        modelId="anthropic.claude-3-sonnet-20240229-v1:0",
        accept="application/json",
        contentType="application/json",
    )
    response_body = json.loads(response.get("body").read())
    print(json.dumps(response_body, indent=2, ensure_ascii=False).replace("\n", "\r"))

    if response_body["type"] != "message":
        print("ERROR: The response is not message")
        return None

    if response_body["stop_reason"] != "end_turn":
        print("ERROR: The stop_reason is not end_turn")
        return None

    text = response_body["content"][0]["text"]

    # save text to file
    filepath_txt = filepath_image.with_suffix(".txt")
    with open(filepath_txt, "w") as f:
        f.write(text)

    # put object to S3 bucket
    s3_key_text = filepath_txt.relative_to("/tmp")
    if s3_key_prefix_output is not None and s3_key_prefix_output != "":
        s3_key_text = Path(s3_key_prefix_output) / s3_key_text

    bucket = s3.Bucket(s3_bucket_name)
    with open(filepath_txt, "rb") as data:
        bucket.put_object(Key=s3_key_text.as_posix(), Body=data)

    return filepath_txt, s3_key_text

def read_images_to_text(
    filepaths_image: list[Path],  # temp file path of images
    s3_bucket_name: str,  # bucket name to output. ex: "bucket-name"
    s3_key_prefix_output: str | None,  # folder to output in bucket. if set None, same folder as input object. ex: "output"
):
    # check input
    if not all([filepath_image.exists() for filepath_image in filepaths_image]):
        print("WARNING: The input file is not image file")
        return None

    # read image with claude 3
    executor = ThreadPoolExecutor(max_workers=5)
    futures: list[Future[tuple[Path, Path] | None]] = []
    for filepath_image in filepaths_image:
        future = executor.submit(read_image_to_text, filepath_image, s3_bucket_name, s3_key_prefix_output)
        futures.append(future)

    results: list[tuple[Path, Path] | None] = []
    for future in futures:
        result = future.result()
        results.append(result)

    return results

def handler(event, context):
    print(event)

    # get input
    s3_object_uri_input = event.get("s3_object_uri_input")
    s3_key_prefix_output = event.get("s3_key_prefix_output")
    if not s3_object_uri_input:
        return "ERROR: The input file is not set"

    # check input
    s3_object_uri_pattern = re.search(r"s3://(.+?)/(.+)", s3_object_uri_input)
    if not s3_object_uri_pattern:
        print("WARNING: The input file is not s3 object uri")
        return None
    s3_bucket_name = s3_object_uri_pattern.group(1)
    s3_key_pptx = s3_object_uri_pattern.group(2)
    print(f"Bucket: {s3_bucket_name}")
    print(f"Key: {s3_key_pptx}")

    # convert pptx to pdf
    result_pptx_to_pdf = convert_pptx_to_pdf(s3_bucket_name, s3_key_pptx, s3_key_prefix_output)
    if result_pptx_to_pdf is None:
        return "ERROR: The convert (pptx to pdf) failed"
    filepath_pdf, s3_key_pdf = result_pptx_to_pdf

    # convert pdf to image
    result_pdf_to_images = convert_pdf_to_images(filepath_pdf, s3_bucket_name, s3_key_prefix_output)
    if result_pdf_to_images is None:
        return "ERROR: The convert (pdf to images) failed"
    filepaths_image, s3_keys_image = result_pdf_to_images

    # read image with claude 3
    result_images_to_text = read_images_to_text(filepaths_image, s3_bucket_name, s3_key_prefix_output)
    if result_images_to_text is None:
        return "ERROR: The read (images to text) failed"
    for result in result_images_to_text:
        print(result)

    # rm tmp files
    if filepath_pdf.exists():
        filepath_pdf.unlink()
    for filepath_image in filepaths_image:
        if filepath_image.exists():
            filepath_image.unlink()
    for result_images_to_text in result_images_to_text:
        if result_images_to_text is not None:
            filepath_txt = result_images_to_text[0]
            if result_images_to_text is not None and filepath_txt.exists():
                filepath_txt.unlink()

    return ""