ゴミ出し日の前日に通知する仕組みを作りたい（pdfからデータ抽出編）

はじめに

ここ数か月間、特に紙類のゴミ出しを忘れ続けています。（おかげで段ボールがたまりにたまって…
そこでゴミ収集の前日に明日は何のゴミ収集日か通知する仕組みを作りたいと考えています。
これなら月一しか収集されない紙類の日も忘れずに済みます^^

通知をするだけならLINEや他アプリに通知を登録する、というのが手っ取り早いですがそれだ面白くないので勉強がてらクラウドサービスを使いつつゼロから作ることにします。
今回Flutterによるモバイルアプリとの連携も考えているため、クラウドサービスはFirebaseを使用します。

今回の記事では一旦データ抽出と加工までを行いますので、タイトルに似合わずデータ分析寄りになります。

全体方針

1. 自治体が公開しているごみ収集カレンダーpdfを読み込む
2. pdfを解析し、カテゴリごとにごみ収集日をJSON化する　（<--本稿ではここまで）
   ※ 3.で扱うFirestoreはドキュメント指向DBであるためここでJSON化する
3. JSONデータをFirestoreへ送信・保存する
4. Firebase Cloud Functionsがゴミ出し日を判定
5. Firebase Cloud Messagingでスマホへ通知を送信する

データ抽出

うちの自治体のゴミ収集日程は自治体HPからPDFで確認することができます。
まずはこのPDFを表形式のデータへと変換させます。

環境

Google Colabとpythonで対話的に処理していきます。
また今回tabula-pyを主に使用してデータを抽出していきます。

tabula-pyとは

PDFの解析にはtabula-pyを使用します。
tabula-pyとはPDFから表を読み込み、pandasのDataFrameやCSV、JSONなどへ変換することができます。
tabula-py公式
これはもともとJavaのライブラリでtabula-pyはそれをpythonで扱えるようにしたライブラリのようです。
そのため実行にはJavaが必要になります。

処理開始

tabula-pyとjavaをGoogle Colab環境へインストールします

!pip install tabula-py
!apt-get install openjdk-11-jdk -y

必要ライブラリをインポートします

import requests
import tabula
from tabulate import tabulate
import pandas as pd

自治体HPからPDFを読み込みます

url = "https://www.city.○○□□××.jp/_res/projects/default_project/_page_/001/016/487/r7_13-14.pdf"
r = requests.get(url)

requests.get()ではオブジェクトを返すため、そのままではtabulaで読み込めません。
そのため一時的にファイル化した後、tabulaでPDFをDataFrameとして読み込みます。
ちなみにlattice=Trueを指定すると表の罫線を基準に区切りを判定します。
またoutput_format="json"を指定するとjsonの形式で出力されますが、表のセル座標を含む構造のため扱いが難しくなります。
そのため今回は通常のDataFrameで読み込み、一度出力させてみます。

# 一時的にファイル化
with open("calender.pdf", "wb") as f:
    f.write(r.content)
# PDFをDataFrame化
dfs = tabula.read_pdf("calender.pdf", pages="all", lattice=True)
dfs

すると以下のように出力されます。

5月の次である6月から急に段落が変わったり、分別名の表示位置に少し違和感があったりとやや視認性に欠けていますが、必要な情報は抽出できていそうです。

次にこのDataFrameをtabulateを使って表を分かりやすく表示させてみます。
そのためにまずはdfsの型を確認します

print(type(dfs))    # 1行目
for df in dfs:
    print(type(df)) # 2行目

実行結果は以下の通りです。
dfsの型はlist型でその中の要素はDataFrameになっています。つまりlist[DataFrame]です。

<class 'list'>
<class 'pandas.core.frame.DataFrame'>
・・・省略

list[DataFrame]から一つ一つDataFrameを取り出しtabulateに渡していい感じに表示させます。

for df in dfs:
    print(tabulate(df, headers=df.columns, showindex=False, tablefmt="grid"))

表示結果は以下の通りです。いいかんじに表示されましたね!

JSON化

ここからドキュメント指向DBであるFirestoreに格納できるようJSON形式へ変換します。
最終的にはの次のような形で出力されるイメージです。(JSONの設計はやや不安ですが)

[
    {
        // 不燃・有害ごみは日程で指定されており、曜日指定はない
        "category": "不燃,有害",
        "schedule": {
            "1月": 〇,
            "2月": 〇,
            ・・・
        }
        "dayOfWeek": null
    },
    {
        // 燃えるゴミは曜日指定であり、日程指定はない
        "category": "燃えるゴミ",
        "schedule": null,
        "dayOfWeek": "火,金"
    },
    ・・・
]

しかし上で求めたDataFrameをこのようなJSONに落とし込むのはなかなかハードだったのでチャッピーに実行コードを全部書いてもらいました^^

import json
import re

CATEGORY_NAMES = [
    "びん,かん,ペットボトル",
    "不燃・有害",
    "布類",
    "紙類",
    "可燃ごみ",
    "プラスチック製容器"
]

