Textractを使用してブログに使う画像をマスキングしてみた

import json
import boto3
import os
import re
import io
import urllib.parse
import unicodedata
from PIL import Image, ImageDraw

s3 = boto3.client('s3')
# Textractを指定
textract = boto3.client('textract', region_name='us-east-1')

def normalize_text(text):
    """
    判定精度を上げるためにテキストを掃除する関数
    1. 全角を半角に変換 (例: １２３ -> 123)
    2. スペースを削除 (例: 12 34 -> 1234)
    """
    # NFKC正規化（全角英数などを半角に）
    text = unicodedata.normalize('NFKC', text)
    # スペース削除
    text_nospace = text.replace(" ", "").replace("　", "")
    return text, text_nospace

def lambda_handler(event, context):
    input_bucket = event['Records'][0]['s3']['bucket']['name']
    key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')
    print(f"Processing: {key}")

    try:
        # 画像取得
        response = s3.get_object(Bucket=input_bucket, Key=key)
        image_bytes = response['Body'].read()
        image = Image.open(io.BytesIO(image_bytes))

        # Textract実行
        textract_response = textract.detect_document_text(Document={'Bytes': image_bytes})

        draw = ImageDraw.Draw(image)
        width, height = image.size

        # --- マスキング条件 ---
        patterns = [
            # 1. AWS Account ID (12桁の数字)
            # ハイフンやスペースが混じっていても、数字が12個並んでいれば検知
            r'\d[\d\s-]{10,}\d',

            # 2. Email Address
            r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',

            # 3. 特定のキーワード 
            r'Admin', 
            r'Secret',
            r'AccessKey',
            r'Password'
        ]

        # 検出されたブロックを走査
        for block in textract_response['Blocks']:
            if block['BlockType'] == 'LINE':
                raw_text = block['Text']

                # 正規化
                _, text_nospace = normalize_text(raw_text)

                # 判定ロジック
                is_hit = False

                for pattern in patterns:
                    # 元のテキストでチェック 
                    if re.search(pattern, raw_text, re.IGNORECASE):
                        is_hit = True
                    # スペース無し版でチェック
                    elif re.search(pattern, text_nospace, re.IGNORECASE):
                        is_hit = True

                if is_hit:
                    print(f"Masking HIT: {raw_text}")
                    box = block['Geometry']['BoundingBox']

                    # 座標計算
                    left = width * box['Left']
                    top = height * box['Top']
                    w = width * box['Width']
                    h = height * box['Height']

                    # マージンを少し広めにとる (上下左右 +7px)
                    margin = 7
                    shape = [
                        left - margin,
                        top - margin,
                        left + w + margin,
                        top + h + margin
                    ]
                    draw.rectangle(shape, fill="black")

        # 保存処理
        output_buffer = io.BytesIO()
        image.save(output_buffer, format='PNG')
        output_buffer.seek(0)

        output_bucket = os.environ['OUTPUT_BUCKET']
        # ファイル名に "masked_" をつける
        output_key = f"masked_{key.split('/')[-1]}"
        if not output_key.endswith('.png'):
            output_key += '.png'

        s3.put_object(
            Bucket=output_bucket, 
            Key=output_key, 
            Body=output_buffer, 
            ContentType='image/png'
        )

        return "Success"

    except Exception as e:
        print(f"Error: {e}")
        raise e

{
	"Version": "2012-10-17",
	"Statement": [
		{
			"Effect": "Allow",
			"Action": [
				"s3:GetObject",
				"s3:PutObject"
			],
			"Resource": [
				"arn:aws:s3:::blog-masking-input-source/*",
				"arn:aws:s3:::blog-masking-output-dest/*"
			]
		},
		{
			"Effect": "Allow",
			"Action": [
				"textract:DetectDocumentText"
			],
			"Resource": "*"
		}
	]
}