Alfrescoのファイルに対して、Archivematicaを使ってAIPを作成する
概要
Alfrescoのファイルに対して、Archivematicaを使ってAIPを作成する方法の一例です。
以下が成果物のデモ動画です。
システム構成
今回は以下のようなシステム構成とします。複数のクラウドサービスを利用していることに特に意味はありません。
Alfrescoは、以下の記事を参考に、Azure上に構築したものを使用します。
Archivematicaとオブジェクトストレージはmdx.jpを使用し、分析環境はGakuNin RDMを使用します。
オブジェクトストレージへのファイルアップロード
Alfrescoからファイルをダウンロード
Alfrescoからのファイルダウンロードにあたっては、REST APIを使用します。
OpenAPIに準拠しており、以下などを参考にしました。
例えば以下により、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
})
オブジェクトストレージにファイルをアップロード
boto3
と、オブジェクトストレージのENDPOINT_URL
、ACCESS_KEY
、SECRET_KEY
およびBUCKET_NAME
などを使用して、ファイルのアップロード(とダウンロード)を行います。
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)
ただし、本記事執筆時点の最新のboto3を使用すると、以下のエラーが発生してしまいました。
boto3==1.35.99
のバージョンを使用することで、とりあえずエラーを回避できました。
Archivematicaを用いたAIPの作成と保存
以下の記事を参考に、指定したprocessing_config
オプションに基づき、transferを開始します。
例えば以下のように、ArchivematicaのHOST
, USER
, API_KEY
を用い、さらに/api/v2beta/package/
メソッドを用いることで実現できます。
import os
from dotenv import load_dotenv
import requests
from .core import ApiClient
import base64
# %% ../nbs/02_am.ipynb 4
class AmClient:
def __init__(self, verbose=False):
"""
AmClientの初期化
Args:
verbose (bool): デバッグ情報を出力するかどうか
"""
self.verbose = verbose
# load .env
load_dotenv(override=True)
self.host = os.getenv("AM_HOST")
api_key = os.getenv("AM_USER") + ":" + os.getenv("AM_API_KEY")
# APIキーの形式を修正
self.headers = {
"Authorization": f"ApiKey {api_key}",
"Content-Type": "application/json"
}
ApiClient.debug("環境変数の設定:", {
"host": self.host,
"api_key": "***"
})
ApiClient.debug("ヘッダーの設定:", {
"Authorization": "ApiKey ***",
"Content-Type": self.headers["Content-Type"]
})
def trasfer(self, location_uuid: str,name: str, path: str, transfer_type: str = "standard", accession: str = None, processing_config: str = None):
"""Archivematicaで転送を開始する
Args:
location_uuid (str): ロケーションのUUID
name (str): 転送の名前
path (str): 転送するファイルのパス
transfer_type (str, optional): 転送タイプ. デフォルトは "standard"
accession (str, optional): アクセッション番号
processing_config (str, optional): 処理設定
Returns:
dict: API レスポンス
"""
url = f"{self.host}/api/v2beta/package/"
# location_uuidとpathを組み合わせてbase64エンコード
location_path = f"{location_uuid}:{path}"
if self.verbose:
ApiClient.debug("location_path:", {
"location_uuid": location_uuid,
"path": path
})
encoded_path = base64.b64encode(location_path.encode()).decode()
# データの準備
data = {
"name": name,
"type": transfer_type,
"path": encoded_path,
"processing_config": processing_config
}
# アクセッション番号が指定されている場合は追加
if accession:
data["accession"] = accession
if self.verbose:
ApiClient.debug("転送データ:", data)
response = requests.post(url, headers=self.headers, json=data)
return response.json()
AIPの可視化
作成されたAIPに対する可視化処理は、以下の記事で紹介したツールを使用しました。
ウェブアプリ化
上記の一連の処理を使用するGradioアプリの例は以下です。
import gradio as gr
from alfresco.task import TaskClient
def download_task(task_id: str):
"""ファイルIDに基づいてファイルをダウンロードする
Args:
task_id (str): ファイルのID
Returns:
str: 処理結果のメッセージ
"""
try:
client = TaskClient(verbose=True)
output_dir = f"./tmp/{task_id}"
output_path = client.download(task_id, output_dir)
return [
output_path,
f"✅ ファイル {task_id} のAIPの作成が完了しました。"
]
except Exception as e:
# エラー時: (None, エラーメッセージ)
return (
None,
f"❌ エラーが発生しました: {str(e)}"
)
# Gradioインターフェースの作成
demo = gr.Interface(
fn=download_task,
inputs=[
gr.Textbox(
label="ファイルID",
placeholder="例: 7bb704a2-40b5-4d53-b704-a240b53d5390",
info="AIPを作成したいファイルのIDを入力してください"
)
],
outputs=[
gr.File(label="ダウンロードファイル"),
gr.Textbox(label="実行結果")
],
title="Alfresco AIPダウンローダ",
description="ファイルIDを入力して、対応するファイルのAIPをダウンロードします。",
examples=[
["a295a2a0-f79d-4f0c-95a2-a0f79d0f0cb0"]
],
allow_flagging="never"
)
demo.launch()
まとめ
考慮不足の点が多いかと思いますが、APIを用いたArchivematicaの利用にあたり、参考になりましたら幸いです。
Discussion