""" 爬蟲狀態回報模組 (Collector Status Reporter) 提供統一的狀態回報機制，讓所有爬蟲都能回報執行狀態到雲端監控服務。 使用方式： ----------- # 方式 1: Context Manager（推薦） from status import StatusReporter reporter = StatusReporter("parking-collector") with reporter.run() as run: # 你的爬蟲邏輯 records = collect_data() run.add_records(len(records)) run.add_metric("cities", 6) # 結束時自動回報成功/失敗 # 方式 2: 手動回報 reporter.report( success=True, records=2153, metrics={"cities": 6}, message="收集完成" ) # 方式 3: 簡單心跳（daemon 模式用） reporter.heartbeat() 環境變數設定： -------------- COLLECTOR_MONITOR_API - 雲端監控 API 端點（必填） COLLECTOR_MONITOR_TOKEN - API 認證 Token（必填） SERVER_TAG - 機器識別標籤（選填，預設 🖥️） 版本記錄： ---------- 2026-02-04: v1.0.0 - 初始版本 - StatusReporter 類別 - RunContext 資料類別 - Context Manager 支援 - 失敗安全設計（監控失敗不影響主程式） """ import os import sys import json import socket import logging from datetime import datetime from contextlib import contextmanager from dataclasses import dataclass, field from typing import Optional, Dict, Any, List # 設定 UTF-8 編碼 if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8') # 嘗試導入 requests，如果失敗則使用 urllib try: import requests HAS_REQUESTS = True except ImportError: import urllib.request import urllib.error HAS_REQUESTS = False # Logger 設定 logger = logging.getLogger(__name__) # ============================================================ # 設定 # ============================================================ # 雲端 API 端點（從環境變數讀取） MONITOR_API = os.environ.get( "COLLECTOR_MONITOR_API", "" # 必須設定，否則不會發送 ) # API 認證 Token MONITOR_TOKEN = os.environ.get("COLLECTOR_MONITOR_TOKEN", "") # 請求逾時（秒） REQUEST_TIMEOUT = 10 # 是否啟用本地備份（當雲端回報失敗時） ENABLE_LOCAL_BACKUP = True LOCAL_BACKUP_DIR = os.environ.get( "COLLECTOR_STATUS_BACKUP_DIR", os.path.join(os.path.dirname(__file__), "status_backup") ) # ============================================================ # 資料類別 # ============================================================ @dataclass class RunContext: """ 執行期間的上下文，用於累積資訊。 在 with reporter.run() as run: 區塊中使用。 """ records: int = 0 metrics: Dict[str, Any] = field(default_factory=dict) errors: List[str] = field(default_factory=list) def add_records(self, count: int): """新增收集的記錄數量""" self.records += count def add_metric(self, key: str, value: Any): """新增自訂指標""" self.metrics[key] = value def add_error(self, error: str): """新增錯誤訊息""" self.errors.append(error) def set_records(self, count: int): """設定收集的記錄數量（覆蓋）""" self.records = count # ============================================================ # 主類別 # ============================================================ class StatusReporter: """ 爬蟲狀態回報器。 每個爬蟲建立一個 StatusReporter 實例，用於回報執行狀態。 Parameters ---------- collector_name : str 爬蟲識別名稱，需與雲端監控服務註冊的名稱一致 Examples -------- >>> reporter = StatusReporter("parking-collector") >>> with reporter.run() as run: ... data = fetch_data() ... run.add_records(len(data)) """ def __init__(self, collector_name: str): self.collector_name = collector_name self.host = socket.gethostname() self.server_tag = os.environ.get("SERVER_TAG", "🖥️") self._api_url = MONITOR_API self._token = MONITOR_TOKEN @contextmanager def run(self): """ Context manager，自動追蹤執行狀態。 自動記錄： - 執行開始/結束時間 - 執行成功/失敗 - 例外捕捉 Yields ------ RunContext 用於記錄執行期間的資訊 Examples -------- >>> with reporter.run() as run: ... run.add_records(100) ... run.add_metric("source", "API") """ ctx = RunContext() start_time = datetime.now() success = False error_message = None try: yield ctx # 如果沒有錯誤，標記為成功 success = len(ctx.errors) == 0 except Exception as e: error_message = str(e) ctx.add_error(error_message) # 重新拋出例外，讓呼叫者處理 raise finally: duration = (datetime.now() - start_time).total_seconds() self.report( success=success, records=ctx.records, metrics=ctx.metrics, duration=duration, errors=ctx.errors if ctx.errors else None, message=error_message ) def report( self, success: bool, records: int = 0, metrics: Optional[Dict[str, Any]] = None, duration: Optional[float] = None, errors: Optional[List[str]] = None, message: Optional[str] = None ) -> bool: """ 發送狀態到雲端監控服務。 Parameters ---------- success : bool 執行是否成功 records : int, optional 收集的記錄數量 metrics : dict, optional 自訂指標 duration : float, optional 執行時間（秒） errors : list, optional 錯誤訊息列表 message : str, optional 附加訊息 Returns ------- bool 回報是否成功 """ payload = { "collector": self.collector_name, "host": self.host, "server_tag": self.server_tag, "timestamp": datetime.now().isoformat(), "success": success, "records": records, "metrics": metrics or {}, "duration_seconds": duration, "errors": errors, "message": message, "python_version": sys.version.split()[0], } # 如果沒有設定 API，只做本地備份 if not self._api_url: logger.warning("[StatusReporter] COLLECTOR_MONITOR_API 未設定，僅本地備份") self._local_backup(payload) return False try: if HAS_REQUESTS: resp = requests.post( f"{self._api_url}/api/heartbeat", json=payload, headers={ "Authorization": f"Bearer {self._token}", "Content-Type": "application/json" }, timeout=REQUEST_TIMEOUT ) resp.raise_for_status() else: # 使用 urllib 作為備選 data = json.dumps(payload).encode('utf-8') req = urllib.request.Request( f"{self._api_url}/api/heartbeat", data=data, headers={ "Authorization": f"Bearer {self._token}", "Content-Type": "application/json" }, method='POST' ) with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp: if resp.status >= 400: raise Exception(f"HTTP {resp.status}") logger.info(f"[StatusReporter] 狀態回報成功: {self.collector_name}") return True except Exception as e: # 監控失敗不應影響主程式 logger.error(f"[StatusReporter] 回報失敗: {e}") # 本地備份 if ENABLE_LOCAL_BACKUP: self._local_backup(payload) return False def heartbeat(self) -> bool: """ 發送簡單心跳，表示爬蟲還活著。 適用於長時間運行的 daemon 模式。 Returns ------- bool 回報是否成功 """ return self.report(success=True, message="heartbeat") def _local_backup(self, payload: Dict[str, Any]): """ 當雲端回報失敗時，將狀態存到本地檔案。 這些備份可以之後手動上傳或重試。 """ try: os.makedirs(LOCAL_BACKUP_DIR, exist_ok=True) # 檔名格式：collector_YYYYMMDD_HHMMSS.json timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{self.collector_name}_{timestamp}.json" filepath = os.path.join(LOCAL_BACKUP_DIR, filename) with open(filepath, 'w', encoding='utf-8') as f: json.dump(payload, f, ensure_ascii=False, indent=2) logger.info(f"[StatusReporter] 本地備份已儲存: {filepath}") except Exception as e: logger.error(f"[StatusReporter] 本地備份失敗: {e}") # ============================================================ # 便利函數 # ============================================================ def create_reporter(collector_name: str) -> StatusReporter: """ 建立 StatusReporter 實例的便利函數。 Parameters ---------- collector_name : str 爬蟲識別名稱 Returns ------- StatusReporter """ return StatusReporter(collector_name) def quick_report( collector_name: str, success: bool, records: int = 0, message: Optional[str] = None ) -> bool: """ 快速回報狀態的便利函數。 適用於不需要 Context Manager 的簡單場景。 Parameters ---------- collector_name : str 爬蟲識別名稱 success : bool 執行是否成功 records : int, optional 收集的記錄數量 message : str, optional 附加訊息 Returns ------- bool 回報是否成功 Examples -------- >>> quick_report("my-collector", True, records=100) """ reporter = StatusReporter(collector_name) return reporter.report(success=success, records=records, message=message) # ============================================================ # CLI 入口 # ============================================================ def main(): """命令列入口，用於測試或手動回報。""" import argparse parser = argparse.ArgumentParser( description="爬蟲狀態回報工具", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 範例： # 回報成功 python status.py parking-collector --success --records 2153 # 回報失敗 python status.py parking-collector --failed --message "API 逾時" # 發送心跳 python status.py parking-collector --heartbeat # 檢查設定 python status.py --check """ ) parser.add_argument("collector", nargs="?", help="爬蟲名稱") parser.add_argument("--success", action="store_true", help="回報成功") parser.add_argument("--failed", action="store_true", help="回報失敗") parser.add_argument("--heartbeat", action="store_true", help="發送心跳") parser.add_argument("--records", type=int, default=0, help="記錄數量") parser.add_argument("--message", type=str, help="附加訊息") parser.add_argument("--check", action="store_true", help="檢查環境設定") args = parser.parse_args() # 檢查設定 if args.check: print("=== 狀態回報模組設定檢查 ===") print(f"COLLECTOR_MONITOR_API: {MONITOR_API or '(未設定)'}") print(f"COLLECTOR_MONITOR_TOKEN: {'*' * 8 if MONITOR_TOKEN else '(未設定)'}") print(f"SERVER_TAG: {os.environ.get('SERVER_TAG', '🖥️ (預設)')}") print(f"本機名稱: {socket.gethostname()}") print(f"本地備份目錄: {LOCAL_BACKUP_DIR}") print(f"requests 套件: {'已安裝' if HAS_REQUESTS else '未安裝（使用 urllib）'}") if not MONITOR_API: print("\n⚠️ 警告：COLLECTOR_MONITOR_API 未設定，回報功能將無法使用") return # 需要爬蟲名稱 if not args.collector: parser.print_help() return reporter = StatusReporter(args.collector) if args.heartbeat: success = reporter.heartbeat() print(f"心跳回報: {'成功' if success else '失敗'}") elif args.failed: success = reporter.report( success=False, records=args.records, message=args.message or "手動回報失敗" ) print(f"失敗回報: {'已送出' if success else '送出失敗'}") else: # 預設為成功 success = reporter.report( success=True, records=args.records, message=args.message ) print(f"成功回報: {'已送出' if success else '送出失敗'}") if __name__ == "__main__": main()