はじめに
HarinaはDiscordで動作するレシート解析Botです。ユーザーがチャットにレシートの画像を送信すると、最新のClaude 3 Haikuモデルを使って高速かつ高精度にレシートの情報を抽出します。抽出されたデータはGoogle Apps Script経由でGoogle Spreadsheetに自動保存され、家計簿アプリなどに活用できます。
デモ動画
Harina –
Claude 3 Haiku を使ったレシート解析Discord Bot アプリ
~日本語レシート認識のコスパ最強の構成~次はこれにData Interpreter(DI)を接続して家計の分析をしてもらう予定です!
---
作り方はこちら
🔗https://t.co/Lr129MFLBZ pic.twitter.com/7pZ6FqxZIM— Maki@Sunwood AI Labs. (@hAru_mAki_ch) March 26, 2024
Harinaの主な特徴は以下の通りです。
- Claude 3 Haikuモデルによる高速・高精度なレシート解析
- マルチチャンネル対応とカスタマイズ可能な設定
- Google Spreadsheetとのシームレスな連携
- Dockerによる簡単セットアップ
こちらの記事もおすすめ
セットアップ手順
1. リポジトリのクローンとディレクトリ移動
git clone https://github.com/your-username/Harina.git
cd Harina
2. Google Apps Scriptの設定
gas
フォルダ内のJSON to Spreadsheet Integrator.js
をGoogle Apps Script(GAS)にコピーします。
以下の箇所を自分の環境に合わせて書き換えてください。
var sheet = SpreadsheetApp.openById('スプレッドシートのID').getSheetByName('シート名');
編集したスクリプトをウェブアプリとして公開し、公開URLをコピーしておきます。
3. 環境変数の設定
以下の3つの環境変数を.env
ファイルまたはOSの環境変数として設定します。
DISCORD_BOT_TOKEN_HARINA
: Discord BotのTokenANTHROPIC_API_KEY
: Anthropic APIのAPIキーGAS_JSON2GSS_URL
: 手順2で公開したGASのURL
4. Dockerコンテナの起動
docker-compose up
上記のコマンドを実行するとDockerコンテナが起動し、設定したDiscordチャンネルでBotが稼働を開始します。
使い方
Harinaを使うのはとてもシンプルです。
- 設定したDiscordチャンネルにレシート画像を送信
- Botがレシートを解析し、Claude 3 Haikuモデルで画像からJSONデータを抽出
- 抽出したデータをチャットに整形して表示
- 同時に抽出データをGoogle Spreadsheetに自動保存
これだけで簡単にレシートの情報をデジタルデータ化できます。
コード解説
次に主要な2つのファイルのコードを解説します。
bot/Kakeibo_HARINA.py
コードの概要
このファイルはDiscord Botの本体です。主な役割は以下の通りです。
- Discord Botの初期化と設定
- メッセージイベントの処理
- 画像の保存とレシート解析の実行
- 解析結果の表示とスプレッドシートへの送信
注目すべきはprocess_attachments
関数の部分です。
async def process_attachments(message):
target_dir = get_target_dir(message.channel.id)
save_dir = f'data/img/{target_dir}'
for attachment in message.attachments:
url = attachment.url
date = datetime.datetime.now()
file_name = f"{date.strftime('%Y%m%d%H%M%S')}_{attachment.filename}".replace(".jpg.jpg", ".jpg")
if await save_image(url, save_dir, file_name):
await message.channel.send(f"画像を保存しました: {file_name}")
logger.info(f"画像を保存しました: {file_name}")
else:
await message.channel.send("画像の保存に失敗しました。")
logger.info("画像の保存に失敗しました。")
receipt = analyzer.analyze_receipts(save_dir, target_dir)
await message.channel.send(f"```{receipt}```")
この関数では、送信された画像を保存した後、ReceiptAnalyzer
クラスのanalyze_receipts
メソッドを呼び出して解析を実行しています。解析結果はチャットにJSON形式で表示されます。
詳細な解説
import discord
from discord.ext import commands
from loguru import logger
import urllib.request
import datetime
import os
import sys
import pprint
import requests
import json
import shutil
まず必要なライブラリをインポートしています。
discord
,discord.ext
: DiscordボットのAPIを使用するためのライブラリloguru
: ログ出力用のライブラリurllib
,requests
: HTTP通信を行うためのライブラリdatetime
,os
,sys
,json
,shutil
: 各種ユーティリティライブラリ
intents = discord.Intents.default()
intents.members = True
intents.message_content = True
Discordボットに必要なintents(ボットが受け取るイベントの種類)を設定しています。
ここでは、デフォルトのintentsに加えて、メンバー情報とメッセージ内容を受け取れるようにしています。
logger.add("log/debug.log", format="{time} {level} {message}", level="DEBUG")
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
pprint.pprint(sys.path)
from api.ConfigLoader import ConfigLoader
from api.CategoryLoader import CategoryLoader
from api.ReceiptAnalyzer import ReceiptAnalyzer
ログ出力の設定と、カスタムモジュールのインポートを行っています。
logger
: loguruライブラリを使ってロギングの設定をしています。ログはDEBUGレベル以上がlog/debug.log
に出力されます。sys.path.append
: カレントディレクトリの親ディレクトリをPythonのモジュール検索パスに追加しています。api/ConfigLoader.py
,api/CategoryLoader.py
,api/ReceiptAnalyzer.py
: レシート解析に必要なカスタムモジュールをインポートしています。
config_loader = ConfigLoader()
category_loader = CategoryLoader("data/category.csv")
analyzer = ReceiptAnalyzer(config_loader, category_loader)
TOKEN = os.getenv('DISCORD_BOT_TOKEN_HARINA')
bot = commands.Bot(command_prefix='!', intents=intents, help_command=None)
設定ファイルとカテゴリ定義ファイルをロードし、レシート解析器を初期化しています。
環境変数からDiscordボットのトークンを取得し、commands.Bot
でボットを初期化しています。
コマンドプレフィックスは!
に設定され、help_commandは無効化されています。
def send_data_to_gas(json_data, tag, file_name):
url = os.environ['GAS_JSON2GSS_URL']
headers = {'Content-Type': 'application/json'}
data = {
'json_data': json_data,
'tag': tag,
'file_name': file_name
}
response = requests.post(url, data=json.dumps(data), headers=headers)
logger.info(f"{file_name} is {response.text}")
Google Apps Script(GAS)にデータを送信する関数です。
環境変数からGASのURLを取得し、jsonデータ、タグ、ファイル名をPOSTリクエストで送信しています。
レスポンスの内容はログに出力されます。
def get_target_dir(channel_id):
return "maki" if channel_id == 1208743619029368836 else "sample"
Discordのチャンネルに応じて画像の保存先ディレクトリ名を決定する関数です。
channel_id
が特定の値なら"maki"
、それ以外なら"sample"
を返します。
async def save_image(url, save_dir, file_name):
os.makedirs(save_dir, exist_ok=True)
try:
opener = urllib.request.build_opener()
opener.addheaders = [('User-agent', 'Mozilla/5.0')]
urllib.request.install_opener(opener)
urllib.request.urlretrieve(url, os.path.join(save_dir, file_name))
return True
except Exception as e:
logger.error(f"画像の保存に失敗しました: {e}")
return False
URLから画像をダウンロードして保存する非同期関数です。
保存先ディレクトリを作成し、URLから画像をダウンロードしています。
User-Agentヘッダを設定することで、一部の画像URLからのダウンロードを可能にしています。
ダウンロードに成功したらTrueを、失敗したらFalseを返します。
async def process_attachments(message):
target_dir = get_target_dir(message.channel.id)
save_dir = f'data/img/{target_dir}'
for attachment in message.attachments:
url = attachment.url
date = datetime.datetime.now()
file_name = f"{date.strftime('%Y%m%d%H%M%S')}_{attachment.filename}".replace(".jpg.jpg", ".jpg")
if await save_image(url, save_dir, file_name):
await message.channel.send(f"画像を保存しました: {file_name}")
logger.info(f"画像を保存しました: {file_name}")
else:
await message.channel.send("画像の保存に失敗しました。")
logger.info("画像の保存に失敗しました。")
receipt = analyzer.analyze_receipts(save_dir, target_dir)
await message.channel.send(f"```{receipt}```")
output_folder = 'output'
archive_folder = 'archive'
os.makedirs(archive_folder, exist_ok=True)
for root, dirs, files in os.walk(output_folder):
for file_name in files:
if file_name.endswith('.json'):
file_path = os.path.join(root, file_name)
with open(file_path, 'r') as file:
json_data = json.load(file)
tag = os.path.basename(os.path.dirname(file_path))
send_data_to_gas(json_data, tag, file_name)
# JSONファイルをarchiveフォルダに移動
archive_path = os.path.join(archive_folder, tag)
os.makedirs(archive_path, exist_ok=True)
shutil.move(file_path, os.path.join(archive_path, file_name))
# imgフォルダ内の画像ファイルを削除
for root, dirs, files in os.walk(save_dir):
for file_name in files:
if file_name.endswith('.jpg'):
file_path = os.path.join(root, file_name)
os.remove(file_path)
Discordの添付ファイル(画像)を処理する非同期関数です。かなり長いので、ステップバイステップで説明します。
- チャンネルIDから画像の保存先ディレクトリを決定
- メッセージの添付ファイルを順にループ
- 画像URLをダウンロードして指定ディレクトリに保存
- 保存結果をチャットとログに出力
- 保存した画像をレシート解析器で解析し、結果をチャットに出力
output
ディレクトリ内のJSONファイルを処理- JSONファイルを読み込んでGASに送信
- 処理済みのJSONファイルを
archive
ディレクトリに移動
img
ディレクトリ内の画像ファイルを削除
これにより、毎回新しい画像だけを解析できるようになります。
@bot.event
async def on_ready():
logger.info("ログインしました")
ボットがDiscordに接続してログインした時に呼ばれるイベントハンドラです。
ここではログにログインメッセージを出力しています。
@bot.event
async def on_message(message):
if message.author.bot:
return
if message.content == '/harina':
await message.channel.send('ちゅん')
logger.info(f"channel id: {message.channel.id}")
if message.attachments:
logger.info(f"channel id: {message.channel.id}")
await process_attachments(message)
else:
await message.channel.send("添付ファイルが見つかりません。")
Discordのチャットにメッセージが投稿された時に呼ばれるイベントハンドラです。
- メッセージの投稿者がbotの場合は無視します
- メッセージ内容が
"/harina"
の場合は、"ちゅん"
とチャットに返信 - メッセージに添付ファイルがある場合は、
process_attachments
関数で処理 - 添付ファイルがない場合は、エラーメッセージをチャットに送信
bot.run(TOKEN)
最後にbotを実行しています。環境変数から取得したトークンを使ってDiscordに接続します。
コードの全体
import discord
from discord.ext import commands
from loguru import logger
import urllib.request
import datetime
import os
import sys
import pprint
import requests
import json
import shutil
intents = discord.Intents.default()
intents.members = True
intents.message_content = True
logger.add("log/debug.log", format="{time} {level} {message}", level="DEBUG")
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
pprint.pprint(sys.path)
from api.ConfigLoader import ConfigLoader
from api.CategoryLoader import CategoryLoader
from api.ReceiptAnalyzer import ReceiptAnalyzer
config_loader = ConfigLoader()
category_loader = CategoryLoader("data/category.csv")
analyzer = ReceiptAnalyzer(config_loader, category_loader)
TOKEN = os.getenv('DISCORD_BOT_TOKEN_HARINA')
bot = commands.Bot(command_prefix='!', intents=intents, help_command=None)
def send_data_to_gas(json_data, tag, file_name):
url = os.environ['GAS_JSON2GSS_URL']
headers = {'Content-Type': 'application/json'}
data = {
'json_data': json_data,
'tag': tag,
'file_name': file_name
}
response = requests.post(url, data=json.dumps(data), headers=headers)
logger.info(f"{file_name} is {response.text}")
def get_target_dir(channel_id):
return "maki" if channel_id == 1208743619029368836 else "sample"
async def save_image(url, save_dir, file_name):
os.makedirs(save_dir, exist_ok=True)
try:
opener = urllib.request.build_opener()
opener.addheaders = [('User-agent', 'Mozilla/5.0')]
urllib.request.install_opener(opener)
urllib.request.urlretrieve(url, os.path.join(save_dir, file_name))
return True
except Exception as e:
logger.error(f"画像の保存に失敗しました: {e}")
return False
async def process_attachments(message):
target_dir = get_target_dir(message.channel.id)
save_dir = f'data/img/{target_dir}'
for attachment in message.attachments:
url = attachment.url
date = datetime.datetime.now()
file_name = f"{date.strftime('%Y%m%d%H%M%S')}_{attachment.filename}".replace(".jpg.jpg", ".jpg")
if await save_image(url, save_dir, file_name):
await message.channel.send(f"画像を保存しました: {file_name}")
logger.info(f"画像を保存しました: {file_name}")
else:
await message.channel.send("画像の保存に失敗しました。")
logger.info("画像の保存に失敗しました。")
receipt = analyzer.analyze_receipts(save_dir, target_dir)
await message.channel.send(f"```{receipt}```")
output_folder = 'output'
archive_folder = 'archive'
os.makedirs(archive_folder, exist_ok=True)
for root, dirs, files in os.walk(output_folder):
for file_name in files:
if file_name.endswith('.json'):
file_path = os.path.join(root, file_name)
with open(file_path, 'r') as file:
json_data = json.load(file)
tag = os.path.basename(os.path.dirname(file_path))
send_data_to_gas(json_data, tag, file_name)
# JSONファイルをarchiveフォルダに移動
archive_path = os.path.join(archive_folder, tag)
os.makedirs(archive_path, exist_ok=True)
shutil.move(file_path, os.path.join(archive_path, file_name))
# imgフォルダ内の画像ファイルを削除
for root, dirs, files in os.walk(save_dir):
for file_name in files:
if file_name.endswith('.jpg'):
file_path = os.path.join(root, file_name)
os.remove(file_path)
@bot.event
async def on_ready():
logger.info("ログインしました")
@bot.event
async def on_message(message):
if message.author.bot:
return
if message.content == '/harina':
await message.channel.send('ちゅん')
logger.info(f"channel id: {message.channel.id}")
if message.attachments:
logger.info(f"channel id: {message.channel.id}")
await process_attachments(message)
else:
await message.channel.send("添付ファイルが見つかりません。")
bot.run(TOKEN)
api/ReceiptAnalyzer.py
コードの概要
このファイルが実際のレシート解析処理を担当しています。
analyze_receipts
メソッドでは以下のような処理を行っています。
- レシート画像をbase64エンコード
- システムプロンプトの生成
- Claude APIへのリクエスト送信
- レスポンスのJSONデータを整形して保存
def analyze_receipts(self, image_folder, save_dir):
"""
指定されたフォルダ内のすべてのレシート画像を解析します。
"""
# 保存ディレクトリのパスを生成
save_path = os.path.join("output", save_dir)
# 保存ディレクトリが存在しない場合は作成
os.makedirs(save_path, exist_ok=True)
image_paths = glob.glob(f"{image_folder}/*.jpg")
print(f"image_paths:{image_paths}")
for image_path in image_paths:
image_name = os.path.basename(image_path)
output_file_name = os.path.join(save_path, f"receipt_{image_name.replace('.jpg', '.json')}")
# 出力フォルダに同名のファイルが存在する場合はスキップ
if os.path.exists(output_file_name):
print(f"Skipping {image_name} as output JSON already exists.")
continue
base64_image = ImageEncoder.encode_image(image_path)
system_prompt = self.create_system_prompt(self.category_loader.categories)
message = self.client.messages.create(
model="claude-3-haiku-20240307",
max_tokens=1024,
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": system_prompt
},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": base64_image
}
}
],
}
],
)
# 応答データの処理
content = message.content[0].text
_receipt = content.replace("```json", "").replace("```", "")
receipt = json.loads(_receipt)
with open(output_file_name, 'w', encoding='utf-8') as f:
json.dump(receipt, f, ensure_ascii=False, indent=2)
print(f"Output saved to {output_file_name}")
return receipt
create_system_prompt
メソッドでは、予め用意したJSONの解析フォーマットとカテゴリの定義リストをプロンプトとして生成しています。
詳細な解説
import os
import json
import pprint
import glob
from api.APIRequestSender import APIRequestSender
from api.ImageEncoder import ImageEncoder
import anthropic
まず、必要なライブラリとモジュールをインポートしています。
os
,json
,glob
: ファイルやディレクトリを操作するための標準ライブラリapi.APIRequestSender
,api.ImageEncoder
: カスタムモジュール(詳細は不明)anthropic
: AnthropicのAPIを使用するためのライブラリ
class ReceiptAnalyzer:
def __init__(self, config_loader, category_loader):
self.config_loader = config_loader
self.category_loader = category_loader
self.client = anthropic.Anthropic()
ReceiptAnalyzer
クラスのコンストラクタです。以下の3つのインスタンス変数を初期化しています。
config_loader
: 設定をロードするためのオブジェクトcategory_loader
: カテゴリ定義をロードするためのオブジェクトclient
: AnthropicのAPIクライアント
def create_system_prompt(self, categories):
"""
システムプロンプトを生成します。
"""
# ユーザーがアップロードしたファイルを読み込んで内容を確認する
receipt_temp_path = 'data/RECEIPT_JSON_FORMAT.txt'
with open(receipt_temp_path, 'r', encoding='utf-8') as file:
receipt_json_format = file.read()
system_prompt = f"""
画像のレシートを下記のJSONフォーマットで解析してください。
カテゴリは下記のリストから適切なカテゴリを選択してください。
リストにないカテゴリは選択しないでください。
{categories}
## 出力フォーマット (JSON形式):
{receipt_json_format}
"""
return system_prompt
create_system_prompt
メソッドは、AIモデルに与えるシステムプロンプトを生成します。
data/RECEIPT_JSON_FORMAT.txt
ファイルからJSON出力フォーマットのテンプレートを読み込みます- 引数の
categories
と組み合わせて、レシート解析のための指示を含むプロンプトを生成します - 生成されたプロンプトを返します
def analyze_receipts(self, image_folder, save_dir):
"""
指定されたフォルダ内のすべてのレシート画像を解析します。
"""
# 保存ディレクトリのパスを生成
save_path = os.path.join("output", save_dir)
# 保存ディレクトリが存在しない場合は作成
os.makedirs(save_path, exist_ok=True)
image_paths = glob.glob(f"{image_folder}/*.jpg")
print(f"image_paths:{image_paths}")
for image_path in image_paths:
image_name = os.path.basename(image_path)
output_file_name = os.path.join(save_path, f"receipt_{image_name.replace('.jpg', '.json')}")
# 出力フォルダに同名のファイルが存在する場合はスキップ
if os.path.exists(output_file_name):
print(f"Skipping {image_name} as output JSON already exists.")
continue
base64_image = ImageEncoder.encode_image(image_path)
system_prompt = self.create_system_prompt(self.category_loader.categories)
message = self.client.messages.create(
model="claude-3-haiku-20240307",
max_tokens=1024,
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": system_prompt
},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": base64_image
}
}
],
}
],
)
# 応答データの処理
content = message.content[0].text
_receipt = content.replace("```json", "").replace("```", "")
receipt = json.loads(_receipt)
with open(output_file_name, 'w', encoding='utf-8') as f:
json.dump(receipt, f, ensure_ascii=False, indent=2)
print(f"Output saved to {output_file_name}")
return receipt
analyze_receipts
メソッドは、指定されたフォルダ内のすべてのレシート画像を解析します。
- 出力ディレクトリのパスを生成し、ディレクトリを作成
- 指定フォルダ内のJPEG画像ファイルのパスを取得
- 各画像ファイルに対して以下の処理を実行
- 出力JSONファイルのパスを生成
- 出力ファイルが既に存在する場合はスキップ
- 画像をBase64エンコード
- システムプロンプトを生成
- AnthropicのAPIを使用してレシート解析を実行
- モデルは"claude-3-haiku-20240307"を使用
- システムプロンプトと画像をAPIに送信
- APIの応答からJSONデータを抽出
- 抽出したJSONデータを出力ファイルに保存
- 最後に解析されたレシートのJSONデータを返す
以上がコードの詳細な解説になります。
ReceiptAnalyzer
クラスは、指定されたフォルダ内のレシート画像を順次解析し、結果をJSONファイルとして出力します。解析にはAnthropicのClaude 3 Haikuモデルが使用されています。
gas/JSON to Spreadsheet Integrator.js
JSONデータの受信と解析
function doPost(e) {
var data = JSON.parse(e.postData.contents);
var jsonData = data.json_data;
var tag = data.tag;
var fileName = data.file_name;
// ...
}
この部分では、POSTリクエストで受け取ったデータを処理しています。
e.postData.contents
には、リクエストのボディに含まれるJSONデータが文字列として格納されています。JSON.parse()
を使ってJSONデータをJavaScriptオブジェクトに変換します。data
オブジェクトから、json_data
、tag
、file_name
の値を取り出します。
スプレッドシートの選択
var sheet = SpreadsheetApp.openById('スプレッドシートのID').getSheetByName('シート名');
この行では、データの書き込み先となるGoogle Spreadsheetsのシートを選択しています。
'スプレッドシートのID'
には、実際のスプレッドシートのIDを指定します。'シート名'
には、データを書き込むシートの名前を指定します。
JSONデータの展開
var storeData = jsonData.store;
var transactionData = jsonData.transaction;
var itemsData = jsonData.items;
var totalData = jsonData.total;
var paymentData = jsonData.payment;
ここでは、受け取ったJSONデータを、店舗情報、取引情報、商品情報、合計情報、支払情報に分解しています。これにより、データを扱いやすくなります。
スプレッドシートへのデータ書き込み
for (var i = 0; i < itemsData.length; i++) {
var itemData = itemsData[i];
var rowData = [
tag,
fileName,
storeData.name,
transactionData.date,
transactionData.time,
itemData.item_name,
itemData.unit_price,
itemData.quantity,
itemData.unit,
itemData.category,
itemData.total_price,
totalData.amount,
totalData.points_earned,
totalData.points_used,
paymentData.payment_method
];
sheet.appendRow(rowData);
}
この部分が、JSONデータをスプレッドシートに書き込む主要な処理です。
- 商品情報の配列
itemsData
をループ処理します。 - 各商品について、スプレッドシートの1行分のデータを
rowData
配列に格納します。- タグ、ファイル名、店舗名、取引日時など、関連するデータを配列に追加します。
sheet.appendRow(rowData)
を呼び出して、rowData
配列をスプレッドシートの新しい行として追加します。
これにより、JSONデータに含まれる各商品の情報が、スプレッドシートの行として追加されます。
レスポンスの返信
return ContentService.createTextOutput('Data received successfully');
最後に、クライアントに対して処理が成功したことを示すレスポンスを返します。
ContentService.createTextOutput()
を使って、テキストレスポンスを作成します。- レスポンスのボディには、
'Data received successfully'
というメッセージを含めます。
以上が、このGoogle Apps Scriptコードの詳細な解説になります。JSONデータを受信し、スプレッドシートに書き込む一連の流れが理解できたでしょうか。
おわりに
以上がHarina Discord Botの概要と主要コードの解説でした。
最先端の言語モデルを活用することで、手軽にレシートのデジタル化を実現できるのがHarinaの強みです。
ぜひ皆さんもHarinaを使って、家計管理やビジネスの効率化に役立ててみてください。
導入や利用方法でご不明な点があれば、お気軽にお問い合わせください。
コメント