Linebot 画像送信機能の追加

Linebot 画像送信機能の追加

{
  "token": "",
  "image_file_path": "runs/detect",
  "line_bot_channel_access_token": "",
  "channel_secret": "",
  "line_bot_user_id": ""
}

というように
Config.jsonの内容を追加

import json
import cv2
import configparser
from ultralytics import YOLO
from collections import defaultdict
from datetime import datetime
import os

from line_bot_sender import LineBotSender  # LineBotSender をインポート
from inventory_database_module import save_detection_to_db  # データベース保存用の関数をインポート

# 設定ファイルの読み込み
config = configparser.ConfigParser()
config.read('config.ini')

# 設定ファイルからモデルパスと画像ディレクトリを取得
model_path = config['Settings']['model_path']
image_directory = config['Settings']['image_directory']

# ラベルマッピングファイルのパス
label_mapping_path = 'label_mapping.json'

# JSONファイルからクラスラベルのマッピングを読み込み
with open(label_mapping_path, 'r', encoding='utf-8') as f:
    label_mapping = json.load(f)

# LINE Bot 送信クラスのインスタンスを作成
line_bot = LineBotSender("config.json")  # `config.json` を指定

# YOLOv8モデルのロード
model = YOLO(model_path)  # 設定ファイルからモデルパスを使用

# 画像ディレクトリ内の全画像ファイルを処理
for image_filename in os.listdir(image_directory):
    image_path = os.path.join(image_directory, image_filename)
    if os.path.isfile(image_path) and image_path.lower().endswith(('.png', '.jpg', '.jpeg')):
        # 画像のロード
        image = cv2.imread(image_path)

        # 画像の検出
        results = model(image, save=True, conf=0.2, iou=0.5)

        # 検出結果の取得
        detections = results[0]  # 最初の結果を取得
        classes = detections.boxes.cls

        # 検出物体のカウント
        object_counts = defaultdict(int)
        for cls in classes:
            class_label = model.names[int(cls)]
            if class_label in label_mapping:
                label = label_mapping[class_label]
            else:
                label = class_label
            object_counts[label] += 1

        # 検出結果のフィルタリング（1以下のもの）
        filtered_object_counts = {label: count for label, count in object_counts.items() if count <= 1}

        # フィルタリングされた検出結果のメッセージ生成
        message_lines = [f'{label}: {count}個' for label, count in filtered_object_counts.items()]
        message = '\n'.join(message_lines)

        # 現在の時刻を取得
        current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        message = f"{message}\n\nMessage sent at: {current_time}"

        # 検出結果の表示
        for line in message_lines:
            print(line)

        # LINE Bot にメッセージを送信（フィルタリングされた結果のみ）
        if message_lines:
            line_bot.send_message(message)  # `LineBotSender` を使用
            save_detection_to_db(filtered_object_counts)  # データベースに検出結果を保存
        else:
            print("No objects with counts of 1 or less detected in file:", image_filename)

というように

cp count_inventory.py count_inventory_send_bot.py

で
Bot送信のテストのためソースを分ける

変更点は
send_line_notify(message) を
line_bot.send_message(message) に変更

LineBotSender を line_bot_sender.py からインポートして、インスタンスを作成
config.json を LINE Bot の設定情報取得用 に使用

from line_bot_sender import LineBotSender  # LineBotSender をインポート

line_bot = LineBotSender("config.json")  # `config.json` を指定

# 変更前（LINE Notify を使っていた部分）
# send_line_notify(message)

# 変更後（LINE Bot を使用）
line_bot.send_message(message)

そしてDBへの保存を
# LINE Bot にメッセージを送信（フィルタリングされた結果のみ）
if message_lines:
    line_bot.send_message(message)  # `LineBotSender` を使用
    save_detection_to_db(filtered_object_counts)  # データベースに検出結果を保存

へ変更

python count_inventory_send_bot.py

で実行すると

0: 640x512 1 baskulin, 74.1ms
Speed: 5.2ms preprocess, 74.1ms inference, 4.6ms postprocess per image at shape (1, 3, 640, 512)
Results saved to runs/detect/predict24
バスクリン: 1個

とbot への送信ができたけど
Notifyの時の画像送信機能を作成していなかったので
これを追加する

import requests
import json
import os
from io import BytesIO
from PIL import Image

