23日前公開・23日前更新・0 pv・13 min read

【個人開発】オクトパスエナジーAPIで電気代を通知するDiscord Botを作ってみた

https://cdn.magicode.io/media/notebox/e7c8b6d6-3e60-4e86-8c5c-999f04d361c4.jpeg

【個人開発】オクトパスエナジーAPIで電気代を通知するDiscord Botを作ってみた

1. はじめに

オクトパスエナジーではさまざまなAPIが提供されています。しかし、その存在を知っている利用者は意外と少ないのではないでしょうか？調べてみると、電力使用量や料金に関するデータを取得できるAPIもあり、これを活用すれば面白いことができそうだと考えてこのAPIを使って、電気代を通知するDiscord Botを作ってみることにした。本当はLINE Notifyを利用したかったのですがサービスが終了しちゃったので…

2. 目標

最終的な目標としてはどこかのホスティングサービスを利用して、定期実行（毎朝通知）みたいなことをやってみたいなと思っています。が、まずはローカルで作成してみます。

3. 開発

DiscordBOTを作成する

DiscordBOTアカウントの作成と設定 DiscordDeveloperページにアクセスします。

ログインができたらApplicationsのページに移動します。上記画面が表示されたら右上にある「New Application」をクリックします。 BOTで使用したい名前を入力して「Create」をクリック。

Botタブの「Reset Token」をクリックしトークンを取得します。次の項目で作成する.envファイルに記載するのでメモしておきます。 ※シークレットトークンなので自分以外の人に教えないようにしてください！漏れてしまうとBOTが悪用されてしまう可能性があります！

次に、BOTとのやり取りを行うための設定として以下の画像と同じになるように
設定を変更してください。

DiscordBOTをサーバに招待する自分だけが参加しているサーバに今回作成したBOTアカウンを招待します。以下の画像のように「Bot」、「Administrator」にチェックをして「Generated URL」の「Copy」をクリックしてコピーされたURLをブラウザに貼り付けて招待完了です！

Dockerで開発環境を作る

以下のファイルを適当なディレクトリに作成します。

Dockerfile

FROM ubuntu:22.04

# タイムゾーンの設定（インタラクティブなプロンプトを避けるため）
ENV DEBIAN_FRONTEND=noninteractive
ENV TZ=Asia/Tokyo

