from __future__ import annotations import os from dataclasses import dataclass, field from datetime import datetime from importlib import import_module from typing import Any, Callable, Dict, List, Optional from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env def _env_bool(name: str, default: bool) -> bool: raw = os.getenv(name) if raw is None: return default return raw.strip().lower() in {"1", "true", "yes", "on"} def _env_int(name: str, default: int) -> int: raw = os.getenv(name) if raw is None: return default try: return int(raw) except Exception: return default def _parse_csv_count(raw: Optional[str]) -> int: if not raw: return 0 return len([part for part in str(raw).split(",") if str(part).strip()]) @dataclass class LoopStatus: key: str label: str configured_enabled: bool started: bool reason: str details: Dict[str, Any] = field(default_factory=dict) @dataclass class RuntimeStatus: started_at: str loops: List[LoopStatus] command_access_mode: str protected_commands: List[str] required_group_chat_id: str def loop_map(self) -> Dict[str, LoopStatus]: return {loop.key: loop for loop in self.loops} class StartupCoordinator: """Centralized startup orchestration + diagnostics snapshot.""" def __init__( self, bot: Any, config: Dict[str, Any], command_access_mode: str, protected_commands: List[str], required_group_chat_id: str, ): self.bot = bot self.config = config self.command_access_mode = command_access_mode self.protected_commands = protected_commands self.required_group_chat_id = required_group_chat_id self._runtime_status = RuntimeStatus( started_at=datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC"), loops=[], command_access_mode=command_access_mode, protected_commands=protected_commands, required_group_chat_id=required_group_chat_id, ) def get_runtime_status(self) -> RuntimeStatus: return self._runtime_status def start_all(self) -> RuntimeStatus: loops = [ self._start_airport_high_freq_loop(), self._start_polygon_wallet_loop(), self._start_polymarket_wallet_activity_loop(), self._start_weekly_reward_loop(), self._start_payment_event_loop(), self._start_payment_confirm_loop(), ] self._runtime_status = RuntimeStatus( started_at=datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC"), loops=loops, command_access_mode=self.command_access_mode, protected_commands=self.protected_commands, required_group_chat_id=self.required_group_chat_id, ) return self._runtime_status def _start_with_validation( self, key: str, label: str, configured_enabled: bool, details: Dict[str, Any], validation_error: Optional[str], starter: Callable[[], Any], ) -> LoopStatus: if not configured_enabled: return LoopStatus( key=key, label=label, configured_enabled=False, started=False, reason="disabled_by_env", details=details, ) if validation_error: return LoopStatus( key=key, label=label, configured_enabled=True, started=False, reason=validation_error, details=details, ) try: thread = starter() except Exception as exc: return LoopStatus( key=key, label=label, configured_enabled=True, started=False, reason=f"starter_error:{exc}", details=details, ) started = thread is not None reason = "started" if started else "starter_returned_none" if started: details = {**details, "thread": getattr(thread, "name", "")} return LoopStatus( key=key, label=label, configured_enabled=True, started=started, reason=reason, details=details, ) def _start_airport_high_freq_loop(self) -> LoopStatus: enabled = _env_bool("TELEGRAM_AIRPORT_PUSH_ENABLED", True) chat_ids = get_telegram_chat_ids_from_env() interval = max(30, _env_int("TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC", 60)) details = { "mode": "airport-periodic", "interval_sec": interval, "cities": ["seoul", "busan", "tokyo", "ankara", "helsinki", "amsterdam", "istanbul", "paris", "hong kong", "lau fau shan", "taipei"], "chat_targets": len(chat_ids), "window": "DEB proximity ≤3°C", } validation_error = None if chat_ids else "missing_TELEGRAM_CHAT_IDS" return self._start_with_validation( key="airport_high_freq_push", label="机场高频推送", configured_enabled=enabled, details=details, validation_error=validation_error, starter=lambda: import_module("src.utils.telegram_push").start_high_freq_airport_push_loop( self.bot, self.config, ), ) def _start_polygon_wallet_loop(self) -> LoopStatus: enabled = _env_bool("POLYGON_WALLET_WATCH_ENABLED", False) chat_ids = get_telegram_chat_ids_from_env() rpc_url = str(os.getenv("POLYGON_RPC_URL") or "").strip() wallets_count = _parse_csv_count(os.getenv("POLYGON_WALLET_WATCH_ADDRESSES")) poll = max(3, _env_int("POLYGON_WALLET_WATCH_INTERVAL_SEC", 8)) details = { "poll_sec": poll, "wallets_count": wallets_count, "polymarket_only": _env_bool("POLYGON_WALLET_WATCH_POLYMARKET_ONLY", True), "chat_targets": len(chat_ids), } validation_error = None if not chat_ids: validation_error = "missing_TELEGRAM_CHAT_IDS" elif not rpc_url: validation_error = "missing_POLYGON_RPC_URL" elif wallets_count == 0: validation_error = "missing_POLYGON_WALLET_WATCH_ADDRESSES" return self._start_with_validation( key="polygon_wallet_watch", label="Polygon 钱包监听", configured_enabled=enabled, details=details, validation_error=validation_error, starter=lambda: import_module( "src.onchain.polygon_wallet_watcher" ).start_polygon_wallet_watch_loop(self.bot), ) def _start_polymarket_wallet_activity_loop(self) -> LoopStatus: return LoopStatus( key="polymarket_wallet_activity", label="Polymarket 钱包异动监听(已停用)", configured_enabled=False, started=False, reason="retired_replaced_by_market_monitor", details={ "note": "wallet activity watcher retired", }, ) def _start_weekly_reward_loop(self) -> LoopStatus: enabled = _env_bool("POLYWEATHER_WEEKLY_REWARD_ENABLED", True) chat_ids = get_telegram_chat_ids_from_env() settle_weekday = min( 7, max(1, _env_int("POLYWEATHER_WEEKLY_REWARD_SETTLE_WEEKDAY", 1)) ) settle_hour = min(23, max(0, _env_int("POLYWEATHER_WEEKLY_REWARD_SETTLE_HOUR", 0))) settle_minute = min( 59, max(0, _env_int("POLYWEATHER_WEEKLY_REWARD_SETTLE_MINUTE", 5)) ) check_interval = max( 30, _env_int("POLYWEATHER_WEEKLY_REWARD_CHECK_INTERVAL_SEC", 300) ) details = { "timezone": str( os.getenv("POLYWEATHER_WEEKLY_REWARD_TIMEZONE") or "Asia/Shanghai" ).strip(), "settle_weekday": settle_weekday, "settle_time": f"{settle_hour:02d}:{settle_minute:02d}", "check_interval_sec": check_interval, "announce": _env_bool("POLYWEATHER_WEEKLY_REWARD_ANNOUNCE_ENABLED", True), "chat_targets": len(chat_ids), } announce_enabled = bool(details["announce"]) validation_error = None if announce_enabled and not chat_ids: validation_error = "missing_TELEGRAM_CHAT_IDS" return self._start_with_validation( key="weekly_reward", label="周榜奖励结算", configured_enabled=enabled, details=details, validation_error=validation_error, starter=lambda: import_module("src.bot.weekly_reward_loop").start_weekly_reward_loop( self.bot ), ) def _start_payment_confirm_loop(self) -> LoopStatus: enabled = _env_bool("POLYWEATHER_PAYMENT_CONFIRM_LOOP_ENABLED", True) details = { "interval_sec": max( 5, _env_int("POLYWEATHER_PAYMENT_CONFIRM_LOOP_INTERVAL_SEC", 20) ), "batch_size": max( 1, min(200, _env_int("POLYWEATHER_PAYMENT_CONFIRM_LOOP_BATCH_SIZE", 20)) ), "payment_enabled": _env_bool("POLYWEATHER_PAYMENT_ENABLED", False), "chain_id": _env_int("POLYWEATHER_PAYMENT_CHAIN_ID", 137), "confirmations": max(1, _env_int("POLYWEATHER_PAYMENT_CONFIRMATIONS", 2)), } validation_error = None if not bool(details["payment_enabled"]): validation_error = "payment_service_disabled" return self._start_with_validation( key="payment_confirm", label="支付自动补单", configured_enabled=enabled, details=details, validation_error=validation_error, starter=lambda: import_module( "src.payments.confirm_loop" ).start_payment_confirm_loop(), ) def _start_payment_event_loop(self) -> LoopStatus: enabled = _env_bool("POLYWEATHER_PAYMENT_EVENT_LOOP_ENABLED", True) details = { "interval_sec": max( 5, _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_INTERVAL_SEC", 20) ), "lookback_blocks": max( 500, _env_int( "POLYWEATHER_PAYMENT_EVENT_LOOP_START_LOOKBACK_BLOCKS", 5000, ), ), "step_blocks": min( 49999, max(100, _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_STEP_BLOCKS", 2000)), ), "max_events": max( 10, _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_MAX_EVENTS_PER_CYCLE", 200) ), "payment_enabled": _env_bool("POLYWEATHER_PAYMENT_ENABLED", False), "chain_id": _env_int("POLYWEATHER_PAYMENT_CHAIN_ID", 137), } validation_error = None if not bool(details["payment_enabled"]): validation_error = "payment_service_disabled" return self._start_with_validation( key="payment_event", label="支付事件监听", configured_enabled=enabled, details=details, validation_error=validation_error, starter=lambda: import_module( "src.payments.event_loop" ).start_payment_event_loop(), ) def render_runtime_status_html(status: RuntimeStatus) -> str: lines = [ "🧭 Bot 启动诊断", f"启动时间: {status.started_at}", "", f"命令准入: {status.command_access_mode}", f"受保护命令: {', '.join(status.protected_commands) or '--'}", f"目标群组: {status.required_group_chat_id or '--'}", "", "后台循环:", ] for loop in status.loops: icon = "✅" if loop.started else ("⏸" if not loop.configured_enabled else "⚠️") detail_str = ", ".join(f"{k}={v}" for k, v in sorted(loop.details.items())) lines.append( f"{icon} {loop.label} | enabled={str(loop.configured_enabled).lower()} | " f"started={str(loop.started).lower()} | reason={loop.reason}" ) if detail_str: lines.append(f" {detail_str}") return "\n".join(lines)