173 lines
6.0 KiB
Python
Executable File
173 lines
6.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
import argparse
|
|
import uuid
|
|
import mimetypes
|
|
import logging
|
|
from datetime import datetime
|
|
from dotenv import load_dotenv
|
|
from b2sdk.v2 import InMemoryAccountInfo, B2Api
|
|
|
|
# .envファイルから環境変数を読み込む
|
|
load_dotenv()
|
|
|
|
# ロギングの設定
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s'
|
|
)
|
|
|
|
def validate_environment():
|
|
"""環境変数の確認"""
|
|
required_vars = [
|
|
'B2_APPLICATION_KEY_ID',
|
|
'B2_APPLICATION_KEY',
|
|
'B2_BUCKET_NAME',
|
|
'B2_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 is_valid_file(file_path):
|
|
"""
|
|
アップロード可能なファイルであることを確認
|
|
画像およびドキュメントファイルを許可
|
|
"""
|
|
valid_types = [
|
|
# 画像ファイル
|
|
'image/jpeg', 'image/png', 'image/gif', 'image/webp', 'image/svg+xml',
|
|
# Apple系画像フォーマット
|
|
'image/heic', 'image/heif', 'image/heif-sequence', 'image/heic-sequence',
|
|
'image/x-heic', 'image/x-heif',
|
|
# Apple Live Photos
|
|
'image/mov', 'video/quicktime',
|
|
# ドキュメントファイル
|
|
'application/pdf', 'application/msword',
|
|
'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
|
|
'application/vnd.ms-excel',
|
|
'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
|
'application/vnd.ms-powerpoint',
|
|
'application/vnd.openxmlformats-officedocument.presentationml.presentation',
|
|
'text/plain', 'text/markdown', 'text/csv'
|
|
]
|
|
|
|
# 拡張子からコンテンツタイプを推測
|
|
content_type, _ = mimetypes.guess_type(file_path)
|
|
|
|
# mimetypesモジュールがHEICなどを認識しない場合があるため、拡張子ベースでもチェック
|
|
if content_type is None:
|
|
ext = os.path.splitext(file_path)[1].lower()
|
|
if ext == '.heic':
|
|
return True
|
|
if ext == '.heif':
|
|
return True
|
|
|
|
return content_type in valid_types
|
|
|
|
def check_file_size(file_path, max_size_mb=10):
|
|
"""ファイルサイズが上限以内か確認"""
|
|
file_size = os.path.getsize(file_path)
|
|
max_size_bytes = max_size_mb * 1024 * 1024
|
|
return file_size <= max_size_bytes, file_size / (1024 * 1024) # サイズをMBで返す
|
|
|
|
def upload_to_b2(file_path):
|
|
"""
|
|
Args:
|
|
file_path (str): アップロードするファイルのパス。
|
|
|
|
Returns:
|
|
str: アップロードされたファイルのURL。
|
|
None: アップロードに失敗した場合。
|
|
"""
|
|
|
|
try:
|
|
validate_environment()
|
|
|
|
# ファイルタイプの検証
|
|
if not is_valid_file(file_path):
|
|
logging.error(f"Invalid file type: {file_path}")
|
|
return None
|
|
|
|
# ファイルサイズの検証
|
|
is_size_valid, size_mb = check_file_size(file_path)
|
|
if not is_size_valid:
|
|
logging.error(f"File too large: {file_path} ({size_mb:.2f} MB)")
|
|
return None
|
|
|
|
# B2 APIクライアントの初期化
|
|
info = InMemoryAccountInfo()
|
|
b2_api = B2Api(info)
|
|
b2_api.authorize_account("production",
|
|
os.getenv('B2_APPLICATION_KEY_ID'),
|
|
os.getenv('B2_APPLICATION_KEY'))
|
|
|
|
# バケットの取得
|
|
bucket = b2_api.get_bucket_by_name(os.getenv('B2_BUCKET_NAME'))
|
|
|
|
# 現在の年と月を取得
|
|
now = datetime.now()
|
|
year = now.strftime("%Y")
|
|
month = now.strftime("%m")
|
|
|
|
# ランダムなファイル名を生成（拡張子は元のまま）
|
|
random_filename = str(uuid.uuid4()) + os.path.splitext(file_path)[1]
|
|
|
|
# 年/月/ファイル名 の形式でパスを構築
|
|
file_path_in_bucket = f"{year}/{month}/{random_filename}"
|
|
|
|
# ファイルのContentTypeをファイル名から推測
|
|
content_type, _ = mimetypes.guess_type(file_path)
|
|
if content_type is None:
|
|
# MIMEタイプが不明な場合はoctet-streamを使用
|
|
content_type = 'application/octet-stream'
|
|
|
|
# ファイルのアップロード
|
|
logging.info(f"Uploading {file_path} to {file_path_in_bucket}")
|
|
|
|
# 大きなファイルの場合はチャンクアップロード
|
|
file_size = os.path.getsize(file_path)
|
|
if file_size > 5 * 1024 * 1024: # 5MB以上
|
|
uploaded_file = bucket.upload_local_file(
|
|
local_file=file_path,
|
|
file_name=file_path_in_bucket,
|
|
content_type=content_type,
|
|
min_part_size=5 * 1024 * 1024 # 5MB
|
|
)
|
|
else:
|
|
uploaded_file = bucket.upload_local_file(
|
|
local_file=file_path,
|
|
file_name=file_path_in_bucket,
|
|
content_type=content_type
|
|
)
|
|
|
|
# アップロードされたファイルのURLを返す
|
|
file_url = f"{os.getenv('B2_CUSTOM_URL')}{file_path_in_bucket}"
|
|
logging.info(f"Successfully uploaded: {file_url}")
|
|
return file_url
|
|
|
|
except ValueError as e:
|
|
logging.error(f"Configuration error: {e}")
|
|
return None
|
|
except Exception as e:
|
|
logging.error(f"Unexpected error: {str(e)}")
|
|
return None
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description='Upload files to Backblaze B2.')
|
|
parser.add_argument('--files-path', nargs='+', required=True, help='Paths to files to upload.')
|
|
args = parser.parse_args()
|
|
|
|
for file_path in args.files_path:
|
|
if not os.path.exists(file_path):
|
|
print(f"Error: File not found at {file_path}. Skipping.")
|
|
continue
|
|
|
|
file_url = upload_to_b2(file_path)
|
|
if file_url:
|
|
print(file_url)
|
|
else:
|
|
print(f"File upload failed for {file_path}.") |