# 必要なパッケージのインストール
RUN apt-get update && apt-get install -y \
    tzdata \
    python3 \
    python3-pip \
    python3-venv \
    git \
    iputils-ping \
    vim \
    && ln -fs /usr/share/zoneinfo/Asia/Tokyo /etc/localtime \
    && dpkg-reconfigure -f noninteractive tzdata \
    && echo "Asia/Tokyo" > /etc/timezone \
    && rm -rf /var/lib/apt/lists/*

# 作業ディレクトリの作成
WORKDIR /app

# Pythonの依存関係ファイル
COPY requirements.txt .

# 依存関係のインストール
RUN pip3 install --no-cache-dir -r requirements.txt

# コンテナを終了させないために sleep infinity を実行
CMD ["sleep", "infinity"]

Docker-compose.yaml

version: '3'

services:
discord-bot:
    build: .
    volumes:
    - .:/app
    restart: unless-stopped
    env_file:
    - .env

.env

OCTOPUS_EMAIL=<オクトパスエナジーに登録しているメアド>
OCTOPUS_PW=<オクトパスエナジーのパスワード>
ACCOUNT_NUMBER=<オクトパスエナジーのお客さま番号>
DISCORD_TOKEN=<さっきコピーしたToken>
DISCORD_CHANNEL_ID=<DiscordサーバのID>

requirements.txt

discord.py>=2.0.0
python-dotenv>=0.19.0
requests>=2.32.3

コンテナの作成とアタッチ Docker Desktop のインストールは省略します。
```
docker-compose up --build
docker exec -it <Container name> bash
```

電気使用量の出力プログラムの作成

　電気使用量を取得するには、認証トークンの取得をしてから使用状況データの取得と段階を踏まなければいけません。　また、認証トークンの有効期限は１時間なので実行するたびに取得する必要があります。

auth_token_retriever.py

import requests
import os
from dotenv import load_dotenv

# .env ファイルから環境変数を読み込む
load_dotenv()

def get_auth_token():
    # GraphQL エンドポイント
    GRAPHQL_URL = "https://api.oejp-kraken.energy/v1/graphql/"

    # GraphQL Mutation クエリ
    graphql_query = """
    mutation login($input: ObtainJSONWebTokenInput!) {
        obtainKrakenToken(input: $input) {
            token
        }
    }
    """

    # 送信するデータ
    variables = {
        "input": {
            "email": os.getenv('OCTOPUS_EMAIL'),
            "password": os.getenv('OCTOPUS_PW')
        }
    }

    # HTTP ヘッダー
    headers = {
        "Content-Type": "application/json"
    }

    try:
        response = requests.post(
            GRAPHQL_URL,
            json={"query": graphql_query, "variables": variables},
            headers=headers
        )
        response_data = response.json()

        if "errors" in response_data:
            print("認証エラー:", response_data["errors"])
            return None

        token = response_data["data"]["obtainKrakenToken"]["token"]
        return token

    except Exception as e:
        print("エラーが発生しました:", str(e))
        return None

# 実行
if __name__ == "__main__":
    token = get_auth_token()
    print("取得したトークン:", token)

Electricity_bill_inquiry.py

# bot.py
import requests
import subprocess
import os
from datetime import datetime, timedelta
import json
from dotenv import load_dotenv
import auth_token_retriever

# .env ファイルから環境変数を読み込む
load_dotenv()

# Octopus Energy API の設定
OCTOPUS_TOKEN = auth_token_retriever.get_auth_token()
ACCOUNT_NUMBER = os.getenv('ACCOUNT_NUMBER')

# GraphQL エンドポイント
GRAPHQL_URL = "https://api.oejp-kraken.energy/v1/graphql/"

def get_electricity_consumption():
    # 現在の日時を取得
    now = datetime.utcnow()

    # 前日の日付を計算
    previous_day = now - timedelta(days=1)

    # "YYYY-MM-DDTHH:MM:SS+00:00Z" 形式に変換
    str_date = previous_day.strftime("%Y-%m-%d")
    start_date = previous_day.strftime("%Y-%m-%dT00:00:00+00:00Z")
    end_date = now.strftime("%Y-%m-%dT00:00:00+00:00Z")

    """指定期間の電気使用量を取得する"""

    # GraphQL クエリ
    QUERY = """
    query halfHourlyReadings($accountNumber: String!, $fromDatetime: DateTime, $toDatetime: DateTime) {
    account(accountNumber: $accountNumber) {
        properties {
        electricitySupplyPoints {
            status
            agreements {
            validFrom
            }
            halfHourlyReadings(fromDatetime: $fromDatetime, toDatetime: $toDatetime) {
            startAt
            value
            costEstimate
            consumptionStep
            consumptionRateBand
            }
        }
        }
    }
    }
    """

    # リクエスト用の変数
    VARIABLES = {
        "accountNumber": ACCOUNT_NUMBER,
        "fromDatetime": start_date,
        "toDatetime": end_date
    }

    # リクエストヘッダー（API トークンを含む）
    HEADERS = {
        "Content-Type": "application/json",
        "Authorization": OCTOPUS_TOKEN
    }

    try:
        response = requests.post(GRAPHQL_URL, json={"query": QUERY, "variables": VARIABLES}, headers=HEADERS)

        # ステータスコードをチェック
        if response.status_code != 200:
            print(f"❌ エラー: ステータスコード {response.status_code}")
            print(response.text)
            return None

        # JSON レスポンスを解析
        data = response.json()

        # GraphQL のエラー処理
        if "errors" in data:
            print("❌ GraphQL エラー:", json.dumps(data["errors"], indent=2, ensure_ascii=False))
            return None

        # JSON からデータを抽出
        properties = data["data"]["account"]["properties"]
        electricity_supply_points = properties[0]["electricitySupplyPoints"]
        half_hourly_readings = electricity_supply_points[0]["halfHourlyReadings"]

        # 消費電力とコストを合計
        kwh = sum(float(reading["value"]) for reading in half_hourly_readings)
        cost = sum(float(reading["costEstimate"]) for reading in half_hourly_readings)

        # 結果を整形
        text = f"{str_date}は{kwh:.2f}kWh消費して{cost:.0f}円でした。"
        return text

    except requests.exceptions.RequestException as e:
        print(f"❌ リクエストエラー: {e}")
        return None

if __name__ == "__main__":
    result = get_electricity_consumption()

    if result:
        print("✅ 取得成功！:")
        print(result)

    else:
        print("⚠ データを取得できませんでした。")

DiscordBOTのメインプログラム作成

main.py

# bot.py
import requests
import subprocess
import os
import datetime
import discord
from discord.ext import commands, tasks
from dotenv import load_dotenv

# .env ファイルから環境変数を読み込む
load_dotenv()

# Discord botのトークンを取得
DISCORD_TOKEN = os.getenv('DISCORD_TOKEN')
DISCORD_CHANNEL_ID = int(os.getenv('DISCORD_CHANNEL_ID'))

# Intentを設定
intents = discord.Intents.default()
intents.message_content = True  # メッセージ内容へのアクセスを有効化

# ボットの初期化
intents = discord.Intents.default()
intents.message_content = True
bot = commands.Bot(command_prefix="!", intents=intents)

# login event
@bot.event
async def on_ready():
    print(f'{bot.user.name} としてログインしました！')
    print('------------------------------------')

@bot.event
async def on_message(message):
    print(f"メッセージ受信: {message.content}")
    await bot.process_commands(message)

# test
@bot.command()
async def run(ctx, script_name: str = None):
    if script_name is None:
        await ctx.send("⚠ エラー: スクリプト名を指定してください。\n使用例: `!run sample`")
        return
    
    script_path = os.path.join(os.getcwd(), script_name )

    if not os.path.exists(script_path):
        await ctx.send(f"エラー: `{script_name}` が見つかりません。")
        return

    try:
        # スクリプトを実行し、出力を取得
        result = subprocess.run(
            ["python3", script_path],
            capture_output=True, text=True, timeout=10
        )
        output = result.stdout or "スクリプトの実行が完了しました。"
        await ctx.send(f"✅ `{script_name}` の実行結果:\n```\n{output}\n```")
    except Exception as e:
        await ctx.send(f"❌ スクリプトの実行中にエラーが発生しました:\n```\n{e}\n```")

# Botを実行
if __name__ == "__main__":
    bot.run(DISCORD_TOKEN)

DiscordBOTの起動とコマンドの送信

アタッチしたCMDでメインプログラムを起動します。

python3 main.py

discordのチャット欄からコマンドを送信

!run Electricity_bill_inquiry.py

無事に取得できました！

【個人開発】オクトパスエナジーAPIで電気代を通知するDiscord Botを作ってみた