自宅のSwitchBotをDiscordから操作できるBotを作った話

• By dogon309 • 開発記事 • タグ: , , , ,

こんにちは、DOGONです。
最近体調を崩して引きこもっています。笑

我が家ではほとんどの家電をSwitchBotで操作できるようにしています。
ただ、普段スマホをあまり使わない自分にとって、いちいちスマホで操作するのが面倒なんですよね……。
（というか、SwitchBotの公式アプリってレスポンス悪くないですか？）

ということで、パソコンからも操作できるように、SwitchBot APIを叩くDiscord Botを作ってみました。今回はその開発記録です。

✅ 完成品

実際にBotとやり取りできるDiscordサーバーはこちらです：
👉 https://discord.gg/yRhCUMBswq

🔧 開発環境

  • macbook air M1
  • Docker
  • PostgreSQL
  • Proxmox + Ubuntu Server（自宅サーバー）

🧱 使用ライブラリ

  • discord.py
  • requests
  • sqlalchemy
  • cryptography

🎯 対応機能

すべてのSwitchBot製品に対応するのは大変なので、
今回は赤外線リモコンのみに対応しました。
（カスタムボタンなどには未対応です）

利用者が増えてリクエストがあれば、他デバイスも対応するかも…笑

📡 SwitchBot APIの使い方

公式のAPI仕様書はこちら →
https://github.com/OpenWonderLabs/SwitchBotAPI

APIを叩くためには、以下のHeaderが必要です：

  • Authorization：トークン
  • sign：シークレットで生成した署名（Base64形式）
  • nonce：一意な値（UUID）
  • t：タイムスタンプ（ミリ秒）

署名は以下のように生成しました：

class SwitchBot:
    def __init__(self, token: str, secret: str):
        nonce = str(uuid.uuid4())
        t = int(round(time.time() * 1000))
        string_to_sign = "{}{}{}".format(token, t, nonce)
        string_to_sign = bytes(string_to_sign, "utf-8")
        secret = bytes(secret, "utf-8")
        sign = base64.b64encode(
            hmac.new(secret, msg=string_to_sign, digestmod=hashlib.sha256).digest()
        )

        self.headers = {}
        self.headers["Authorization"] = token
        self.headers["Content-Type"] = "application/json"
        self.headers["charset"] = "utf8"
        self.headers["t"] = str(t)
        self.headers["sign"] = str(sign, "utf-8")
        self.headers["nonce"] = nonce

🔐 セキュリティ対策（トークン暗号化）

from cryptography.fernet import Fernet

FERNET_KEY = os.environ.get("FERNET_KEY")
if not FERNET_KEY:
    raise ValueError("FERNET_KEY is not set")

fernet = Fernet(FERNET_KEY)

def encrypt_token(token: str) -> str:
    return fernet.encrypt(token.encode()).decode()

def decrypt_token(token: str) -> str:
    return fernet.decrypt(token.encode()).decode()

保存されたトークンやシークレットは万が一DBが漏洩した場合に備えて暗号化しています。

今回は cryptography.Fernet を使って、以下のように実装：

  • encrypt_token(token)：トークンを暗号化してDBに保存用の文字列を返す
  • decrypt_token(token)：暗号化された文字列を元のトークンに戻す

鍵（FERNET_KEY）は .env やDockerの環境変数に設定することで、安全に管理しています。
(いまいち、環境変数に保存してたら大丈夫みたいなのもちょっとわからない気がしますが……)

🗃 データベース設計（SQLAlchemy）

今回はORMとして SQLAlchemy を使用しました。

モデル

class User(Base):
    __tablename__ = "users"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    discord_id = Column(String, unique=True, nullable=False)
    switchbot_token = Column(String, nullable=False)
    switchbot_secret = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)

UUID(as_uuid=True) はUUID型のカラムを定義するもので、as_uuid=True をつけると SQLAlchemy で Python の uuid.UUID オブジェクトとして扱えます。
デフォルトの uuid.uuid4 を使って毎回自動でユニークなIDが生成されます。

DiscordのユーザーIDは非常に大きな整数（例: 734256882932547678）ですが、精度の問題を避けるために文字列（String）として保存しています。
また、同じユーザーが複数レコードを持たないように unique=True で一意にしています。

CRUD処理

