チラシの商品名をLINE送信

チラシの商品名をLINE送信

ocr_list.py
でキーワードと一致した商品名の取得ができるので
これを
line_notify.pyにセットして送信するようにする

import json
from google.cloud import vision
import io

# 設定ファイルの読み込み
settings_json = open('settings.json', 'r', encoding='utf_8')
settings = json.load(settings_json)

# OCRで画像からテキストを抽出
def detect_text(image_paths):
    client = vision.ImageAnnotatorClient()

    all_text = ''

    for image_path in image_paths:
        with io.open(image_path, 'rb') as image_file:
            content = image_file.read()

        image = vision.Image(content=content)

        # document_text_detectionを使用して文書全体のテキストを取得
        response = client.document_text_detection(image=image)
        full_text_annotation = response.full_text_annotation

        # テキストの抽出
        all_text += full_text_annotation.text

        if response.error.message:
            raise Exception(
                '{}\nFor more info on error messages, check: '
                'https://cloud.google.com/apis/design/errors'.format(
                    response.error.message))

    return all_text

# キーワード検索
def search_words(all_text):
    hitwords = []
    for keyword in settings["keywords"]:
        if keyword in all_text:
            hitwords.append(keyword)

    return hitwords

# 例として実行
if __name__ == "__main__":
    image_paths = ["images/combined_image_20240805.jpg"]
    extracted_text = detect_text(image_paths)
    hitwords = search_words(extracted_text)
    
    # ヒットしたキーワードのみを表示
    if hitwords:
        print("マッチしたキーワード:", ", ".join(hitwords))
    else:
        print("マッチしたキーワードはありませんでした。")

の中で

import requests
import os
from PIL import Image
from io import BytesIO
from utils import load_config, get_latest_directory, get_image_files

