🍺

Alexaで追加した買い物リストをもとにAmazon系以外のECサイトでも音声ショッピングしてみる

2023/09/18に公開

Google Cloud

はじめに

どうも、趣味で機械学習関連を勉強中の生ビールと申します。備忘録がてら初めてのZenn記事投稿します。
モチベーション湧かない家事はやりたくないめんどくさがり屋です。家事する暇あったらその時間ビールを飲んで過ごしていたいです。

動機

結婚したことをきっかけに、互いの時間を増やすべく家事の時短化を進めていました。初めの一歩として、音声操作のAlexaやSwitchbotを導入。これで少しは楽になったのですが、子供が生まれたのを機にもっと家事の時間を減らすべくトライしてみました。
例えば、「たこ焼きソースないな」て思った”その場で”買い物リストに追加して、それをネット内で自動で買い物カゴに入れてくれたら、買い物リストを目で見てスマホでポチポチ買い物する手間が減るかなと思いまして。「あれ買い忘れとるやん！」という夫婦喧嘩のタネが減るはず。

導入背景

Amazon内ならAlexaから音声で買い物できるけど、他のECサイトでもできたらいいですよね。ネットスーパーとか。
都会でAmazonネットスーパーに対応している地域なら音声で日用品や食料品買い物できるんでしょうけど(多分。知らんけど)うちは田舎なのでAmazonネットスーパーがないのでございます。そもそも日用品ならAmazon定期購入すればいい、って話もありますが段ボール片付けるのさえもめんどくさいんです。そして家も広くないのでまとめ買いもしたくないのでございます。
本題のECサイト音声ショッピング化にあたり、IFTTTでGoogle Tasks(ToDo)をトリガーにクラウドBOTを動かすことも検討したのですが、残念ながら私が利用したいECサイト(ネットスーパー)がクラウドBOTをアクセス拒否してた模様。しかたなくローカル実装しました。

検証環境

OS: macOS Ventura 13.5.2
Python: 3.11.5
Selenium: 4.12.0
chromdriver 117.0.5938.62

実装概要

1. Alexa設定

「アレクサ、買い物リストに追加して」でアレクサ標準の買い物リストに追加してもらいます。初期設定してれば特にやることないです。

2. Google Tasksとの連携

※2023/11/20　IFTTT仕様変更に伴う追記
~~IFTTTを使って、~~ Alexaの買い物リストはTodoistと連携し、IFTTTでTodoistからGoogle Tasks(ToDo)と連携。これで妻と買い物リストを共有。Alexa側のリストを共有でもいいのですが、予定表の共有含めGoogleだけで完結するほうが楽なのでこうしてます。
※元々IFTTTでAlexa買物リストとGoogle Tasksを連携していましたが2023年11月からAlexa関連のトリガーが廃止され、困っていたところ下記情報発見し活用させていただきました。助かります。

3. Google Tasks API取得

Google Tasks APIを呼び出すため、GCPプロジェクトを作成しOAuth2.0の認証情報JSONファイルをダウンロード。

4. Seleniumで自動操作

ダウンロードしたOAuth認証ファイルを元にGoogle Tasksの買い物リストを取得→seleniumでネットスーパーにログイン＆取得した買い物リストの文字列と部分一致する商品をカートに追加。Selenium自体のコードはChrome拡張機能のSelenium IDEで録画し、pythonコードをエクスポート。細かい処理はChatGPT先輩にコードを出力してもらい.pyファイル作成。

5. Automator✖️カレンダーで定期実行

Mac標準アプリAutomatorでシェルスクリプト実行で上記pythonファイルを実行するアプリ作成し、カレンダーで毎週決まった時間にpythonスクリプトを実行してもらいます。
pythonのscheduleライブラリ使ってもいいと思いますが、ずっとスクリプト立ち上げっぱなしはリソースもったいないかなと思ったのでこのやり方にしました。

実装詳細

上記1,2,3,5の詳細は割愛して、4のpythonコードを載せます。

import pickle
import os.path
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from google.auth.transport.requests import Request
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options
from dotenv import load_dotenv

# 環境変数のロード
load_dotenv() #同ディレクトリに.envファイル作成の前提

