diff --git a/.env.example b/.env.example index 51d41df2..559660f5 100644 --- a/.env.example +++ b/.env.example @@ -59,7 +59,7 @@ TELEGRAM_AIRPORT_PUSH_LANGUAGE=both POLYWEATHER_BOT_CPUS=0.75 POLYWEATHER_BOT_MEM_LIMIT=768m POLYWEATHER_BOT_MEMSWAP_LIMIT=1g -POLYWEATHER_BOT_AIRPORT_PUSH_INTERVAL_SEC=180 +POLYWEATHER_BOT_AIRPORT_PUSH_INTERVAL_SEC=60 POLYWEATHER_BOT_AIRPORT_PUSH_MAX_WORKERS=1 ######################################## diff --git a/docker-compose.yml b/docker-compose.yml index 149f4a82..73176cfe 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -33,7 +33,7 @@ services: POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false' POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false' POLYWEATHER_SERVICE_ROLE: bot - TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC: ${POLYWEATHER_BOT_AIRPORT_PUSH_INTERVAL_SEC:-180} + TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC: ${POLYWEATHER_BOT_AIRPORT_PUSH_INTERVAL_SEC:-60} TELEGRAM_AIRPORT_PUSH_MAX_WORKERS: ${POLYWEATHER_BOT_AIRPORT_PUSH_MAX_WORKERS:-1} healthcheck: interval: 60s @@ -105,7 +105,7 @@ services: POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY:-2} POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS:-8000} POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC:-60} - POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180} + POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-60} POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC:-60} POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false' POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC:-600} @@ -156,7 +156,7 @@ services: environment: POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ${POLYWEATHER_COLLECTOR_PATCH_ENDPOINT:-http://polyweather_web:8000/api/internal/collector-patch} POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC:-60} - POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180} + POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-60} POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC:-60} POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'true' POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC:-600} diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index da2eba88..8d00141f 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -287,6 +287,22 @@ def _parse_iso_datetime_utc(value: Any) -> Optional[datetime]: 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()) @@ -598,6 +614,9 @@ HIGH_FREQ_AIRPORT_CITIES = { "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", @@ -1381,13 +1400,7 @@ _AIRPORT_PUSH_INTERVAL = { _AIRPORT_PUSH_INTERVAL.update({ "seoul": 60, "busan": 60, - "beijing": 180, - "shanghai": 180, - "guangzhou": 180, - "qingdao": 180, - "chengdu": 180, - "chongqing": 180, - "wuhan": 180, + **{city: 60 for city in CHINA_HIGH_FREQ_AIRPORT_CITIES}, }) @@ -1534,6 +1547,10 @@ def _process_airport_city( """ 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 @@ -1601,6 +1618,7 @@ def _process_airport_city( 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 @@ -1656,7 +1674,20 @@ def _process_airport_city( return None except Exception: return None - elif current_obs_time and last_obs_time and current_obs_time == last_obs_time: + 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 = ( @@ -1684,7 +1715,15 @@ def _process_airport_city( 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}) + return ( + city, + { + "ts": now_ts, + "active": True, + "obs_time": current_obs_time, + "obs_ts": current_obs_ts, + }, + ) return None @@ -1695,7 +1734,10 @@ def _due_airport_cities( last_by_city: Dict[str, Any], ) -> List[str]: due: List[str] = [] - for city in sorted(cities): + 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) diff --git a/tests/test_deployment_runtime_config.py b/tests/test_deployment_runtime_config.py index 687f31be..ccc29958 100644 --- a/tests/test_deployment_runtime_config.py +++ b/tests/test_deployment_runtime_config.py @@ -110,7 +110,8 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services(): in collector_block ) assert "command: python -m web.observation_collector_worker" in collector_block - assert "POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180}" in collector_block + assert "POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-60}" in collector_block + assert "TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC: ${POLYWEATHER_BOT_AIRPORT_PUSH_INTERVAL_SEC:-60}" in bot_block assert "POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300}" in collector_block diff --git a/tests/test_telegram_hashtags.py b/tests/test_telegram_hashtags.py index 7b834a9f..2c228e51 100644 --- a/tests/test_telegram_hashtags.py +++ b/tests/test_telegram_hashtags.py @@ -4,6 +4,8 @@ from src.utils.telegram_push import ( _AIRPORT_PUSH_INTERVAL, _build_airport_status_message, _compute_slope_15m, + _due_airport_cities, + _parse_observation_time_epoch, _run_high_freq_airport_cycle, _telegram_push_language, ) @@ -170,16 +172,31 @@ def test_shenzhen_is_in_high_freq_push_as_hko_station(): assert HIGH_FREQ_AIRPORT_ICAO["shenzhen"] == "LFS" -def test_china_airport_push_defaults_to_three_minute_city_interval(): +def test_china_airport_push_defaults_to_one_minute_city_interval(): assert _AIRPORT_PUSH_INTERVAL["seoul"] == 60 assert _AIRPORT_PUSH_INTERVAL["busan"] == 60 - assert _AIRPORT_PUSH_INTERVAL["shanghai"] == 180 - assert _AIRPORT_PUSH_INTERVAL["beijing"] == 180 - assert _AIRPORT_PUSH_INTERVAL["guangzhou"] == 180 - assert _AIRPORT_PUSH_INTERVAL["qingdao"] == 180 - assert _AIRPORT_PUSH_INTERVAL["chengdu"] == 180 - assert _AIRPORT_PUSH_INTERVAL["chongqing"] == 180 - assert _AIRPORT_PUSH_INTERVAL["wuhan"] == 180 + assert _AIRPORT_PUSH_INTERVAL["shanghai"] == 60 + assert _AIRPORT_PUSH_INTERVAL["beijing"] == 60 + assert _AIRPORT_PUSH_INTERVAL["guangzhou"] == 60 + assert _AIRPORT_PUSH_INTERVAL["qingdao"] == 60 + assert _AIRPORT_PUSH_INTERVAL["chengdu"] == 60 + assert _AIRPORT_PUSH_INTERVAL["chongqing"] == 60 + assert _AIRPORT_PUSH_INTERVAL["wuhan"] == 60 + + +def test_airport_push_prioritizes_china_markets(): + due = _due_airport_cities( + {"paris", "shanghai", "wuhan", "ankara", "beijing"}, + now_ts=1000, + last_by_city={}, + ) + + assert due[:3] == ["beijing", "shanghai", "wuhan"] + + +def test_airport_push_normalizes_observation_times_for_stale_rejection(): + assert _parse_observation_time_epoch("1781161200") == 1781161200 + assert _parse_observation_time_epoch("2026-06-11T07:10:00+00:00") == 1781161800 def test_high_freq_airport_push_prefers_fresh_city_cache(monkeypatch): @@ -321,6 +338,38 @@ def test_high_freq_airport_cycle_skips_cities_before_interval(monkeypatch): assert calls == [] +def test_airport_push_rejects_observation_older_than_last_push(monkeypatch): + import src.utils.telegram_push as telegram_push + + monkeypatch.setattr( + telegram_push, + "_load_airport_city_weather_for_push", + lambda _city: { + "local_time": "15:22", + "current": {"temp": 31.0}, + "deb": {"prediction": 32.0}, + "airport_primary": { + "temp": 31.0, + "obs_time": "2026-06-11T06:56:00+00:00", + }, + }, + ) + + result = telegram_push._process_airport_city( + "shanghai", + now_ts=1781162600, + last_city={ + "ts": 1781161800, + "obs_time": "2026-06-11T07:10:00+00:00", + "obs_ts": 1781161800, + }, + chat_ids=["chat-1"], + bot=object(), + ) + + assert result is None + + def test_high_freq_airport_push_workers_default_to_one_for_shared_cpu(monkeypatch): source = Path("src/utils/telegram_push.py").read_text(encoding="utf-8") assert 'TELEGRAM_AIRPORT_PUSH_MAX_WORKERS", 1' in source