""" バニラ自動設定スクリプト analysis_suggestions から最新の suggest_times を取得し、 バニラ（qzin.jp）の「店舗上位表示」に自動入力する。 Usage: python tools/set_vanilla.py """ import argparse import json import logging import os import sys from datetime import datetime from pathlib import Path from playwright.sync_api import Page, sync_playwright sys.path.append(str(Path(__file__).resolve().parents[1])) from config.database import execute_query, execute_update # load_dotenv() は自動実行 # --------------------------------------------------------------------------- # 定数 # --------------------------------------------------------------------------- VANILLA_URL = "https://qzin.jp/entry/" MEDIA_NAME = "vanilla" # --------------------------------------------------------------------------- # ロガー設定 # --------------------------------------------------------------------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # DB ヘルパー # --------------------------------------------------------------------------- def get_media_settings(client_id: int) -> dict | None: """カゴヤDB の media_settings から vanilla の設定を取得する。""" rows = execute_query( """ SELECT login_id, login_pass, is_active FROM media_settings WHERE client_id = %s AND media_name = %s LIMIT 1 """, (client_id, MEDIA_NAME), local=False, ) if not rows: return None row = rows[0] return { "login_id": row["login_id"], "login_pass": row["login_pass"], "is_active": row["is_active"], } def _to_hhmm(value) -> str: """任意の時刻値を HH:MM 形式の文字列に変換する。""" if isinstance(value, str): if ":" in value: return value.strip() value = float(value) if isinstance(value, (int, float)): if value >= 100: h = int(value) // 100 m = int(value) % 100 else: h = int(value) m = round((value - h) * 60) return f"{h:02d}:{m:02d}" return str(value) def get_latest_suggest_times(client_id: int) -> list[str]: """ analysis_suggestions から最新のバニラ suggest_times を取得する。 Returns: ["09:00", "12:00", ...] などの HH:MM リスト """ rows = execute_query( """ SELECT suggest_time FROM analysis_suggestions WHERE client_id = %s AND media_name = %s AND DATE(suggested_at) = ( SELECT DATE(suggested_at) FROM analysis_suggestions WHERE client_id = %s AND media_name = %s ORDER BY suggested_at DESC LIMIT 1 ) ORDER BY suggest_time """, (client_id, MEDIA_NAME, client_id, MEDIA_NAME), local=True, ) if not rows: return [] converted = [row["suggest_time"] for row in rows if row["suggest_time"]] logger.info("suggest_times 取得結果: %s", converted) return converted def save_update_history(client_id: int, status: int) -> None: """update_histories に実行結果を記録する（local=True）。""" execute_update( """ INSERT INTO update_histories (client_id, media_name, executed_at, status) VALUES (%s, %s, %s, %s) """, (client_id, MEDIA_NAME, datetime.now(), status), local=True, ) def save_error_log(client_id: int, message: str) -> None: """error_logs にエラーを記録する（local=False）。""" try: execute_update( """ INSERT INTO error_logs (client_id, error_type, error_message, occurred_at) VALUES (%s, %s, %s, %s) """, (client_id, "vanilla_error", message, datetime.now()), local=False, ) except Exception as e: logger.error("エラーログのDB保存も失敗: %s", e) # --------------------------------------------------------------------------- # ブラウザ起動 # --------------------------------------------------------------------------- def build_browser(p): """HEADLESS 環境変数に従いブラウザを起動し、(browser, page) を返す。""" headless = os.getenv("HEADLESS", "true").lower() != "false" if headless: browser = p.chromium.launch( headless=True, args=["--no-sandbox", "--disable-blink-features=AutomationControlled"], ) else: browser = p.chromium.launch( headless=False, executable_path=r"C:\Program Files\Google\Chrome\Application\chrome.exe", slow_mo=500, args=["--no-sandbox", "--disable-blink-features=AutomationControlled"], ) context = browser.new_context( user_agent=( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ), locale="ja-JP", timezone_id="Asia/Tokyo", ) page = context.new_page() return browser, page # --------------------------------------------------------------------------- # バニラ自動設定本体 # --------------------------------------------------------------------------- def run_vanilla(page: Page, settings: dict, suggest_times: list[str]) -> None: login_id = settings["login_id"] login_pass = settings["login_pass"] # ① アクセス logger.info("バニラにアクセス: %s", VANILLA_URL) page.goto(VANILLA_URL, timeout=30_000, wait_until="domcontentloaded") page.wait_for_timeout(2000) # ② ログイン logger.info("ログイン処理") page.fill("input[name='username']", str(login_id)) page.fill("input[name='password']", str(login_pass)) page.click("input[type='submit']") page.wait_for_timeout(3000) # ③ 「店舗上位表示を設定する」ボタンをクリック logger.info("「店舗上位表示を設定する」ボタンをクリック") allup_selectors = [ "#allupform img", "#allupform input[type='image']", "form#allupform button", "form#allupform input[type='submit']", ] allup_clicked = False for sel in allup_selectors: el = page.locator(sel).first if el.count() > 0 and el.is_visible(): logger.info("上位表示ボタンをクリック: %s", sel) el.click(force=True) allup_clicked = True break if not allup_clicked: raise RuntimeError("「店舗上位表示を設定する」ボタンが見つかりません") page.wait_for_timeout(3000) # ④ 全削除ボタンをクリック logger.info("全削除ボタンをクリック") def _handle_dialog(dialog): logger.info("ダイアログ承認: %s", dialog.message) dialog.accept() page.on("dialog", _handle_dialog) page.locator("input[name='js-all_delete']").click(force=True) page.wait_for_timeout(3000) page.remove_listener("dialog", _handle_dialog) # ⑤ suggest_times の時刻を1つずつ追加 logger.info("時刻を追加します: %s", suggest_times) # max_count を取得 max_count_el = page.locator("input[name='max_count']") max_count = None if max_count_el.count() > 0: try: max_count = int(max_count_el.get_attribute("value") or "0") logger.info("max_count: %d", max_count) except (ValueError, TypeError): logger.warning("max_count の取得に失敗しました") for idx, t in enumerate(suggest_times): if max_count is not None and idx >= max_count: logger.warning("max_count=%d に達したため時刻 %s をスキップします", max_count, t) continue hh, mm = t.split(":") logger.info(" [%d] %s を追加 (hour=%s, min=%s)", idx + 1, t, hh, mm) page.locator("select[name='select_hour']").select_option(hh) page.locator("select[name='select_min']").select_option(mm) page.locator("input[name='add_select']").click(force=True) page.wait_for_timeout(1000) # ⑥ 保存ボタンをクリック logger.info("保存ボタンをクリック") save_selectors = [ "input[value='保存']", "button:has-text('保存')", ] save_clicked = False for sel in save_selectors: el = page.locator(sel).first if el.count() > 0 and el.is_visible(): logger.info("保存ボタンをクリック: %s", sel) el.click(force=True) save_clicked = True break if not save_clicked: logger.warning("保存ボタンが見つかりませんでした") page.wait_for_timeout(3000) logger.info("バニラ設定完了") # --------------------------------------------------------------------------- # エントリポイント # --------------------------------------------------------------------------- def main() -> None: parser = argparse.ArgumentParser(description="バニラ自動設定") parser.add_argument("client_id", type=int, help="クライアントID") args = parser.parse_args() client_id = args.client_id logger.info("バニラ自動設定開始 | client_id=%d", client_id) # media_settings 取得 try: settings = get_media_settings(client_id) except Exception as e: logger.error("media_settings 取得エラー: %s", e) save_error_log(client_id, f"media_settings取得失敗: {e}") sys.exit(1) if settings is None: logger.error("client_id=%d の media_settings(vanilla) が見つかりません", client_id) save_error_log(client_id, f"media_settingsが見つかりません (client_id={client_id})") sys.exit(1) if settings["is_active"] == 0: logger.info("is_active=0 のためスキップします (client_id=%d)", client_id) sys.exit(0) # suggest_times 取得 try: suggest_times = get_latest_suggest_times(client_id) except Exception as e: logger.error("suggest_times 取得エラー: %s", e) save_error_log(client_id, f"suggest_times取得失敗: {e}") sys.exit(1) if not suggest_times: logger.error("suggest_times が空です (client_id=%d)", client_id) save_error_log(client_id, f"suggest_timesが空のため設定をスキップ (client_id={client_id})") sys.exit(1) # Playwright で設定実行 try: with sync_playwright() as p: browser, page = build_browser(p) try: run_vanilla(page, settings, suggest_times) save_update_history(client_id, 1) logger.info("完了 | client_id=%d status=1（成功）", client_id) finally: browser.close() except Exception as e: error_msg = f"バニラ設定失敗 (client_id={client_id}): {e}" logger.error(error_msg) save_error_log(client_id, error_msg) try: save_update_history(client_id, 0) except Exception as db_err: logger.error("update_histories 記録失敗: %s", db_err) sys.exit(1) if __name__ == "__main__": main()