""" RankRadar スクレイピング停止監視スクリプト scraping_results の最終取得時刻を確認し、STALE_MINUTES 以上更新がなければ Gmail SMTP でアラートメールを送信する。 - 連続通知抑制: 停止継続中は1時間ごとに再通知（alert_state.json で管理） - 復旧通知: alerting=True だった場合のみ復旧メールを1通送信 - DB接続失敗も停止の兆候として通知対象 Usage: python check_scraping_alert.py # 通常チェック python check_scraping_alert.py --test # テストメール送信（閾値無視） 設定ファイル: python/monitoring/alert_config.example.py をコピーして python/monitoring/alert_config.py を作成し実値を記入すること。 """ import argparse import importlib.util import json import smtplib import sys from datetime import datetime, timedelta from email.mime.text import MIMEText from pathlib import Path import pymysql # --------------------------------------------------------------------------- # 定数 # --------------------------------------------------------------------------- SCRIPT_DIR = Path(__file__).parent STATE_FILE = SCRIPT_DIR / "alert_state.json" CONFIG_FILE = SCRIPT_DIR / "alert_config.py" RESEND_INTERVAL_HOURS = 1 # 停止継続中の再通知間隔 # --------------------------------------------------------------------------- # 設定ロード # --------------------------------------------------------------------------- def load_config(): if not CONFIG_FILE.exists(): print( f"[ERROR] 設定ファイルが見つかりません: {CONFIG_FILE}\n" f" alert_config.example.py をコピーして alert_config.py を作成し、実値を記入してください。\n" f" cp alert_config.example.py alert_config.py" ) sys.exit(1) spec = importlib.util.spec_from_file_location("alert_config", CONFIG_FILE) cfg = importlib.util.module_from_spec(spec) spec.loader.exec_module(cfg) return cfg # --------------------------------------------------------------------------- # 状態管理 (alert_state.json) # --------------------------------------------------------------------------- def load_state() -> dict: if STATE_FILE.exists(): try: return json.loads(STATE_FILE.read_text(encoding="utf-8")) except Exception: pass return {} def save_state(state: dict) -> None: STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8") def state_key(client_id: int, media: str) -> str: return f"{client_id}:{media}" # --------------------------------------------------------------------------- # DB # --------------------------------------------------------------------------- def get_latest_fetched_at(cfg, client_id: int, media: str) -> datetime | None: """scraping_results から最新の fetched_at を取得する。""" conn = pymysql.connect( host=cfg.DB_HOST, port=getattr(cfg, "DB_PORT", 3306), user=cfg.DB_USER, password=cfg.DB_PASS, database=cfg.DB_NAME, charset="utf8mb4", connect_timeout=10, ) try: with conn.cursor() as cur: cur.execute( "SELECT MAX(fetched_at) FROM scraping_results " "WHERE client_id = %s AND media_name = %s", (client_id, media), ) row = cur.fetchone() return row[0] if row and row[0] else None finally: conn.close() # --------------------------------------------------------------------------- # メール送信 # --------------------------------------------------------------------------- def send_mail(cfg, subject: str, body: str) -> None: msg = MIMEText(body, "plain", "utf-8") msg["Subject"] = subject msg["From"] = cfg.MAIL_FROM msg["To"] = cfg.MAIL_TO with smtplib.SMTP(cfg.SMTP_HOST, cfg.SMTP_PORT, timeout=30) as smtp: smtp.ehlo() smtp.starttls() smtp.login(cfg.SMTP_USER, cfg.SMTP_APP_PASSWORD) smtp.sendmail(cfg.MAIL_FROM, cfg.MAIL_TO, msg.as_bytes()) def try_send_mail(cfg, subject: str, body: str) -> bool: """送信を試みる。失敗してもクラッシュしない。成功したら True を返す。""" try: send_mail(cfg, subject, body) print(f"[MAIL] 送信完了: {subject}") return True except Exception as e: print(f"[MAIL ERROR] 送信失敗: {e}") return False # --------------------------------------------------------------------------- # 本体 # --------------------------------------------------------------------------- def check_target(cfg, client_id: int, media: str, state: dict, now: datetime) -> None: key = state_key(client_id, media) entry = state.setdefault(key, {"alerting": False, "last_notified": None}) # ── DB から最新 fetched_at を取得 ──────────────────────────────────────── latest: datetime | None = None db_error: str | None = None try: latest = get_latest_fetched_at(cfg, client_id, media) except Exception as e: db_error = str(e) print(f"[DB ERROR] client_id={client_id} media={media}: {e}") # ── 停止判定 ───────────────────────────────────────────────────────────── if db_error: # DB自体に繋がらない → 停止と同等に扱う is_stale = True stale_reason = f"DB接続失敗: {db_error}" latest_str = "(不明)" elapsed_min = -1 elif latest is None: is_stale = True stale_reason = "scraping_results にデータが存在しません" latest_str = "(なし)" elapsed_min = -1 else: elapsed = now - latest elapsed_min = int(elapsed.total_seconds() / 60) is_stale = elapsed_min >= cfg.STALE_MINUTES stale_reason = f"最終取得から {elapsed_min} 分経過（閾値: {cfg.STALE_MINUTES} 分）" latest_str = latest.strftime("%Y-%m-%d %H:%M:%S") print( f"[CHECK] client_id={client_id} media={media} | " f"latest={latest_str} | elapsed={elapsed_min}min | stale={is_stale}" ) # ── 停止中の通知ロジック ────────────────────────────────────────────────── if is_stale: last_notified = entry.get("last_notified") already_alerting = entry.get("alerting", False) # 前回通知から RESEND_INTERVAL_HOURS 以上経過しているか should_notify = True if last_notified: last_dt = datetime.fromisoformat(last_notified) if (now - last_dt) < timedelta(hours=RESEND_INTERVAL_HOURS): should_notify = False if should_notify: subject = f"[RankRadar警告] client_id={client_id} {media} のスクレイピングが停止" body = ( f"■ スクレイピング停止アラート\n\n" f" 対象 : client_id={client_id} / {media}\n" f" 最新取得 : {latest_str}\n" f" 停止時間 : {elapsed_min} 分以上\n" f" 判定理由 : {stale_reason}\n" f" サーバー時刻: {now.strftime('%Y-%m-%d %H:%M:%S')}\n\n" f"monitor.py の再起動を確認してください。\n" ) if try_send_mail(cfg, subject, body): entry["alerting"] = True entry["last_notified"] = now.isoformat() else: remaining = RESEND_INTERVAL_HOURS * 60 - int((now - datetime.fromisoformat(last_notified)).total_seconds() / 60) print(f" → 停止継続中・次回通知まで約 {remaining} 分") # ── 復旧通知ロジック ────────────────────────────────────────────────────── else: if entry.get("alerting", False): subject = f"[RankRadar復旧] client_id={client_id} {media} 復旧しました" body = ( f"■ スクレイピング復旧通知\n\n" f" 対象 : client_id={client_id} / {media}\n" f" 最新取得 : {latest_str}\n" f" 経過時間 : {elapsed_min} 分\n" f" サーバー時刻: {now.strftime('%Y-%m-%d %H:%M:%S')}\n\n" f"正常に取得が再開されています。\n" ) try_send_mail(cfg, subject, body) entry["alerting"] = False entry["last_notified"] = None else: print(f" → 正常稼働中") def run_test(cfg) -> None: """--test: 閾値に関わらずテストメールを1通送る。""" now = datetime.now() subject = "[RankRadar] テストメール" body = ( f"■ RankRadar メール設定テスト\n\n" f" このメールが届いていれば SMTP 設定は正常です。\n" f" 送信時刻: {now.strftime('%Y-%m-%d %H:%M:%S')}\n" ) print("[TEST] テストメールを送信します...") ok = try_send_mail(cfg, subject, body) sys.exit(0 if ok else 1) def main() -> None: parser = argparse.ArgumentParser(description="RankRadar スクレイピング停止監視") parser.add_argument("--test", action="store_true", help="テストメールを送信して終了") args = parser.parse_args() cfg = load_config() now = datetime.now() if args.test: run_test(cfg) state = load_state() errors: list[str] = [] for client_id, media in cfg.TARGETS: try: check_target(cfg, client_id, media, state, now) except Exception as e: msg = f"client_id={client_id} media={media}: {e}" print(f"[ERROR] 監視エラー: {msg}") errors.append(msg) save_state(state) # 監視スクリプト自体のエラーをメール通知（SMTPが生きていれば） if errors: subject = "[RankRadar警告] 監視スクリプト自体でエラーが発生" body = ( f"■ 監視スクリプトエラー\n\n" f" 以下のターゲットで予期しないエラーが発生しました:\n\n" + "\n".join(f" - {e}" for e in errors) + f"\n\n サーバー時刻: {now.strftime('%Y-%m-%d %H:%M:%S')}\n" ) try_send_mail(cfg, subject, body) if __name__ == "__main__": main()