def resize_image_if_needed(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

def send_line_notify(message, config_path='config.json'):
    # 設定ファイルを読み込む
    config = load_config(config_path)

    # 設定ファイルからトークンとディレクトリパスを取得
    token = config['token']
    base_path = config['image_file_path']

    # 最新のpredictディレクトリを取得
    latest_dir = get_latest_directory(base_path)
    image_files = get_image_files(latest_dir)

    url = 'https://notify-api.line.me/api/notify'

    headers = {'Authorization': f"Bearer {token}"}
    params = {'message': message}

    # 最新のpredictディレクトリ内の全ての画像ファイルに対してLINE Notify APIにリクエストを送信
    for image_file_path in image_files:
        with open(image_file_path, 'rb') as img_file:
            img_data = img_file.read()
            img_data = resize_image_if_needed(img_data)

            # ファイルデータをバイトデータとして用意
            files = {'imageFile': BytesIO(img_data)}
            files['imageFile'].name = os.path.basename(image_file_path)

            # LINE Notify APIにリクエストを送信
            res = requests.post(url, headers=headers, params=params, files=files)

            # レスポンスを出力
            print(f"File: {image_file_path}")
            print(res.status_code)
            print(res.text)

を呼び出し
hit words を line notifyで送信したい

その前に

import requests
import os
from PIL import Image
from io import BytesIO
from utils import load_config, get_latest_directory, get_image_files

def resize_image_if_needed(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

def send_line_notify(message, config_path='config.json'):
    # 設定ファイルを読み込む
    config = load_config(config_path)

    # 設定ファイルからトークンとディレクトリパスを取得
    token = config['token']
    base_path = config['image_file_path']

    # 最新のpredictディレクトリを取得
    latest_dir = get_latest_directory(base_path)
    image_files = get_image_files(latest_dir)

    url = 'https://notify-api.line.me/api/notify'

    headers = {'Authorization': f"Bearer {token}"}
    params = {'message': message}

    # 最新のpredictディレクトリ内の全ての画像ファイルに対してLINE Notify APIにリクエストを送信
    for image_file_path in image_files:
        with open(image_file_path, 'rb') as img_file:
            img_data = img_file.read()
            img_data = resize_image_if_needed(img_data)

            # ファイルデータをバイトデータとして用意
            files = {'imageFile': BytesIO(img_data)}
            files['imageFile'].name = os.path.basename(image_file_path)

            # LINE Notify APIにリクエストを送信
            res = requests.post(url, headers=headers, params=params, files=files)

            # レスポンスを出力
            print(f"File: {image_file_path}")
            print(res.status_code)
            print(res.text)

で今回はyolov8は使ってないので
まずはテキストのみにする

import requests
import os
from utils import load_config

def send_line_notify(message, config_path='config.json'):
    # 設定ファイルを読み込む
    config = load_config(config_path)

    # 設定ファイルからトークンを取得
    token = config['token']

    url = 'https://notify-api.line.me/api/notify'

    headers = {'Authorization': f"Bearer {token}"}
    params = {'message': message}

    # LINE Notify APIにリクエストを送信
    res = requests.post(url, headers=headers, params=params)

    # レスポンスを出力
    print(res.status_code)
    print(res.text)

# 例として実行
if __name__ == "__main__":
    message = "マッチしたキーワード: サンプルキーワード"
    send_line_notify(message)

とりあえず main 部分は削除して
ocr_list.pyの中で呼び出したい

import json
from google.cloud import vision
import io
from line_notify import send_line_notify

# 設定ファイルの読み込み
def load_settings(file_path='settings.json'):
    with open(file_path, 'r', encoding='utf_8') as settings_json:
        return json.load(settings_json)

# OCRで画像からテキストを抽出
def detect_text(image_paths):
    client = vision.ImageAnnotatorClient()

    all_text = ''

    for image_path in image_paths:
        with io.open(image_path, 'rb') as image_file:
            content = image_file.read()

        image = vision.Image(content=content)

        # document_text_detectionを使用して文書全体のテキストを取得
        response = client.document_text_detection(image=image)
        full_text_annotation = response.full_text_annotation

        # テキストの抽出
        all_text += full_text_annotation.text

        if response.error.message:
            raise Exception(
                '{}\nFor more info on error messages, check: '
                'https://cloud.google.com/apis/design/errors'.format(
                    response.error.message))

    return all_text

# キーワード検索
def search_words(all_text, keywords):
    hitwords = []
    for keyword in keywords:
        if keyword in all_text:
            hitwords.append(keyword)

    return hitwords

# 例として実行
if __name__ == "__main__":
    settings = load_settings()
    image_paths = ["images/combined_image_20240805.jpg"]
    extracted_text = detect_text(image_paths)
    hitwords = search_words(extracted_text, settings["keywords"])
    
    # ヒットしたキーワードをLINE Notifyで送信
    if hitwords:
        message = "マッチしたキーワード: " + ", ".join(hitwords)
        send_line_notify(message)
    else:
        print("マッチしたキーワードはありませんでした。")

これを実行したらLINEで送信されたのでOK

このままだとメッセージが分かりにくいので
マッチしたキーワードから
特売リスト
にメッセージを変更

そしてOCRした画像ファイルも一緒にLINE送信するように
line_notify.pyのソースを変更

import requests
import os
from utils import load_config

def send_line_notify(message, config_path='config.json'):
    # 設定ファイルを読み込む
    config = load_config(config_path)

    # 設定ファイルからトークンを取得
    token = config['token']

    url = 'https://notify-api.line.me/api/notify'

    headers = {'Authorization': f"Bearer {token}"}
    params = {'message': message}

    # LINE Notify APIにリクエストを送信
    res = requests.post(url, headers=headers, params=params)

    # レスポンスを出力
    print(res.status_code)
    print(res.text)

を

import requests
from utils import load_config
from io import BytesIO
from PIL import Image

def resize_image_if_needed(image_data, max_size=3 * 1024 * 1024):
    """
    画像が指定されたサイズを超える場合は、画像のサイズを縮小する。

    Args:
        image_data (bytes): 画像データ。
        max_size (int): 最大ファイルサイズ（バイト）。

    Returns:
        bytes: サイズ変更後の画像データ。
    """
    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

def send_line_notify(message, image_path=None, config_path='config.json'):
    """
    LINE Notifyを使用してメッセージとオプションで画像を送信する関数。

    Args:
        message (str): 送信するメッセージ。
        image_path (str): 送信する画像のパス。
        config_path (str): 設定ファイルのパス。

    Returns:
        None
    """
    # 設定ファイルを読み込む
    config = load_config(config_path)

    # 設定ファイルからトークンを取得
    token = config['token']

    url = 'https://notify-api.line.me/api/notify'

    headers = {'Authorization': f"Bearer {token}"}
    params = {'message': message}

    # 画像がある場合は読み込み
    files = None
    if image_path:
        with open(image_path, 'rb') as img_file:
            img_data = img_file.read()
            img_data = resize_image_if_needed(img_data)
            files = {'imageFile': BytesIO(img_data)}
            files['imageFile'].name = os.path.basename(image_path)

    # LINE Notify APIにリクエストを送信
    res = requests.post(url, headers=headers, params=params, files=files)

    # レスポンスを出力
    print(res.status_code)
    print(res.text)

しかし

WARNING: All log messages before absl::InitializeLog() is called are written to STDERR
I0000 00:00:1723040085.190147 14567125 config.cc:230] gRPC experiments enabled: call_status_override_on_cancellation, event_engine_dns, event_engine_listener, http2_stats_fix, monitoring_experiment, pick_first_new, trace_record_callops, work_serializer_clears_time_cache
Traceback (most recent call last):
  File "/Users/snowpool/aw10s/store_adversting_list/ocr_list.py", line 57, in <module>
    send_line_notify(message,image_path)
  File "/Users/snowpool/aw10s/store_adversting_list/line_notify.py", line 54, in send_line_notify
    with open(image_path, 'rb') as img_file:
TypeError: expected str, bytes or os.PathLike object, not list

となる

send_line_notify関数は単一の画像パスを期待しているため、リストから1つずつ画像パスを取り出して送信する必要があります。

ソースを

import requests
from utils import load_config
from io import BytesIO
from PIL import Image
import os

def resize_image_if_needed(image_data, max_size=3 * 1024 * 1024):
    """
    画像が指定されたサイズを超える場合は、画像のサイズを縮小する。

    Args:
        image_data (bytes): 画像データ。
        max_size (int): 最大ファイルサイズ（バイト）。

    Returns:
        bytes: サイズ変更後の画像データ。
    """
    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

def send_line_notify(message, image_paths=None, config_path='config.json'):
    """
    LINE Notifyを使用してメッセージとオプションで画像を送信する関数。
    複数の画像パスをリストとして受け取ることができます。

    Args:
        message (str): 送信するメッセージ。
        image_paths (list): 送信する画像のパスのリスト。
        config_path (str): 設定ファイルのパス。

    Returns:
        None
    """
    # 設定ファイルを読み込む
    config = load_config(config_path)

    # 設定ファイルからトークンを取得
    token = config['token']

    url = 'https://notify-api.line.me/api/notify'

    headers = {'Authorization': f"Bearer {token}"}
    params = {'message': message}

    # 画像がリストとして渡されている場合に対応
    if image_paths is not None:
        if not isinstance(image_paths, list):
            image_paths = [image_paths]

        for image_path in image_paths:
            if image_path is not None:
                with open(image_path, 'rb') as img_file:
                    img_data = img_file.read()
                    img_data = resize_image_if_needed(img_data)
                    files = {'imageFile': BytesIO(img_data)}
                    files['imageFile'].name = os.path.basename(image_path)

                    # LINE Notify APIにリクエストを送信
                    res = requests.post(url, headers=headers, params=params, files=files)

                    # レスポンスを出力
                    print(f"Sent {image_path}: {res.status_code} {res.text}")
    else:
        # 画像がない場合はメッセージのみ送信
        res = requests.post(url, headers=headers, params=params)
        print(f"Sent message: {res.status_code} {res.text}")

とすることで複数の画像に対応

リストに対応: image_paths引数がリスト型で渡されても対応できるようにしました。リストでない場合は、単一の画像パスとしてリストに変換して扱います。
画像のループ処理: 画像が渡された場合はループでそれぞれの画像をLINE Notifyに送信します。
画像なしの場合: 画像が渡されなかった場合は、メッセージのみを送信します。