1
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@yoshiasa1117(Aki Yoshi)

請求書の作成をLINEで済ませよう ～プログラム編～

1
Posted at

こんにちは。しがないメーカーエンジニアのよしあきです。

個人事業主にとってめんどうな請求書作成を自動化するため、LINE Botの開発をしています。
この記事はその第二回となります。

前回の記事では、アカウントの開設やローカルサーバーの起動を行いました。

今回はコードを（Claudeが）書いて、ユーザーとのやり取りを通して請求書を作成できるようにします。

今回やりたいこと

今回のプログラムでは次の流れを自動化します。

  1. 公式LINEで「請求書作成」を受信
  2. ユーザーごとに必要な情報を聞き取り
  3. 集めた情報でPDF請求書を作成
  4. 生成したPDFのURLをLINEに送信

PDFファイルを送信するにはLINE WORKS（企業向けLINE）が必要です。
一般のLINE Botである。LINE Messaging APIではファイル送信用SDKであるFile Messageが使えないため、今回は作成したPDFのURLを文字列で送信するようにします。

各実装の免責事項です。

  • Webhook受信用サーバー： Flask を使ってPC上に建てます。生成したPDFはローカルの静的配信パスで XXXX（例：/static/invoices/...）で一時公開するか、S3等にアップロードして公開URLを取得する。
  • 状態管理（ステートマシン）：ユーザーごとに現在のステップと収集済みデータを保持（簡易はメモリ辞書、実運用はRedisやDB）。
  • PDF生成：Pythonの reportlab モジュールを使います。
  • PDF送信：LINEの FileMessage を使い、original_content_url に公開URLを指定して送信。ローカル開発では ngrok の公開URL を使う。

プログラムの実装

app.pyにこれから説明するプログラムを実装していきます。

モジュールインポート

今回使用するモジュールは主に以下の4つです。

  • dotenv: APIは .envファイルに保存して、さらにそれを.gitignoreに書いておくことで外部に公開しないよう対策しましょう。最近のAIは、勝手にcommitメッセージをつけてpushまでしてくれますが、APIキーなど外部に漏れてはいけない情報までpushされてしまいます。必ずこういった対策を取りましょう。
  • flask: PC上にローカルサーバーを建てるためのモジュールです。インスタンス化して実行すると、Flaskが起動します。さらにそれをgrokを通して外部に接続できるようにすると、LINEとPC上のPythonのやり取りができるわけです。
  • linebot: 公式LINE用のSDKです。Webhookを受け取り、それに対するアクションを定義することができます。
  • reportlab: PDFを作成するためのモジュールです。

モジュールをインストールし、インポートしましょう。

pip install flask line-bot-sdk python-dotenv reportlab

import os
from dotenv import load_dotenv
from flask import Flask, request
from linebot import LineBotApi, WebhookHandler
from linebot.models import MessageEvent, TextMessage, TextSendMessage
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas

初期設定

.envで定義したAPIの読み込みと、Flaskのインスタンス化を行います。

load_dotenv()

LINE_CHANNEL_SECRET = os.getenv("LINE_CHANNEL_SECRET")
LINE_CHANNEL_ACCESS_TOKEN = os.getenv("LINE_CHANNEL_ACCESS_TOKEN")

app = Flask(__name__, static_folder="static", static_url_path="/static")

handler = WebhookHandler(LINE_CHANNEL_SECRET)
line_bot_api = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)

ユーザー状態管理

このLINEはユーザーのヒアリングを通して、ユーザーの状態が変わっていきます。
（情報の聞き取り段階なのか、PDFの作成段階なのか…）
ユーザーごとにこの状態を管理して、適切な応答を返すようにしましょう。

user_states = {}

def init_user_state(user_id):
    user_states[user_id] = {
        "step": "await_confirm",
        "data": {}
    }

PDF作成

PDFを作成するための関数も定義しておきます。
生成されるPDFを見つつ、この辺りは調整ですね。

def generate_invoice_pdf(user_id, data):
    out_dir = os.path.join("static", "invoices")
    os.makedirs(out_dir, exist_ok=True)

    filename = f"invoice_{user_id}.pdf"
    path = os.path.join(out_dir, filename)

    c = canvas.Canvas(path, pagesize=A4)
    width, height = A4

    c.setFont("Helvetica-Bold", 16)
    c.drawString(40, height - 60, "請求書")

    c.setFont("Helvetica", 12)
    c.drawString(40, height - 100, f"宛先: {data['name']}")
    c.drawString(40, height - 120, f"業務内容: {data['item']}")
    c.drawString(40, height - 140, f"金額: ¥{data['amount']:,}")

    c.save()
    return path

