【Python】KDPのレポートをスクレイピングしてLINE通知する

アプリ開発

本ブログは、筆者が日々作成しているプログラムについて掲載していますが、時々、本ブログで取り扱うには大きすぎるプログラムを作ってしまうことがあります。

そんな時は、電子書籍としてKindleで出版することにしています。

筆者が執筆する電子書籍はニッチな内容のため、それほど売り上げは多くありませんが、それでも読んで下さる方がいて、感謝の気持ちでいっぱいになります。

それが行動として現れてしまうのが、Webレポートの見過ぎです。。σ(￣∇￣; )

Kindle出版の売り上げ状況はWebで確認できますが、売り上げがあるとうれしい気持ちになるため、つい高頻度で確認してしまいます。。

当然、Webレポート確認時に、売り上げが増えていることもあれば、変わらないこともあります。

増えているときは、「ありがとうございますっ！」とテンションが上がりますが、変わらないときは「また無駄なことをしてしまった」という気持ちになります。

この確認行為による時間の無駄と気分変動を防ぐため、売り上げがあったタイミングでLINE通知する方法について考えました。

KDP(Kindle Direct Publishing)のレポート機能

KDP(Kindle Direct Publishing)のレポート機能は、Kindleの書籍売り上げをリアルタイムに確認できる機能です。

KDPのアカウントにログインすることで、推計ロイヤリティ、注文数、既読KENPCを確認できます。

既読KENPCは、Kindle Unlimited会員の方が、無料で読んだページ数で、既読KENPCの数に応じて報酬が支払われます。有料で注文されたものと、既読KENPCによる報酬を合わせて推計ロイヤリティが計算されています。

KDPのサイトに売り上げ通知機能はありませんし、今回のようなニッチな要望に応える機能は無くて良いと思います。笑

通知機能の実現方法

Webスクレイピングで、定期的に(1時間に一度)KDPレポートを見に行き、ダッシュボードの推計ロイヤリティーと注文数を取得します。前回確認時と変化があった場合のみ、LINE通知で知らせます。

実際のLINE通知はこちらです。

「今月の予測値」は、このままの売り上げが続いた場合の１ヵ月の売り上げの推定値です。

過去１ヵ月の売り上げ推移の折れ線グラフ画像も同時に送付しています。

Amazonのスクレイピングは許可されているのか？

Amazonは利用規約でスクレイピングが許可されていないとWebで見かけることがあります。

結論から言うと、今回の利用は、個人利用であり、商業目的または第三者のために行っていないため問題ありません。

1時間に一度Webブラウザで見に行ってるだけですから、自動でやっても手動でやるのと変わらないということです。

利用規約は読み解くのが難しいですが、下記サイトが分かりやすく解説されています。

Amazonサイトのスクレイピングについて
Amazonのサイトはその「利用規約」により条件付きでスクレイピングが禁止されています。「アマゾンまたはコンテンツ提供者は、アマゾンサービスを限定的、非独占的、非商業的および個人的に利用する権利をお客様に許諾します（譲渡およびサブライセンス
kykddaddyblog.com

Selenium実行環境の整備

Seleniumの実行環境整備方法については他サイトに譲りますが、Webスクレイピングで利用するブラウザ(Chrome)とプログラムで利用するブラウザを操作するためのドライバの互換性には注意が必要です。

ブラウザは、日々の利用で最新版にアップデートされますので、追従してドライバのアップデートも必要になります。具体的には、タスクスケジューラやCron等を利用してドライバのアップデートコマンドを定期的に実行する必要があります。

筆者は、Dockerの動作環境が手元にあるため、Dockerイメージ「selenium/standalone-chrome」を利用しています。こちらを利用すれば、ブラウザとドライバがセットになっているため、バージョン互換の問題が発生しません。

Web版VSCodeを自宅サーバのDocker上に構築する
Web版VSCode(Coder)を自宅サーバのDocker上に構築しましたので、備忘録としてこの記事を書くことにしました。Webベースで、プログラム開発、サーバ構築、デプロイまでを完結させます。
truewalk.net
2022.03.28

Pythonプログラムの作成

処理の流れ

処理の流れは下記のようになります。

  1. ブラウザでKDPレポートのサイトにアクセス
  2. KDPサイトにログイン
  3. ダッシュボードの推計ロイヤリティーと注文数を取得
  4. (データ取得に失敗した場合)失敗した旨をLINE通知する
  5. (朝8時台の場合)1日の売り上げデータをグラフデータとして保存
  6. (通知時間外の場合)処理を終了
  7. (前回確認時と変わらない場合)処理を終了
  8. 1ヵ月の予測値を算出
  9. グラフデータからグラフ画像を作成
  10. LINE Notify送付