def clean_category(cell):
    """カテゴリセルの整形"""
    if pd.isna(cell):
        return ""
    text = (
        str(cell)
        .replace("\r", ",")
        .replace("\n", "")
        .replace(" ", "")
        .strip()
    )
    return text

def parse_schedule(row, category):
    schedule = {}
    for month in range(1, 12 + 1):
        month_col = f"{month}月"

        if month_col not in row.index:
            continue

        vals = str(row[month_col]).strip()
        if not vals or vals == "nan":
            continue

        # 生の数字抽出（例： "19  " → ["19", "0"] などのノイズあり）
        nums_raw = re.findall(r"\d+", vals)

        # ---- ここで余計な 0 を削除 ----
        nums = [n for n in nums_raw if n != "0"]

        # すべて 0 しかなかった場合は無視
        if not nums:
            continue

        if "びん,かん,ペットボトル" in category:
            # 複数日 → リスト
            nums = [int(n) for n in nums]
            schedule[month_col] = nums

        else:
            # 単一日 → 整数
            schedule[month_col] = int(nums[0])

    return schedule


result = []
course_no = 1

for df in dfs:
    df = df.fillna("")
    current_category = ""
    data_per_course = []

    for _, row in df.iterrows():
        cat = clean_category(row[0])

        # カテゴリ更新
        if cat:
            current_category = cat

        # 特例カテゴリ処理
        day_of_week = []
        note = None
        schedule = None

        if "可燃" in current_category:
            day_of_week = ["火曜", "金曜"]
            note = "火・金曜日（祝日も収集します。年末は12/30(火)まで、年始は1/6(火)から）"
            current_category = "可燃ごみ"

        elif "プラスチック" in current_category:
            day_of_week = ["月曜"]
            current_category = "プラスチック製容器"

        else:
            # 通常カテゴリのスケジュール解析
            schedule = parse_schedule(row, current_category)

        data_per_course.append({
            "収集コース": course_no,
            "category": current_category,
            "schedule": schedule,
            "dayOfWeek": day_of_week,
            "note": note
        })

        # カテゴリが揃ったらコース確定
        if len(data_per_course) >= len(CATEGORY_NAMES):
            result.extend(data_per_course)
            data_per_course = []
            current_category = ""
            course_no += 1

# JSON出力
json_text = json.dumps(result, ensure_ascii=False, indent=4)
print(json_text)

いい感じに出力されました^^

[
    {
        "収集コース": 1,
        "category": "びん,かん,ペットボトル",
        "schedule": {
            "1月": [
                15,
                29
            ],
            "2月": [
                19
            ],
            "3月": [
                5,
                19
            ],
            "4月": [
                10,
                24
            ],
            "5月": [
                15,
                29
            ],
            "6月": [
                12,
                26
            ],
            "7月": [
                10,
                24
            ],
            "8月": [
                7,
                21
            ],
            "9月": [
                4,
                18
            ],
            "10月": [
                2,
                16,
                30
            ],
            "11月": [
                13,
                27
            ],
            "12月": [
                11,
                25
            ]
        },
        "dayOfWeek": [],
        "note": null
    },
    {
        "収集コース": 1,
        "category": "不燃·有害",
        "schedule": {
            "1月": 8,
            "2月": 12,
            "3月": 12,
            "4月": 17,
            "5月": 22,
            "6月": 19,
            "7月": 17,
            "8月": 14,
            "9月": 11,
            "10月": 9,
            "11月": 6,
            "12月": 4
        },
        "dayOfWeek": [],
        "note": null
    },
    {
        "収集コース": 1,
        "category": "布類",
        "schedule": {
            "1月": 22,
            "2月": 26,
            "3月": 26,
            "4月": 3,
            "5月": 8,
            "6月": 5,
            "7月": 3,
            "8月": 28,
            "9月": 25,
            "10月": 23,
            "11月": 20,
            "12月": 18
        },
        "dayOfWeek": [],
        "note": null
    },
    {
        "収集コース": 1,
        "category": "紙類",
        "schedule": {
            "1月": 7,
            "2月": 4,
            "3月": 4,
            "4月": 2,
            "5月": 7,
            "6月": 4,
            "7月": 2,
            "8月": 6,
            "9月": 3,
            "10月": 1,
            "11月": 5,
            "12月": 3
        },
        "dayOfWeek": [],
        "note": null
    },
    {
        "収集コース": 1,
        "category": "可燃ごみ",
        "schedule": null,
        "dayOfWeek": [
            "火曜",
            "金曜"
        ],
        "note": "火・金曜日（祝日も収集します。年末は12/30(火)まで、年始は1/6(火)から）"
    },
    {
        "収集コース": 1,
        "category": "プラスチック製容器",
        "schedule": null,
        "dayOfWeek": [
            "月曜"
        ],
        "note": null
    },
    続く...
]

これならFirestoreに格納できそうですたぶん

終わりに

ゴミ出し前日に通知する仕組みを作るにあたり、まずはゴミ出しカレンダーのデータを抽出・加工するところまで行いました
続きはまたやる気があるときに取り組みます。