ユーザー投稿のメディアを Gemini に渡す前段を固める — 安全弁としての検証パイプライン実装メモ

import os
 
INLINE_LIMIT = 18 * 1024 * 1024     # インライン送信の安全圏（公称20MBより手前で止める）
HARD_LIMIT = 480 * 1024 * 1024      # これ以上は受け付けない
 
def check_size(path: str) -> int:
    size = os.path.getsize(path)
    if size == 0:
        raise ValueError("空のファイルです")
    if size > HARD_LIMIT:
        raise ValueError(f"ファイルが大きすぎます: {size} bytes")
    return size

SIGNATURES = (
    (b"\xff\xd8\xff", "image/jpeg"),
    (b"\x89PNG\r\n\x1a\n", "image/png"),
    (b"RIFF", "image/webp"),         # 9〜12バイト目の "WEBP" で最終確認する
    (b"\x00\x00\x00\x18ftyp", "video/mp4"),
    (b"\x00\x00\x00\x20ftyp", "video/mp4"),
)
 
ALLOWED = {"image/jpeg", "image/png", "image/webp", "video/mp4"}
 
def sniff_mime(path: str) -> str | None:
    with open(path, "rb") as f:
        head = f.read(32)
    for sig, mime in SIGNATURES:
        if head.startswith(sig):
            if mime == "image/webp" and head[8:12] != b"WEBP":
                continue
            return mime
    return None
 
def resolve_mime(path: str) -> str:
    mime = sniff_mime(path)
    if mime not in ALLOWED:   # None も許可外も同じく拒否する
        raise ValueError("許可されない、または判定できない形式です")
    return mime

from PIL import Image, ImageFile
 
Image.MAX_IMAGE_PIXELS = 40_000_000   # 約4000万画素を超えたら異常とみなす
ImageFile.LOAD_TRUNCATED_IMAGES = False  # 壊れた画像は黙って通さない
 
def sanitize_image(src_path: str, dst_path: str) -> None:
    with Image.open(src_path) as img:
        img.verify()                  # まず構造の妥当性だけ確認する
    with Image.open(src_path) as img:  # verify 後は開き直しが必要
        clean = Image.new(img.mode, img.size)
        clean.putdata(list(img.getdata()))   # ピクセルだけを移し替える
        clean.save(dst_path, format="PNG")    # 付帯情報を持ち越さない

import time
from google import genai
 
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
 
def upload_and_wait(path: str, timeout: int = 180):
    f = client.files.upload(file=path)
    started = time.time()
    while f.state.name == "PROCESSING":
        if time.time() - started > timeout:
            client.files.delete(name=f.name)
            raise TimeoutError("動画の処理がタイムアウトしました")
        time.sleep(3)
        f = client.files.get(name=f.name)
    if f.state.name != "ACTIVE":
        client.files.delete(name=f.name)
        raise RuntimeError(f"ファイルが利用可能になりませんでした: {f.state.name}")
    return f

def build_part(path: str, mime: str, size: int):
    if size <= INLINE_LIMIT and mime.startswith("image/"):
        with open(path, "rb") as f:
            return genai.types.Part.from_bytes(data=f.read(), mime_type=mime), None
    uploaded = upload_and_wait(path)
    return uploaded, uploaded   # 第2要素は後で削除するための参照
 
def analyze(path: str, prompt: str) -> str:
    size = check_size(path)
    mime = resolve_mime(path)
    part, to_cleanup = build_part(path, mime, size)
    try:
        resp = client.models.generate_content(
            model="gemini-3.5-flash",   # 2026-06 時点の GA。マルチモーダル入力に十分速い
            contents=[part, prompt],
            config=genai.types.GenerateContentConfig(
                safety_settings=[
                    genai.types.SafetySetting(
                        category="HARM_CATEGORY_DANGEROUS_CONTENT",
                        threshold="BLOCK_MEDIUM_AND_ABOVE",
                    ),
                ],
                max_output_tokens=1024,
            ),
        )
        return mask_pii(resp.text)
    finally:
        if to_cleanup is not None:
            client.files.delete(name=to_cleanup.name)   # 使い終わったら必ず消す

import re
 
def mask_pii(text: str) -> str:
    text = re.sub(r"[\w.+-]+@[\w-]+\.[\w.-]+", "[email]", text)
    text = re.sub(r"\d{2,4}-\d{2,4}-\d{4}", "[phone]", text)
    return text