# Chromeのオプション設定
chrome_options = Options()
chrome_options.add_argument("--headless")
chrome_options.add_argument('window-size=1400,600')
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
chrome_options.add_experimental_option('useAutomationExtension', False)
chrome_options.add_experimental_option('prefs', {
    'credentials_enable_service': False,
    'profile': {'password_manager_enabled': False}
})

# 環境変数から認証情報を取得
CREDENTIALS_PATH = os.environ['CREDENTIALS_PATH']#Google Tasks API取得のための認証JSONファイルのパスを指定。
TOKEN＿PATH = os.environ['TOKEN_PATH']#認証情報をキャッシュするファイルパス
USER_ID = os.environ['USER_ID']#ECサイトのID
USER_PASS = os.environ['USER_PASS']#ECサイトのパスワード

# Google Tasksの認証済みサービスを取得する関数
def get_service():
    SCOPES = ['https://www.googleapis.com/auth/tasks.readonly']
    creds = None
    #既に保存された認証情報があるかどうかを確認
    if os.path.exists(TOKEN_PATH):
        with open(TOKEN_PATH, 'rb') as token:
            creds = pickle.load(token)

    # 有効な認証情報がない場合は新しい認証を行う
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = google_auth_oauthlib.flow.InstalledAppFlow.from_client_secrets_file(CREDENTIALS_PATH, SCOPES)
            creds = flow.run_local_server(port=0)
            # 新しい認証情報を保存
            with open(TOKEN_PATH, 'wb') as token:
                pickle.dump(creds, token)
    service = build('tasks', 'v1', credentials=creds)
    return service

# Google Tasksからすべてのタスクのタイトルを取得する関数
def get_all_task_titles(service):
    tasklists = service.tasklists().list().execute()
    tasklist_id = tasklists['items'][6]['id']　#買い物リストに使うGoogle TasksのIDを指定
    results = service.tasks().list(tasklist=tasklist_id, maxResults=20).execute()
    tasks = results.get('items', [])
    return [task['title'] for task in tasks] if tasks else []

# メインの処理
def main():
    # サービスとタスクのタイトルを取得
    service = get_service()
    task_titles = get_all_task_titles(service)
    
    # ブラウザの操作開始
    driver = webdriver.Chrome(options=chrome_options)
    ###ここからは利用したいECサイトによってアレンジください
    driver.get("利用したい/ECサイトの/アドレス")
    WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.CSS_SELECTOR, ".c-login-body-ecsite .c-input__input"))).click()
    driver.find_element(By.CSS_SELECTOR, ".c-login-body-ecsite .c-input__input").send_keys(USER_ID)
    driver.find_element(By.CSS_SELECTOR, ".c-card:nth-child(1) .c-pwd-input__input:nth-child(1)").send_keys(USER_PASS)
    driver.find_element(By.CSS_SELECTOR, ".c-login-body-ecsite .c-button").click()
    WebDriverWait(driver, 30).until(EC.element_to_be_clickable((By.LINK_TEXT, "お気に入り"))).click()
    WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.LINK_TEXT, "よく買うリスト"))).click()

    # 買い物リストに基づいて商品をカートに追加
    for title in task_titles:
        try:
            product_element = driver.find_element(By.XPATH, f"//a[contains(@title, '{title}')]")
            add_to_cart_button = product_element.find_element(By.XPATH, ".//following::button[@title='カゴ追加']")
            add_to_cart_button.click()
            WebDriverWait(driver, 10).until(EC.staleness_of(add_to_cart_button))
        except Exception as e:
            print(f"Error with {title}: {e}")
            continue

    # ログアウトしてブラウザを閉じる
    driver.find_element(By.LINK_TEXT, "ログアウト").click()
    driver.quit()

if __name__ == '__main__':
    main()

おわりに

必要だと思ったその場で買い物リストに追加して、ある程度ネットスーパーで自動で買い物してくれるようにしてみました。週に1回ネットスーパーで買い物をするので、週に1回動くように運用してます。
とりあえず文字列部分一致するものしか買い物カゴに入れてくれないので、今後はセマンティック検索で一致する仕様も検討してみようと思います。
前からアイデアとしては考えていたものの、実装するまで至りませんでした。が、ChatGPT先輩の登場で重い腰を上げてみました。本当にGPT先輩ありがたいです。乾杯🍺