📝
S3 に保存されたファイル名をリネームする

2025/10/05に公開
S3 バケットに保存されているファイルの名前を変更したい状況があったのでやってみました。シンプルなサーバレスクラウドアプリの練習、権限周りなどの理解・復習にも良い気がします。

 前提S3 バケット（サムネイル画像）は以下の設定で作成済みの前提で進めます。
このリネームに関しては実行側のLambdaのIAMに権限を付与しているため、S3側では特に権限などを何かに付与する必要はありません。
したがってS3コンソール画面で以下の設定でも大丈夫です。
アクセス許可タブのブロックパブリックアクセスは「パブリックアクセスをすべて ブロックをオン」
プロパティタブの暗号化タイプは「Amazon S3 マネージドキーを使用したサーバー側の暗号化 (SSE-S3)」
このバケット（サムネイル画像）に保存されているファイルを変更します。

 自分の状況MediaConvert でサムネイルを書き出す際に、細かいファイル指定がどうやらできなさそうという状況がありました。そこで AI に相談しつつ以下の方法でいけそうだとなり、実装し実現できました。
https://zenn.dev/isosa/articles/f42fc3126168d3

 全体構成｜サーバレスアプリケーションLambda-A → MediaConvert(mp4 からサムネイル画像を作成)

↓

EventBridge → SNS（MediaConvert 完了をメール送信）

↓

Lambda-B(サムネイルのファイル名を変換する)
※メール通知のSNS部分はこの記事では解説しておりません。

 Lambda 関数設定ファイル名変換用（Lambda-B）の関数を作成します。

 関数の基本設定ランタイム: Python 3.13
タイムアウト: 3 秒
メモリ: 128MB

 コンソールから設定今回はコンソールから設定していきます。
AWS コンソールにログインして Lambda の画面に飛びます。
「関数を作成」というボタンをクリックします。
関数の作成セクションで「一から作成」を選択します。
「関数名」を任意の値に設定します。
ランタイムを選択します。今回は Python3.13 を選択します。
アーキテクチャは x86_64 を選択します。
その他の項目はデフォルトで進めます。入力が完了したら「関数の作成」をクリックします。
※デフォルトの実行ロールの変更という項目は「基本的な Lambda アクセス権限で新しいロールを作成」が選択されていると思います。これによりこの Lambda の IAM ロールが作成されてアタッチされます。もし既存のロールを設定したい場合は「既存のロールを使用する」を選択してもらえれば使用できます。

 Lambda 関数のコードLambda 関数自体を作成できたら以下のコードを貼り付けます。以下のコードは別の Lambda 関数で MediaConvert を実行し、MediaConvert のジョブが完了したら実行するという前提です。単体で動かす場合には適宜書き換えてみてください。
lambda_rename

import os
import re
import json
import time
import logging
from urllib.parse import urlparse

import boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

s3 = boto3.client("s3")

NAME_MODIFIER = os.environ.get("NAME_MODIFIER", "_thumbnail_test")
DRY_RUN = os.environ.get("DRY_RUN", "false").lower() == "true"
LIST_RETRIES = int(os.environ.get("LIST_RETRIES", "3"))

THUMB_FILE_RE = re.compile(
    rf"^(?P<name>.+){re.escape(NAME_MODIFIER)}\.(?P<num>\d+)\.(?P<ext>jpe?g)$",
    re.IGNORECASE,
)

def parse_s3_url(s3_url: str):
    u = urlparse(s3_url)
    return u.netloc, u.path.lstrip("/")

def list_keys(bucket: str, prefix: str):
    paginator = s3.get_paginator("list_objects_v2")
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for obj in page.get("Contents", []):
            yield obj["Key"]