def get_token(db: Session, discord_user_id: str):
    user = db.query(User).filter(User.discord_id == discord_user_id).first()
    if user:
        user.switchbot_token = decrypt_token(user.switchbot_token)
        user.switchbot_secret = decrypt_token(user.switchbot_secret)
    return user

def save_token(db: Session, discord_user_id: str, switchbot_token: str, switchbot_secret: str):
    encrypted_token = encrypt_token(switchbot_token)
    encrypted_secret = encrypt_token(switchbot_secret)
    user = db.query(User).filter(User.discord_id == discord_user_id).first()
    if user:
        user.switchbot_token = encrypted_token
        user.switchbot_secret = encrypted_secret
    else:
        user = User(
            discord_id=discord_user_id,
            switchbot_token=encrypted_token,
            switchbot_secret=encrypted_secret,
        )
        db.add(user)
    db.commit()
    db.refresh(user)
    return user

🤖 Discord Bot本体の実装

ここからは、実際にDiscord Botを動かすための「メイン部分」と「Cog（機能モジュール）」の実装例を紹介します。

  • main.py などのメイン実行ファイル
  • cogs/switchcog.py（SwitchBot操作用Cog）
  • cogs/reportcog.py（バグ・要望報告用Cog）

1. main.py（Bot起動・イベント定義・Cog読み込み）

TOKEN = os.getenv("DISCORD_TOKEN")

intents = discord.Intents.default()
intents.members = True

INITIAL_EXTENSIONS = [
    "cogs.switchcog",
    "cogs.reportcog"
]

class SwitchBot(commands.Bot):
    def __init__(self):
        super().__init__(
            command_prefix="/",
            intents=intents,
            status=discord.Status.online,
            activity=discord.Activity(
                type=discord.ActivityType.competing,
                name="SwitchBot-bot",
            ),
        )

    async def setup_hook(self):
        for cog in INITIAL_EXTENSIONS:
            try:
                await self.load_extension(cog)
            except Exception:
                traceback.print_exc()
            else:
                pass
        
    async def on_ready(self):
        for cog in INITIAL_EXTENSIONS:
            try:
                await self.reload_extension(cog)
            except Exception:
                traceback.print_exc()
            else:
                pass
        await self.tree.sync()

bot = SwitchBot()

# サーバー参加者にDMを送信する
@bot.event
async def on_member_join(member):
    try:
        response = discord.Embed()
        response.title = "ようこそ!!"
        response.description = f"{member.name}さん、サーバーへようこそ!!\nコマンドはこちらのDMへお願いします。\n(サーバーへのコマンド送信はセキュリティーがとても弱くなるのでおやめください)"
        response.color = discord.Color.green()
        await member.send(embed=response)
    except discord.Forbidden:
        print(f"{member.name}にDMを送信できませんでした。")

@bot.command()
@commands.is_owner()
async def reload(ctx):
    for cog in INITIAL_EXTENSIONS:
        try:
            await bot.reload_extension(cog)
        except Exception:
            traceback.print_exc()
        else:
            print(f"extension [{cog}] is loaded")
    await bot.tree.sync()

bot.run(TOKEN)
  • intents.members = True
    サーバー内のメンバー参加イベント（on_member_join）を拾うためには、BotのIntentsで members を許可する必要があります。Discord開発者ポータルでも「Server Members Intent」を有効化しておきましょう。
  • setup_hook と on_ready の違い
    • setup_hook: Botが内部的に起動するタイミングで一度だけ呼ばれるメソッド。Cogの初期ロードに使います。
    • on_ready: Discordサーバーへの接続が完了し、Botが稼働した状態になったときに呼ばれるイベント。スラッシュコマンド同期や再ロード処理に使います。
  • commands.is_owner()
    /reload コマンドをBotオーナー（自分）だけ実行できるようにするデコレータ。不要なら外してください。

2. cogs/switchcog.py（SwitchBot連携Cog）

def get_user(user_id: str) -> User | None:
    session = SessionLocal()

    try:
        user = session.query(User).filter_by(discord_id=user_id).first()
        if not user or not user.switchbot_token or not user.switchbot_secret:
            return None
        return user
        
    except Exception as e:
        print("Error: {e}")
        return None