class LineBotSender:
    def __init__(self, config_path):
        with open(config_path, 'r') as file:
            config = json.load(file)
        self.channel_access_token = config.get('line_bot_channel_access_token')
        self.channel_secret = config.get('channel_secret')
        self.user_id = config.get('line_bot_user_id')  # ユーザーID
        self.api_endpoint = 'https://api.line.me/v2/bot/message/push'  # メッセージ送信用API
        self.image_api_endpoint = 'https://api.line.me/v2/bot/message/push'  # 画像送信用API

    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}")

    def send_image(self, image_path, message="画像を送信します"):
        headers = {
            'Authorization': f'Bearer {self.channel_access_token}'
        }
        with open(image_path, 'rb') as img_file:
            image_data = img_file.read()
            image_data = self.resize_image_if_needed(image_data)

            files = {'imageFile': BytesIO(image_data)}
            files['imageFile'].name = os.path.basename(image_path)

            payload = {
                "to": self.user_id,
                "messages": [
                    {
                        "type": "text",
                        "text": message
                    },
                    {
                        "type": "image",
                        "originalContentUrl": f"https://your-server.com/images/{os.path.basename(image_path)}",
                        "previewImageUrl": f"https://your-server.com/images/{os.path.basename(image_path)}"
                    }
                ]
            }

            response = requests.post(self.image_api_endpoint, headers=headers, json=payload, files=files)

            if response.status_code != 200:
                raise Exception(f"Error sending image to LINE bot: {response.status_code}, {response.text}")

    def resize_image_if_needed(self, image_data, max_size=3 * 1024 * 1024):
        if len(image_data) > max_size:
            image = Image.open(BytesIO(image_data))
            new_size = (image.width // 2, image.height // 2)
            image = image.resize(new_size, Image.LANCZOS)

            output = BytesIO()
            image_format = image.format if image.format else 'JPEG'
            image.save(output, format=image_format)
            return output.getvalue()
        return image_data

✅ send_image() メソッドを追加し、画像を送信できるようにした
✅ resize_image_if_needed() で 3MB 超過時にリサイズする処理を追加
✅ originalContentUrl と previewImageUrl は 公開 URL に設定する必要があるため、適宜サーバーにアップロードする処理が必要

しかし
また config.json で “image_file_path”: “runs/detect”,
が設定してあるので
コードを変更する

import requests
import json
import os
from io import BytesIO
from PIL import Image

class LineBotSender:
    def __init__(self, config_path):
        with open(config_path, 'r') as file:
            config = json.load(file)
        self.channel_access_token = config.get('line_bot_channel_access_token')
        self.channel_secret = config.get('channel_secret')
        self.user_id = config.get('line_bot_user_id')  # ユーザーID
        self.image_file_path = config.get('image_file_path')  # YOLOの検出画像ディレクトリ
        self.api_endpoint = 'https://api.line.me/v2/bot/message/push'

    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}")

    def send_image(self, image_path, message="検出された画像を送信します"):
        headers = {
            'Authorization': f'Bearer {self.channel_access_token}',
            'Content-Type': 'application/json'
        }

        payload = {
            "to": self.user_id,
            "messages": [
                {
                    "type": "text",
                    "text": message
                },
                {
                    "type": "image",
                    "originalContentUrl": f"https://your-server.com/images/{os.path.basename(image_path)}",
                    "previewImageUrl": f"https://your-server.com/images/{os.path.basename(image_path)}"
                }
            ]
        }

        response = requests.post(self.api_endpoint, headers=headers, json=payload)

        if response.status_code != 200:
            raise Exception(f"Error sending image to LINE bot: {response.status_code}, {response.text}")

    def get_latest_detected_image(self):
        """ `image_file_path` ディレクトリから最新の検出画像を取得 """
        if not os.path.exists(self.image_file_path):
            print("Error: 指定されたディレクトリが存在しません:", self.image_file_path)
            return None

        images = sorted(
            [os.path.join(self.image_file_path, f) for f in os.listdir(self.image_file_path)
             if f.lower().endswith(('.png', '.jpg', '.jpeg'))],
            key=os.path.getmtime, reverse=True
        )

        return images[0] if images else None

なお
https://kotovuki.co.jp/archives/17715
によれば
1つの画像の最大ファイルサイズは 10MBらしい

