12 日間 (10 日目) - 添付ファイルをスペースに保存する
12 日間の DigitalOcean の 10 日目 へようこそ!昨日、私たちはあなたのアプリに、DigitalOcean の GenAI エージェントを使用して電子メールのコンテンツから情報を抽出するよう教えました。これは大きな一歩でしたが、正直に言って、領収書や請求書がメール本文に常に存在するとは限りません。多くの場合、それらは添付ファイルです。
今日はそれを扱います。これらの添付ファイルを抽出し、DigitalOcean Spaces に安全に保存し、各ファイルのパブリック URL を生成する方法をアプリに教えます。これらの URL は最終的にデータベースに保存され、経費を確認するときに添付ファイルをプレビューできるようになります。
飛び込んでみましょう。
🚀 学ぶ内容
今日のセッションが終わるまでに、次の方法がわかるようになります。
- 添付ファイルを保存するための DigitalOcean Space を作成します。
- 消印メールから Base64 でエンコードされた添付ファイルを抽出してデコードします。
- boto3 を使用して添付ファイルを DigitalOcean Spaces にアップロードします。
- 上書きを防ぐために、uuid を使用して一意のファイル名を生成します。
- ワークフロー全体を調整して、複数の添付ファイルをシームレスに処理します。
🛠 必要なもの
このチュートリアルを最大限に活用するには、次のことを前提としています。
- すでに DigitalOcean にデプロイされている Flask アプリ: Flask アプリをまだデプロイしていない場合は、「7 日目 - 電子メールベースの受信プロセッサの構築とデプロイ」の手順に従うことができます。
- 電子メール テスト用に構成された消印: 電子メールから受信までの処理パイプラインをテストするには、メールを Flask アプリに転送するように消印を設定する必要があります。ステップバイステップのガイドについては、「8 日目 - 消印を Flask アプリに接続する」を参照してください。
- DigitalOcean Spaces のセットアップ: 処理された添付ファイルを DigitalOcean Space に保存します。まだスペースをお持ちでない場合は、このチュートリアルでスペースの作成方法を説明します。
<$> [情報] 注: すべてを設定していない場合でも、次の方法を学習できます。
添付ファイルを保存するための DigitalOcean Space を作成します。
Base64 でエンコードされた添付ファイルをプログラムでデコードします。
boto3
を使用してファイルを DigitalOcean Spaces にアップロードします。添付ファイルの処理を Flask アプリにシームレスに統合します。
<$>
ステップ 1: DigitalOcean スペースを作成する
まず、添付ファイルを保管する場所が必要です。 DigitalOcean Spaces は、領収書や請求書などのファイルを安全に処理するのに最適なオブジェクト ストレージ サービスです。スケーラブルで安全で、アプリとシームレスに統合されます。
スペースを作成する
DigitalOcean ダッシュボードにログインし、スペース オブジェクト ストレージをクリックします。
次に、[バケットの作成] をクリックします。
ユーザーに近い地域を選択します (例: ニューヨークの場合は
nyc3
)。スペースに名前を付けます (例:
email-receipts
)
これにより、email-receipts
という名前のバケットが作成され、https://email-receipts.nyc3.digitaloceanspaces.com
のような URL で利用可能になります。
アクセスキーの生成
プログラムで (たとえば、boto3
経由で) スペースと対話するには、アクセス キーと秘密キーが必要です。
スペースを開き、[設定] をクリックし、アクセス キーまでスクロールします。
[アクセス キーの作成] をクリックします。
権限をすべての権限に設定して、アプリがファイルの読み取り、書き込み、削除をできるようにします。
-
キーに名前を付け(またはデフォルトを使用)、アクセス キーの作成をクリックします。
アクセス キーと秘密キーを保存します。秘密キーが表示されるのはこのときだけです。
環境変数の更新
DigitalOcean アプリ プラットフォーム ダッシュボードで:
[設定] > [環境変数]に移動します。
以下を追加します。
SPACES_ACCESS_KEY
: スペース アクセス キー ID。
SPACES_SECRET_KEY
: Spaces の秘密キー。SPACES_BUCKET_NAME
: スペースの名前 (例:email-receipts
)。SPACES_REGION
: スペースの地域 (例:nyc3
)。
ステップ 2: 添付ファイルを処理して DigitalOcean Spaces にアップロードする
アプリ内の添付ファイルを処理するために、app.py
を更新し、いくつかの新しい関数を作成します。各機能は、添付ファイルのデコードから DigitalOcean Spaces へのアップロードまで、特定の目的を果たします。これらを一つずつ見ていきましょう。
注: アプリをまだセットアップしていない場合は、「7 日目 - 電子メールベースの受信プロセッサの構築とデプロイ」の手順に従って、アプリを作成し、DigitalOcean のアプリ プラットフォームにデプロイします。
添付ファイルをデコードして保存する
Postmark は、JSON ペイロード内の Base64 でエンコードされたデータとして添付ファイルを送信します。最初のステップは、Python の base64
ライブラリを使用してこのデータをデコードし、ローカルに保存することです。この関数は、uuid
ライブラリを利用して各ファイルに一意の名前を付けるようにします。
Base64 とは何ですか? バイナリ ファイル (PDF など) のトランスレータのようなものです。それらを Web 経由で安全に送信できるプレーン テキスト形式に変換します。デコードしてバイナリに戻すと、通常のファイルと同じように扱うことができます。
ファイルはどこに保存されますか?: デコードされたファイルは一時的に /tmp
に保存されます。これは、ほとんどのシステムで利用できる短期保存ディレクトリです。これはスクラッチパッドのようなものだと考えてください。短期間の使用に最適で、アプリの実行が停止するとすべてが消去されます。
これは、添付ファイルをデコードし、ファイル名が一意であることを確認し (uuid
のおかげで)、それを /tmp
に保存する関数です。
import os
import base64
import uuid
def decode_and_save_attachment(attachment):
"""Decode base64-encoded attachment and save it locally with a unique name."""
file_name = attachment.get("Name")
encoded_content = attachment.get("Content")
if not file_name or not encoded_content:
logging.warning("Invalid attachment, skipping.")
return None
unique_file_name = f"{uuid.uuid4()}_{file_name}"
file_path = os.path.join("/tmp", unique_file_name)
try:
with open(file_path, "wb") as file:
file.write(base64.b64decode(encoded_content))
logging.info(f"Attachment saved locally: {file_path}")
return file_path
except Exception as e:
logging.error(f"Failed to decode and save attachment {file_name}: {e}")
return None
DigitalOcean Spaces への添付ファイルのアップロード
ファイルをデコードして保存したので、次のステップはファイルを DigitalOcean Spaces にアップロードすることです。アップロードを処理するには、AWS 互換 API を操作するための強力な Python SDK である boto3
を使用します。 Spaces は S3 バケットと同じように機能するため、完璧に適合します。
この関数はファイルをスペースにアップロードし、パブリック URL を返します。
import boto3
def upload_attachment_to_spaces(file_path):
"""Upload a file to DigitalOcean Spaces and return its public URL."""
file_name = os.path.basename(file_path)
object_name = f"email-receipt-processor/{file_name}"
try:
s3_client.upload_file(file_path, SPACES_BUCKET, object_name, ExtraArgs={"ACL": "public-read"})
file_url = f"https://{SPACES_BUCKET}.{SPACES_REGION}.cdn.digitaloceanspaces.com/{object_name}"
logging.info(f"Attachment uploaded to Spaces: {file_url}")
return file_url
except Exception as e:
logging.error(f"Failed to upload attachment {file_name} to Spaces: {e}")
return None
複数の添付ファイルの処理
全部まとめてみましょう。この関数はすべてを調整します。
- 各添付ファイルをデコードします。
- それをスペースにアップロードします。
- アップロードされたファイルの URL を収集します。
def process_attachments(attachments):
"""Process all attachments and return their URLs."""
attachment_urls = []
for attachment in attachments:
file_path = decode_and_save_attachment(attachment)
if file_path:
file_url = upload_attachment_to_spaces(file_path)
if file_url:
attachment_urls.append({"file_name": os.path.basename(file_path), "url": file_url})
os.remove(file_path) # Clean up local file
return attachment_urls
/inbound
ルートを更新します。
最後に、添付ファイルの処理を含めるように /inbound
ルートを更新します。このルートは、電子メールのコンテンツの処理、添付ファイルのデコードとアップロード、および最終応答の返しを処理するようになります。
@app.route('/inbound', methods=['POST'])
def handle_inbound_email():
"""Process inbound emails and return extracted JSON."""
logging.info("Received inbound email request.")
data = request.json
email_content = data.get("TextBody", "")
attachments = data.get("Attachments", [])
if not email_content:
logging.error("No email content provided.")
return jsonify({"error": "No email content provided"}), 400
extracted_data = extract_text_from_email(email_content)
attachment_urls = process_attachments(attachments)
response_data = {
"extracted_data": extracted_data,
"attachments": attachment_urls
}
# Log the final combined data
logging.info("Final Response Data: %s", response_data)
return jsonify(response_data)
最終的な完全なコード
すべての更新を含む完全な app.py
ファイルは次のとおりです。
from flask import Flask, request, jsonify
import os
import base64
import uuid
import boto3
from dotenv import load_dotenv
from openai import OpenAI
import logging
Load environment variables
load_dotenv()
Initialize Flask app
app = Flask(__name__)
Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
Initialize DigitalOcean GenAI client
SECURE_AGENT_KEY = os.getenv("SECURE_AGENT_KEY")
AGENT_BASE_URL = os.getenv("AGENT_BASE_URL")
AGENT_ENDPOINT = f"{AGENT_BASE_URL}/api/v1/"
client = OpenAI(base_url=AGENT_ENDPOINT, api_key=SECURE_AGENT_KEY)
DigitalOcean Spaces credentials
SPACES_ACCESS_KEY = os.getenv("SPACES_ACCESS_KEY")
SPACES_SECRET_KEY = os.getenv("SPACES_SECRET_KEY")
SPACES_BUCKET = os.getenv("SPACES_BUCKET_NAME")
SPACES_REGION = os.getenv("SPACES_REGION")
SPACES_ENDPOINT = f"https://{SPACES_BUCKET}.{SPACES_REGION}.digitaloceanspaces.com"
Initialize DigitalOcean Spaces client
session = boto3.session.Session()
s3_client = session.client(
's3',
region_name=SPACES_REGION,
endpoint_url=SPACES_ENDPOINT,
aws_access_key_id=SPACES_ACCESS_KEY,
aws_secret_access_key=SPACES_SECRET_KEY
)
def extract_text_from_email(email_content):
"""Extract relevant details from the email content using DigitalOcean GenAI."""
logging.debug("Extracting details from email content.")
prompt = (
"Extract the following details from the email:\n"
"- Date of transaction\n"
"- Amount\n"
"- Currency\n"
"- Vendor name\n\n"
f"Email content:\n{email_content}\n\n"
"Ensure the output is in JSON format with keys: date, amount, currency, vendor."
)
response = client.chat.completions.create(
model="your-model-id", # Replace with your GenAI model ID
messages=[{"role": "user", "content": prompt}]
)
logging.debug("GenAI processing completed.")
return response.choices[0].message.content
def decode_and_save_attachment(attachment):
"""Decode base64-encoded attachment and save it locally with a unique name."""
file_name = attachment.get("Name")
encoded_content = attachment.get("Content")
if not file_name or not encoded_content:
logging.warning("Invalid attachment, skipping.")
return None
unique_file_name = f"{uuid.uuid4()}_{file_name}"
file_path = os.path.join("/tmp", unique_file_name)
try:
with open(file_path, "wb") as file:
file.write(base64.b64decode(encoded_content))
logging.info(f"Attachment saved locally: {file_path}")
return file_path
except Exception as e:
logging.error(f"Failed to decode and save attachment {file_name}: {e}")
return None
def upload_attachment_to_spaces(file_path):
"""Upload a file to DigitalOcean Spaces and return its public URL."""
file_name = os.path.basename(file_path)
object_name = f"email-receipt-processor/{file_name}"
try:
s3_client.upload_file(file_path, SPACES_BUCKET, object_name, ExtraArgs={"ACL": "public-read"})
file_url = f"https://{SPACES_BUCKET}.{SPACES_REGION}.cdn.digitaloceanspaces.com/{object_name}"
logging.info(f"Attachment uploaded to Spaces: {file_url}")
return file_url
except Exception as e:
logging.error(f"Failed to upload attachment {file_name} to Spaces: {e}")
return None
def process_attachments(attachments):
"""Process all attachments and return their URLs."""
attachment_urls = []
for attachment in attachments:
file_path = decode_and_save_attachment(attachment)
if file_path:
file_url = upload_attachment_to_spaces(file_path)
if file_url:
attachment_urls.append({"file_name": os.path.basename(file_path), "url": file_url})
os.remove(file_path) # Clean up local file
return attachment_urls
@app.route('/inbound', methods=['POST'])
def handle_inbound_email():
"""Process inbound emails and return extracted JSON."""
logging.info("Received inbound email request.")
data = request.json
email_content = data.get("TextBody", "")
attachments = data.get("Attachments", [])
if not email_content:
logging.error("No email content provided.")
return jsonify({"error": "No email content provided"}), 400
extracted_data = extract_text_from_email(email_content)
attachment_urls = process_attachments(attachments)
response_data = {
"extracted_data": extracted_data,
"attachments": attachment_urls
}
# Log the final combined data
logging.info("Final Response Data: %s", response_data)
return jsonify(response_data)
if __name__ == "__main__":
logging.info("Starting Flask application.")
app.run(port=5000)
ステップ 3: DigitalOcean にデプロイする
更新された Flask アプリをデプロイするには、7 日目の手順に従います。簡単な概要は次のとおりです。
更新されたコードを GitHub にプッシュする: Flask アプリに必要な変更を加えた後、更新されたコードをコミットして GitHub にプッシュします。これにより、DigitalOcean のアプリ プラットフォームへの自動デプロイメントがトリガーされます。
git add . git commit -m "Add attachment processing with DigitalOcean Spaces" git push origin main
デプロイメントの監視: アプリのダッシュボードのデプロイメントセクションで進行状況を追跡できます。
デプロイメントを確認する: デプロイメントが完了したら、アプリのパブリック URL に移動し、その機能をテストします。ダッシュボードのランタイム ログをチェックして、アプリが正常に起動したことを確認することもできます。
ステップ 4: ワークフロー全体をテストする
アプリが完全に構成され準備が整ったので、ワークフロー全体をテストします。メール本文が処理され、添付ファイルがデコードされて DigitalOcean にアップロードされることを保証します
最終出力には必要なものがすべて含まれています。
段階的にテストする方法は次のとおりです。
テストメールを送信: テキスト本文と添付ファイルを含むメールを消印まで送信します。 Postmark の設定方法がわからない場合は、「Day 8: Postmark を Flask アプリに接続する」を参照してください。ここでは、メールをアプリに転送するための Postmark の設定について説明しました。
消印アクティビティ JSON を確認: 消印ダッシュボードで、アクティビティ タブに移動します。送信した電子メールを見つけて、JSON ペイロードにテキスト本文と Base64 でエンコードされた添付データが含まれていることを確認します。これにより、Postmark がメール データをアプリに正しく転送していることが確認できます。
ログを監視: DigitalOcean アプリ プラットフォーム ダッシュボードのランタイム ログをチェックして、アプリが JSON ペイロードを処理していることを確認します。 9 日目では、ランタイム ログにアクセスする方法について説明しました。
スペースのアップロードを確認: DigitalOcean Space にアクセスして、ファイルが正常にアップロードされたことを確認します。バケット内に添付ファイルが表示されるはずです。
最終出力を確認する: アプリは、抽出されたデータと添付ファイルの URL をログに記録する必要があります。これらのログには次のものが含まれます。
- 電子メール本文から抽出された詳細。
- アップロードされた添付ファイルの公開 URL。
実行時ログの検査に関するヒントについては、9 日目を参照してください。
これらの手順が完了すると、ワークフローでデータをデータベースに保存する準備が整います。これについては次に取り組みます。
🎁 まとめ
今日は、添付ファイルをプロのように処理する方法をアプリに教えました。私たちがやったことは次のとおりです。
- 安全でスケーラブルなストレージのための DigitalOcean Space を作成しました。
- Postmark JSON から Base64 でエンコードされた添付ファイルをデコードしました。
uuid
を使用して一意のファイル名を確保しました。boto3
を使用して添付ファイルを DigitalOcean Spaces にアップロードしました。- ファイルごとにパブリック URL が生成され、レシート プロセッサですぐに使用できます。
次はこのデータをデータベースに統合します。これにより、抽出された電子メールの詳細と添付ファイルの URL を長期使用できるように保存できるようになり、レシート プロセッサがさらに強力になります。乞うご期待!