class AirConditionerConfigView(discord.ui.View):
    def __init__(self, device_id: str, command: str, user: User):
        super().__init__(timeout=None)
        self.device_id = device_id
        self.command = command
        self.user = user

        self.temperature = None
        self.mode = None
        self.fan_speed = None
        self.power = None

        self.add_item(self.TemperatureSelect(self))
        self.add_item(self.ModeSelect(self))
        self.add_item(self.FanSpeedSelect(self))
        self.add_item(self.PowerSelect(self))
        self.add_item(self.ConfirmButton(self))

    class TemperatureSelect(discord.ui.Select):
        def __init__(self, parent_view):
            options = [
                discord.SelectOption(label=f"{i}℃", value=str(i))
                for i in range(16, 31)
            ]
            super().__init__(placeholder="温度を選択", options=options)
            self._parent_view = parent_view

        async def callback(self, interaction: discord.Interaction):
            self._parent_view.temperature = self.values[0]
            await interaction.response.defer()
        
    class ModeSelect(discord.ui.Select):
        def __init__(self, parent_view):
            options = [
                discord.SelectOption(label=f"{item['name']}", value=f"{item['value']}")
                for item in AIR_CONDITIONER_MODE
            ]
            super().__init__(placeholder="モードを選択", options=options)
            self._parent_view = parent_view
        
        async def callback(self, interaction: discord.Interaction):
            self._parent_view.mode = self.values[0]
            await interaction.response.defer()

    class FanSpeedSelect(discord.ui.Select):
        def __init__(self, parent_view):
            options = [
                discord.SelectOption(label=f"{item['name']}", value=f"{item['value']}")
                for item in AIR_CONDITIONER_FAN_SPEED
            ]
            super().__init__(placeholder="ファン速度を選択", options=options)
            self._parent_view = parent_view
        
        async def callback(self, interaction: discord.Interaction):
            self._parent_view.fan_speed = self.values[0]
            await interaction.response.defer()

    class PowerSelect(discord.ui.Select):
        def __init__(self, parent_view):
            options = [
                discord.SelectOption(label=item["name"], value=item["value"])
                for item in AIR_CONDITIONER_POWER
            ]
            super().__init__(placeholder="電源を選択", options=options)
            self._parent_view = parent_view

        async def callback(self, interaction: discord.Interaction):
            self._parent_view.power = self.values[0]
            await interaction.response.defer()

    class ConfirmButton(discord.ui.Button):
        def __init__(self, parent_view):
            super().__init__(label="設定を送信", style=discord.ButtonStyle.green)
            self._parent_view = parent_view

        async def callback(self, interaction: discord.Interaction):
            if None in (self._parent_view.mode, self._parent_view.fan_speed, self._parent_view.power):
                await interaction.response.send_message("すべての項目を選択してください。")
                return

            parameter = f"{self._parent_view.temperature},{self._parent_view.mode},{self._parent_view.fan_speed},{self._parent_view.power}"
            
            # APIに送る処理を書く
            try: 
                token = decrypt_token(self._parent_view.user.switchbot_token)
                secret = decrypt_token(self._parent_view.user.switchbot_secret)
                sb = SwitchBot(token=token, secret=secret)
                sb.post_command(
                    device_id=self._parent_view.device_id,
                    command=self._parent_view.command,
                    command_type= "command",
                    parameter=parameter,
                )

                await interaction.response.send_message(f"`setAll`: `{parameter}` を送信しました！")

            except Exception as e:
                print(f"ConfirmButton Error: {e}")
                response = command_error()
                await interaction.response.send_message(embed=response)

class CommandSelectView(discord.ui.View):
    def __init__(self, device_id: str, commands: list[dict], user: User, timeout=None):
        super().__init__(timeout=timeout)
        self.add_item(CommandSelect(device_id, commands, user))

