代理店から返信があった納入価をエクセルファイルに書き込む(Gmail version)
以前、Outlookバージョンを記事にしました。が、Outlookは色々と認証回りのセキュリティが厳しくIMAP接続出来なくなってきました。そこで、Gmailにしたので記事にしたいと思います。
実現したこと
以前、研究室のネットワークドライブに用意している発注表からメールで注文や見積を飛ばすPythonスクリプトを紹介しました。
今回は、見積を依頼するメールに対して代理店から納入価の返信が来たら、価格情報などをエクセルファイルに飛ばすPythonスクリプトを紹介します。
スクリプトの概要
メールの取得と解析:
IMAPを使用して指定されたメールボックスに接続します。
直近1週間以内の「見積のお願い」という件名のメールを検索します。
見つかったメールの本文を抽出し、送信者の情報と合わせて保存します。
データの抽出:
OpenAI GPT-4o-mini APIを利用して、メール本文から製品名、会社名、品番、定価、納入価格などの重要情報を抽出します。
抽出されたデータは構造化され、後の処理に備えます。
代理店情報の紐付け:
事前に定義された代理店とメールアドレスのマッピングを使用して、各メールに対応する代理店を特定します。
Excelファイルへの出力:
抽出されたデータをpandasデータフレームに変換します。
既存のExcelファイルに新しいデータを追加します。各項目は指定された列に書き込まれます。
環境構築
必要なものをインストール
conda install -c conda-forge pandas openpyxl openai python-dotenvスクリプト
import imaplib
import email
from email.header import decode_header
import os
import datetime
import pandas as pd
from openai import OpenAI
from openai import BadRequestError
from email.utils import parseaddr
from pathlib import Path
import re
from dotenv import load_dotenv
# .envファイルから環境変数を読み込む
load_dotenv()
# 環境変数からセンシティブな情報を取得
username = os.getenv('GMAIL_USERNAME')
password = os.getenv('GMAIL_PASSWORD')
imap_url = 'imap.gmail.com' # Gmailの IMAP サーバー
# 実行しているOSを確認
if os.name == 'nt': # Windowsの場合
excel_file_path = Path(os.getenv('EXCEL_FILE_PATH_KAKAKU_WINDOWS'))
elif os.name == 'posix': # macOSの場合
excel_file_path = Path(os.getenv('EXCEL_FILE_PATH_KAKAKU_MAC'))
else:
raise EnvironmentError("Unsupported Operating System")
# OpenAI APIの設定
client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
# 代理店とメールアドレスのマッピング
agency_mapping = {
'代理店A': {'email1@example.com', 'email2@example.com'},
'代理店B': {'email3@example.com'},
'代理店C': {'email4@example.com'},
'代理店D': {'email5@example.com'},
'代理店E': {'email6@example.com', 'email7@example.com'},
}
def decode_mime_words(s):
return u''.join(
word.decode(encoding or 'utf-8') if isinstance(word, bytes) else word
for word, encoding in decode_header(s))
# メールからデータを取得する関数
def get_mail_data(username, password, imap_url, retries=200):
attempt = 0
while attempt < retries:
try:
mail = imaplib.IMAP4_SSL(imap_url)
mail.login(username, password)
mail.select("INBOX")
# 直近1週間の日付を計算
one_week_ago = (datetime.datetime.now() - datetime.timedelta(days=7)).strftime("%d-%b-%Y")
# 直近1週間のメールの検索と取得
status, messages = mail.search(None, f'SINCE "{one_week_ago}"')
messages = messages[0].split(b' ')
extracted_data = []
if not messages or messages == [b'']:
print("No matching emails found.")
mail.logout()
return extracted_data
for mail_id in messages:
res, msg = mail.fetch(mail_id, '(RFC822)')
for response in msg:
if isinstance(response, tuple):
msg = email.message_from_bytes(response[1])
subject = decode_mime_words(msg.get("Subject"))
if '見積のお願い' in subject:
from_address = decode_mime_words(msg.get("From"))
# メールアドレスの部分だけを抽出
from_address = parseaddr(from_address)[1]
# デバッグ用：取得したメールアドレスを出力
print("From address:", from_address)
body = ""
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == "text/plain":
body = part.get_payload(decode=True)
charset = part.get_content_charset()
body = body.decode(charset, errors='replace') if charset else body.decode('utf-8', errors='replace')
else:
body = msg.get_payload(decode=True)
charset = msg.get_content_charset()
body = body.decode(charset, errors='replace') if charset else body.decode('utf-8', errors='replace')
body = body.replace("\r\n", " ")
agency = next((key for key, val in agency_mapping.items() if from_address in val), "不明")
extracted_data.append((body, agency))
mail.logout()
return extracted_data
except imaplib.IMAP4.error as e:
print(f"Login attempt {attempt + 1} failed: {e}")
attempt += 1
print("All login attempts failed.")
return []
# OpenAI APIを使ってデータを抽出する関数
def extract_data_with_gpt(text_data, agency_mapping):
parsed_data = []
print("Input Data:", text_data) # デバッグ用：入力データ
for text, agency in text_data:
# テキストの前処理
text = str(text)
text = text.encode('utf-8', errors='ignore').decode('utf-8')
text = re.sub(r'[^\x20-\x7Eぁ-んァ-ン一-龥ー]', ' ', text)
if len(text) > 5000:
print("Text too long, truncating")
text = text[:5000]
print("Sending text to API:", text[:500]) # デバッグ用：APIに送信するテキスト
try:
chat_completion = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "以下のテキストから製品名、会社、品番、定価、納入価を抽出してください。その際、定価と納入価については数字の価格情報のみにして下さい。最後に得られた情報をPandas dataframeに適した「セミコロン区切り形式」にしてください。データフレーム構造は「製品名;会社名;品番;定価;納入価格」ですのでヘッダーとして0行目に配置してください。レスポンスにはセミコロン区切り形式のデータだけ含めてください。"},
{"role": "user", "content": text}
]
)
print("API Response:", chat_completion) # デバッグ用：APIレスポンス
extracted_info = chat_completion.choices[0].message.content
print("Extracted info:", extracted_info) # デバッグ用：抽出した情報
if ';' in extracted_info:
# セミコロン区切りの場合
data_rows = [item.split(';') for item in extracted_info.split("\n") if item.strip() and "製品名" not in item]
else:
# 改行区切りの場合
product_name = [line.split(': ')[1] for line in extracted_info.split("\n") if line.startswith('製品名')]
company_name = [line.split(': ')[1] for line in extracted_info.split("\n") if line.startswith('会社名')]
product_number = [line.split(': ')[1] for line in extracted_info.split("\n") if line.startswith('品番')]
list_price = [line.split(': ')[1] for line in extracted_info.split("\n") if line.startswith('定価')]
delivery_price = [line.split(': ')[1] for line in extracted_info.split("\n") if line.startswith('納入価')]
data_rows = list(zip(product_name, company_name, product_number, list_price, delivery_price))
# デバッグ用：data_rowsの内容を出力
print("Data rows before adding agency:", data_rows)
# 代理店情報を各行に追加
for row in data_rows:
row.append(agency)
# デバッグ用：data_rowsの内容を出力
print("Data rows after adding agency:", data_rows)
# DataFrameのカラムに「代理店」を含める
df = pd.DataFrame(data_rows, columns=["製品名", "会社名", "品番", "定価", "納入価格", "代理店"])
parsed_data.append(df)
except BadRequestError as e:
print("Error with API request:", e)
return pd.concat(parsed_data, ignore_index=True) if parsed_data else pd.DataFrame()
def write_to_excel(df, excel_file_path):
with pd.ExcelWriter(excel_file_path, engine='openpyxl', mode='a', if_sheet_exists='overlay') as writer:
if 'Sheet1' in writer.book.sheetnames:
start_row = writer.book['Sheet1'].max_row
else:
start_row = 0 # シートが空の場合
# 各カラムを特定の位置に書き込む
column_positions = {"製品名": 0, "会社名": 1, "品番": 2, "定価": 4, "納入価格": 5, "代理店": 8} # エクセルのカラム位置
for column in df.columns:
col_idx = column_positions.get(column, 0)
df[column].to_excel(writer, sheet_name='Sheet1', startcol=col_idx, startrow=start_row, index=False, header=False)
# メインの処理
if __name__ == "__main__":
mail_data = get_mail_data(username, password, imap_url)
df = extract_data_with_gpt(mail_data, agency_mapping)
print(df)
write_to_excel(df, excel_file_path).env
# メールサーバー設定
GMAIL_USERNAME=your_email@example.com
GMAIL_PASSWORD=your_email_password
# OpenAI API設定
OPENAI_API_KEY=your_openai_api_key
# Excelファイルのパス
EXCEL_FILE_PATH_WINDOWS=C:\path\to\your\excel\file.xlsx
EXCEL_FILE_PATH_MAC=/path/to/your/excel/file.xlsxまとめ
これで、各代理店毎の納入価格を自動的にデータベース化できます。
ちなみに処理した後は、INBOXからDONEというフォルダにメール移動してます。
これで教育・研究活動に時間が使えるようになりハッピーになると良いですね。
他にも色々と仕組みを動かしていますので、都度紹介します！
質問や改善点があれば、ぜひコメントで教えてください。