#!/usr/bin/env python3 """Fetch a month of seminars from seminar.zenko-sai.or.jp. The site is an authenticated Nuxt application. This script deliberately uses the normal browser flow instead of hard-coding the site's private API login. """ from __future__ import annotations import argparse import getpass import json import os import re import sys from datetime import datetime, timezone from pathlib import Path from typing import Any from playwright.sync_api import Error as PlaywrightError from playwright.sync_api import Page, TimeoutError as PlaywrightTimeoutError, sync_playwright TOP_URL = "https://seminar.zenko-sai.or.jp/top" LOGIN_URL = "https://www.zenko-sai.or.jp/login/" def parse_month(value: str) -> tuple[int, int]: match = re.fullmatch(r"(\d{4})-(0[1-9]|1[0-2])", value) if not match: raise argparse.ArgumentTypeError("month must be YYYY-MM") return int(match.group(1)), int(match.group(2)) def clean(value: Any) -> Any: if isinstance(value, str): return re.sub(r"\s+", " ", value).strip() if isinstance(value, list): return [clean(item) for item in value] if isinstance(value, dict): return {key: clean(item) for key, item in value.items()} return value def login(page: Page, username: str | None, password: str | None) -> None: page.goto(LOGIN_URL, wait_until="domcontentloaded") user = page.locator("#login_user_name") passwd = page.locator("#login_password") if not user.count() or not passwd.count(): raise RuntimeError("ログイン画面の入力欄が見つかりません。サイト側の変更を確認してください。") username = username or input("全厚済ログインID: ") password = password or getpass.getpass("全厚済パスワード: ") user.fill(username) passwd.fill(password) page.locator("form").filter(has=user).locator("button, input[type=submit]").first.click() page.wait_for_load_state("domcontentloaded") # The HP login normally completes SSO when the seminar site is opened. page.goto(TOP_URL, wait_until="domcontentloaded") def choose_month(page: Page, year: int, month: int) -> None: """Fill date controls and submit the site's seminar search form.""" page.wait_for_load_state("networkidle") target = f"{year:04d}/{month:02d}/01" inputs = page.locator("input") date_inputs = [] for i in range(inputs.count()): item = inputs.nth(i) typ = (item.get_attribute("type") or "").lower() name = (item.get_attribute("name") or "").lower() if typ in {"date", "month"} or any(word in name for word in ("from", "to", "date", "year", "month")): date_inputs.append(item) if len(date_inputs) >= 2: date_inputs[0].fill(target) date_inputs[1].fill(f"{year:04d}/{month:02d}/{min(28, 1):02d}") else: # The current UI uses a calendar picker in some accounts. Selecting # the first day by text lets the app itself update its search state. page.get_by_text(f"{year}年{month}月", exact=False).first.click() buttons = page.get_by_role("button") candidates = ["検索", "Search"] for label in candidates: if buttons.filter(has_text=label).count(): buttons.filter(has_text=label).first.click() page.wait_for_load_state("networkidle") return raise RuntimeError("セミナー検索ボタンが見つかりません。サイト側の画面変更を確認してください。") def api_payloads(page: Page, year: int, month: int) -> list[Any]: """Read the rendered page and capture the site's seminar API payloads.""" payloads: list[Any] = [] page.on("response", lambda response: _capture(response, payloads)) choose_month(page, year, month) page.wait_for_timeout(1500) return payloads def _capture(response: Any, payloads: list[Any]) -> None: if "/api/" not in response.url or "seminar" not in response.url: return try: data = response.json() except Exception: return if data not in payloads: payloads.append(data) def extract_records(payloads: list[Any], page: Page) -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] seen: set[str] = set() def visit(value: Any) -> None: if isinstance(value, dict): ident = value.get("seminar_id") or value.get("seminarId") or value.get("id") if ident is not None and any(key in value for key in ("start_at", "start_datetime", "datetime", "seminar_date", "date")): sid = str(ident) if sid not in seen: seen.add(sid) item = clean(value) item["seminar_id"] = sid item.setdefault("detail_url", f"https://seminar.zenko-sai.or.jp/seminar/{sid}") records.append(item) for child in value.values(): visit(child) elif isinstance(value, list): for child in value: visit(child) for payload in payloads: visit(payload) # API field names have changed in past releases. Keep a useful fallback # when the response is only exposed through rendered seminar links. if not records: for link in page.locator('a[href*="/seminar/"]').all(): match = re.search(r"/seminar/(\d+)", link.get_attribute("href") or "") if match and match.group(1) not in seen: sid = match.group(1) seen.add(sid) records.append({"seminar_id": sid, "detail_url": link.get_attribute("href")}) return records def original_table(page: Page) -> dict[str, Any]: """Preserve a lightweight copy of the visible calendar table.""" days = [] for index, row in enumerate(page.locator("table tr").all(), start=1): cells = [clean(cell.inner_text()) for cell in row.locator("th,td").all()] if cells: days.append({"day_index": index, "original_cells": cells, "events": []}) return {"days": days} def main() -> int: parser = argparse.ArgumentParser(description="全厚済セミナーを月単位で取得します") parser.add_argument("month", type=parse_month, help="取得対象月 (YYYY-MM)") parser.add_argument("--output-dir", default="docs/json") parser.add_argument("--storage-state", default=".seminar-storage-state.json") parser.add_argument("--headed", action="store_true", help="ブラウザ画面を表示する") args = parser.parse_args() year, month = args.month output_dir = Path(args.output_dir) output_dir.mkdir(parents=True, exist_ok=True) state = Path(args.storage_state) with sync_playwright() as playwright: browser = playwright.chromium.launch(headless=not args.headed) context_args: dict[str, Any] = {"locale": "ja-JP"} if state.exists(): context_args["storage_state"] = str(state) context = browser.new_context(**context_args) page = context.new_page() page.goto(TOP_URL, wait_until="domcontentloaded") if "zenko-sai.or.jp/login" in page.url or not page.locator("body").inner_text().strip(): login(page, os.getenv("ZENKO_SAI_USERNAME"), os.getenv("ZENKO_SAI_PASSWORD")) if "login" in page.url: raise RuntimeError("ログインできませんでした。認証情報またはSSO状態を確認してください。") payloads = api_payloads(page, year, month) records = extract_records(payloads, page) if not records: raise RuntimeError("セミナーが取得できませんでした。検索結果またはAPIレスポンスを確認してください。") context.storage_state(path=str(state)) captured_at = datetime.now(timezone.utc).isoformat(timespec="milliseconds") result = {"captured_at": captured_at, "records": records} (output_dir / f"{year:04d}-{month:02d}.json").write_text( json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) (output_dir / f"{year:04d}-{month:02d}-original-table.json").write_text( json.dumps(original_table(page), ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) print(f"取得完了: {year:04d}-{month:02d} / {len(records)}件") browser.close() return 0 if __name__ == "__main__": try: raise SystemExit(main()) except PlaywrightTimeoutError as exc: print(f"タイムアウト: {exc}", file=sys.stderr) raise SystemExit(1) except PlaywrightError as exc: if "Executable doesn't exist" in str(exc): print("ブラウザが未インストールです。`python3 -m playwright install chromium` を実行してください。", file=sys.stderr) else: print(f"Playwrightエラー: {exc}", file=sys.stderr) raise SystemExit(1) except RuntimeError as exc: print(f"エラー: {exc}", file=sys.stderr) raise SystemExit(1)