class CommandSelect(discord.ui.Select):
    def __init__(self, device_id: str, commands: list[dict], user: User):
        self.device_id = device_id
        self.user = user
        options = [
            discord.SelectOption(label=cmd["name"], value=cmd["command"])
            for cmd in commands
        ]
        super().__init__(placeholder="コマンドを選択してください。", min_values=1, max_values=1, options=options)

    async def callback(self, interaction: discord.Interaction):
        selected_command = self.values[0]

        # エアコンのsetAllの場合は設定
        if selected_command == "setAll":
            print(f"selected_command: {selected_command}")

            view = AirConditionerConfigView(device_id=self.device_id, command=selected_command, user=self.user)
            await interaction.response.send_message(view=view)
            return 

        # ユーザーを取得
        token = decrypt_token(self.user.switchbot_token)
        secret = decrypt_token(self.user.switchbot_secret)

        sb = SwitchBot(token=token, secret=secret)

        # デバイス情報を取得してtypeを判定
        device_info = sb.devices_by_device_id(device_id=self.device_id)
        is_remote = device_info.get("type") == "remote"
        command_type = "command"

        sb.post_command(
            device_id=self.device_id,
            command=selected_command,
            command_type=command_type,
            parameter="default",
        )

        await interaction.response.send_message(f"選択されたコマンド: `{selected_command}`を実行します。")

