94 lines
2.9 KiB
Python
Executable File
94 lines
2.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
import argparse
|
|
from dotenv import load_dotenv
|
|
import boto3
|
|
import mimetypes
|
|
import uuid
|
|
|
|
# .envファイルから環境変数を読み込む
|
|
load_dotenv()
|
|
|
|
# .envファイルからCloudflare R2のエンドポイントを読み込む
|
|
ENDPOINT = os.getenv('R2_ENDPOINT')
|
|
|
|
def validate_environment():
|
|
"""環境変数の確認"""
|
|
required_vars = [
|
|
'R2_ENDPOINT',
|
|
'R2_ACCESS_KEY_ID',
|
|
'R2_SECRET_ACCESS_KEY',
|
|
'R2_BUCKET_NAME',
|
|
'R2_CUSTOM_URL'
|
|
]
|
|
|
|
missing_vars = [var for var in required_vars if not os.getenv(var)]
|
|
if missing_vars:
|
|
raise ValueError(f"Missing required environment variables: {', '.join(missing_vars)}")
|
|
|
|
def upload_to_r2(image_path):
|
|
"""
|
|
Args:
|
|
image_path (str): アップロードする画像のパス。
|
|
|
|
Returns:
|
|
str: アップロードされた画像のURL。
|
|
int: アップロードに失敗した場合のエラーコード。
|
|
"""
|
|
|
|
try:
|
|
validate_environment()
|
|
|
|
# R2のクライアントを作成
|
|
s3 = boto3.client(
|
|
's3',
|
|
endpoint_url=ENDPOINT,
|
|
aws_access_key_id=os.getenv('R2_ACCESS_KEY_ID'),
|
|
aws_secret_access_key=os.getenv('R2_SECRET_ACCESS_KEY'),
|
|
config=boto3.session.Config(signature_version='s3v4')
|
|
)
|
|
|
|
# ランダムなファイル名を生成（拡張子は元のまま）
|
|
random_filename = str(uuid.uuid4()) + os.path.splitext(image_path)[1]
|
|
|
|
# 画像のContentTypeをファイル名から推測
|
|
content_type, _ = mimetypes.guess_type(image_path)
|
|
|
|
# 画像をアップロード
|
|
with open(image_path, 'rb') as f:
|
|
bucket_name = os.getenv('R2_BUCKET_NAME')
|
|
s3.put_object(Bucket=bucket_name, Key=random_filename, Body=f, ContentType=content_type)
|
|
|
|
# アップロードされた画像のURLを返す
|
|
return f"{os.getenv('R2_CUSTOM_URL')}{random_filename}"
|
|
|
|
except ValueError as e:
|
|
print(f"Configuration error: {e}")
|
|
return None
|
|
except boto3.exceptions.BotoServerError as e:
|
|
print(f"R2 server error: {e}")
|
|
return None
|
|
except boto3.exceptions.ClientError as e:
|
|
print(f"R2 client error: {e}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"Unexpected error: {e}")
|
|
return None
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description='Upload images to Cloudflare R2.')
|
|
parser.add_argument('--images-path', nargs='+', required=True, help='Paths to image files.')
|
|
args = parser.parse_args()
|
|
|
|
for image_path in args.images_path:
|
|
if not os.path.exists(image_path):
|
|
print(f"Error: Image not found at {image_path}. Skipping.")
|
|
continue
|
|
|
|
image_url = upload_to_r2(image_path)
|
|
if image_url:
|
|
print(image_url)
|
|
else:
|
|
print(f"Image upload failed for {image_path}.") |