工夫したところは、KDPのサイトへのログイン部分です。

KDPのサイトはAmazonのシングルサインオンを利用していますが、高頻度でログインを行うと、ロボットと認識され、ログインクイズの画像が表示されます。

これを防ぐには、ログイン回数を減らす必要があるため、クッキーを保存し、ログインセッションを維持するように構成しています。

また、クッキーを持っている場合でも、セッションが期限切れとなった場合には、IDは入力済で、パスワードのみ再入力が必要な状態となります。そういった場合の細かい処理は試行錯誤しました。

スクリプト全体

kdpreport.py

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
import os
import time
import json
import pickle
import datetime
import calendar
import requests
import matplotlib.pyplot as plt
import matplotlib.ticker as mtick

# ---------------------------------------------------------------------
# ログイン情報
# ---------------------------------------------------------------------
# KDP
url = "https://kdpreports.amazon.co.jp/dashboard"
username = "[KDPログインユーザ名]"
password = "[KDPログインパスワード]"

# LINE
line_notify_api = "https://notify-api.line.me/api/notify"
line_notify_token = "[LINE Notifyトークン]"

# ---------------------------------------------------------------------
# クッキーファイル読み込み
# ---------------------------------------------------------------------
cookies_file = 'tmp/stay_loggedin.pkl'
cookies = []
if os.path.isfile(cookies_file):
    cookies = pickle.load(open(cookies_file, "rb"))

# ---------------------------------------------------------------------
# Chromeドライバ取得
# ---------------------------------------------------------------------
options = webdriver.ChromeOptions()
#options.add_argument("--headless")
options.add_argument("--disable-dev-shm-usage")
driver = [Webドライバーの取得コード]
driver.implicitly_wait(10)

# ---------------------------------------------------------------------
# KDPログイン
# ---------------------------------------------------------------------
# URL接続
driver.get(url)

# クッキーを設定する
for c in cookies:
    if c["domain"] == ".amazon.co.jp":
        driver.add_cookie(c)

# URL接続
driver.get(url)

# 新規ログイン実行
if not os.path.isfile(cookies_file):
    print("新規ログイン実行")
    driver.find_element(By.NAME, "email").send_keys(username)
    time.sleep(1)
    driver.find_element(By.NAME, "password").send_keys(password)
    time.sleep(1)
    driver.find_element(By.NAME, "rememberMe").click()
    time.sleep(1)
    driver.find_element(By.NAME, "password").send_keys(Keys.RETURN)
    time.sleep(5)
    # クッキーを保存
    cookies = driver.get_cookies()
    pickle.dump(cookies, open(cookies_file, "wb"))
# 新規ログインではない
else:
    # クッキーの有効期限切れの場合
    try:
        driver.find_element(By.NAME, "password").send_keys(password)
        time.sleep(1)
        driver.find_element(By.NAME, "password").send_keys(Keys.RETURN)
        time.sleep(5)
        # クッキーを保存
        print("クッキーを更新します。")
        cookies = driver.get_cookies()
        pickle.dump(cookies, open(cookies_file, "wb"))
    # クッキーが有効期限内の場合
    except:
        print("ログイン画面ではありません。続行します。")

# ---------------------------------------------------------------------
# データ取得
# ---------------------------------------------------------------------
now_data = {}
try:
    values = driver.find_elements(By.CLASS_NAME, "metric-value")
    now_data["royalty"] = values[0].text.replace("￥", "").replace("*", "").replace(",", "")
    now_data["order_num"] = values[1].text.replace(",", "")
except:
    msg = "\nサイトログインに失敗しました。"
    headers = {'Authorization': f'Bearer {line_notify_token}'}
    data = {'message': msg}
    requests.post(line_notify_api, headers=headers, data=data)
finally:
    time.sleep(5)
    driver.quit()

# ---------------------------------------------------------------------
# 1日1度のバッチ処理
# ---------------------------------------------------------------------
now_date = datetime.datetime.today()
if now_date.hour == 8:
    # グラフデータ読み込み
    f = open("tmp/graph.json", "r", encoding='utf-8')
    graph = json.load(f)
    f.close()
    # 今日のデータを追加
    add_data = {}
    add_data["date"] = str(now_date.month) + "/" + str(now_date.day)
    sum = 0
    for g in graph:
        if g["date"].startswith(str(now_date.month) + "/"):
            sum = sum + int(g["royalty"])
    add_data["royalty"] = str(int(now_data["royalty"]) - sum)
    graph.append(add_data)
    # データ数が30を超えたら1番古いデータを削除
    if len(graph) > 30:
        graph.pop(0)
    # グラフデータ書き込み
    f = open("tmp/graph.json", "w", encoding='utf-8')
    f.write(json.dumps(graph, ensure_ascii=False))
    f.close()

