import hashlib import json import os import re import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Set, Tuple import requests as requests_lib from loguru import logger from src.database.db_manager import DBManager from src.database.runtime_state import ( STATE_STORAGE_SQLITE, TelegramAlertStateRepository, get_state_storage_mode, ) from src.data_collection.city_registry import ALIASES from src.data_collection.city_registry import CITY_REGISTRY from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env, parse_telegram_chat_ids from src.utils.telegram_i18n import ( copy_text as _copy, is_bilingual as _is_bilingual, is_zh as _is_zh, normalize_push_language as _normalize_push_language, telegram_push_language as _resolve_telegram_push_language, ) from web.services.canonical_temperature import build_city_weather_from_canonical # Forum topic routing: maps city_key -> message_thread_id for the push forum group. # Created by scripts/create_forum_topics.py, stored in the runtime data dir. _CITY_THREAD_IDS_PATH = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "data", "city_thread_ids.json", ) _DEFAULT_FORUM_CHAT_ID = "-1003927451869" _city_thread_ids: dict = {} _CITY_THREAD_IDS_LOCK = threading.Lock() # Shared HTTP session for AROME and auxiliary queries (connection reuse) _HTTP_SESSION: Optional[requests_lib.Session] = None _HTTP_SESSION_LOCK = threading.Lock() # Bot send_message rate limiter: max N messages per second across all threads _SEND_MSG_LOCK = threading.Lock() _SEND_MSG_LAST_TS: float = 0.0 _SEND_MSG_MIN_INTERVAL_SEC = float(os.getenv("TELEGRAM_SEND_RATE_LIMIT_SEC", "1.1")) def _get_http_session() -> requests_lib.Session: global _HTTP_SESSION if _HTTP_SESSION is None: with _HTTP_SESSION_LOCK: if _HTTP_SESSION is None: _HTTP_SESSION = requests_lib.Session() return _HTTP_SESSION # Reusable executor for airport push cycles (avoids thread pool churn) _AIRPORT_EXECUTOR: Optional[ThreadPoolExecutor] = None _AIRPORT_EXECUTOR_LOCK = threading.Lock() _AIRPORT_EXECUTOR_MAX_WORKERS: int = 0 def _get_airport_executor(max_workers: int) -> ThreadPoolExecutor: global _AIRPORT_EXECUTOR, _AIRPORT_EXECUTOR_MAX_WORKERS if _AIRPORT_EXECUTOR is None or _AIRPORT_EXECUTOR_MAX_WORKERS != max_workers: with _AIRPORT_EXECUTOR_LOCK: if _AIRPORT_EXECUTOR is None or _AIRPORT_EXECUTOR_MAX_WORKERS != max_workers: if _AIRPORT_EXECUTOR is not None: _AIRPORT_EXECUTOR.shutdown(wait=False) _AIRPORT_EXECUTOR = ThreadPoolExecutor(max_workers=max_workers) _AIRPORT_EXECUTOR_MAX_WORKERS = max_workers return _AIRPORT_EXECUTOR def _rate_limited_send(bot: Any, chat_id: str, message: str, **kwargs: Any) -> None: """Throttle bot.send_message calls to avoid hitting Telegram rate limits.""" global _SEND_MSG_LAST_TS with _SEND_MSG_LOCK: now = time.time() wait = _SEND_MSG_MIN_INTERVAL_SEC - (now - _SEND_MSG_LAST_TS) if wait > 0: time.sleep(wait) _SEND_MSG_LAST_TS = time.time() bot.send_message(chat_id, message, **kwargs) def _load_city_thread_ids() -> dict: global _city_thread_ids if _city_thread_ids: return _city_thread_ids paths = [ _CITY_THREAD_IDS_PATH, "/var/lib/polyweather/city_thread_ids.json", "/app/data/city_thread_ids.json", ] for path in paths: if os.path.isfile(path): try: with open(path, "r", encoding="utf-8") as f: _city_thread_ids = json.load(f) logger.info("loaded city_thread_ids from {}: {} cities", path, len(_city_thread_ids)) return _city_thread_ids except Exception as exc: logger.warning("failed to load city_thread_ids from {}: {}", path, exc) return {} def normalize_airport_push_city(raw: str) -> str: city = str(raw or "").strip().lower().replace("-", " ") city = re.sub(r"\s+", " ", city) compact = city.replace(" ", "") city = ALIASES.get(city, ALIASES.get(compact, city)) if city not in HIGH_FREQ_AIRPORT_CITIES: city = { candidate.replace(" ", ""): candidate for candidate in HIGH_FREQ_AIRPORT_CITIES }.get(compact, city) if city in HIGH_FREQ_AIRPORT_CITIES: return city return "" def _city_thread_ids_write_path() -> str: env_path = str(os.getenv("POLYWEATHER_CITY_THREAD_IDS_PATH") or "").strip() if env_path: return env_path for path in ( "/var/lib/polyweather/city_thread_ids.json", "/app/data/city_thread_ids.json", _CITY_THREAD_IDS_PATH, ): if os.path.isfile(path): return path return "/var/lib/polyweather/city_thread_ids.json" def record_city_thread_id(city: str, thread_id: int) -> dict: """Persist a forum topic mapping and update the in-process cache.""" global _city_thread_ids normalized_city = normalize_airport_push_city(city) if not normalized_city: raise ValueError(f"unsupported airport push city: {city}") try: normalized_thread_id = int(thread_id) except Exception as exc: raise ValueError(f"invalid message_thread_id: {thread_id}") from exc if normalized_thread_id <= 0: raise ValueError(f"invalid message_thread_id: {thread_id}") with _CITY_THREAD_IDS_LOCK: path = _city_thread_ids_write_path() mapping = dict(_load_city_thread_ids()) mapping[normalized_city] = normalized_thread_id os.makedirs(os.path.dirname(path), exist_ok=True) tmp_path = f"{path}.tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(mapping, f, ensure_ascii=False, indent=2, sort_keys=True) f.write("\n") os.replace(tmp_path, path) _city_thread_ids = mapping logger.info( "recorded forum topic mapping city={} thread_id={} path={} total={}", normalized_city, normalized_thread_id, path, len(mapping), ) return { "city": normalized_city, "thread_id": normalized_thread_id, "path": path, "total": len(mapping), } def _forum_chat_ids() -> Set[str]: return set( parse_telegram_chat_ids( os.getenv("TELEGRAM_FORUM_CHAT_ID"), os.getenv("POLYWEATHER_TELEGRAM_TOPICS_GROUP_ID"), os.getenv("POLYWEATHER_TELEGRAM_GROUP_ID"), _DEFAULT_FORUM_CHAT_ID, ) ) def _is_forum_chat_id(chat_id: Any) -> bool: chat_key = str(chat_id or "").strip() return bool(chat_key and chat_key in _forum_chat_ids()) def _resolve_thread_id(chat_id: str, city: str) -> int: """Return message_thread_id for a given chat and city, or 0 if not a forum topic.""" if not _is_forum_chat_id(chat_id): return 0 mapping = _load_city_thread_ids() city_key = (city or "").strip().lower() return int(mapping.get(city_key) or 0) SEVERITY_RANK = { "none": 0, "low": 1, "medium": 2, "high": 3, } _telegram_state_repo = TelegramAlertStateRepository() 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 _env_float(name: str, default: float) -> float: raw = os.getenv(name) if raw is None: return default try: return float(raw) except Exception: return default def _telegram_push_language() -> str: return _resolve_telegram_push_language( "TELEGRAM_AIRPORT_PUSH_LANGUAGE", "TELEGRAM_PUSH_LANGUAGE", "POLYWEATHER_TELEGRAM_PUSH_LANGUAGE", ) def _norm_prob(v: Any) -> Optional[float]: if v is None: return None try: n = float(v) except Exception: return None if n > 1.0: n = n / 100.0 return max(0.0, min(1.0, n)) def _fmt_cents(value: Any) -> Optional[str]: numeric = _norm_prob(value) if numeric is None: return None cents = numeric * 100.0 rounded = round(cents, 1) text = f"{rounded:.1f}".rstrip("0").rstrip(".") return f"{text}c" def _safe_float(v: Any) -> Optional[float]: if v is None: return None try: return float(v) except Exception: return None def _bucket_value(row: Dict[str, Any]) -> Optional[float]: if not isinstance(row, dict): return None for key in ("value", "temp"): n = _safe_float(row.get(key)) if n is not None: return n label = str(row.get("label") or "").strip() m = re.search(r"(-?\d+(?:\.\d+)?)", label) if not m: return None return _safe_float(m.group(1)) def _bucket_bounds(row: Dict[str, Any]) -> Optional[Tuple[Optional[float], Optional[float]]]: value = _bucket_value(row) if value is None: return None label = str(row.get("label") or "").strip().lower() is_upper_tail = any(key in label for key in ("+", "or higher", "or above", "and above")) is_lower_tail = any(key in label for key in ("<=", "or lower", "or below", "and below")) if is_upper_tail and not is_lower_tail: return value, None if is_lower_tail and not is_upper_tail: return None, value return value, value def _observed_settlement_floor(alert_payload: Dict[str, Any]) -> Optional[float]: evidence = alert_payload.get("evidence") or {} if not isinstance(evidence, dict): evidence = {} inputs = evidence.get("inputs") or {} if not isinstance(inputs, dict): inputs = {} suppression = alert_payload.get("suppression") or {} if not isinstance(suppression, dict): suppression = {} rules = alert_payload.get("rules") or {} if not isinstance(rules, dict): rules = {} breakthrough = rules.get("forecast_breakthrough") or {} if not isinstance(breakthrough, dict): breakthrough = {} floor_candidates: List[float] = [] for raw in ( inputs.get("wu_settle"), suppression.get("max_so_far"), inputs.get("current_temp"), suppression.get("current_temp"), breakthrough.get("current_temp"), ): n = _safe_float(raw) if n is not None: floor_candidates.append(n) if not floor_candidates: return None return max(floor_candidates) def _optional_bool(value: Any) -> Optional[bool]: if value is None: return None if isinstance(value, bool): return value if isinstance(value, (int, float)): return bool(value) if isinstance(value, str): normalized = value.strip().lower() if normalized in {"1", "true", "yes", "on"}: return True if normalized in {"0", "false", "no", "off"}: return False return bool(value) def _parse_iso_datetime_utc(value: Any) -> Optional[datetime]: if not value: return None text = str(value).strip() if not text: return None if "T" not in text: return None try: dt = datetime.fromisoformat(text.replace("Z", "+00:00")) except Exception: return None if dt.tzinfo is None: return dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc) def _parse_observation_time_epoch(value: Any) -> Optional[int]: if value is None: return None text = str(value).strip() if not text: return None try: numeric = float(text) if numeric >= 1_000_000_000: return int(numeric) except (TypeError, ValueError): pass parsed = _parse_iso_datetime_utc(text) return int(parsed.timestamp()) if parsed is not None else None def _parse_city_list(raw: Optional[str]) -> List[str]: if not raw: return list(CITY_REGISTRY.keys()) out: List[str] = [] for part in raw.split(","): city = part.strip().lower() if city and city in CITY_REGISTRY: out.append(city) return out or list(CITY_REGISTRY.keys()) def _state_file() -> str: root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) return os.path.join(root, "data", "telegram_alert_state.json") def _load_state(path: str) -> Dict[str, Any]: mode = get_state_storage_mode() if mode == STATE_STORAGE_SQLITE: try: return _telegram_state_repo.load_state() except Exception as exc: logger.error(f"failed to load telegram push state from sqlite: {exc}") if not os.path.exists(path): return {"last_by_city": {}, "by_signature": {}} try: with open(path, "r", encoding="utf-8") as fh: data = json.load(fh) if isinstance(data, dict): data.setdefault("last_by_city", {}) data.setdefault("by_signature", {}) return data except Exception as exc: logger.warning(f"failed to load telegram push state: {exc}") return {"last_by_city": {}, "by_signature": {}} def _save_state(path: str, state: Dict[str, Any]) -> None: mode = get_state_storage_mode() if mode == STATE_STORAGE_SQLITE: _telegram_state_repo.save_state(state) if mode == STATE_STORAGE_SQLITE: return os.makedirs(os.path.dirname(path), exist_ok=True) tmp_path = f"{path}.tmp" with open(tmp_path, "w", encoding="utf-8") as fh: json.dump(state, fh, ensure_ascii=False, indent=2) os.replace(tmp_path, path) def _cleanup_state(state: Dict[str, Any], now_ts: int, keep_sec: int = 7 * 86400) -> None: for bucket_name in ("by_signature",): bucket = state.get(bucket_name, {}) if not isinstance(bucket, dict): state[bucket_name] = {} continue stale = [key for key, value in bucket.items() if now_ts - int(value or 0) > keep_sec] for key in stale: bucket.pop(key, None) last_by_city = state.get("last_by_city", {}) if not isinstance(last_by_city, dict): state["last_by_city"] = {} return stale_city = [] for city, row in last_by_city.items(): ts = int((row or {}).get("ts") or 0) if now_ts - ts > keep_sec: stale_city.append(city) for city in stale_city: last_by_city.pop(city, None) def _severity_ok(alert_payload: Dict[str, Any], min_severity: str, min_trigger_count: int) -> bool: triggered_alerts = alert_payload.get("triggered_alerts") or [] if any(alert.get("force_push") for alert in triggered_alerts): return True trigger_count = int(alert_payload.get("trigger_count") or 0) if trigger_count < min_trigger_count: return False severity = str(alert_payload.get("severity") or "none").lower() return SEVERITY_RANK.get(severity, 0) >= SEVERITY_RANK.get(min_severity, 0) def _market_price_cap_ok( alert_payload: Dict[str, Any], ) -> bool: market = alert_payload.get("market_snapshot") or {} if not isinstance(market, dict) or not market.get("available"): return True primary_market = market.get("primary_market") or {} if not isinstance(primary_market, dict): primary_market = {} market_slug = ( str(market.get("selected_slug") or "").strip() or str(primary_market.get("slug") or "").strip() or "--" ) active = market.get("market_active") if active is None: active = primary_market.get("active") active = _optional_bool(active) closed = market.get("market_closed") if closed is None: closed = primary_market.get("closed") closed = _optional_bool(closed) accepting_orders = market.get("market_accepting_orders") if accepting_orders is None: accepting_orders = primary_market.get( "accepting_orders", primary_market.get("acceptingOrders"), ) accepting_orders = _optional_bool(accepting_orders) market_tradable = _optional_bool(market.get("market_tradable")) tradable_reason = str( market.get("market_tradable_reason") or primary_market.get("tradable_reason") or "" ).strip() ended_at = str( market.get("market_ended_at_utc") or primary_market.get("ended_at_utc") or "" ).strip() ended_dt = _parse_iso_datetime_utc(ended_at) is_past_end = ended_dt is not None and ended_dt <= datetime.now(timezone.utc) if ( market_tradable is False or closed is True or active is False or accepting_orders is False or is_past_end ): reason = tradable_reason or ("past_end_time" if is_past_end else "market_not_tradable") logger.info( "trade alert skipped: market not tradable city={} slug={} reason={} active={} closed={} accepting_orders={} ended_at={}".format( alert_payload.get("city"), market_slug, reason, active, closed, accepting_orders, ended_at or "--", ) ) return False # Strict rule: use the bucket mapped from multi-model anchor settlement. forecast_bucket = market.get("forecast_bucket") or {} settle_ref = market.get("anchor_settlement") if settle_ref is None: settle_ref = market.get("open_meteo_settlement") anchor_model = str(market.get("anchor_model") or "").strip() or "--" yes_buy = None bucket_label = None if isinstance(forecast_bucket, dict): yes_buy = _norm_prob(forecast_bucket.get("yes_buy")) bucket_label = str(forecast_bucket.get("label") or "").strip() or None observed_floor = _observed_settlement_floor(alert_payload) bucket_bounds = _bucket_bounds(forecast_bucket) if isinstance(forecast_bucket, dict) else None if observed_floor is not None and bucket_bounds is not None: _lower, upper = bucket_bounds if upper is not None and observed_floor > upper + 1e-9: logger.info( "trade alert skipped: mapped bucket invalidated by observed high city={} bucket={} observed_floor={} upper_bound={} anchor_model={} anchor_settle={}".format( alert_payload.get("city"), bucket_label or "--", round(observed_floor, 2), round(upper, 2), anchor_model, settle_ref, ) ) return False if yes_buy is None or yes_buy <= 0.0: logger.info( "trade alert skipped: no actionable mapped bucket quote city={} bucket={} anchor_model={} anchor_settle={}".format( alert_payload.get("city"), bucket_label or "--", anchor_model, settle_ref, ) ) return False return True def _trigger_type_key(alert_payload: Dict[str, Any]) -> str: trigger_types = sorted( str(alert.get("type") or "").strip() for alert in (alert_payload.get("triggered_alerts") or []) if alert.get("type") ) market = alert_payload.get("market_snapshot") or {} if isinstance(market, dict) and market.get("available"): signal = str(market.get("signal_label") or "").strip() bucket = str(market.get("selected_bucket") or "").strip() if signal: trigger_types.append(f"mkt:{signal}:{bucket}") return "|".join(trigger_types) def _evidence_brief(alert_payload: Dict[str, Any]) -> str: evidence = alert_payload.get("evidence") or {} if not isinstance(evidence, dict): return "--" trigger_summary = evidence.get("trigger_summary") or {} rules = evidence.get("rules") or {} market = evidence.get("market") or {} momentum = rules.get("momentum_spike") or {} advection = rules.get("advection") or {} breakthrough = rules.get("forecast_breakthrough") or {} parts: List[str] = [] trigger_types = trigger_summary.get("trigger_types") if isinstance(trigger_types, list) and trigger_types: parts.append(f"triggers={','.join(str(t) for t in trigger_types)}") slope = momentum.get("slope_30m") if slope is not None: parts.append(f"slope_30m={slope}") lead_delta = advection.get("lead_delta") if lead_delta is not None: parts.append(f"lead_delta={lead_delta}") margin = breakthrough.get("margin") if margin is not None: parts.append(f"break_margin={margin}") edge = market.get("edge_percent") if edge is not None: parts.append(f"edge_pct={edge}") forecast_bucket = market.get("forecast_bucket") or {} if isinstance(forecast_bucket, dict): label = str(forecast_bucket.get("label") or "").strip() yes_buy = forecast_bucket.get("yes_buy") if label: parts.append(f"bucket={label}") if yes_buy is not None: parts.append(f"yes_buy={yes_buy}") if not parts: return "--" return "; ".join(parts) def _alert_signature(alert_payload: Dict[str, Any]) -> str: rules = alert_payload.get("rules") or {} center_deb = rules.get("ankara_center_deb_hit") or {} momentum = rules.get("momentum_spike") or {} breakthrough = rules.get("forecast_breakthrough") or {} advection = rules.get("advection") or {} suppression = alert_payload.get("suppression") or {} market = alert_payload.get("market_snapshot") or {} signature_payload = { "city": alert_payload.get("city"), "target_date": alert_payload.get("target_date"), "severity": alert_payload.get("severity"), "trigger_types": sorted( alert.get("type") for alert in (alert_payload.get("triggered_alerts") or []) if alert.get("type") ), "center_temp": round(float(((center_deb.get("center_station") or {}).get("temp")) or 0.0), 1), "center_deb_prediction": round(float(center_deb.get("deb_prediction") or 0.0), 1), "center_airport_gap": round(float(center_deb.get("center_lead_vs_airport") or 0.0), 1), "momentum_direction": momentum.get("direction"), "momentum_slope_30m": round(float(momentum.get("slope_30m") or 0.0), 1), "breakthrough_margin": round(float(breakthrough.get("margin") or 0.0), 1), "lead_station": (advection.get("lead_station") or {}).get("name"), "lead_delta": round(float(advection.get("lead_delta") or 0.0), 1), "suppressed": bool(suppression.get("suppressed")), "suppression_reason": suppression.get("reason"), "suppression_peak_time": suppression.get("max_temp_time"), "suppression_rollback": round(float(suppression.get("rollback") or 0.0), 1), "market_available": bool(market.get("available")), "market_bucket": market.get("selected_bucket"), "market_top_bucket": market.get("top_bucket"), "market_top_bucket_prob": round(float(market.get("top_bucket_prob") or 0.0), 3), "market_prob": round(float(market.get("market_prob") or 0.0), 3), "model_prob": round(float(market.get("model_prob") or 0.0), 3), "market_yes_buy": round(float(market.get("yes_buy") or 0.0), 3), "market_yes_sell": round(float(market.get("yes_sell") or 0.0), 3), "market_spread": round(float(market.get("spread") or 0.0), 3), "market_edge_percent": round(float(market.get("edge_percent") or 0.0), 2), "market_signal": market.get("signal_label"), "market_confidence": market.get("confidence"), } raw = json.dumps(signature_payload, sort_keys=True, ensure_ascii=True) return hashlib.sha1(raw.encode("utf-8")).hexdigest() # ── high-freq airport push loop ── HIGH_FREQ_AIRPORT_CITIES = { "seoul", "singapore", "busan", "tokyo", "ankara", "helsinki", "amsterdam", "istanbul", "paris", "hong kong", "taipei", "beijing", "shanghai", "guangzhou", "qingdao", "chengdu", "chongqing", "wuhan", "shenzhen", "new york", "los angeles", "chicago", "denver", "atlanta", "miami", "san francisco", "houston", "dallas", "austin", "seattle", "tel aviv", } CHINA_HIGH_FREQ_AIRPORT_CITIES = { "beijing", "shanghai", "guangzhou", "qingdao", "chengdu", "chongqing", "wuhan", } HIGH_FREQ_AIRPORT_ICAO = { "seoul": "RKSI", "singapore": "WSSS", "busan": "RKPK", "tokyo": "44166", "ankara": "17128", "helsinki": "EFHK", "amsterdam": "EHAM", "istanbul": "17058", "paris": "LFPB", "hong kong": "HKO", "taipei": "466920", "beijing": "ZBAA", "shanghai": "ZSPD", "guangzhou": "ZGGG", "qingdao": "ZSQD", "chengdu": "ZUUU", "chongqing": "ZUCK", "wuhan": "ZHHH", "shenzhen": "LFS", "new york": "KLGA", "los angeles": "KLAX", "chicago": "KORD", "denver": "KBKF", "atlanta": "KATL", "miami": "KMIA", "san francisco": "KSFO", "houston": "KHOU", "dallas": "KDAL", "austin": "KAUS", "seattle": "KSEA", "tel aviv": "LLBG", } # Settlement runway mapping — matches settlement anchor stations. # Format: (low_number, high_number) order-independent; stored sorted for lookup. SETTLEMENT_RUNWAY_PAIRS: Dict[str, Set[Tuple[str, str]]] = { "shanghai": {("17L", "35R")}, "beijing": {("01", "19")}, "guangzhou": {("02L", "20R")}, "chengdu": {("02L", "20R")}, "chongqing": {("02L", "20R")}, "wuhan": {("04", "22")}, "qingdao": {("16", "34")}, "seoul": {("15R", "33L")}, } SETTLEMENT_RUNWAY_TARGETS: Dict[str, str] = { "shanghai": "35R", "chengdu": "02L", "chongqing": "02L", "guangzhou": "02L", "wuhan": "04", "beijing": "01", "qingdao": "34", } # All cities with active runway observation data (AMSC AWOS / AMOS). RUNWAY_OBSERVATION_CITIES = { "shanghai", "beijing", "guangzhou", "chengdu", "chongqing", "wuhan", "qingdao", "seoul", "busan", } # Wind regime sectors per airport (approximate, based on runway orientation + coastline). # Values: {sea_breeze: (from_deg, to_deg), warm_advection: (from_deg, to_deg)} WIND_REGIME: Dict[str, Dict[str, Tuple[int, int]]] = { "shanghai": {"sea_breeze": (45, 140), "warm_advection": (180, 260)}, "seoul": {"sea_breeze": (270, 350), "warm_advection": (150, 230)}, "busan": {"sea_breeze": (120, 200), "warm_advection": (250, 340)}, "qingdao": {"sea_breeze": (90, 180), "warm_advection": (200, 300)}, "beijing": {"sea_breeze": (120, 200), "warm_advection": (220, 320)}, "guangzhou": {"sea_breeze": (120, 200), "warm_advection": (200, 300)}, "chengdu": {"sea_breeze": (0, 0), "warm_advection": (0, 0)}, "chongqing": {"sea_breeze": (0, 0), "warm_advection": (0, 0)}, "wuhan": {"sea_breeze": (0, 0), "warm_advection": (0, 0)}, } # Legacy alias for backward compat with existing _select_focus_runway_obs / _focus_runway_pairs_for_city FOCUS_RUNWAY_PAIRS = SETTLEMENT_RUNWAY_PAIRS _FUNCTION_HASHTAGS_ZH = { "runway": "#跑道观测", "airport": "#机场观测", "trade": "#交易机会", } _FUNCTION_HASHTAGS_EN = { "runway": "#RunwayObs", "airport": "#AirportObs", "trade": "#TradeAlert", } def _city_hashtag(city: Optional[str]) -> Optional[str]: text = (city or "").strip() if not text: return None parts = [part for part in re.split(r"[^A-Za-z0-9]+", text.title()) if part] if not parts: return None return "#" + "".join(parts) def _station_hashtag(station: Optional[str]) -> Optional[str]: text = re.sub(r"[^A-Za-z0-9]+", "", (station or "").upper()) return f"#{text}" if text else None def _build_telegram_hashtag_line( kind: str, *, city: Optional[str] = None, station: Optional[str] = None, extra: Optional[List[str]] = None, language: Optional[str] = None, ) -> str: tags: List[str] = [] tag_maps = ( (_FUNCTION_HASHTAGS_EN, _FUNCTION_HASHTAGS_ZH) if _is_bilingual(language) else ((_FUNCTION_HASHTAGS_ZH,) if _is_zh(language) else (_FUNCTION_HASHTAGS_EN,)) ) for tag_map in tag_maps: primary = tag_map.get(kind) if primary and primary not in tags: tags.append(primary) for item in extra or []: added = False for tag_map in tag_maps: tag = tag_map.get(item) if tag and tag not in tags: tags.append(tag) added = True if not added and item and item not in tags: tags.append(item) city_tag = _city_hashtag(city) if city_tag and city_tag not in tags: tags.append(city_tag) station_tag = _station_hashtag(station) if station_tag and station_tag not in tags: tags.append(station_tag) return " ".join(tags) def _fmt(value: Any) -> str: """Format a single temperature reading; returns '--' for missing values.""" if value is None: return "--" try: f = float(value) if not -80.0 < f < 80.0: return "--" return f"{f:.1f}" except (TypeError, ValueError): return "--" def _normalize_runway_label(value: Any) -> str: return re.sub(r"[^0-9A-Z]+", "", str(value or "").strip().upper()) def _runway_pair_key(r1: Any, r2: Any) -> Tuple[str, str]: a = _normalize_runway_label(r1) b = _normalize_runway_label(r2) return tuple(sorted((a, b))) # type: ignore[return-value] def _focus_runway_pairs_for_city(city: str) -> Set[Tuple[str, str]]: return {_runway_pair_key(a, b) for a, b in FOCUS_RUNWAY_PAIRS.get(city, set())} def _settlement_runway_target_for_city(city: str) -> str: city_key = (city or "").strip().lower() return _normalize_runway_label(SETTLEMENT_RUNWAY_TARGETS.get(city_key)) def _runway_pair_from_point(pair: Any, point: Any) -> Tuple[str, str]: if isinstance(point, dict): rw = str(point.get("runway") or "") parts = [_normalize_runway_label(p) for p in rw.split("/") if p.strip()] if len(parts) >= 2: return parts[0], parts[1] try: r1, r2 = pair return _normalize_runway_label(r1), _normalize_runway_label(r2) except Exception: return "", "" def _settlement_endpoint_for_point( city: str, pair: Any, point: Any, ) -> Optional[Dict[str, Any]]: if not isinstance(point, dict): point = {} r1, r2 = _runway_pair_from_point(pair, point) if not r1 or not r2 or not _is_settlement_runway(city, r1, r2): return None target = _settlement_runway_target_for_city(city) if target: direct_temp = _safe_float(point.get("settlement_runway_temp")) direct_runway = _normalize_runway_label(point.get("settlement_runway")) if direct_temp is not None and (not direct_runway or direct_runway == target): return { "temp": direct_temp, "pair": f"{r1}/{r2}", "runway": target, "position": point.get("settlement_runway_position") or "settlement", "label": "settle", } tdz = _safe_float(point.get("tdz_temp")) end = _safe_float(point.get("end_temp")) if target == r1: temp = tdz if tdz is not None else end position = "tdz" if tdz is not None else "end_fallback" elif target == r2: temp = end if end is not None else tdz position = "end" if end is not None else "tdz_fallback" else: return None if temp is None: return None return { "temp": temp, "pair": f"{r1}/{r2}", "runway": target, "position": position, "label": "settle", } tmax = _safe_float(point.get("target_runway_max")) if tmax is None: return None return { "temp": tmax, "pair": f"{r1}/{r2}", "runway": "", "position": "max", "label": "max", } def _settlement_endpoint_from_obs( city: str, runway_pairs: List[Any], point_temps: Optional[List[Any]] = None, ) -> Optional[Dict[str, Any]]: points = point_temps or [] for i, pair in enumerate(runway_pairs or []): point = points[i] if i < len(points) else {} endpoint = _settlement_endpoint_for_point(city, pair, point) if endpoint is not None: return endpoint return None def _select_focus_runway_obs( city: str, runway_pairs: List[Any], runway_temps: List[Any], point_temps: Optional[List[Any]] = None, ) -> Tuple[List[Any], List[Any], List[Any]]: """Return only market-relevant runway pairs when configured for the city. If a configured focus pair is not present in the upstream payload, fall back to the original lists so the push still carries useful airport evidence. """ focus_pairs = _focus_runway_pairs_for_city(city) if not focus_pairs or not runway_pairs or not runway_temps: return runway_pairs, runway_temps, point_temps or [] selected_pairs: List[Any] = [] selected_temps: List[Any] = [] selected_points: List[Any] = [] points = point_temps or [] for i, (pair, temp) in enumerate(zip(runway_pairs, runway_temps)): try: r1, r2 = pair except Exception: continue if _runway_pair_key(r1, r2) not in focus_pairs: continue selected_pairs.append(pair) selected_temps.append(temp) if i < len(points): selected_points.append(points[i]) if selected_pairs: return selected_pairs, selected_temps, selected_points return runway_pairs, runway_temps, points def _settlement_runway_for_city(city: str) -> Optional[Tuple[str, str]]: """Return the settlement runway pair for a city, if configured.""" pairs = SETTLEMENT_RUNWAY_PAIRS.get((city or "").strip().lower(), set()) return next(iter(pairs)) if pairs else None def _is_settlement_runway(city: str, r1: str, r2: str) -> bool: """Check if a runway pair is the settlement anchor for this city.""" pair_set = SETTLEMENT_RUNWAY_PAIRS.get((city or "").strip().lower(), set()) return _runway_pair_key(r1, r2) in pair_set def _wind_regime_label(city: str, wind_dir: Optional[int], language: Optional[str] = None) -> Optional[str]: """Classify wind direction into a thermal regime label.""" if wind_dir is None: return None regimes = WIND_REGIME.get((city or "").strip().lower(), {}) sea = regimes.get("sea_breeze") warm = regimes.get("warm_advection") if sea and sea[0] != sea[1] and sea[0] <= wind_dir <= sea[1]: return _copy(language, "sea-breeze cooling", "海风降温") if warm and warm[0] != warm[1] and warm[0] <= wind_dir <= warm[1]: return _copy(language, "warm advection", "暖平流增强") return None def _runway_row_temp_for_city(city: str, row: Dict[str, Any]) -> Optional[float]: endpoint = _settlement_endpoint_for_point(city, row.get("runway"), row) if endpoint is not None: return _safe_float(endpoint.get("temp")) target_runway_max = _safe_float(row.get("target_runway_max")) if target_runway_max is not None: return target_runway_max return _safe_float(row.get("tdz_temp")) def _compute_slope_15m(icao: str, current_temp: float, city: str = "") -> Optional[float]: """Estimate 15-minute temperature trend from runway_obs_log.""" try: db = DBManager() rows = db.get_runway_obs_recent(icao, minutes=20) temps = [] for r in rows: t = _runway_row_temp_for_city(city, r) if t is not None: temps.append(float(t)) if len(temps) >= 2: # Compare latest vs earliest in ~15 min window return round(current_temp - temps[0], 1) except Exception: pass return None def _runway_heat_signal( current_temp: float, slope_15m: Optional[float], wind_dir: Optional[int], city: str, language: Optional[str] = None, ) -> str: """Compute a simple runway heat signal label.""" if slope_15m is None: return "" regime = _wind_regime_label(city, wind_dir, language) warm_regime = _copy(language, "warm advection", "暖平流增强") sea_regime = _copy(language, "sea-breeze cooling", "海风降温") if slope_15m >= 1.0: return _copy(language, "🚀 Peak push strengthening", "🚀 冲顶增强") if regime == warm_regime else _copy(language, "🔥 Warming", "🔥 升温中") if slope_15m >= 0.5: return _copy(language, "🔥 Warming", "🔥 升温中") if slope_15m >= -0.2: return _copy(language, "⚠️ Watch sea-breeze cooling", "⚠️ 高位观察") if regime == sea_regime else _copy(language, "⏸️ Holding near high", "⏸️ 高位横盘") return _copy(language, "🧊 Peak-risk cooling", "🧊 过峰风险") def _focused_runway_max(city: str, city_weather: Dict[str, Any]) -> Optional[float]: amos = city_weather.get("amos") or {} runway_obs = (amos.get("runway_obs") or {}) if isinstance(amos, dict) else {} runway_pairs = runway_obs.get("runway_pairs") or [] runway_temps = runway_obs.get("temperatures") or [] runway_pairs, runway_temps, _points = _select_focus_runway_obs( city, runway_pairs, runway_temps, runway_obs.get("point_temperatures") or [], ) endpoint = _settlement_endpoint_from_obs(city, runway_pairs, _points) if endpoint is not None: return float(endpoint["temp"]) del runway_pairs valid = [float(t) for (t, _d) in runway_temps if t is not None] return max(valid) if valid else None _AIRPORT_PUSH_STATE_PATH = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "data", "airport_push_state.json", ) def _load_airport_state() -> Dict[str, Any]: path = _AIRPORT_PUSH_STATE_PATH if not os.path.exists(path): return {"last_by_city": {}} try: with open(path, "r", encoding="utf-8") as fh: data = json.load(fh) if isinstance(data, dict): data.setdefault("last_by_city", {}) return data except Exception: pass return {"last_by_city": {}} def _save_airport_state(state: Dict[str, Any]) -> None: path = _AIRPORT_PUSH_STATE_PATH os.makedirs(os.path.dirname(path), exist_ok=True) tmp = f"{path}.tmp" with open(tmp, "w", encoding="utf-8") as fh: json.dump(state, fh, ensure_ascii=False, indent=2) os.replace(tmp, path) _AROME_CACHE: Dict[str, Any] = {} _AROME_CACHE_TTL_SEC = 600 # AROME HD updates every 15 min; cache 10 min def _fetch_arome_temp() -> Optional[float]: """Fetch latest AROME France HD 15-min temperature for LFPB from Open-Meteo. Cached for 10 minutes since the model only updates every 15 minutes. """ now = time.time() cached = _AROME_CACHE.get("value") cached_at = _AROME_CACHE.get("ts", 0) if cached is not None and (now - cached_at) < _AROME_CACHE_TTL_SEC: return cached try: url = ( "https://api.open-meteo.com/v1/forecast?" "latitude=48.9673&longitude=2.4277" "&models=meteofrance_arome_france_hd" "&minutely_15=temperature_2m" "&timezone=Europe/Paris" "&forecast_minutely_15=2" ) resp = _get_http_session().get(url, timeout=8) data = resp.json() temps = (data.get("minutely_15") or {}).get("temperature_2m") or [] result = float(temps[-1]) if temps else None _AROME_CACHE["value"] = result _AROME_CACHE["ts"] = now return result except Exception: return _AROME_CACHE.get("value") _LAST_AEROWEB: Dict[str, Any] = {} # {"temp": 31.0, "max_so_far": 31.0, "max_time": "14:00", "ts": epoch} def _build_narrative( current_temp: Optional[float], max_so_far: Optional[float], deb_pred: Optional[float], models: Dict[str, Any], city_weather: Dict[str, Any], ) -> str: """Generate a market-structure interpretation based on current state.""" if current_temp is None: return "" vals = sorted([v for v in models.values() if isinstance(v, (int, float))]) model_lo = vals[0] if vals else None model_hi = vals[-1] if vals else None h = 12 try: h = int(str(city_weather.get("local_time") or "12")[:2]) except ValueError: pass lines = [] # Delta vs daily high if max_so_far is not None: d = current_temp - max_so_far if d >= 0.3: lines.append(f"🔺 当前已创今日新高（+{d:.1f}°），持续冲高") elif d >= -1.0: lines.append(f"紧贴日高，距峰值仅 {abs(d):.1f}°") else: lines.append(f"低于日高 {abs(d):.1f}°") # DEB relation if deb_pred is not None: if current_temp > deb_pred: lines.append("DEB 已被突破，模型偏保守") elif max_so_far is not None and max_so_far >= deb_pred: pass # already covered by 日高已触及 elif current_temp > deb_pred - 2.0: lines.append("DEB 仍在可达范围") # Model position + time context if model_lo is not None and model_hi is not None: if current_temp < model_lo: if h >= 17: lines.append("低于所有模型，晚间升温窗口有限") else: lines.append("低于主流模型，日间仍有升温空间") elif current_temp <= model_hi: lines.append("位于模型区间内，市场在预期路径上") else: lines.append("已超出最热模型，市场进入超预期定价") elif max_so_far is not None and current_temp < max_so_far - 2.0: if h < 14: lines.append("日间仍可能二次冲高") else: lines.append("已脱离日内峰值") return "\n".join(lines) def _build_airport_status_message( city: str, city_weather: Dict[str, Any], deb_pred: Optional[float], local_time: str = "", state: str = "", source_label: str = "", arome_temp: Optional[float] = None, aeroweb_available: bool = False, language: Optional[str] = None, ) -> str: language = _normalize_push_language(language or _telegram_push_language()) _AIRPORT_EN = {"seoul": "Incheon", "singapore": "Changi", "busan": "Gimhae", "tokyo": "Haneda", "ankara": "Esenboğa", "helsinki": "Vantaa", "amsterdam": "Schiphol", "istanbul": "Airport", "paris": "Le Bourget", "hong kong": "Observatory", "taipei": "Songshan", "beijing": "Capital", "shanghai": "Pudong", "guangzhou": "Baiyun", "qingdao": "Jiaodong", "chengdu": "Shuangliu", "chongqing": "Jiangbei", "wuhan": "Tianhe", "shenzhen": "Lau Fau Shan", "new york": "LaGuardia", "los angeles": "LAX", "chicago": "O'Hare", "denver": "Buckley", "atlanta": "Hartsfield", "miami": "Intl", "san francisco": "SFO", "houston": "Hobby", "dallas": "Love Field", "austin": "Bergstrom", "seattle": "Sea-Tac", "tel aviv": "Ben Gurion"} en_name = city.title() ap_name = _AIRPORT_EN.get(city, "") time_suffix = f" · {local_time}" if local_time else "" amos = city_weather.get("amos") or {} runway_data = amos.get("runway_obs") or {} runway_pairs = runway_data.get("runway_pairs") or [] runway_temps = runway_data.get("temperatures") or [] point_temps = runway_data.get("point_temperatures") or [] is_amsc = amos.get("source") in ("amsc_awos", "amos") has_runway = bool(runway_pairs and (runway_temps or point_temps)) amos_icao = amos.get("icao") or HIGH_FREQ_AIRPORT_ICAO.get(city, "") settlement_pair = _settlement_runway_for_city(city) settlement_endpoint = _settlement_endpoint_from_obs(city, runway_pairs, point_temps) # ── Display temp: settlement endpoint first, then airport temp ── display_temp: Optional[float] = None if settlement_endpoint is not None: display_temp = float(settlement_endpoint["temp"]) if display_temp is None: if point_temps: valid_tmax = [float(p.get("target_runway_max")) for p in point_temps if p.get("target_runway_max") is not None] display_temp = max(valid_tmax) if valid_tmax else None if display_temp is None: station_temp = None mgm_nearby = city_weather.get("mgm_nearby") or [] airport_icao = HIGH_FREQ_AIRPORT_ICAO.get(city, "") for row in mgm_nearby: if str(row.get("istNo") or "") == airport_icao or str(row.get("icao") or "") == airport_icao: station_temp = row.get("temp") break if station_temp is None and mgm_nearby: logger.warning( "airport message fallback city={}: station {} not found in mgm_nearby, falling back to current.temp", city, airport_icao, ) if station_temp is None: station_temp = (city_weather.get("current") or {}).get("temp") display_temp = station_temp # ── Heat model ── wind_dir = amos.get("wind_dir") if is_amsc else None slope_15m = _compute_slope_15m(amos_icao, display_temp, city) if is_amsc and display_temp is not None else None heat_signal = _runway_heat_signal(display_temp or 0, slope_15m, wind_dir, city, language) if is_amsc else "" wind_label = _wind_regime_label(city, wind_dir, language) if is_amsc and wind_dir is not None else None max_so_far, max_temp_time = _get_airport_daily_high(city_weather) # ── Build message ── lines: List[str] = [] # Header hashtag_line = _build_telegram_hashtag_line( "runway" if has_runway else "airport", city=city, language=language, ) icao_display = f"{amos_icao} · " if amos_icao else "" settlement_pair_label = ( str(settlement_endpoint.get("pair")) if settlement_endpoint is not None and settlement_endpoint.get("pair") else (f"{settlement_pair[0]}/{settlement_pair[1]}" if settlement_pair else "") ) settlement_str = f" · ★{settlement_pair_label}" if settlement_pair_label else "" header = f"{icao_display}{en_name} / {ap_name}{settlement_str}{time_suffix}" if ap_name else f"{icao_display}{en_name}{settlement_str}{time_suffix}" lines.append(hashtag_line) lines.append("") lines.append(header) # Heat signal if heat_signal: lines.append("") lines.append(heat_signal) if state: lines.append(state) # All runway detail block if has_runway: lines.append("") for i, ((r1, r2), (t, _d)) in enumerate(zip(runway_pairs, runway_temps)): if t is None: continue pts = point_temps[i] if i < len(point_temps) else {} tdz = pts.get("tdz_temp") mid = pts.get("mid_temp") end = pts.get("end_temp") is_settlement = _is_settlement_runway(city, r1, r2) marker = f" {_copy(language, '★Settlement', '★结算')}" if is_settlement else "" if tdz is not None or mid is not None or end is not None: line = f"{r1}/{r2}{marker} TDZ:{_fmt(tdz)} MID:{_fmt(mid)} END:{_fmt(end)}" settlement_line_endpoint = _settlement_endpoint_for_point(city, (r1, r2), pts) if is_settlement else None if settlement_line_endpoint is not None: line += f" settle:{float(settlement_line_endpoint['temp']):.1f}" lines.append(line) else: temp_symbol = str(city_weather.get("temp_symbol") or "°C").strip() lines.append(f"{r1}/{r2}{marker} {t:.1f}{temp_symbol}") # Summary stats lines.append("") temp_symbol = str(city_weather.get("temp_symbol") or "°C").strip() cur_str = f"{display_temp:.1f}{temp_symbol}" if display_temp is not None else "--" if city == "paris": if aeroweb_available and display_temp is not None: lines.append(f"{_copy(language, 'Current observation', '当前实况')}: {cur_str}") else: lines.append(_copy(language, "Current observation: unavailable", "当前实况：暂无")) else: if has_runway: current_label = ( _copy(language, "Settlement runway now", "结算跑道当前") if settlement_pair else _copy(language, "Runway now", "跑道当前") ) lines.append(f"{current_label}: {cur_str}") else: lines.append(f"{_copy(language, 'Current', '当前')}: {cur_str}") if city == "paris": if aeroweb_available and max_so_far is not None: time_str = f" ({max_temp_time})" if max_temp_time and not _is_zh(language) else (f"（{max_temp_time}）" if max_temp_time else "") lines.append(f"{_copy(language, 'Observed high', '日高实况')}: {max_so_far:.1f}{temp_symbol}{time_str}") elif not aeroweb_available: last_temp = _LAST_AEROWEB.get("temp") if last_temp is not None: last_time = _LAST_AEROWEB.get("max_time", "") time_str = f" ({last_time})" if last_time and not _is_zh(language) else (f"（{last_time}）" if last_time else "") lines.append(f"{_copy(language, 'Latest observation', '最近实况')}: {last_temp:.1f}{temp_symbol}{time_str}") elif max_so_far is not None: time_str = f" ({max_temp_time})" if max_temp_time and not _is_zh(language) else (f"（{max_temp_time}）" if max_temp_time else "") if has_runway: high_label = _copy(language, "Today's runway high", "今日跑道高点") lines.append(f"{high_label}: {max_so_far:.1f}{temp_symbol}{time_str}") else: high_label = _copy(language, "Today's high", "日高") lines.append(f"{high_label}: {max_so_far:.1f}{temp_symbol}{time_str}") if slope_15m is not None: sign = "+" if slope_15m >= 0 else "" lines.append(f"{_copy(language, '15m trend', '15分钟趋势')}: {sign}{slope_15m:.1f}°") if wind_dir is not None: wind_str = f"{_copy(language, 'Wind dir', '风向')}: {wind_dir}°" if wind_label: wind_str += f" {wind_label}" lines.append(wind_str) # --- AMSC METAR temp + time for Chinese cities (Beijing time) --- if is_amsc: raw_metar = amos.get("raw_metar") or "" if raw_metar: parts = raw_metar.split() # Extract temp/dew: "20/17" → 20 metar_temp = None for p in parts: m = re.match(r"^(M?\d{2})/(M?\d{2})$", p) if m: t = m.group(1) metar_temp = str(int(t.replace("M", "-"))) break # Extract time: "211930Z" → Beijing time (UTC+8) metar_time = None for p in parts: m = re.match(r"^(\d{2})(\d{2})(\d{2})Z$", p) if m: _day, hh, mm = int(m.group(1)), int(m.group(2)), m.group(3) bj_h = hh + 8 if bj_h >= 24: bj_h -= 24 metar_time = f"{_copy(language, 'Beijing time', '北京时')} {bj_h:02d}:{mm}" break if metar_temp or metar_time: bits = [] if metar_temp: bits.append(f"{metar_temp}{temp_symbol}") if metar_time: bits.append(metar_time) lines.append(f"{_copy(language, 'METAR', '报文')}: {' '.join(bits)}") if deb_pred is not None: if city == "paris": if aeroweb_available and display_temp is not None: diff = display_temp - deb_pred sign = "+" if diff >= 0 else "" suffix = _copy(language, f" (vs obs {sign}{diff:.1f}°)", f"（距实况 {sign}{diff:.1f}°）") lines.append(f"DEB: {deb_pred:.1f}{temp_symbol}{suffix}" if not _is_zh(language) else f"DEB：{deb_pred:.1f}{temp_symbol}{suffix}") elif not aeroweb_available: last_temp = _LAST_AEROWEB.get("temp") if last_temp is not None: diff = last_temp - deb_pred sign = "+" if diff >= 0 else "" suffix = _copy(language, f" (vs latest obs {sign}{diff:.1f}°)", f"（距最近实况 {sign}{diff:.1f}°）") lines.append(f"DEB: {deb_pred:.1f}{temp_symbol}{suffix}" if not _is_zh(language) else f"DEB：{deb_pred:.1f}{temp_symbol}{suffix}") else: lines.append(f"DEB: {deb_pred:.1f}{temp_symbol}" if not _is_zh(language) else f"DEB：{deb_pred:.1f}{temp_symbol}") else: lines.append(f"DEB: {deb_pred:.1f}{temp_symbol}" if not _is_zh(language) else f"DEB：{deb_pred:.1f}{temp_symbol}") elif display_temp is not None and display_temp > deb_pred: suffix = _copy(language, f" (already above +{display_temp - deb_pred:.1f}°)", f"（已突破 +{display_temp - deb_pred:.1f}°）") lines.append(f"DEB: {deb_pred:.1f}{temp_symbol}{suffix}" if not _is_zh(language) else f"DEB：{deb_pred:.1f}{temp_symbol}{suffix}") else: lines.append(f"DEB: {deb_pred:.1f}{temp_symbol}" if not _is_zh(language) else f"DEB：{deb_pred:.1f}{temp_symbol}") if city == "paris": lines.append("") if aeroweb_available and display_temp is not None: label = _copy(language, "AEROWEB airport obs", "AEROWEB 机场实况") lines.append(f"📡 {label}: {display_temp:.1f}{temp_symbol} · Météo-France") else: lines.append(_copy(language, "📡 AEROWEB airport obs: unavailable", "📡 AEROWEB 机场实况：暂不可用")) if arome_temp is not None: if aeroweb_available and display_temp is not None: d = arome_temp - display_temp sign = "+" if d >= 0 else "" label = _copy(language, "AROME HD 15m nowcast", "AROME HD 15分钟临近预报") suffix = _copy(language, f" (vs obs {sign}{d:.1f}°)", f"（较实况 {sign}{d:.1f}°）") lines.append(f"🕐 {label}: {arome_temp:.1f}{temp_symbol}{suffix}") else: label = _copy(language, "AROME HD 15m nowcast", "AROME HD 15分钟临近预报") lines.append(f"🕐 {label}: {arome_temp:.1f}{temp_symbol}") else: label = _copy(language, "AROME HD 15m nowcast", "AROME HD 15分钟临近预报") lines.append(f"🕐 {label}: --") if not aeroweb_available: lines.append("") lines.append(_copy(language, "Note: showing model reference only; observed high is not updating.", "提示：当前仅显示模型参考，不更新实况日高。")) elif source_label: lines.append("") lines.append(f"📡 {source_label}") # Model summary (compact) models = city_weather.get("multi_model") or {} if isinstance(models, dict) and len(models) >= 2: vals = sorted([(v, k) for k, v in models.items() if isinstance(v, (int, float))]) if len(vals) >= 2: lo, hi = vals[0][0], vals[-1][0] spread = hi - lo if spread <= 2.0: spread_label = _copy(language, "low dispersion", "低分歧") elif spread <= 4.0: spread_label = _copy(language, "moderate dispersion", "中等分歧") else: spread_label = _copy(language, "high dispersion", "高分歧") range_label = _copy(language, "Model range", "模型区间") lines.append(f"{range_label}: {lo:.1f}~{hi:.1f}{temp_symbol} {spread_label}") return "\n".join(lines) def _get_airport_daily_high(city_weather: Dict[str, Any]): """Get today's observed high from METAR/AMOS airport history.""" airport = city_weather.get("airport_current") or {} max_so_far = airport.get("max_so_far") max_time = airport.get("max_temp_time") if max_so_far is not None: try: max_so_far = round(float(max_so_far), 1) except Exception: max_so_far = None return max_so_far, max_time # Per-city push interval. The loop wakes every minute, but Telegram should # read recent city cache instead of acting as an upstream observation producer. _AIRPORT_PUSH_INTERVAL = { city: 600 for city in HIGH_FREQ_AIRPORT_CITIES } _AIRPORT_PUSH_INTERVAL.update({ "seoul": 60, "busan": 60, **{city: 60 for city in CHINA_HIGH_FREQ_AIRPORT_CITIES}, }) def _airport_push_cache_max_age_sec(city: str) -> int: interval = int(_AIRPORT_PUSH_INTERVAL.get((city or "").strip().lower(), 600) or 600) return max(90, interval * 2) def _cached_payload_observation_epoch(payload: Dict[str, Any]) -> Optional[int]: amos = payload.get("amos") or {} airport_primary = payload.get("airport_primary") or {} airport_current = payload.get("airport_current") or {} current = payload.get("current") or {} candidates = [ (payload.get("canonical_temperature") or {}).get("observed_at"), amos.get("observation_time"), airport_primary.get("obs_time"), airport_current.get("obs_time"), current.get("observed_at"), current.get("obs_time"), ] parsed = [ timestamp for timestamp in (_parse_observation_time_epoch(value) for value in candidates) if timestamp is not None ] return max(parsed) if parsed else None def _read_cached_airport_city_weather(city: str, max_age_sec: Optional[int] = None) -> Optional[Dict[str, Any]]: """Read web city cache for Telegram without triggering collection.""" normalized_city = (city or "").strip().lower() if not normalized_city: return None max_age = int(max_age_sec if max_age_sec is not None else _airport_push_cache_max_age_sec(normalized_city)) now_ts = time.time() candidates: List[Tuple[int, int, float, str, Dict[str, Any]]] = [] try: db = DBManager() for kind in ("full", "panel"): entry = db.get_city_cache(kind, normalized_city) if not isinstance(entry, dict): continue updated_at_ts = float(entry.get("updated_at_ts") or 0.0) payload = entry.get("payload") if updated_at_ts <= 0 or not isinstance(payload, dict): continue age_sec = now_ts - updated_at_ts is_fresh = age_sec <= max_age if age_sec > max_age: logger.debug( "airport push cache stale city={} kind={} age_sec={} max_age_sec={}", normalized_city, kind, round(age_sec, 1), max_age, ) observation_ts = _cached_payload_observation_epoch(payload) candidates.append( ( int(observation_ts or 0), 1 if is_fresh else 0, updated_at_ts, kind, dict(payload), ) ) except Exception as exc: logger.debug("airport push city cache read failed city={}: {}", normalized_city, exc) if candidates: observation_ts, is_fresh, updated_at_ts, kind, payload = max( candidates, key=lambda item: (item[0], item[1], item[2]), ) logger.debug( "airport push cache selected city={} kind={} age_sec={} observation_ts={} fresh={}", normalized_city, kind, round(max(0.0, now_ts - updated_at_ts), 1), observation_ts or None, bool(is_fresh), ) return payload return None def _attach_latest_raw_observation_payload( db: Any, city: str, payload: Dict[str, Any], ) -> Dict[str, Any]: canonical = payload.get("canonical_temperature") or {} source = str(canonical.get("source") or "").strip().lower() if source != "amsc_awos": return payload getter = getattr(db, "get_latest_raw_observation", None) if not callable(getter): return payload try: row = getter("amsc_awos", city) except Exception as exc: logger.debug("airport push latest raw observation read failed city={}: {}", city, exc) return payload raw_payload = row.get("payload") if isinstance(row, dict) else None if isinstance(raw_payload, dict) and raw_payload: payload = dict(payload) payload["amos"] = dict(raw_payload) return payload def _read_canonical_airport_city_weather(city: str, db: Optional[Any] = None) -> Optional[Dict[str, Any]]: normalized_city = (city or "").strip().lower() if not normalized_city: return None try: db = db or DBManager() getter = getattr(db, "get_canonical_temperature", None) if not callable(getter): return None row = getter(normalized_city) except Exception as exc: logger.debug("airport push canonical latest read failed city={}: {}", normalized_city, exc) return None if not isinstance(row, dict): return None canonical = row.get("payload") or row if not isinstance(canonical, dict): return None payload = build_city_weather_from_canonical(normalized_city, canonical) if not isinstance(payload, dict): return None return _attach_latest_raw_observation_payload(db, normalized_city, payload) def _merge_airport_push_context( latest_payload: Dict[str, Any], cached_payload: Optional[Dict[str, Any]], ) -> Dict[str, Any]: if not isinstance(cached_payload, dict) or not cached_payload: return latest_payload merged = dict(latest_payload) def has_useful_context(value: Any) -> bool: if isinstance(value, dict): return any(item not in (None, "", [], {}) for item in value.values()) if isinstance(value, list): return bool(value) return value is not None for key in ( "deb", "multi_model", "multi_model_daily", "forecast", "probabilities", "peak", "local_time", "local_date", "temp_symbol", "risk", ): value = cached_payload.get(key) if not has_useful_context(value): continue if key not in merged or not has_useful_context(merged.get(key)): merged[key] = value return merged def _load_airport_city_weather_for_push(city: str) -> Dict[str, Any]: cached = _read_cached_airport_city_weather(city) canonical = _read_canonical_airport_city_weather(city) if canonical is not None: cached_ts = _cached_payload_observation_epoch(cached or {}) canonical_ts = _cached_payload_observation_epoch(canonical) if cached is None or (canonical_ts or 0) > (cached_ts or 0): return _merge_airport_push_context(canonical, cached) if cached is not None: return cached raise RuntimeError(f"no cached city weather for airport push city={city}") # Per-city temperature window threshold (°C below DEB predicted high) # Continental airports: wider window (temp rises steadily over land) # Maritime airports: narrower (sea breeze moderates temp) # Strong sea breeze: tightest (marine air suppresses peak) _AIRPORT_HEAT_THRESHOLD = { "seoul": 3.0, "ankara": 3.0, "istanbul": 3.0, "paris": 3.0, "busan": 2.0, "tokyo": 2.0, "amsterdam": 2.0, "helsinki": 2.0, "hong kong": 1.5, "taipei": 1.5, } # 部分城市 Open-Meteo 算出的 peak 窗口偏窄，用 fallback 拓宽 # （例如沿海城市受海风影响，高温窗口被压缩） _AIRPORT_PEAK_FALLBACK = { "busan": (12, 16), } def _in_peak_time_window(city: str, city_weather: Dict[str, Any]) -> bool: """Check if current local time is within the expected peak temperature window.""" peak = city_weather.get("peak") or {} first_h = peak.get("first_h") last_h = peak.get("last_h") fallback = _AIRPORT_PEAK_FALLBACK.get(city) if fallback and ((first_h is None) or (last_h is not None and last_h - first_h < 3)): first_h, last_h = fallback local_time = city_weather.get("local_time") or "" if first_h is None or last_h is None or not local_time: return False try: current_h, current_m = int(local_time[:2]), int(local_time[3:5]) current_minutes = current_h * 60 + current_m # Window: first_h - 4h to last_h + 2h start_min = max(0, (first_h - 4) * 60) end_min = min(24 * 60 - 1, (last_h + 2) * 60) return start_min <= current_minutes <= end_min except Exception: return False def _check_rising_trend(icao: str) -> bool: """Check if temperature has been rising over the last 30-60 minutes.""" try: from src.database.db_manager import DBManager db = DBManager() obs = db.get_airport_obs_recent(icao, minutes=60) if not obs: return False temps = [r.get("temp_c") for r in obs if r.get("temp_c") is not None] if len(temps) < 4: return False # Check: last 3 readings are increasing recent = temps[-3:] if recent[2] > recent[1] > recent[0]: return True # Or: current > 30 min ago if len(temps) >= 4: mid = len(temps) // 2 if temps[-1] > temps[mid]: return True return False except Exception: return False def _process_airport_city( city: str, now_ts: int, last_city: dict, chat_ids: List[str], bot: Any, ) -> Optional[Tuple[str, dict]]: """Process one airport city and return (city, new_state_entry) or None. This is the per-city unit used by the concurrent thread pool in ``_run_high_freq_airport_cycle``. """ last_city_ts = int(last_city.get("ts") or 0) last_obs_time = str(last_city.get("obs_time") or "") last_obs_ts = ( _parse_observation_time_epoch(last_city.get("obs_ts")) or _parse_observation_time_epoch(last_obs_time) ) city_interval = _AIRPORT_PUSH_INTERVAL.get(city, 600) if now_ts - last_city_ts < city_interval: return None city_weather: Dict[str, Any] = {} deb_pred: Optional[float] = None try: city_weather = _load_airport_city_weather_for_push(city) deb_raw = (city_weather.get("deb") or {}).get("prediction") if deb_raw is not None: deb_pred = float(deb_raw) except Exception: logger.exception("airport analyze failed for city={}", city) return None # Extract airport-level temperature amos = city_weather.get("amos") or {} mgm_nearby = city_weather.get("mgm_nearby") or [] airport_icao = HIGH_FREQ_AIRPORT_ICAO.get(city, "") airport_row = None for row in mgm_nearby: if str(row.get("istNo") or "") == airport_icao or str(row.get("icao") or "") == airport_icao: airport_row = row break if not airport_row: airport_primary = city_weather.get("airport_primary") or {} if airport_primary.get("temp") is not None: airport_row = airport_primary else: current_fallback = city_weather.get("current") or {} if current_fallback.get("temp") is not None: airport_row = current_fallback else: logger.warning( "airport push skipped city={}: station {} not found in mgm_nearby, " "airport_primary, or current (mgm={} rows)", city, airport_icao, len(mgm_nearby), ) return None station_temp = airport_row.get("temp") if airport_row else None current_obs_time = str(airport_row.get("obs_time") or "") runway_obs = (amos.get("runway_obs") or {}) runway_pairs = runway_obs.get("runway_pairs") or [] runway_temps = runway_obs.get("temperatures") or [] runway_pairs, runway_temps, point_temps = _select_focus_runway_obs( city, runway_pairs, runway_temps, runway_obs.get("point_temperatures") or [], ) if runway_pairs and (runway_temps or point_temps): endpoint = _settlement_endpoint_from_obs(city, runway_pairs, point_temps) if endpoint is not None: station_temp = float(endpoint["temp"]) else: valid_temps = [t for (t, _d) in runway_temps if t is not None] if valid_temps: station_temp = max(valid_temps) amos_obs_time = amos.get("observation_time") or "" if amos_obs_time: current_obs_time = amos_obs_time current_temp = station_temp if current_temp is None: airport_primary = city_weather.get("airport_primary") or {} current_temp = airport_primary.get("temp") or (city_weather.get("current") or {}).get("temp") if not current_obs_time: current_obs_time = str(airport_primary.get("obs_time") or "") current_obs_ts = _parse_observation_time_epoch(current_obs_time) source_label = "" # human-readable data source for Paris messages arome_temp_val = None # AROME HD temperature for display (always fetched for comparison) aeroweb_available = False if city == "paris": arome_temp_val = _fetch_arome_temp() airport_primary = city_weather.get("airport_primary") or {} # Detect AEROWEB availability if airport_primary.get("source_code") == "aeroweb" and airport_primary.get("temp") is not None: aeroweb_available = True source_label = "AEROWEB 机场实况 · Météo-France" # Update last known AEROWEB cache _LAST_AEROWEB["temp"] = float(airport_primary["temp"]) _LAST_AEROWEB["ts"] = time.time() # Get daily high from airport data aero_max = (city_weather.get("airport_current") or {}).get("max_so_far") aero_max_time = (city_weather.get("airport_current") or {}).get("max_temp_time") if aero_max is not None: _LAST_AEROWEB["max_so_far"] = float(aero_max) _LAST_AEROWEB["max_time"] = str(aero_max_time or "")[11:16] if aero_max_time else "" # Allow Paris pushes even when AEROWEB is down (show cached/last-known data) if city == "paris" and not aeroweb_available and deb_pred is not None: pass elif current_temp is None or deb_pred is None: return None # Dedup: same observation → skip (with delayed retry for HKO) _CITIES_WITH_DELAYED_API = {"hong kong"} if (current_obs_time and last_obs_time and current_obs_time == last_obs_time and city in _CITIES_WITH_DELAYED_API and now_ts - last_city_ts > 540): time.sleep(4) try: city_weather = _load_airport_city_weather_for_push(city) deb_raw2 = (city_weather.get("deb") or {}).get("prediction") if deb_raw2 is not None: deb_pred = float(deb_raw2) mgm_nearby2 = city_weather.get("mgm_nearby") or [] row2 = None for r in mgm_nearby2: if str(r.get("istNo") or "") == airport_icao or str(r.get("icao") or "") == airport_icao: row2 = r break if not row2 and mgm_nearby2: row2 = mgm_nearby2[0] retry_obs = str(row2.get("obs_time") or "") if row2 else "" if retry_obs and retry_obs != last_obs_time: current_obs_time = retry_obs station_temp = row2.get("temp") if row2 else None current_temp = station_temp or (city_weather.get("current") or {}).get("temp") if current_temp is None or deb_pred is None: return None else: return None except Exception: return None current_obs_ts = _parse_observation_time_epoch(current_obs_time) if ( current_obs_ts is not None and last_obs_ts is not None and current_obs_ts <= last_obs_ts ): logger.debug( "airport push skipped stale observation city={} current_obs_time={} last_obs_time={}", city, current_obs_time, last_obs_time, ) return None if current_obs_time and last_obs_time and current_obs_time == last_obs_time: return None obs_local = ( ((city_weather.get("amos") or {}).get("observation_time_local") or "")[11:16] if len(str((city_weather.get("amos") or {}).get("observation_time_local") or "")) >= 16 else (city_weather.get("airport_current") or {}).get("obs_time") or city_weather.get("local_time") or "" ) message = _build_airport_status_message(city, city_weather, deb_pred, obs_local, state="", source_label=source_label, arome_temp=arome_temp_val, aeroweb_available=aeroweb_available) # Send to all target chats sent = False for chat_id in chat_ids: thread_id = 0 try: kwargs = {} thread_id = _resolve_thread_id(chat_id, city) if thread_id: kwargs["message_thread_id"] = thread_id elif _is_forum_chat_id(chat_id): logger.warning( "airport push skipped missing forum thread mapping city={} chat_id={} mapping_cities={}", city, chat_id, len(_load_city_thread_ids()), ) continue _rate_limited_send(bot, chat_id, message, **kwargs) sent = True except Exception as exc: if thread_id and "message thread not found" in str(exc).lower(): logger.warning( "airport push skipped missing forum thread city={} chat_id={} thread_id={}", city, chat_id, thread_id, ) continue logger.warning("airport push failed city={} chat_id={}: {}", city, chat_id, exc) if sent: logger.info("airport status pushed city={} temp={} deb={} obs_time={}", city, current_temp, deb_pred, current_obs_time) return ( city, { "ts": now_ts, "active": True, "obs_time": current_obs_time, "obs_ts": current_obs_ts, }, ) return None def _due_airport_cities( cities: Set[str], now_ts: int, last_by_city: Dict[str, Any], ) -> List[str]: due: List[str] = [] for city in sorted( cities, key=lambda item: (0 if item in CHINA_HIGH_FREQ_AIRPORT_CITIES else 1, item), ): last_city = last_by_city.get(city) or {} last_city_ts = int(last_city.get("ts") or 0) city_interval = _AIRPORT_PUSH_INTERVAL.get(city, 600) if now_ts - last_city_ts >= city_interval: due.append(city) return due def _run_high_freq_airport_cycle( bot: Any, config: Dict[str, Any], chat_ids: List[str], state: Dict[str, Any], ) -> bool: state_dirty = False now_ts = int(time.time()) last_by_city = state.setdefault("last_by_city", {}) max_workers = max(1, min(4, _env_int("TELEGRAM_AIRPORT_PUSH_MAX_WORKERS", 1))) cities = _due_airport_cities(HIGH_FREQ_AIRPORT_CITIES, now_ts, last_by_city) logger.info( "airport cycle tick cities={} due={} max_workers={}", len(HIGH_FREQ_AIRPORT_CITIES), len(cities), max_workers, ) if not cities: return False pool = _get_airport_executor(max_workers) futures = { pool.submit( _process_airport_city, city, now_ts, last_by_city.get(city) or {}, chat_ids, bot, ): city for city in cities } for future in as_completed(futures): try: result = future.result() except Exception: logger.exception("airport city task crashed city={}", futures[future]) continue if result is None: continue city, entry = result last_by_city[city] = entry state_dirty = True return state_dirty def start_high_freq_airport_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[threading.Thread]: enabled = _env_bool("TELEGRAM_AIRPORT_PUSH_ENABLED", True) chat_ids = get_telegram_chat_ids_from_env() if not enabled: logger.info("airport high-freq push loop disabled") return None if not chat_ids: logger.warning("airport high-freq push loop skipped: TELEGRAM_CHAT_IDS is not set") return None interval_sec = max(30, _env_int("TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC", 60)) def _runner() -> None: state = _load_airport_state() logger.info( "airport high-freq push loop started cities={} interval={}s chat_targets={}", len(HIGH_FREQ_AIRPORT_CITIES), interval_sec, len(chat_ids), ) while True: cycle_started = time.time() try: state = _load_airport_state() if _run_high_freq_airport_cycle( bot=bot, config=config, chat_ids=chat_ids, state=state, ): _save_airport_state(state) except Exception: logger.exception("airport push cycle crashed") elapsed = time.time() - cycle_started sleep_sec = max(5, interval_sec - int(elapsed)) time.sleep(sleep_sec) thread = threading.Thread( target=_runner, name="airport-high-freq-pusher", daemon=True, ) thread.start() logger.info("airport high-freq push loop thread started") return thread