class SwitchCog(commands.Cog):
    def __init__(self, bot):
        self.bot = bot

    @commands.Cog.listener()
    async def on_ready(self):
        print("[Cogs] Switch is ready")

    @app_commands.command(name="register", description="SwitchBotのトークンを登録します")
    async def register(self, interaction: discord.Interaction, token: str = None, secret: str = None):
        # レスポンスEmbed
        response = discord.Embed()

        # DMのみに制限
        if not isinstance(interaction.channel, discord.DMChannel):
            response = guild_error()
            await interaction.response.send_message(embed=response)
            return 

        # tokenとsecret
        if token is None:
            response = command_error()
            await interaction.response.send_message(embed=response)
            return 

        if secret is None:
            response = command_error()
            await interaction.response.send_message(embed=response)
            return

        # DB登録処理
        session = SessionLocal()
        try:
            user_id = str(interaction.user.id)
            existing = session.query(User).filter_by(discord_id=user_id).first()
            if existing:
                response.title = "登録済み"
                response.description = "あなたのアカウントにはすでにトークンが登録されています"
                response.color = discord.Color.orange()
            else:
                encrypted_token = encrypt_token(token)
                encrypted_secret = encrypt_token(secret)
                new_entry = User(
                    discord_id=user_id,
                    switchbot_token=encrypted_token,
                    switchbot_secret=encrypted_secret,
                )
                session.add(new_entry)
                session.commit()

                response.title = "登録完了"
                response.description = "SwitchBotのトークンを暗号化して登録されました。"
                response.color = discord.Color.green()
        
        except SQLAlchemyError as e:
            print(f"DB Error: {e}")
            response = except_response()
        finally:
            session.close()
        
        await interaction.response.send_message(embed=response)

    @app_commands.command(name="devices", description="操作可能なデバイスの一覧を返します")
    async def devices(self, interaction: discord.Interaction):
        # レスポンスEmbed
        response = discord.Embed()

        # DMのみに制限
        if not isinstance(interaction.channel, discord.DMChannel):
            response = guild_error()
            await interaction.response.send_message(embed=response)
            return 
        
        # トークンの存在確認
        session = SessionLocal()
        try:
            user_id = str(interaction.user.id)
            user = session.query(User).filter_by(discord_id=user_id).first()
            if not user or not user.switchbot_token or not user.switchbot_secret:
                response = user_error()
                await interaction.response.send_message(embed=response)
                return
            if user:
                token = decrypt_token(user.switchbot_token)
                secret = decrypt_token(user.switchbot_secret)
                sb = SwitchBot(token, secret)
                data = sb.devices()
        except Exception as e:
            print(f"Error: {e}")
            response = device_error()
            return await interaction.response.send_message(embed=response)
        
        response.title = "【SwitchBot デバイス一覧】"

        device_list = data["body"].get("deviceList", [])
        infrared_list = data["body"].get("infraredRemoteList", [])

        # デバイス
        devices_text = ""
        for d in device_list:
            devices_text += f"・{d['deviceName']} ({d['deviceType']})\n"
        if not devices_text:
            devices_text = "・なし\n"
        response.add_field(name="■ デバイス", value=devices_text, inline=False)

        # 赤外線リモコン
        infrared_text = ""
        for d in infrared_list:
            infrared_text += f"・{d['deviceName']} ({d['remoteType']})\n"
        if not infrared_text:
            infrared_text = "・なし\n"
        response.add_field(name="■ 赤外線リモコン", value=infrared_text, inline=False)

        await interaction.response.send_message(embed=response)

    async def devices_autocomplete(self, interaction: discord.Interaction, current: str):
        session = SessionLocal()
        try:
            user_id = str(interaction.user.id)
            user = session.query(User).filter_by(discord_id=user_id).first()
            if not user or not user.switchbot_token or not user.switchbot_secret:
                return []
            
            token = decrypt_token(user.switchbot_token)
            secret = decrypt_token(user.switchbot_secret)

            sb = SwitchBot(token=token, secret=secret)
            result = sb.devices()

            device_choices = []

            for device in result["body"].get("deviceList", []):
                name = f"{device.get('deviceName')} ({device.get('deviceType')})"
                if current.lower() in name.lower():
                    device_choices.append(app_commands.Choice(
                        name=name,
                        value=device["deviceId"]
                    ))

            for remote in result["body"].get("infraredRemoteList", []):
                name = f"{remote.get('deviceName')} ({remote.get('remoteType')})"
                if current.lower() in name.lower():
                    device_choices.append(app_commands.Choice(
                        name=name,
                        value=remote["deviceId"]
                    ))

            return device_choices[:25]
        
        except Exception as e:
            print(f"Autocomplete error: {e}")
            return []

        finally:
            session.close()


    @app_commands.command(name="support_commands", description="各デバイスで対応しているコマンドの一覧を返します")
    @app_commands.describe(device="コマンドを知りたいデバイス")
    @app_commands.autocomplete(device=devices_autocomplete)
    async def support_commands(self, interaction: discord.Interaction, device: str):
        # レスポンスEmbed
        response = discord.Embed()

        # ユーザー取得
        try:
            user = get_user(user_id=str(interaction.user.id))

            if user is None:
                response = user_error()
                await interaction.response.send_message(embed=response)
                return
            if user:
                token = decrypt_token(user.switchbot_token)
                secret = decrypt_token(user.switchbot_secret)

                sb = SwitchBot(token=token, secret=secret)
                result = sb.devices_by_device_id(device_id=device)
                device_type = result["deviceType"] if "deviceType" in result else result["remoteType"]

                commands = sb.get_supported_commands(device_type, is_remote=(result["type"] == "remote"))

                print(commands)

                response.title = f"[{result['deviceName']}の対応コマンド一覧]"
                
                commands_text = ""
                for command in commands:
                    commands_text += f"・{command['name']} ({command['command']})\n"
                if not commands_text:
                    commands_text = f"・なし\n"
                response.add_field(name="■ コマンド", value=commands_text, inline=False)
            
        except Exception as e:
            print(f"Error: {e}")
            response = except_response()
            return await interaction.response.send_message(embed=response)
        
        await interaction.response.send_message(embed=response)

    @app_commands.command(name="command", description="コマンドを実行するためのセレクビューを返します。")
    @app_commands.describe(device="コマンドを実行するデバイス")
    @app_commands.autocomplete(device=devices_autocomplete)
    async def command(self, interaction: discord.Interaction, device: str):
        # レスポンスEmbed
        response = discord.Embed()

        # DM制限
        if not isinstance(interaction.channel, discord.DMChannel):
            response = guild_error()
            await interaction.response.send_message(embed=response)

        # ユーザー取得
        try:
            user = get_user(user_id=str(interaction.user.id))

            if user is None:
                response = user_error()
                await interaction.response.send_message(embed=response)
                return
            if user:
                token = decrypt_token(user.switchbot_token)
                secret = decrypt_token(user.switchbot_secret)

                sb = SwitchBot(token=token, secret=secret)
                result = sb.devices_by_device_id(device_id=device)
                device_type = result["deviceType"] if "deviceType" in result else result["remoteType"]

                commands = sb.get_supported_commands(device_type, is_remote=(result["type"] == "remote"))

                response.title = f"[{result['deviceName']}]のコマンド操作"

                view = CommandSelectView(device_id=result["deviceId"], commands=commands, user=user)
        
        except Exception as e:
            print(f"Error: {e}")
            response = except_response()
            return await interaction.response.send_message(embed=response)

        await interaction.response.send_message(embed=response, view=view)

