きっかけは「また手入力ですか」のため息
EC案件に関わるようになって半年くらい経った頃、チーム内でずっと繰り返されていた作業がありました。取引先から届く受注メールの内容を、社内の管理システムに手で転記するという作業です。
1日あたり50〜80件。月にすると約80時間。
正直なところ、最初は「まあ仕方ないよね」と思っていました。取引先ごとにメールのフォーマットが全然違うんです。ある会社はCSV添付、別の会社は本文にテーブルっぽく書いてくる、さらに別の会社はExcel添付——「統一してください」とお願いできる立場でもないので、人間が目で読んで入力するしかなかった。
でもある日、入力ミスが原因で納品先を間違えるインシデントが発生して、「これは自動化しないとまずい」という空気になりました。
最初のアプローチ：正規表現で全部吸収する（破綻編）
私が最初に書いたのは、メール本文を正規表現でパースするスクリプトでした。
import re
def parse_order_email(body: str) -> dict:
patterns = {
"order_id": r"注文番号[：:]\s*(\S+)",
"product": r"商品名[：:]\s*(.+)",
"quantity": r"数量[：:]\s*(\d+)",
"shipping_address": r"送付先[：:]\s*(.+)",
}
result = {}
for key, pattern in patterns.items():
match = re.search(pattern, body)
if match:
result[key] = match.group(1).strip()
return result取引先A社のメールではうまくいきました。B社もなんとかいけた。でもC社のメールが来た瞬間に破綻しました。
C社のフォーマットは「商品名」ではなく「品名」だったんですよね。D社は「お届け先」。E社にいたっては全角コロンと半角コロンが混在していて、しかも行の途中で改行が入る。パターンを追加するたびに既存のパターンと干渉して、3社目あたりで正規表現が人間に読めない代物になっていました。
今思えば、「全部のフォーマットを1つの正規表現群で吸収しよう」という発想自体が間違いだったんですが、渦中にいるとなかなか気づけないものです。
構造化パーサーに方針転換する
ここで方針を変えて、取引先ごとにパーサークラスを分離する設計にしました。
from abc import ABC, abstractmethod
from dataclasses import dataclass
@dataclass
class OrderData:
order_id: str
product: str
quantity: int
shipping_address: str
raw_source: str # 元メール本文（トラブル時のデバッグ用）
class BaseParser(ABC):
@abstractmethod
def can_parse(self, body: str, sender: str) -> bool:
...
@abstractmethod
def parse(self, body: str) -> OrderData:
...
class CompanyAParser(BaseParser):
def can_parse(self, body: str, sender: str) -> bool:
return "companya.co.jp" in sender
def parse(self, body: str) -> OrderData:
# A社固有のパースロジック
lines = body.strip().split("\n")
# ... A社のフォーマットに特化した処理
return OrderData(
order_id=extract_field(lines, "注文番号"),
product=extract_field(lines, "商品名"),
quantity=int(extract_field(lines, "数量")),
shipping_address=extract_field(lines, "送付先"),
raw_source=body,
)各パーサーが can_parse で「自分が処理すべきメールか」を判定し、該当すれば parse で構造化データに変換する。新しい取引先が増えたら新しいパーサークラスを1つ追加するだけです。既存のパーサーには一切触れなくていい。
この設計に切り替えてから、パーサーの追加作業が1社あたり30分〜1時間程度で済むようになりました。正規表現の時代は1社追加するたびに既存ロジックのデグレを確認する作業が発生していたので、体感で3倍は楽になった。
ただ、正直に言うとパーサー選択の部分は can_parse をループで回しているだけなので、取引先が100社を超えたらもう少し効率的な振り分けを考えたほうがいいかもしれません。今は20社程度なので問題にはなっていませんが。
Lambda + SESで回す運用構成
パーサーができたら、次は「どうやって自動で動かすか」です。受注メールは専用のメールアドレスに転送される仕組みにして、Amazon SESで受信 → S3に保存 → Lambdaが起動、という構成にしました。
import json
import boto3
import email
def lambda_handler(event, context):
s3 = boto3.client("s3")
# SES経由でS3に保存されたメールを取得
bucket = event["Records"][0]["s3"]["bucket"]["name"]
key = event["Records"][0]["s3"]["object"]["key"]
obj = s3.get_object(Bucket=bucket, Key=key)
raw_email = obj["Body"].read().decode("utf-8")
msg = email.message_from_string(raw_email)
body = extract_body(msg)
sender = msg.get("From", "")
# パーサーチェーンで処理
parsers = [CompanyAParser(), CompanyBParser(), CompanyCParser()]
for parser in parsers:
if parser.can_parse(body, sender):
order = parser.parse(body)
save_to_database(order)
return {
"statusCode": 200,
"body": json.dumps({"order_id": order.order_id}),
}
# どのパーサーにもマッチしなかった場合
notify_slack(f"未対応フォーマットのメール: {sender}")
return {"statusCode": 200, "body": "no parser matched"}ここで地味に大事だったのがエラーハンドリングです。パースに失敗したメールを握りつぶすと、気づいたときには受注が何件か漏れている——という最悪のシナリオがあり得ます。なので「パースできなかったメールはSlackに通知 + 元メールをエラー用S3バケットに退避」というフローにしました。
実際に運用を始めてみると、パース自体の失敗よりも「メールの文字エンコーディングが想定外」というケースのほうが多かったです。ISO-2022-JPで送ってくる取引先がいて、decode のところで落ちるという、わりと古典的な問題に遭遇した経験があります。chardet で自動判定させるようにしてからは安定しましたが、こういうエッジケースは事前に想像しにくいんですよね。
振り返って
この仕組みを入れてから3ヶ月が経ちましたが、月80時間の手作業はほぼゼロになりました。手入力のミスも当然なくなって、納品先間違いのインシデントもゼロです。
ただ、すべてがうまくいっているわけではなくて、新しい取引先が増えるたびにパーサーを書く作業は残っています。ここは今後、メール本文の構造をLLMに解析させることで、パーサーを個別に書かなくても済む仕組みにできないかと考えているところです。
あと、extract_body まわりのmultipartメールの扱いがまだ雑で、添付ファイル付きのケースをもう少し丁寧にハンドリングしたいという課題もあります。完璧ではないけれど、「月80時間の手作業を消す」という当初の目的は達成できたので、まずはよかったかなと。