# ---------------------------------------------------------------------
# 通知可能な時間帯を制限する
# ---------------------------------------------------------------------
hour_from = 21
hour_to = 6
if now_date.hour >= hour_from or now_date.hour <= hour_to:
    print("寝ている時間です。通知をやめときます。")
    exit()

# ---------------------------------------------------------------------
# 前回取得データと比較
# ---------------------------------------------------------------------
f = open("tmp/pre_data.json", "r", encoding='utf-8')
pre_data = json.load(f)
f.close()
# 前回取得データと同じ場合は何もせず終了
if pre_data["royalty"] == now_data["royalty"] and pre_data["order_num"] == now_data["order_num"]:
    print("前回確認時と同じでした。終了します。")
    exit()
# 前回データと違う場合はJSONファイル更新
f = open("tmp/pre_data.json", "w", encoding='utf-8')
f.write(json.dumps(now_data, ensure_ascii=False))
f.close()

# ---------------------------------------------------------------------
# 1ヵ月の予測値を算出
# ---------------------------------------------------------------------
last_day = calendar.monthrange(now_date.year, now_date.month)[1]
predicted = int((int(now_data["royalty"]) / now_date.day) * last_day)

# ---------------------------------------------------------------------
# グラフ画像作成
# ---------------------------------------------------------------------
# グラフデータ読み込み
f = open("tmp/graph.json", "r", encoding='utf-8')
graph = json.load(f)
f.close()
# 最新データ追加
add_data = {}
add_data["date"] = str(now_date.month) + "/" + str(now_date.day)
sum = 0
for g in graph:
    if g["date"].startswith(str(now_date.month) + "/"):
        sum = sum + int(g["royalty"])
add_data["royalty"] = str(int(now_data["royalty"]) - sum)
graph.append(add_data)
# グラフ化
x_l = [d.get("date") for d in graph]
y_l = [int(d.get("royalty")) for d in graph]
fig, ax = plt.subplots()
ax.yaxis.set_major_formatter(mtick.StrMethodFormatter('¥{x:,.0f}')) 
ax.plot(x_l, y_l)
plt.xticks(x_l[::5])
plt.savefig("tmp/graph.png", format="png")

# ---------------------------------------------------------------------
# メッセージ送信
# ---------------------------------------------------------------------
msg = "\n今月のロイヤリティ(現在)： ￥" + f'{int(now_data["royalty"]):,}' + "\n今月の注文数：　" + f'{int(now_data["order_num"]):,}' + "\n今月の予測値：　￥" + f'{predicted:,}'
headers = {'Authorization': f'Bearer {line_notify_token}'}
data = {'message': msg}
files = {'imageFile': open("tmp/graph.png", "rb")}
requests.post(line_notify_api, headers=headers, data=data, files=files)

スクリプト修正

上記スクリプトは、そのままでは動作しません。4か所修正が必要です。

  • [KDPログインユーザ名]
  • [KDPログインパスワード]
  • [LINE Notifyトークン]
  • [Webドライバーの取得コード]

LINE Notifyトークンの取得方法はこちらに記載しています。

【WordPress】コメントのスパム対策とLINE Notify設定
WordPressでブログを運営しているとスパムコメントが大量に来ます。当ブログでは定期的にスパムマークをつけるようにしていましたが、通常のコメントが増えてくると見分けるのが大変です。本記事では当ブログで行ったスパムコメント対策とコメントが来たときのLINE通知設定の方法を説明します。
truewalk.net
2024.04.30

[Webドライバーの取得コード]は、Selenium実行環境によって異なるため明記していませんが、Windowsで実施される場合は、こちらが参考になると思います。

[Python] Windows環境でseleniumを動かすまで - Qiita
PythonインストールPython公式から最新版インストーラをダウンロードしてインストールしてください。参考はこちら。インストールが終わったら、以下のコマンドでバージョン確認します。pyth…
qiita.com

実行フォルダ整備

Pythonスクリプト「kdpreport.py」を置いたフォルダに「tmp」フォルダを作成します。

「tmp」フォルダ内には、JSONファイル「graph.json」を作成し、下記内容を記入して保存します。

[]

↑半角の大括弧です。JSONファイルでは、空の配列を意味します。この配列にグラフデータを自動で格納していくことになります。

定期実行設定

タスクスケジューラやCron等で、1時間に一度定期実行させれば完成です。

必要に応じてシェルスクリプトやバッチファイルでラップしましょう。

シェルスクリプトの例です。

run_kdpreport.sh

#!/bin/bash

python kdpreport.py

exit 0

コメント

タイトルとURLをコピーしました