コールバック処理

コールバック処理とは一般的に「外部サービスからの通知を受け取って、プログラム側で処理する仕組み」を指します。
ここで外部サービス=LINE, プログラム=app.pyを指します。

実際にプログラムを見ながら理解していきましょう。

@app.route("/callback", methods=["POST"])
def callback():
    signature = request.headers["X-Line-Signature"]
    body = request.get_data(as_text=True)
    handler.handle(body, signature)
    return "OK"

まず一行目の @app.route("/callback", methods=["POST"])は「/callbackというURLにPOSTが来たら、この関数（callback）を実行してね」とFlaskに教えているコードです。

@app.route() はFlaskの機能で、/callbackはLINE Developersに登録するWebhook URLです。
メッセージを送ると、LINEがこのURLにPOSTを送ります。そのPOSTを受け取って、デコレーターがついたcallback()が受け口になるといった挙動です。

signature = request.headers["X-Line-Signature"]は署名（signature）を検証して改ざんされていないかチェックします。

handler.handle(body, signature)がメッセージイベントを解析します。
解析してイベントをSDKに渡す処理は、LINE SDKが自動で行ってくれます。
受け取ったメッセージに応じて、処理を定義してあげましょう。

メッセージ処理

こちらがメッセージ処理の流れ一覧です。

@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
    user_id = event.source.user_id
    text = event.message.text.strip()

    # 初回
    if user_id not in user_states:
        init_user_state(user_id)
        line_bot_api.reply_message(
            event.reply_token,
            TextSendMessage(text="請求書作成を開始しますか？「はい」で開始します。")
        )
        return

    state = user_states[user_id]
    step = state["step"]

    # ステップごとの処理
    if step == "await_confirm":
        if "はい" in text:
            state["step"] = "await_name"
            line_bot_api.reply_message(event.reply_token, TextSendMessage(text="先方の名前を教えてください。"))
        else:
            line_bot_api.reply_message(event.reply_token, TextSendMessage(text="いつでも「請求書作成」と送ってください。"))

    elif step == "await_name":
        state["data"]["name"] = text
        state["step"] = "await_item"
        line_bot_api.reply_message(event.reply_token, TextSendMessage(text="業務内容を教えてください。"))

    elif step == "await_item":
        state["data"]["item"] = text
        state["step"] = "await_amount"
        line_bot_api.reply_message(event.reply_token, TextSendMessage(text="金額を数字で教えてください。"))

    elif step == "await_amount":
        try:
            amount = int(text.replace(",", ""))
            state["data"]["amount"] = amount

            line_bot_api.reply_message(event.reply_token, TextSendMessage(text="PDFを作成しています…"))

            # PDF生成
            pdf_path = generate_invoice_pdf(user_id, state["data"])

            # ngrok公開URL（あなたのURLに置き換える）
            NGROK_BASE_URL = os.environ["NGROK_BASE_URL"]
            public_url = f"{NGROK_BASE_URL}/static/invoices/invoice_{user_id}.pdf"

            # PDFのURLをテキストで送信
            line_bot_api.push_message(
                user_id,
                TextSendMessage(text=f"請求書を作成しました。こちらから確認できます: {public_url}")
            )

            line_bot_api.push_message(user_id, TextSendMessage(text="請求書を送信しました。"))
            state["step"] = "done"

        except ValueError:
            line_bot_api.reply_message(event.reply_token, TextSendMessage(text="金額は数字で入力してください。"))

    else:
        line_bot_api.reply_message(event.reply_token, TextSendMessage(text="新しい請求書を作る場合は「請求書作成」と送ってください。"))

処理を一行ずつ、確認していきます。

ユーザーIDとテキストの抽出

  • @handler.add(MessageEvent, message=TextMessage): 「TextMessageのMessageEventが来たら、この関数を呼んでね」とLINE SDKに登録しています
  • user_id = event.source.user_id: メッセージを送ってきたユーザーを一意に識別するためのIDです。ステート管理（誰がどこまで入力したか）に使われます
  • text = event.message.text.strip(): ユーザーが送ったテキストメッセージです。.strip()で前後の空白を削って扱いやすくしています