1回のメッセージには吹き出しの上限が5つ
テキストの場合は 1つの吹き出しには最大5000文字まで入力可能

とりあえずサイズを縮小する機能が必要

縦横 50% に縮小

new_size = (image.width // 2, image.height // 2)  # 縦横 50% 縮小
image = image.resize(new_size, Image.LANCZOS)

JPEG/PNG を維持して保存

image_format = image.format if image.format else 'JPEG'
image.save(output, format=image_format)

リサイズ後の画像データを返す

return output.getvalue()

画像サイズ リサイズ処理
2MB そのまま送信
4MB リサイズして 50% 縮小
8MB リサイズして 50% 縮小（さらに 3MB 超えていれば再縮小）

✅ 3MB 以上の画像は自動で縮小
✅ JPEG/PNG フォーマットを保持
✅ LINE Bot に適したサイズで送信可能

ということで

import requests
import json
import os
from io import BytesIO
from PIL import Image

class LineBotSender:
    def __init__(self, config_path):
        """設定ファイルからLINE Botの情報を読み込む"""
        with open(config_path, 'r') as file:
            config = json.load(file)
        self.channel_access_token = config.get('line_bot_channel_access_token')
        self.channel_secret = config.get('channel_secret')
        self.user_id = config.get('line_bot_user_id')  # ユーザーID
        self.image_file_path = config.get('image_file_path')  # YOLOの検出画像ディレクトリ
        self.api_endpoint = 'https://api.line.me/v2/bot/message/push'

    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}")

    def send_image(self, image_path, message="検出された画像を送信します"):
        """画像を送信（3MB超えた場合はリサイズ）"""
        if not os.path.exists(image_path):
            print(f"Error: 画像が見つかりません: {image_path}")
            return
        
        headers = {
            'Authorization': f'Bearer {self.channel_access_token}',
            'Content-Type': 'application/json'
        }

        with open(image_path, 'rb') as img_file:
            image_data = img_file.read()
            image_data = self.resize_image_if_needed(image_data)  # 3MB超えたらリサイズ

            # LINE Bot の API 用データを作成
            payload = {
                "to": self.user_id,
                "messages": [
                    {
                        "type": "text",
                        "text": message
                    },
                    {
                        "type": "image",
                        "originalContentUrl": f"https://your-server.com/images/{os.path.basename(image_path)}",
                        "previewImageUrl": f"https://your-server.com/images/{os.path.basename(image_path)}"
                    }
                ]
            }

            response = requests.post(self.api_endpoint, headers=headers, json=payload)

            if response.status_code != 200:
                raise Exception(f"Error sending image to LINE bot: {response.status_code}, {response.text}")

    def resize_image_if_needed(self, image_data, max_size=3 * 1024 * 1024):
        """画像が max_size (3MB) を超える場合はリサイズ"""
        while len(image_data) > max_size:
            image = Image.open(BytesIO(image_data))
            new_size = (image.width // 2, image.height // 2)  # 縦横 50% 縮小
            image = image.resize(new_size, Image.LANCZOS)

            output = BytesIO()
            image_format = image.format if image.format else 'JPEG'
            image.save(output, format=image_format)
            image_data = output.getvalue()

        return image_data  # 3MB 以下になった画像を返す

    def get_latest_detected_image(self):
        """ `image_file_path` ディレクトリから最新の検出画像を取得 """
        if not os.path.exists(self.image_file_path):
            print("Error: 指定されたディレクトリが存在しません:", self.image_file_path)
            return None

        images = sorted(
            [os.path.join(self.image_file_path, f) for f in os.listdir(self.image_file_path)
             if f.lower().endswith(('.png', '.jpg', '.jpeg'))],
            key=os.path.getmtime, reverse=True
        )

        return images[0] if images else None


if __name__ == "__main__":
    sender = LineBotSender("config.json")  # 設定ファイルのパスを指定
    sender.send_message("こんにちは！")  # LINE Bot で送信

    latest_image = sender.get_latest_detected_image()
    if latest_image:
        sender.send_image(latest_image, "最新の検出画像を送信します")

これだけで実行しても画像がつかない

とりあえず
GAS の Webhook エンドポイントを利用し、ローカルの画像を LINE に送信 できるようにする