def build_prefix_from_sample_key(key: str):
    if "/" in key:
        dirprefix, filename = key.rsplit("/", 1)
        dirprefix += "/"
    else:
        dirprefix, filename = "", key

    idx = filename.lower().rfind(NAME_MODIFIER.lower())
    if idx == -1:
        return None, None, None

    base = filename[:idx]
    list_prefix = f"{dirprefix}{base}{NAME_MODIFIER}."
    return dirprefix, base, list_prefix

def new_key_from_old(key: str):
    if "/" in key:
        dirprefix, filename = key.rsplit("/", 1)
        dirprefix += "/"
    else:
        dirprefix, filename = "", key

    m = THUMB_FILE_RE.match(filename)
    if not m:
        return None

    name = m.group("name")
    num = int(m.group("num"))
    ext = m.group("ext").lower()
    new_filename = f"{name}{num:05d}{NAME_MODIFIER}.{ext}"
    return f"{dirprefix}{new_filename}"

def copy_then_delete(bucket: str, src_key: str, dst_key: str):
    if DRY_RUN:
        logger.info(f"[DRY_RUN] s3://{bucket}/{src_key} -> s3://{bucket}/{dst_key}")
        return
    s3.copy_object(
        Bucket=bucket,
        Key=dst_key,
        CopySource={"Bucket": bucket, "Key": src_key},
        MetadataDirective="COPY",
    )
    s3.delete_object(Bucket=bucket, Key=src_key)
    logger.info(f"RENAMED s3://{bucket}/{src_key} -> s3://{bucket}/{dst_key}")

def stems_from_event(event):
    stems = []
    detail = event.get("detail", {})
    for og in detail.get("outputGroupDetails", []):
        for od in og.get("outputDetails", []):
            for path in od.get("outputFilePaths", []):
                if not re.search(r"\.(jpe?g)$", path, flags=re.IGNORECASE):
                    continue
                bucket, key = parse_s3_url(path)
                dp, base, list_prefix = build_prefix_from_sample_key(key)
                if dp is None:
                    continue
                stems.append((bucket, dp, base, list_prefix))
    uniq = []
    seen = set()
    for t in stems:
        k = (t[0], t[3])
        if k not in seen:
            uniq.append(t)
            seen.add(k)
    return uniq

def lambda_handler(event, context):
    logger.info("Event: %s", json.dumps(event))
    stems = stems_from_event(event)

    if not stems:
        logger.warning("No JPEG stems detected in event; nothing to do.")
        return {"renamed": 0, "message": "no jpg stems"}

    total = 0
    for bucket, dirprefix, base, list_prefix in stems:
        keys = []
        for attempt in range(LIST_RETRIES):
            keys = list(list_keys(bucket, list_prefix))
            if keys:
                break
            sleep = 2 ** attempt
            logger.info("No keys yet under %s; retrying in %ss", list_prefix, sleep)
            time.sleep(sleep)

        for old_key in sorted(keys):
            new_key = new_key_from_old(old_key)
            if not new_key or new_key == old_key:
                continue
            copy_then_delete(bucket, old_key, new_key)
            total += 1

    return {"renamed": total}


 IAM ポリシーの設定｜ Lambda 関数用まずこのポリシーがどうして必要かを整理します。前述の関数を作成したときにすでに一つロールとポリシーが作成されています。このポリシーには Lambda 関数の実行結果を Cloud Watch に出力できるようにする設定が書かれています。
この関数ではさらに Lambda 関数から S3 にアクセスしてファイル名を取得、コピーして、元のファイルを削除するという手順を実行するために権限を追加していきます。
AWS のコンソールで IAM の画面に移動します。
左メニューから「ポリシー」をクリックします。
「ポリシーの作成」をクリックします。
ポリシーエディタで「JSON」をクリックします。そして以下のカスタムポリシーを Lambda 関数作成時に生成されたロールにアタッチします。
iam_policy
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "S3ListBucket",
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::BUCKET_NAME",
      "Condition": { "StringLike": { "s3:prefix": ["PREFIX/*"] } }
    },
    {
      "Sid": "S3ObjectRW",
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:PutObject", "s3:DeleteObject"],
      "Resource": "arn:aws:s3:::BUCKET_NAME/PREFIX/*"
    }
  ]
}
注意: バケット名やプレフィックス(任意の文字列、厳密には違いますがフォルダのようなものです)を実際の S3 の名前に置き換えてください。
入力が完了してエラーが出なければ「次へ」を押してください。
次の画面で任意の「ポリシー名」を入力します。後から見てわかりやすいようにネーミングして説明も追加するとより親切ですね。
「ポリシーの作成」をクリックします。

 Lambda 関数にロールを割り当てLambda 関数のコンソール画面で「設定」タブを選択
