今年からLINEでSALE中だよと毎週教えてもらう

目次

• 概要
• 構成
• 実装内容
• 実行結果
• 感想
• 参考資料

概要

DMM.comで商品を買い始めてから、かれこれ10年ほど経過いたしました。
10年で延べ十数万ほど購入させていただき、いろいろとお世話になってきました。
しかし、今年から出費を抑える必要が出てきましたので、SALE中の商品以外は購入しないという縛りを今年から課すことになりました。

今回はそんな縛りの中でも人生をenjoyしていくため、欲しい商品がSALE中であることを通知するツールを作成し、SALEを見逃すことなく購入できるようにした話です。
本ツールはGithubに公開しております。

構成

汚くて申し訳ないですが、ざっくり以下のような構成になっております。

①毎週金曜日の20:00にジョブをキック
②Airtableから購入予定の商品情報を取得
③DMM.com提供の商品検索APIを通じて、その商品がSALE中であるか確認
④SALE中の商品情報を通知

本ツールは、以下のサービスを借用し作成しております。

• Google Cloud Platform
• Airtable
• DMM.com 商品検索API
• LINE Messaging API

実装内容

• main.py
  ジョブIDに紐づくクラスを、リフレクションして主処理を実行しています。

main.py

from .jobs.abstract_job import AbstractJob
from flask import Request, Response
from importlib import import_module

import functions_framework
import json
import yaml

@functions_framework.http
def main(request: Request) -> Response:
    request_json = request.get_json(silent=True)
    job_id = request_json["job_id"]
    result = get_instance(job_id).execute()
    return Response(response=json.dumps({"status": result.name}))

def get_instance(job_id: str) -> AbstractJob:
    with open("config.yml", "r", encoding="utf-8") as yml:
        config = yaml.safe_load(yml)
        file_name = config["jobs"][f"{job_id}"]["file-name"]
        job_name = config["jobs"][f"{job_id}"]["job-name"]
        module = import_module(f"jobs.{file_name}")
        cls = getattr(module, job_name)
        return cls()

• sale_notification_job.py
  こちらは構成に書いている内容と同じことを書いています。

sale_notification_job.py

import sys
sys.path.append('../')

from .abstract_job import AbstractJob
from constants.job_status import JobStatus
from utils.dmm_util import DmmUtil
from utils.air_table_util import AirTableUtil
from utils.line_util import LineUtil

from itertools import filterfalse

class SaleNotificationJob(AbstractJob):
    
    def init(self):
        self.logger.info(f"{self.get_name()} init start.")
        self.air_table_util = AirTableUtil()
        self.dmm_util = DmmUtil()
        self.line_util = LineUtil()
        self.logger.info(f"{self.get_name()} init end.")

    def invoke(self) -> JobStatus:
        self.logger.info(f"{self.get_name()} invoke start.")
        
        favorite_items = self.air_table_util.fetch_favorite_items()
        not_bought_favorite_items = list(filterfalse(lambda item : item.is_bought, favorite_items))
        items = self.dmm_util.fetch_items(not_bought_favorite_items)
        sale_items = list(filter(lambda item : item.is_sale, items))
        
        if sale_items:
            self.line_util.push_announcement(sale_items)
        else:
            self.line_util.push_disappointment_message()
        
        self.logger.info(f"{self.get_name()} invoke end.")
        return JobStatus.OK

    def get_name(self) -> str:
        return "Job001(SALE情報配信)"

• dmm_util.py
  応答にcampaignという項目があるので、その有無でSALE中であるか判断しています。

dmm_util.py

import sys
sys.path.append('../')

import os
import requests

from models.item import Item
from models.favorite_item import FavoriteItem
from utils.logger_util import LoggerUtil

class DmmUtil(object):
    
    def __init__(self):
        self.api_id = os.getenv("DMM_API_ID")
        self.affiliate_id = os.getenv("DMM_AFFILIATE_ID")
        self.item_list_api_url = os.getenv("DMM_ITEM_LIST_API_URL")
        self.logger = LoggerUtil().get_logger(__name__)
    
    def fetch_items(self, not_bought_favorite_items: list[FavoriteItem]) -> list[Item]:
        items = []
        if not not_bought_favorite_items:
            return items
        
        try:
            self.logger.info("DMM.com 商品検索API ver3 実行開始")
            for item in not_bought_favorite_items:
                response = requests.get(self.item_list_api_url, params = {
                    'api_id':  self.api_id,
                    'affiliate_id': self.affiliate_id,
                    'site': item.site,
                    'service': item.service,
                    'floor': item.floor,
                    'cid': item.content_id
                })
                response.raise_for_status()

                data = response.json()
                results = data['result']['items']
                items += [
                    Item(
                        content_id = result['content_id'], 
                        title = result['title'], 
                        affiliate_url = result['affiliateURL'], 
                        sample_image_url = result['imageURL']['list'], 
                        price = result['prices']['price'], 
                        is_sale = 'campaign' in result.keys()
                    )
                    for result in results
                ]
        except Exception as e:
            self.logger.exception("DMM接続関連エラー")
            raise e
        finally:
            self.logger.info("DMM.com 商品検索API ver3 実行終了")

        return items

実行結果

• LINE

  • SALE品無しパターン
    

  • SALE品有りパターン
    ※諸事象のため一部モザイクをかけておりますが、モザイクは本ツールの機能ではありません。
    

• ローカル
  

• GCP

  • Cloud Scheduler
    Cloud IAMの認証がありますが、念のためマスキングしております。
    

  • Cloud Logging
    

感想

• アプリ
  本ツールを作成したことで、SALE中だよと定期的に教えてもらえるようになりました。
  商品によりますが、定価とSALE価格で幅がすごいのでかなりの効果が期待できそうです。

• 基盤
  GCPは比較的使用しやすいので、何かサービスを作りたい場合はおすすめできます。
  覚えることも多くないです。
  サービス廃止や名称変更等がたまに傷ですが...

参考資料