linebotへ送信 – Linux & Android Dialy

linebotへ送信

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がわからなかったため検索

開発者が自分自身のユーザーIDを取得する

を参考に

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設定ができていないのでエラーになる

コメントを残す コメントをキャンセル

コメントを残すコメントをキャンセル