「アクセス権限」セクションで「ロール名」をクリックしましょう。これが関数作成時に生成されたロールです。
「許可を追加 → ポリシーをアタッチ」でポリシーを追加します。

※すでに一つポリシーも生成されてアタッチされていると思います。これが Lambda 関数のログを CloudWatch に出力するためのものです。
次の画面で検索窓に作成したポリシー名を入力して「チェックを入れ」「許可を追加」をクリックします。

 EventBridge のルール設定今回はもう一つの Lambda 関数から実行されている MediaConvert 完了イベントをキャッチして Lambda 関数を起動する EventBridge ルールを設定します。

 EventBridge ルールの作成EventBridge コンソールで左メニューからルールをクリックし「ルールを作成」をクリックします。
「名前」に任意の値を入れます。あとはデフォルトで「次へ」をクリックします。
イベントパターンセクションの「カスタムパターン」を選択して以下の JSON を入力します。
{
  "source": ["aws.mediaconvert"],
  "detail-type": ["MediaConvert Job State Change"],
  "detail": {
    "status": ["COMPLETE"]
  }
}
入力したら「次へ」をクリックします。

※イベントセクションはイベントパターンセクションを選択すると自動的に設定されるので触る必要はありません。
ターゲットタイプは「AWS のサービス」を選択、ターゲットを選択のプルダウンから Lambda 関数を選択します。
すると項目が現れるので「このアカウントのターゲット」を選択、関数のプルダウンから「リネームを実行する Lambda 関数名」を選択し「次へ」をクリックします。
タグの設定は必要ならば設定し「次へ」をクリックします。
次の画面で一通り確認して間違いがあれば編集ボタンから修正します。間違いなさそうであれば「ルールの作成」をクリックします。
これで Lambda 関数 A → MediaConvert → Lambda 関数 B（今回のリネームするもの）の設定が完了しました。

 動作確認MediaConvert で mp4 からサムネイル生成ジョブを実行
ジョブ完了後、S3 バケットを確認
ファイル名が ファイル名_thumbnail_test.0000001.jpg から ファイル名00001_thumbnail_test.jpg に変更されていることを確認
CloudWatch Logs で Lambda 関数の実行ログを確認

 トラブルシューティング以下トラブルシューティング例です。
Lambda 関数が実行されない場合: EventBridge ルールの設定と IAM ロールの権限を確認
S3 操作でエラーが発生する場合: IAM ポリシーの S3 権限とバケット名を確認
ファイル名変換が期待通りでない場合: 正規表現パターンと入力ファイル名の形式を確認

 あとがき今回は別のLambda関数が既にあってその中でのAWSリソースのイベントをもとに発火する形でしたが、S3にアップロードした瞬間にネーミングを自動的に調整するアプリなども少しアレンジすればできると思います。サーバ管理なくアプリ部分だけ実装できるのもこのようなサーバレス構成でのメリットですね。
前提

自分の状況

全体構成｜サーバレスアプリケーション

Lambda 関数設定

関数の基本設定

コンソールから設定

Lambda 関数のコード

IAM ポリシーの設定｜ Lambda 関数用

Lambda 関数にロールを割り当て

EventBridge のルール設定

EventBridge ルールの作成

動作確認

トラブルシューティング

あとがき

Discussion
