チラシの商品名を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に送信します。
画像なしの場合: 画像が渡されなかった場合は、メッセージのみを送信します。