[Python]楽天市場の価格を自動監視して値下げ時にLINEに通知するツールを作った

1.はじめに

楽天市場で「欲しい商品、もう少し安くならないかな...」と思ったことありませんか？

毎日価格をチェックするのは面倒なので、Pythonで値段を自動監視して、値下げしたらLINEに通知をするツールを作成しました。

この記事では

• 複数商品に対応したスクレイピング方法
• LINE Messaging APIを使用した自動通知
• cronで定期実行をする方法

について解説していきます。

私は2025年12月からBizCodeXでPythonの学習を始めています。
今回は日々の学習でのアウトプットも兼ねてツールを作成しました。
より多くの方に見ていただけると幸いです。

2.完成イメージ

(1).LINEに通知を送る

指定した商品の情報を取得して、値下がりをしたらLINEに通知を送ります。

(最近ゴルフを始めたのでゴルフ用品を取得しています☺️)

(2).商品情報を自動で取得する

cron(windowsはタスクスケジューラー)で定期実行を行い、値下がり時にLINEへ通知を送るように設定します。

今回は1時間ごとに自動取得しようと思います。
cronの設定はこちらの記事を参考にしました。

ターミナル

crontab -e

0 0 * * * cd /Users/{実行ファイルの保存先} && {python3の絶対パス} app.py

3.ファイル構成

rakuten_price_monitor
├── app.py
├── config.py
├── rakuten_product_1.csv
├── rakuten_product_2.csv
│
├── notifier
│   ├── __init__.py
│   └── line_notifier.py
│
├── rakuten
│   ├── __init__.py
│   ├── product.py
│   └── scraper.py
│
└── storage
    ├── __init__.py
    └── csv_handler.py

4.使用したライブラリ、技術

ライブラリ、技術	使用目的
Python	ツール全体の基本構成
LINE Messaging API	LINEへの通知処理
cron	ツールの定期実行
requests	Webデータの取得
BeautifulSoup	スクレイピング処理
os	ファイルの存在確認
csv	データ保存
pandas	csvファイルの読み込み
datetime	日付表示

5.コード概要

5-1.メイン(app.py)

コード実装のメインファイルです。

各モジュール、パッケージをインポートします。
URLをfor文で周して、各商品ごとにCSVファイルを分けることで、前回の商品の値段と比較をしやすくしています。

app.py

from rakuten.scraper import RakutenScraper
from storage.csv_handler import CSVHandler
from notifier.line_notifier import LineNotifier
import config

def create_csv(i):
    return f"rakuten_product_{i+1}.csv"

def main():

    for i, url in enumerate(config.RAKUTEN_URLS):

        print(f"{i+1}件目の情報を取得中...")

        filename = create_csv(i)

        scraper = RakutenScraper(url)
        csv_handler = CSVHandler(filename)
        line_notice = LineNotifier(config.LINE_TOKEN, config.USER_ID)

        product = scraper.get_product()

csvファイルの最後の行を読み取り、一行前のpriceから値下がりをしたら、LINEに通知を送ります。

        last_price = csv_handler.get_last_price()

        csv_handler.save(product)

        if last_price is not None and int(product.price) < int(last_price):
            message = f"""
値下げされました！

商品名: {product.name}
価格: {product.price}円
URL: {product.url}
        """
            
            line_notice.send(filename, message)
        
        print(f"{i+1}件目の処理が正常に終了しました")
        
        
if __name__ == "__main__":
    main()

5-2.URL、TOKEN、IDの情報(config.py)

main.pyに直接書くことを避けることで、汎用性を高くします。

config.py

RAKUTEN_URLS = [
    "URL_1",
    "URL_2"
]

LINE_TOKEN = "YOUR_TOKEN"

USER_ID = "YOUR_ID"

5-3.商品情報をカテゴリ分け(product.py)

スクレイピングで集めた情報を[商品名、値段、URL、取得日]
に仕分けをします。

product.py

class Product(object):
    def __init__(self, name, price, url, date):
        self.name = name
        self.price = price
        self.url = url
        self.date = date
    
    def to_list(self):
        return [self.date, self.name, self.price, self.url]

5-4.スクレイピング(scraper.py)

楽天市場から各商品情報を取得します。
取得した情報をProductに反映します。

scraper.py

from rakuten.product import Product

import requests
from bs4 import BeautifulSoup
from datetime import datetime