初回メッセージかどうかの判別

  • if user_id not in user_states:: user_idがuser_statesに登録されているかどうかで、初回メッセージかどうかの判別をします
  • init_user(user_id): ユーザー用の状態を初期化します
  • reply_message(): 本格的なステート処理に入る前に、まずは開始確認をします

現在のステップの抽出

  • state = user_states[user_id]: そのユーザー専用の状態（辞書）を取り出します
  • step = state["step"]: 今どの段階かを表します
    • 例："await_confirm", "await_name", "await_item", "await_amount" など

ここから先は、stepの値によって分岐していきます。

await_confirm（開始していい？）

    if step == "await_confirm":
        if "はい" in text:
            state["step"] = "await_name"
            line_bot_api.reply_message(event.reply_token, TextSendMessage(text="先方の名前を教えてください。"))
        else:
            line_bot_api.reply_message(event.reply_token, TextSendMessage(text="いつでも「請求書作成」と送ってください。"))

await_confirmの状態のときに、はいと送信された場合に状態をawait_nameに変え、先方の名前を聞きます。

await_name（先方の名前は？）

    elif step == "await_name":
        state["data"]["name"] = text
        state["step"] = "await_item"
        line_bot_api.reply_message(event.reply_token, TextSendMessage(text="業務内容を教えてください。"))

状態がawait_nameであれば、先方の名前を待ちます。
ユーザーが送ったテキストを"name"として保存し、状態をawait_itemに進めます。

await_item（業務内容は？）

    elif step == "await_item":
        state["data"]["item"] = text
        state["step"] = "await_amount"
        line_bot_api.reply_message(event.reply_token, TextSendMessage(text="金額を数字で教えてください。"))

await_itemは業務内容を待ちます。
これは"item"として保存され、状態はawait_amountに進みます。

await_amount（金額は？）& PDF作成

    elif step == "await_amount":
        try:
            amount = int(text.replace(",", ""))
            state["data"]["amount"] = amount

            line_bot_api.reply_message(event.reply_token, TextSendMessage(text="PDFを作成しています…"))

            # PDF生成
            pdf_path = generate_invoice_pdf(user_id, state["data"])

            # ngrok公開URL
            NGROK_BASE_URL = os.environ["NGROK_BASE_URL"]
            public_url = f"{NGROK_BASE_URL}/static/invoices/invoice_{user_id}.pdf"

            # PDFのURLをテキストで送信
            line_bot_api.push_message(
                user_id,
                TextSendMessage(text=f"請求書を作成しました。こちらから確認できます: {public_url}")
            )

            line_bot_api.push_message(user_id, TextSendMessage(text="請求書を作成しました。URLのファイルを確認してください。"))
            state["step"] = "done"

        except ValueError:
            line_bot_api.reply_message(event.reply_token, TextSendMessage(text="金額は数字で入力してください。"))

最後に金額を待ちます。50,000のような入力に対応するため、カンマを消してから整数に変換します。この変換が成功すると、PDF生成に進みます。

  • generate_invoice_pdf(user_id, state["data"]): PDFを作成
  • NGROK_BASE_URL + /static/invoices/...: 生成したPDFを外部からアクセスできるURLに変換
  • line_bot_api.push_message(...): LINE に作成した請求書PDFが保存されているURLを送信
  • state["step"] = "done": このユーザーの請求書作成フローは完了
  • except ValueError:: 数字に変換できなかった場合（例：「五万円」など）はエラー扱いとし、「数字で入力してください」と再入力を促す

例外処理

line_bot_api.reply_message(event.reply_token, TextSendMessage(text="新しい請求書を作る場合は「請求書作成」と送ってください。")): stepがdoneなどの場合は、すでに一連の流れが終わっているので、「また始めたいときは「請求書作成」と送って」と案内しましょう。

すべてのコードをつなげるとこのようになります。

import os
from dotenv import load_dotenv
from flask import Flask, request
from linebot import LineBotApi, WebhookHandler
from linebot.models import MessageEvent, TextMessage, TextSendMessage, FileMessage
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas

# =========================
# 1. 初期設定
# =========================

load_dotenv()

LINE_CHANNEL_SECRET = os.getenv("LINE_CHANNEL_SECRET")
LINE_CHANNEL_ACCESS_TOKEN = os.getenv("LINE_CHANNEL_ACCESS_TOKEN")

