Creating AIPs with Archivematica for Files in Alfresco

# %% ../nbs/00_core.ipynb 3
from dotenv import load_dotenv
import os
import requests
from base64 import b64encode

# %% ../nbs/00_core.ipynb 4
class ApiClient:
    def __init__(self, verbose=False):
        """Alfresco API Client

        Args:
            verbose (bool): デバッグ情報を出力するかどうか
        """
        self.verbose = verbose

        # .envの読み込み
        load_dotenv(override=True)

        # 環境変数の取得
        self.user = os.getenv('ALF_USER')
        self.password = os.getenv('ALF_PASSWORD')
        self.target_host = os.getenv('ALF_TARGET_HOST')

        self._debug("環境変数の設定:", {
            "user": self.user,
            "password": "*" * len(self.password) if self.password else None,
            "target_host": self.target_host
        })

        # Basic認証のヘッダーを作成
        credentials = f"{self.user}:{self.password}"
        encoded_credentials = b64encode(credentials.encode()).decode()

        self.headers = {
            'accept': 'application/json',
            'authorization': f'Basic {encoded_credentials}'
        }

        self._debug("ヘッダーの設定:", {
            "accept": self.headers['accept'],
            "authorization": "Basic ***"
        })

    def _debug(self, message: str, data: dict = None):
        """デバッグ情報を出力する

        Args:
            message (str): メッセージ
            data (dict, optional): 追加のデータ
        """
        if self.verbose:
            print(f"🔍 {message}")
            if data:
                for key, value in data.items():
                    print(f"  - {key}: {value}")

    def get_nodes_nodeId(self, node_id: str):
        """ノードIDでノード情報を取得する

        Args:
            node_id (str): ノードID

        Returns:
            dict: ノード情報
        """
        url = f"{self.target_host}/alfresco/api/-default-/public/alfresco/versions/1/nodes/{node_id}"
        self._debug("APIリクエスト:", {"url": url})

        try:
            response = requests.get(
                url,
                headers=self.headers,
                timeout=float(30)
            )
            response.raise_for_status()
            return response.json()

        except requests.exceptions.Timeout:
            self._debug("エラー:", {"type": "timeout", "message": "リクエストがタイムアウトしました"})
            return None

        except requests.exceptions.RequestException as e:
            self._debug("エラー:", {"type": "request", "message": str(e)})
            return None

    def get_nodes_nodeId_content(self, node_id: str, output_path: str):
        """ノードのコンテンツを取得する

        Args:
            node_id (str): ノードID
            output_path (str): 出力パス
        """
        url = f"{self.target_host}/alfresco/api/-default-/public/alfresco/versions/1/nodes/{node_id}/content"
        self._debug("APIリクエスト:", {
            "url": url,
            "output_path": output_path
        })

        response = requests.get(url, headers=self.headers)
        binary_data = response.content

        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        with open(output_path, "wb") as file:
            file.write(binary_data)

        self._debug("ファイル保存完了:", {
            "size": len(binary_data),
            "path": output_path
        })

import os
from dotenv import load_dotenv
import boto3

# %% ../nbs/04_mdx.ipynb 4
class MdxClient:
    def __init__(self, verbose=False):
        """
        MdxClientの初期化

        Args:
            verbose (bool): デバッグ情報を出力するかどうか
        """

        self.verbose = verbose

        # load .env
        load_dotenv(override=True)

        # 環境変数を取得
        endpoint_url = os.getenv("MDX_ENDPOINT_URL")
        access_key = os.getenv("MDX_ACCESS_KEY")
        secret_key = os.getenv("MDX_SECRET_KEY")



        self.s3_client = boto3.client(
            's3',
            endpoint_url=endpoint_url,
            aws_access_key_id=access_key,
            aws_secret_access_key=secret_key,
        )

        self.bucket_name = os.getenv("MDX_BUCKET_NAME")

    def upload(self, file_path, object_name):
        """
        ファイルをアップロードする

        Args:
            file_path (str): ファイルパス
            object_name (str): オブジェクト名
        """

        try:
            self.s3_client.upload_file(
                file_path,
                self.bucket_name,
                object_name,
                )
        except Exception as e:
            print(f"❌ Upload failed: {e}")
            raise



    def download(self, object_name, file_path):
        """
        ファイルをダウンロードする

        Args:
            object_name (str): オブジェクト名
            file_path (str): ファイルパス
        """
        self.s3_client.download_file(self.bucket_name, object_name, file_path)