class RakutenScraper:
    def __init__(self, url):
        self.url = url

    def get_product(self):
        headers = {
            "User-Agent": "Mozilla/5.0"
        }

        response = requests.get(self.url, headers=headers)
        soup = BeautifulSoup(response.text, "html.parser")

        name_tag = soup.find('meta', itemprop="name")
        name = name_tag["content"]
        name = name.split("｜")[0]

        price_tag = soup.find("meta", itemprop="price")
        price = price_tag["content"]

        date = datetime.now().strftime("%Y/%m/%d %H:%M")

        return Product(name, price, self.url, date)

5-5.csvファイルへ保存(csv_handler.py)

スクレイピングで取得した情報をcsvファイルに保存します。
2回目以降の実行は最後のpriceを取得します。
最新の情報から4つまでデータを残し、古いデータから順に削除をします。

csv_handler.py

from rakuten.product import Product

import csv
import os

class CSVHandler(object):

    def __init__(self, filename):
        self.filename = filename

    def save(self, product):
        if os.path.exists(self.filename):
            with open(self.filename, "r", encoding="utf-8-sig") as f:
                rows = list(csv.reader(f))

        if not rows:
            rows.append(["date", "name", "price", "url"])

        rows.append(product.to_list())

        if len(rows) > 4:
            rows.pop(1)

        with open(self.filename, "w", newline="", encoding="utf-8-sig") as f:
            writer = csv.writer(f)
            writer.writerows(rows)

    def get_last_price(self):
        if not os.path.exists(self.filename):
            return None
        
        with open(self.filename, "r", encoding="utf-8-sig") as f:
            rows = list(csv.reader(f))

            if len(rows) <= 1:
                return None

            return int(rows[-1][2])

CSVファイルイメージ

5-6.LINE通知処理(line_notifier.py)

csvファイルを読み取り、前回取得した情報よりも値段が下がっていたらLINEに通知を送る処理をします。

また、LINE Messaging APIについてはこちらの記事を参考に作成しました。

line_notifier.py

import config

import pandas as pd

from linebot import LineBotApi
from linebot.models import TextSendMessage

class LineNotifier(object):

    def __init__(self, token, user_id):
        self.token = token
        self.user_id = user_id

    def send(self, filename, message):
        df = pd.read_csv(filename)
        df.columns = ['date', 'name', 'price', "url"]

        if len(df) >= 1:
            old_price = df.iloc[-2]["price"]
            new_price = df.iloc[-1]["price"]
    
            if int(new_price) < int(old_price):
                line_bot_api = LineBotApi(config.LINE_TOKEN)
                line_bot_api.push_message(
                self.user_id,
                messages = TextSendMessage(text = message)
                )

6.開発を通しての振り返り

6-1.苦労した点

モジュールとパッケージ

今回のアウトプットでモジュールとパッケージを本格的に使用しました。__init__の使い方やどうやって他のモジュールのデータを反映させるかは相当悩みました。相変わらず基本的な部分でつまづいてしまって悔しい思いをしました...。

cron

定期実行のためにcronを利用してみたのですが、設定しても動かなかったり、色々いじって仮想環境を壊してしまったり散々でした。YouTubeやいろんな記事を見て試行錯誤しながらなんとか実装に至りました。

6-2.学んだこと

前回のアウトプットの反省を活かして、細かくデバッグをするようにしました。1つコードを書いたら正常に動作するか確認して進めていくことで、結果作業スピードが上がりました。
また、まだ学習していない分野についても自主的に学ぶことで、より理解度が増したと思います。

7.今後の展望

• 現状はconfig.pyに直接TOKENやIDを書いています。これはセキュリティ上危険だと感じています。ここはGoogle Cloud の Secret ManagerやCloud IAM等を使用して安全性を高めようと思います。
• cronはパソコンを起動していないと定期実行しません。Google Cloud の Cloud SchedulerやCloud Functionsの学習をして、パソコンを起動していなくても定期実行できるようにアップデートしていきます。

8.感想

最後まで本記事をご覧いただき、ありがとうございました！

今回はBizCodeXで学習した内容のアウトプットも含めて、普段使いできるツールを作成してみました。学習を進めていく上で、本ツールもどんどん改善していこうと思います。

今後も記事を更新していく予定ですので、またご覧いただけると幸いです！
(相変わらずツール作成から記事の投稿までかなり時間を要してしまいます)