app = Flask(__name__, static_folder="static", static_url_path="/static")

handler = WebhookHandler(LINE_CHANNEL_SECRET)
line_bot_api = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)

# =========================
# 2. ユーザー状態管理
# =========================

user_states = {}

def init_user_state(user_id):
    user_states[user_id] = {
        "step": "await_confirm",
        "data": {}
    }

# =========================
# 3. PDF生成
# =========================
def generate_invoice_pdf(user_id, data):
    out_dir = os.path.join("static", "invoices")
    os.makedirs(out_dir, exist_ok=True)

    filename = f"invoice_{user_id}.pdf"
    path = os.path.join(out_dir, filename)

    c = canvas.Canvas(path, pagesize=A4)
    width, height = A4

    c.setFont("Helvetica-Bold", 16)
    c.drawString(40, height - 60, "請求書")

    c.setFont("Helvetica", 12)
    c.drawString(40, height - 100, f"宛先: {data['name']}")
    c.drawString(40, height - 120, f"業務内容: {data['item']}")
    c.drawString(40, height - 140, f"金額: ¥{data['amount']:,}")

    c.save()
    return path

# =========================
# 4. コールバック処理
# =========================

@app.route("/callback", methods=["POST"])
def callback():
    signature = request.headers["X-Line-Signature"]
    body = request.get_data(as_text=True)
    handler.handle(body, signature)
    return "OK"

# =========================
# 5. メッセージ処理
# =========================

@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
    user_id = event.source.user_id
    text = event.message.text.strip()

    # 初回
    if user_id not in user_states:
        init_user_state(user_id)
        line_bot_api.reply_message(
            event.reply_token,
            TextSendMessage(text="請求書作成を開始しますか？「はい」で開始します。")
        )
        return

    state = user_states[user_id]
    step = state["step"]

    # ステップごとの処理
    if step == "await_confirm":
        if "はい" in text:
            state["step"] = "await_name"
            line_bot_api.reply_message(event.reply_token, TextSendMessage(text="先方の名前を教えてください。"))
        else:
            line_bot_api.reply_message(event.reply_token, TextSendMessage(text="いつでも「請求書作成」と送ってください。"))

    elif step == "await_name":
        state["data"]["name"] = text
        state["step"] = "await_item"
        line_bot_api.reply_message(event.reply_token, TextSendMessage(text="業務内容を教えてください。"))

    elif step == "await_item":
        state["data"]["item"] = text
        state["step"] = "await_amount"
        line_bot_api.reply_message(event.reply_token, TextSendMessage(text="金額を数字で教えてください。"))

    elif step == "await_amount":
        try:
            amount = int(text.replace(",", ""))
            state["data"]["amount"] = amount

            line_bot_api.reply_message(event.reply_token, TextSendMessage(text="PDFを作成しています…"))

            # PDF生成
            pdf_path = generate_invoice_pdf(user_id, state["data"])

            # ngrok公開URL（あなたのURLに置き換える）
            NGROK_BASE_URL = os.environ["NGROK_BASE_URL"]
            public_url = f"{NGROK_BASE_URL}/static/invoices/invoice_{user_id}.pdf"

            # PDFのURLをテキストで送信
            line_bot_api.push_message(
                user_id,
                TextSendMessage(text=f"請求書を作成しました。こちらから確認できます: {public_url}")
            )

            line_bot_api.push_message(user_id, TextSendMessage(text="請求書を送信しました。"))
            state["step"] = "done"

        except ValueError:
            line_bot_api.reply_message(event.reply_token, TextSendMessage(text="金額は数字で入力してください。"))

    else:
        line_bot_api.reply_message(event.reply_token, TextSendMessage(text="新しい請求書を作る場合は「請求書作成」と送ってください。"))

# =========================
# 6. Flask起動
# =========================

if __name__ == "__main__":
    app.run(port=5000)

プログラムを実行すると、以下のようにやり取りができます。

image.png

ちなみに現状だと、フォーマットも何もなくかつ文字化けしているという全く使い物にならないPDFが入手できます。

ここら辺を修正して、使い物になるようになったら次は第三回のサーバー編に移ろうと思います。
ラズパイを使って常時起動しているローカルサーバーを建てる予定です。

それでは今回もありがとうございました。

1
2
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?