PythonでBOXアップロード/削除してみた

import os
import time
import logging
import math
from boxsdk import BoxAPIException, JWTAuth, Client
from boxsdk.object.folder import Folder
from boxsdk.object.file import File
from boxsdk.config import Proxy

# プロキシ設定
Proxy.URL = 'http://192.168.1.100:8080'

# デフォルトの設定ファイルパス
DEFAULT_JSON_FILE_PATH = '/path/config.json'

# ログの設定
LOG_FILE_PATH = '/var/log/box.log'
logging.basicConfig(filename=LOG_FILE_PATH, level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s')

# Box クライアントの初期化
def init_box_client(json_file_path: str = DEFAULT_JSON_FILE_PATH):
    """
    Box クライアントを初期化します。

    Args:
        json_file_path (str, optional): Box 認証用の JSON ファイルのパス。デフォルトはデフォルトのパスです。
    """
    auth = JWTAuth.from_settings_file(json_file_path)
    return Client(auth)

# ファイルのアップロード
def upload_file(file_path: str, folder_id: str, retry_count: int = 3):
    """
    ファイルを Box にアップロードします。

    Args:
        file_path (str): アップロードするローカルファイルのパス。
        folder_id (str): ファイルをアップロードする Box フォルダの ID。
        client (Client): Box クライアントインスタンス。
        retry_count (int, optional): 失敗した場合のリトライ回数。デフォルトは 3 回。
    """
    client = init_box_client() # BOXクライアント初期化
    # リトライ回数分繰り返す
    for _ in range(retry_count):
        try:
            # ファイルのサイズと名前を取得
            file_size = os.path.getsize(file_path)
            file_name = os.path.basename(file_path)
            # Box フォルダを取得
            folder = client.folder(folder_id).get()
            # プレフライトチェックを実行
            folder.preflight_check(file_size, file_name)
            # ファイルをアップロード
            client.folder(folder_id).upload(file_path, file_name, preflight_check=False)
            return  # アップロード成功のため、関数を終了
        except BoxAPIException as err:
            logging.error(f"ファイルのアップロードに失敗しました: {err}")
            time.sleep(1)  # リトライ前に 1 秒待機
    raise RuntimeError("複数回のリトライ後にファイルをアップロードできませんでした")

# フォルダのアップロード
def upload_folder(local_folder_path: str, box_parent_folder_id: str, min_file_size: int = 1024 * 1024 * 20, retry_count: int = 3):
    """
    フォルダとその内容を Box にアップロードします。

    Args:
        local_folder_path (str): アップロードするローカルフォルダのパス。
        box_parent_folder_id (str): フォルダをアップロードする Box 親フォルダの ID。
        min_file_size (int, optional): チャンクアップロードのための最小ファイルサイズ。デフォルトは 20MB です。
        retry_count (int, optional): 失敗した場合のリトライ回数。デフォルトは 3 回 です。
    """
    client = init_box_client()  # BOXクライアント初期化
    # ローカルフォルダのパスを絶対パスに変換
    local_folder = os.path.abspath(local_folder_path)
    # フォルダ名を取得
    folder_name = os.path.basename(local_folder_path)
    try:
        # Box に新しいフォルダを作成
        new_box_folder = create_box_folder(folder_name, box_parent_folder_id, client)
        # フォルダ内のアイテムを反復処理
        for item in os.listdir(local_folder):
            item_path = os.path.join(local_folder, item)
            # サブフォルダを再帰的にアップロード
            if os.path.isdir(item_path):
                # サブフォルダ内のファイルとサブフォルダをアップロード
                upload_folder(item_path, new_box_folder.id, min_file_size=min_file_size, retry_count=retry_count)
            else:
                try:
                    # ファイルサイズに基づいてファイルまたはチャンクアップロードを実行
                    if os.path.getsize(item_path) < min_file_size:
                        upload_file(item_path, new_box_folder.id, retry_count=retry_count)
                    else:
                        upload_file_chunked(item_path, new_box_folder.id, client, retry_count=retry_count)
                except Exception as e:
                    logging.error(f"{item_path} のアップロードに失敗しました: {e}")
    except Exception as e:
        logging.error(f"フォルダの作成に失敗しました: {e}")

# Box にフォルダを作成
def create_box_folder(folder_name: str, parent_folder_id: str, client: Client) -> Folder:
    """
    Box にフォルダを作成します。

    Args:
        folder_name (str): 作成するフォルダの名前。
        parent_folder_id (str): フォルダが作成される親フォルダの ID。
        client (Client): Box クライアントインスタンス。

    Returns:
        Folder: 作成された Box フォルダオブジェクト。
    """
    try:
        folder = client.folder(parent_folder_id).create_subfolder(folder_name)
    except BoxAPIException as err:
        if err.code == "item_name_in_use":
            folder_id = err.context_info["conflicts"][0]["id"]
            folder = client.folder(folder_id).get()
        else:
            raise err
    return folder

# Box 内のアイテムを削除
def delete_item(item_id: str, item_type: str):
    """
    Box 内のアイテムを削除します。

    Args:
        item_id (str): アイテムの ID。
        item_type (str): アイテムの種類（'file' または 'folder'）。
        client (Client): Box クライアントインスタンス。
    """
    client = init_box_client() # BOXクライアント初期化
    try:
        if item_type == 'file':
            client.file(item_id).delete()
        elif item_type == 'folder':
            client.folder(item_id).delete()
        else:
            raise ValueError("無効なアイテムタイプです。'file' または 'folder' を指定してください。")
    except BoxAPIException as err:
        logging.error(f"{item_type} の削除に失敗しました: {err}")

def upload_file_chunked(file_path: str, folder_id: str, client: Client, chunk_size: int = 1024 * 1024 * 100, retry_count: int = 3):
    """
    ファイルをチャンク単位で分割して Box にアップロードします。

    Args:
        file_path (str): アップロードするローカルファイルのパス。
        folder_id (str): ファイルをアップロードする Box フォルダの ID。
        client (Client): Box クライアントインスタンス。
        chunk_size (int, optional): チャンクのサイズ。デフォルトは 100MB です。
        retry_count (int, optional): 失敗した場合のリトライ回数。デフォルトは 3 回です。
    """
    # リトライ回数分繰り返す
    for _ in range(retry_count):
        try:
            # ファイルのサイズと名前を取得
            file_size = os.path.getsize(file_path)
            file_name = os.path.basename(file_path)
            # Box フォルダを取得
            folder = client.folder(folder_id).get()
            # プレフライトチェックを実行
            folder.preflight_check(file_size, file_name)
            # ファイルをチャンク単位で分割してアップロード
            with open(file_path, 'rb') as file_stream:
                upload_session = folder.create_upload_session(file_name, file_size)
                total_parts = math.ceil(file_size / chunk_size)
                for part_index in range(total_parts):
                    offset = part_index * chunk_size
                    chunk = file_stream.read(chunk_size)
                    upload_session.upload_part_bytes(chunk, offset, part_index + 1, total_parts)
                upload_session.commit()
            return  # アップロード成功のため、関数を終了
        except BoxAPIException as err:
            logging.error(f"アップロードに失敗しました: {err}")
            time.sleep(1)  # リトライ前に 1 秒待機
    raise RuntimeError("複数回のリトライ後にファイルをアップロードできませんでした")

if __name__ == "__main__":
    # アップロードするローカルフォルダのパス
    local_folder_path = "/path/to/local/folder"

    # アップロード先の Box フォルダの ID
    box_folder_id = "0"  # フォルダの ID を指定

    # Box クライアントを初期化
    client = init_box_client()

    # フォルダのアップロードを実行
    upload_folder(local_folder_path, box_folder_id)

python3 -c "import box_exec; box_exec.upload_folder("/tmp/hogehoge", "123456789012")"

python3 -c "import box_exec; box_exec.upload_folder("/tmp/hogehoge", "123456789012")"

python3 -c "import box_exec; box_exec.delete_item("123456789012", "file|folder")"