From c903689a3aef35db9b92da87a22a75f9b4aaa4bc Mon Sep 17 00:00:00 2001 From: Nanten Date: Thu, 20 Mar 2025 23:33:02 +0900 Subject: [PATCH] 20032025 --- upload_b2.py | 173 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 173 insertions(+) create mode 100755 upload_b2.py diff --git a/upload_b2.py b/upload_b2.py new file mode 100755 index 0000000..216dda1 --- /dev/null +++ b/upload_b2.py @@ -0,0 +1,173 @@ +#!/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}.") \ No newline at end of file