linebotへ送信
実行環境
M1 MacbookAir 16GB
Ollamaで修正されたテキストをLINE botに送信するため
既存のLINE Notifyの処理と合わせて、修正された文章をLINE botに送信する処理を実装
LINE botにメッセージを送信するには、LINE Messaging APIを利用する
LINE botにメッセージを送信するためのモジュールを作る
touch line_bot_sender.py
中身は
# line_bot_sender.py
import requests
import json
class LineBotSender:
def __init__(self, config_file_path):
self.config = self._load_config(config_file_path)
self.channel_access_token = self.config["line_bot_channel_access_token"]
self.user_id = self.config["line_bot_user_id"]
def _load_config(self, config_file_path):
with open(config_file_path, 'r') as file:
return json.load(file)
def send_message(self, message):
url = "https://api.line.me/v2/bot/message/push"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.channel_access_token}"
}
payload = {
"to": self.user_id,
"messages": [
{
"type": "text",
"text": message
}
]
}
response = requests.post(url, headers=headers, json=payload)
if response.status_code != 200:
raise Exception(f"Error sending message to LINE bot: {response.status_code}, {response.text}")
として保存
config.json にLINE botの情報を追加
{
"line_notify_token": "YOUR_LINE_NOTIFY_TOKEN",
"ollama_model": "elyza:jp8b",
"line_bot_channel_access_token": "YOUR_LINE_BOT_CHANNEL_ACCESS_TOKEN",
"line_bot_user_id": "TARGET_USER_ID"
}
linebot useridがわからなかったため検索
を参考に
LINE Developersコンソール
で
基本情報のプロバイダーIDを入れてみる
次にOllamaで修正されたテキストをLINE botに送信するために、メインスクリプトにLineBotSenderを組み込む
import sounddevice as sd
from module.module_whisper import FasterWhisperModel
from module.module_recorder import Recorder
import time
from line_notify import LineNotify # 作成したLineNotifyモジュールをインポート
from ollama_text_correction import OllamaTextCorrector # Ollamaによる修正モジュールをインポート
def main():
recorder = Recorder()
fasterWhispermodel = FasterWhisperModel()
# 入力された音声テキストを格納するリスト
recognized_texts = []
# LINE Notifyのモジュールを初期化（config.jsonからトークンを読み込む）
line_notify = LineNotify("config.json")
# Ollamaのテキスト修正モジュールを初期化
text_corrector = OllamaTextCorrector("config.json")
while True:
start_time = time.time() # 処理開始時刻を記録
audio_data = recorder.speech2audio()
# 処理が10秒間行われなかった場合はループを抜ける
if time.time() - start_time >= 10:
print("10秒間音声が入力されなかったため、ループを終了します。")
break
if audio_data is None:
print("無音状態が続いたため、ループを終了します。")
break # 無音でループを抜ける
# 音声をテキストに変換
text = fasterWhispermodel.audio2text(audio_data)
# Ollamaでテキストを構成
corrected_text = text_corrector.correct_text(text)
if corrected_text: # Noneが返された場合はスキップ
recognized_texts.append(corrected_text)
print(corrected_text)
# ループ終了後に、入力した音声テキストを改行付きで一覧表示
if recognized_texts:
message = "\n".join(recognized_texts)
print("\n入力された音声テキスト一覧:")
print(message)
# LINE Notifyでメッセージを送信
line_notify.send(f"入力された音声テキスト一覧:\n{message}")
else:
print("入力メッセージはありませんでした")
if __name__ == "__main__":
main()
を書き換えるとプロンプト対処があとで困るので
これをコピーし main2.pyとしてコードを記述
import sounddevice as sd
from module.module_whisper import FasterWhisperModel
from module.module_recorder import Recorder
import time
from line_notify import LineNotify # 作成したLineNotifyモジュールをインポート
from ollama_text_correction import OllamaTextCorrector # Ollamaによる修正モジュールをインポート
from line_bot_sender import LineBotSender # LINE bot送信用のモジュールをインポート
def main():
recorder = Recorder()
fasterWhispermodel = FasterWhisperModel()
# 入力された音声テキストを格納するリスト
recognized_texts = []
# LINE Notifyのモジュールを初期化（config.jsonからトークンを読み込む）
line_notify = LineNotify("config.json")
# Ollamaのテキスト修正モジュールを初期化
text_corrector = OllamaTextCorrector("config.json")
# LINE bot送信用のモジュールを初期化
line_bot_sender = LineBotSender("config.json")
while True:
start_time = time.time() # 処理開始時刻を記録
audio_data = recorder.speech2audio()
# 処理が10秒間行われなかった場合はループを抜ける
if time.time() - start_time >= 10:
print("10秒間音声が入力されなかったため、ループを終了します。")
break
if audio_data is None:
print("無音状態が続いたため、ループを終了します。")
break # 無音でループを抜ける
# 音声をテキストに変換
text = fasterWhispermodel.audio2text(audio_data)
# Ollamaでテキストを構成
corrected_text = text_corrector.correct_text(text)
if corrected_text: # Noneが返された場合はスキップ
recognized_texts.append(corrected_text)
print(corrected_text)
# ループ終了後に、入力した音声テキストを改行付きで一覧表示
if recognized_texts:
message = "\n".join(recognized_texts)
print("\n入力された音声テキスト一覧:")
print(message)
# LINE Notifyでメッセージを送信
line_notify.send(f"入力された音声テキスト一覧:\n{message}")
# LINE botで修正されたテキストを送信
line_bot_sender.send_message(f"修正された音声テキスト:\n{message}")
else:
print("入力メッセージはありませんでした")
if __name__ == "__main__":
main()
これで実行したら
入力された音声テキスト一覧:
本日の天候
Traceback (most recent call last):
File "/Users/snowpool/aw10s/linebot/main2.py", line 63, in <module>
main()
File "/Users/snowpool/aw10s/linebot/main2.py", line 58, in main
line_bot_sender.send_message(f"修正された音声テキスト:\n{message}")
File "/Users/snowpool/aw10s/linebot/line_bot_sender.py", line 32, in send_message
raise Exception(f"Error sending message to LINE bot: {response.status_code}, {response.text}")
Exception: Error sending message to LINE bot: 400, {"message":"The property, 'to', in the request body is invalid (line: -, column: -)"}
となった
config.jsonに "channel_secret": "YOUR_CHANNEL_SECRET",
を追加する
次に
line_bot_sender.pyを修正する
line_bot_sender.py ファイルで、
send_message メソッド内のリクエストボディに ‘to’ フィールドを追加し、
config.jsonから読み込んだ user_id を設定
def send_message(self, message):
url = "https://api.line.me/v2/bot/message/push"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.channel_access_token}"
}
payload = {
"to": self.user_id,
"messages": [
{
"type": "text",
"text": message
}
]
}
response = requests.post(url, headers=headers, json=payload)
if response.status_code != 200:
raise Exception(f"Error sending message to LINE bot: {response.status_code}, {response.text}")
を
def send_message(self, message):
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {self.channel_access_token}'
}
data = {
"to": self.user_id, # 追加
"messages": [
{
"type": "text",
"text": message
}
]
}
response = requests.post(self.api_endpoint, headers=headers, json=data)
if response.status_code != 200:
raise Exception(f"Error sending message to LINE bot: {response.status_code}, {response.text}")
に変更
line_bot_sender.py のクラス内で user_id を読み込むように変更
# line_bot_sender.py
import requests
import json
class LineBotSender:
def __init__(self, config_path):
with open(config_path, 'r') as file:
config = json.load(file)
self.channel_access_token = config.get('channel_access_token')
self.channel_secret = config.get('channel_secret')
self.user_id = config.get('user_id') # 追加
self.api_endpoint = 'https://api.line.me/v2/bot/message/push' # push APIを使用
def _load_config(self, config_file_path):
with open(config_file_path, 'r') as file:
return json.load(file)
def send_message(self, message):
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {self.channel_access_token}'
}
data = {
"to": self.user_id, # 追加
"messages": [
{
"type": "text",
"text": message
}
]
}
response = requests.post(self.api_endpoint, headers=headers, json=data)
if response.status_code != 200:
raise Exception(f"Error sending message to LINE bot: {response.status_code}, {response.text}")
というようにした
これだけだとまだwebhook設定ができていないのでエラーになる