async def setup(bot: commands.Bot):
    await bot.add_cog(SwitchCog(bot))
  1. get_user 関数
    • with SessionLocal() as session: を使うことで、自動的にセッションがクローズされます。
    • 取得直後に decrypt_token() をかけ、Bot処理内では平文トークンを使えるようにしています。
  2. /register コマンド
    • DMからのみ実行可能に制限することで、トークン漏洩のリスクを低減。
    • 新規は「登録」というフローに分けています。
  3. /devices コマンド
    • SwitchBot API から返ってくる JSONを整形し、Embedの2つのフィールドに分けて表示。
    • 失敗時はあらかじめ用意しておいた device_error() Embedを返します。

3. cogs/reportcog.py（バグ／要望受付用Cog）

解説ポイント

NOTIFICATION_GUILD = os.getenv("NOTIFICATION_GUILD") # バグ・要望を受けるサーバー
NOTIFICATION_CHANNEL = os.getenv("NOTIFICATION_CHANNEL") # バグ・要望を受けるチャンネル

class ReportCog(commands.Cog):
    def __init__(self, bot):
        self.bot = bot

    @commands.Cog.listener()
    async def on_ready(self):
        print("[Cogs] Report is ready")

    @app_commands.command(name="report", description="バグ・不具合・改善点を通報します")
    @app_commands.describe(content="報告内容を入力してください")
    async def report(self, interaction: discord.Interaction, content: str):
        # ユーザーに返信します
        response = discord.Embed()
        response.title = "報告"
        response.description = "ご報告ありがとうございます！開発チームに送信しました。"
        response.color = discord.Color.green()
        await interaction.response.send_message(embed=response)

        # 管理者チャンネルに送信
        response = discord.Embed()
        response.title = "報告"
        response.description = content
        response.color = discord.Color.orange()
        response.set_author(name=interaction.user.name, icon_url=interaction.user.display_avatar.url)
        response.set_footer(text=f"ユーザー名: {interaction.user.id}")

        report_guild = self.bot.get_guild(int(NOTIFICATION_GUILD))
        report_channel = report_guild.get_channel(int(NOTIFICATION_CHANNEL))
        if report_channel:
            await report_channel.send(embed=response)
        else:
            print("⚠️ 通報チャンネルが見つかりません")

async def setup(bot: commands.Bot):
    await bot.add_cog(ReportCog(bot))
  • /report content :で呼び出すと、自分のDMでもサーバーの報告チャンネルでも使えます。
  • ephemeral=True にすると、ユーザー側には「自分だけに見えるメッセージ」として送信されます。

🔌 Docker／Docker Composeでの起動例

最後に、Docker Compose を使ってBotとDBを立ち上げる例を示します。
プロジェクトルートに以下の docker-compose.yml を置いてください。

version: "3.9"

services:
  bot:
    build: .
    container_name: switchbot-bot
    env_file:
      - .env
    environment:
      - DATABASE_URL
      - FERNET_KEY
      - DISCORD_TOKEN
      - NOTIFICATION_GUILD
      - NOTIFICATION_CHANNEL
    volumes:
      - .:/app
    depends_on:
      - db

  db:
    image: postgres:16
    container_name: switchbot-db
    restart: always
    env_file:
      - .env
    environment:
      - POSTGRES_USER
      - POSTGRES_PASSWORD
      - POSTGRES_DB
    ports:
      - "5432:5432"
    volumes:
      - db_data:/var/lib/postgresql/data

volumes:
  db_data:

起動コマンド例

.env に必要な環境変数を定義しておく
docker compose up -d

💬 最後に

スマホを使わずに家電を操作したいというシンプルな動機から始まったこのBot開発。
SwitchBot×Discordという構成が思った以上に快適です！

今後は以下のような機能も追加していければと思っています：

  • カスタムボタンの対応
  • 他のSwitchBotデバイス（カーテン、プラグなど）への拡張

特に玄関のロックを操作できるように作りたいな〜（セキュリティが怖いのでパスワードとか導入するか……）
とりあえず、今はトークンとシークレットを更新する仕組みと追加しようと考えてます笑

Botは以下のサーバーから試せますので、ぜひフィードバックください！
👉 https://discord.gg/yRhCUMBswq

Leave a Comment

Your email address will not be published.