diff --git a/src/bot/analysis/city_analysis_service.py b/src/bot/analysis/city_analysis_service.py index 06b51699..7a7c1ef1 100644 --- a/src/bot/analysis/city_analysis_service.py +++ b/src/bot/analysis/city_analysis_service.py @@ -1,13 +1,136 @@ from __future__ import annotations -from typing import Any +from datetime import datetime, timezone, timedelta +from typing import Any, Dict, Optional + +from src.database.db_manager import DBManager +from src.data_collection.city_registry import CITY_REGISTRY +from web.services.canonical_temperature import build_city_weather_from_canonical + + +def _safe_float(value: Any) -> Optional[float]: + try: + if value is None or value == "": + return None + return float(value) + except (TypeError, ValueError): + return None + + +def _today_for_city(city_name: str) -> str: + meta = CITY_REGISTRY.get(str(city_name or "").strip().lower()) or {} + offset = int(meta.get("tz_offset") or 0) + return datetime.now(timezone.utc).astimezone(timezone(timedelta(seconds=offset))).strftime("%Y-%m-%d") + + +def _first_forecast_high(payload: Dict[str, Any], city_name: str) -> Optional[float]: + deb = payload.get("deb") if isinstance(payload.get("deb"), dict) else {} + value = _safe_float(deb.get("prediction")) + if value is not None: + return value + + local_date = str(payload.get("local_date") or _today_for_city(city_name)).strip() + daily = payload.get("multi_model_daily") + if isinstance(daily, dict): + dates = [local_date] if local_date in daily else sorted(daily.keys()) + for date_key in dates: + entry = daily.get(date_key) + if not isinstance(entry, dict): + continue + daily_deb = entry.get("deb") if isinstance(entry.get("deb"), dict) else {} + value = _safe_float(daily_deb.get("prediction")) + if value is not None: + return value + models = entry.get("models") if isinstance(entry.get("models"), dict) else {} + values = sorted( + value + for value in (_safe_float(item) for item in models.values()) + if value is not None + ) + if values: + return values[len(values) // 2] + + current = payload.get("current") if isinstance(payload.get("current"), dict) else {} + for key in ("max_so_far", "max_temp_so_far", "temp"): + value = _safe_float(current.get(key)) + if value is not None: + return value + return None + + +def _city_cache_payload_to_weather_data(city_name: str, payload: Dict[str, Any]) -> Dict[str, Any]: + city_key = str(city_name or "").strip().lower() + meta = CITY_REGISTRY.get(city_key) or {} + offset = int(meta.get("tz_offset") or 0) + local_date = str(payload.get("local_date") or _today_for_city(city_key)).strip() + current = payload.get("current") if isinstance(payload.get("current"), dict) else {} + airport_current = ( + payload.get("airport_current") + if isinstance(payload.get("airport_current"), dict) + else current + ) + current_temp = _safe_float(current.get("temp")) + if current_temp is None: + current_temp = _safe_float(airport_current.get("temp")) + max_so_far = _safe_float(current.get("max_so_far")) + if max_so_far is None: + max_so_far = _safe_float(current.get("max_temp_so_far")) + if max_so_far is None: + max_so_far = _safe_float(airport_current.get("max_so_far")) + if max_so_far is None: + max_so_far = _safe_float(airport_current.get("max_temp_so_far")) + + forecast_high = _first_forecast_high(payload, city_key) + dates = [local_date] if local_date else [] + highs = [forecast_high] if forecast_high is not None else [] + observed_at = str( + current.get("observed_at") + or current.get("observation_time") + or airport_current.get("observation_time") + or airport_current.get("obs_time") + or "" + ).strip() + metar_current = { + "temp": current_temp, + "max_temp_so_far": max_so_far, + "max_temp_time": current.get("max_time") or current.get("max_temp_time"), + "wind_speed_kt": current.get("wind_speed_kt") or airport_current.get("wind_speed_kt"), + "wind_dir": current.get("wind_dir") or airport_current.get("wind_dir"), + "visibility_mi": current.get("visibility_mi") or airport_current.get("visibility_mi"), + "humidity": current.get("humidity") or airport_current.get("humidity"), + "clouds": current.get("clouds") or airport_current.get("clouds") or [], + "wx_desc": current.get("wx_desc") or airport_current.get("wx_desc"), + } + weather_data = { + "open-meteo": { + "utc_offset": offset, + "current": {"local_time": payload.get("local_time")}, + "daily": { + "time": dates, + "temperature_2m_max": highs, + }, + }, + "metar": { + "observation_time": observed_at, + "current": metar_current, + }, + "settlement_current": { + "observation_time": observed_at, + "current": dict(metar_current), + }, + "amos": payload.get("amos") if isinstance(payload.get("amos"), dict) else {}, + "deb": payload.get("deb") if isinstance(payload.get("deb"), dict) else {}, + "multi_model": payload.get("multi_model") if isinstance(payload.get("multi_model"), dict) else {}, + } + return weather_data class CityAnalysisService: """City analysis adapter with lazy imports to trim cold startup.""" - def __init__(self, weather: Any): + def __init__(self, weather: Any, cache_db: Optional[Any] = None): self.weather = weather + self.cache_db = cache_db or DBManager() def resolve_city(self, city_input: str): from src.analysis.city_query_service import resolve_city_name @@ -19,12 +142,11 @@ class CityAnalysisService: if not coords: raise ValueError(f"未找到城市坐标: {city_name}") - weather_data = self.weather.fetch_all_sources( - city_name, - lat=coords["lat"], - lon=coords["lon"], - force_refresh=True, - ) + payload = self._load_cached_city_payload(city_name) + if not payload: + self._enqueue_refresh(city_name) + raise RuntimeError("城市天气缓存正在初始化，请稍后重试") + weather_data = _city_cache_payload_to_weather_data(city_name, payload) from src.analysis.city_query_service import build_city_query_report @@ -34,3 +156,36 @@ class CityAnalysisService: city_query_cost=city_query_cost, ) + def _load_cached_city_payload(self, city_name: str) -> Optional[Dict[str, Any]]: + city_key = str(city_name or "").strip().lower() + getter = getattr(self.cache_db, "get_city_cache", None) + if callable(getter): + for kind in ("panel", "full", "summary"): + entry = getter(kind, city_key) + payload = entry.get("payload") if isinstance(entry, dict) else None + if isinstance(payload, dict): + return payload + + canonical_getter = getattr(self.cache_db, "get_canonical_temperature", None) + canonical_entry = canonical_getter(city_key) if callable(canonical_getter) else None + canonical = ( + canonical_entry.get("payload") + if isinstance(canonical_entry, dict) and isinstance(canonical_entry.get("payload"), dict) + else canonical_entry + ) + return ( + build_city_weather_from_canonical(city_key, canonical) + if isinstance(canonical, dict) + else None + ) + + def _enqueue_refresh(self, city_name: str) -> None: + enqueue = getattr(self.cache_db, "enqueue_observation_refresh_request", None) + if not callable(enqueue): + return + enqueue( + city=str(city_name or "").strip().lower(), + kind="panel", + priority="high", + reason="bot_city_report", + ) diff --git a/src/utils/daily_weather_report.py b/src/utils/daily_weather_report.py index 1a58e902..5cfd861f 100644 --- a/src/utils/daily_weather_report.py +++ b/src/utils/daily_weather_report.py @@ -3,7 +3,6 @@ from __future__ import annotations import os -import re import threading import time from datetime import datetime @@ -20,9 +19,11 @@ except Exception: ZoneInfo = None # type: ignore[assignment] +from src.database.db_manager import DBManager from src.data_collection.city_registry import CITY_REGISTRY from src.data_collection.weather_sources import WeatherDataCollector from src.utils.telegram_i18n import copy_text, normalize_push_language, telegram_push_language +from web.services.canonical_temperature import build_city_weather_from_canonical TARGET_CITIES: List[str] = [ "beijing", @@ -46,18 +47,7 @@ CITY_NAME_ZH: Dict[str, str] = { "qingdao": "青岛", } -# weather.com.cn city codes -CMA_CITY_CODES: Dict[str, str] = { - "beijing": "101010100", - "shanghai": "101020100", - "guangzhou": "101280101", - "chengdu": "101270101", - "chongqing": "101040100", - "wuhan": "101200101", - "qingdao": "101120201", -} - -_CMA_FORECAST_URL = "http://www.weather.com.cn/weather/{code}.shtml" +_DAILY_REPORT_DB = DBManager() def _env_bool(name: str, default: bool) -> bool: @@ -82,85 +72,119 @@ def _daily_report_language() -> str: ) -def _fetch_cma_forecast(city_key: str) -> Optional[Dict[str, Any]]: - """Scrape today's forecast from weather.com.cn (CMA).""" - code = CMA_CITY_CODES.get(city_key) - if not code: - return None - - url = _CMA_FORECAST_URL.format(code=code) +def _safe_float(value: Any) -> Optional[float]: try: - resp = httpx.get( - url, - headers={ - "User-Agent": ( - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " - "AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/120.0.0.0 Safari/537.36" - ), - }, - timeout=httpx.Timeout(timeout=10.0, connect=5.0, read=10.0), - follow_redirects=True, + if value is None or value == "": + return None + return float(value) + except (TypeError, ValueError): + return None + + +def _enqueue_daily_report_refresh(city_key: str) -> None: + enqueue = getattr(_DAILY_REPORT_DB, "enqueue_observation_refresh_request", None) + if not callable(enqueue): + return + try: + enqueue( + city=city_key, + kind="panel", + priority="normal", + reason="daily_weather_report", ) - resp.raise_for_status() - html = resp.text except Exception as exc: - logger.warning( - "daily_weather_report: CMA fetch failed for {}: {}", city_key, exc + logger.debug("daily_weather_report: refresh enqueue skipped city={}: {}", city_key, exc) + + +def _load_cached_city_payload(city_key: str) -> Optional[Dict[str, Any]]: + getter = getattr(_DAILY_REPORT_DB, "get_city_cache", None) + if callable(getter): + for kind in ("panel", "full", "summary"): + try: + entry = getter(kind, city_key) + except Exception as exc: + logger.debug( + "daily_weather_report: city cache read skipped city={} kind={}: {}", + city_key, + kind, + exc, + ) + continue + payload = entry.get("payload") if isinstance(entry, dict) else None + if isinstance(payload, dict): + return payload + + canonical_getter = getattr(_DAILY_REPORT_DB, "get_canonical_temperature", None) + if callable(canonical_getter): + try: + canonical_entry = canonical_getter(city_key) + except Exception as exc: + logger.debug("daily_weather_report: canonical read skipped city={}: {}", city_key, exc) + canonical_entry = None + canonical = ( + canonical_entry.get("payload") + if isinstance(canonical_entry, dict) and isinstance(canonical_entry.get("payload"), dict) + else canonical_entry ) - return None + payload = ( + build_city_weather_from_canonical(city_key, canonical) + if isinstance(canonical, dict) + else None + ) + if payload: + return payload - # Parse today's weather block from the 7-day forecast page. - # The HTML structure has entries like: - #

晴转多云

- #

25℃ / 19℃

- # We target the first occurrence (today). - - weather = _extract_first(html, r']*class="wea"[^>]*>([^<]+)

') - tem_text = _extract_first(html, r']*class="tem"[^>]*>(.+?)

') - - high_str: Optional[str] = None - low_str: Optional[str] = None - - if tem_text: - # Strip all HTML tags, then find numbers followed by degree - clean = re.sub(r"<[^>]+>", " ", tem_text) - nums = re.findall(r"(-?\d+)\s*(?:℃|°C|°c|°)?", clean) - if len(nums) >= 1: - high_str = nums[0] - if len(nums) >= 2: - low_str = nums[1] - - if not weather and not high_str: - return None - - result: Dict[str, Any] = {"source": "cma"} - if weather: - result["weather"] = weather.strip() - if high_str: - try: - result["forecast_high"] = float(high_str) - except (TypeError, ValueError): - result["forecast_high"] = None - if low_str: - try: - result["forecast_low"] = float(low_str) - except (TypeError, ValueError): - result["forecast_low"] = None - - logger.info( - "daily_weather_report: CMA parsed {} weather={} high={} low={}", - city_key, - result.get("weather"), - result.get("forecast_high"), - result.get("forecast_low"), - ) - return result + _enqueue_daily_report_refresh(city_key) + return None -def _extract_first(html: str, pattern: str) -> Optional[str]: - m = re.search(pattern, html, re.IGNORECASE | re.DOTALL) - return m.group(1) if m else None +def _daily_report_forecast_high(payload: Dict[str, Any]) -> Optional[float]: + deb = payload.get("deb") if isinstance(payload.get("deb"), dict) else {} + value = _safe_float(deb.get("prediction")) + if value is not None: + return value + + local_date = str(payload.get("local_date") or "").strip() + multi_model_daily = payload.get("multi_model_daily") + if isinstance(multi_model_daily, dict): + dates = [local_date] if local_date in multi_model_daily else sorted(multi_model_daily.keys()) + for date_key in dates: + entry = multi_model_daily.get(date_key) + if not isinstance(entry, dict): + continue + daily_deb = entry.get("deb") if isinstance(entry.get("deb"), dict) else {} + value = _safe_float(daily_deb.get("prediction")) + if value is not None: + return value + models = entry.get("models") if isinstance(entry.get("models"), dict) else {} + values = sorted( + value + for value in (_safe_float(item) for item in models.values()) + if value is not None + ) + if values: + return values[len(values) // 2] + + current = payload.get("current") if isinstance(payload.get("current"), dict) else {} + for key in ("max_so_far", "max_temp_so_far", "temp"): + value = _safe_float(current.get(key)) + if value is not None: + return value + return None + + +def _daily_report_weather_text(payload: Dict[str, Any]) -> str: + current = payload.get("current") if isinstance(payload.get("current"), dict) else {} + for value in ( + payload.get("weather"), + payload.get("weather_desc"), + current.get("wx_desc"), + current.get("weather"), + ): + text = str(value or "").strip() + if text: + return text + return "?" def _fetch_city_data( @@ -170,82 +194,32 @@ def _fetch_city_data( info = CITY_REGISTRY.get(city_key) name_en = str((info or {}).get("name") or city_key.title()) - # 1. Try CMA first for weather description - cma = _fetch_cma_forecast(city_key) - - # 2. Get Open-Meteo data for temperature fallback - om_high = None - if info: - try: - results = collector.fetch_all_sources( - city_key, - lat=info["lat"], - lon=info["lon"], - include_taf=False, - include_ensemble=False, - include_multi_model=False, - ) - if isinstance(results, dict): - om = results.get("open-meteo", {}) - daily = om.get("daily", {}) if isinstance(om, dict) else {} - daily_highs = daily.get("temperature_2m_max", []) or [] - om_high = daily_highs[0] if daily_highs else None - except Exception as exc: - logger.warning( - "daily_weather_report: OM fetch failed for {}: {}", city_key, exc - ) - - # Use CMA weather + CMA high, fall back to OM high - weather: Optional[str] = None - forecast_high: Optional[float] = None - - if cma: - weather = cma.get("weather") - forecast_high = cma.get("forecast_high") - if forecast_high is None: - forecast_high = om_high - - if not weather and not forecast_high: + payload = _load_cached_city_payload(city_key) + if not payload: return None + forecast_high = _daily_report_forecast_high(payload) + if forecast_high is None: + _enqueue_daily_report_refresh(city_key) + return None + weather = _daily_report_weather_text(payload) + logger.info( - "daily_weather_report: {} weather={} high={} (cma_ok={})", + "daily_weather_report: {} weather={} high={} cache_only=true", city_key, - weather or "?", + weather, forecast_high, - bool(cma and cma.get("weather")), ) return { "city": city_key, "name": name, "name_en": name_en, - "weather": weather or "?", + "weather": weather, "forecast_high": forecast_high, } -def _wmo_to_weather(code: Any) -> str: - """Translate WMO weather code to Chinese (fallback only).""" - try: - c = int(code or 0) - except (TypeError, ValueError): - return "未知" - if c == 0: - return "晴" - if 1 <= c <= 3: - return "多云" - if c in (45, 48): - return "雾" - if 51 <= c <= 67: - return "雨" - if 71 <= c <= 86: - return "雪" - if 95 <= c <= 99: - return "雷暴" - return "阴" - - def _build_ai_prompt( cities_data: List[Dict[str, Any]], report_date: str, diff --git a/tests/test_cache_warmer_service.py b/tests/test_cache_warmer_service.py index 2babdbbf..ee03317e 100644 --- a/tests/test_cache_warmer_service.py +++ b/tests/test_cache_warmer_service.py @@ -1,6 +1,6 @@ from datetime import datetime, timezone -from web.cache_warmer_service import CacheWarmer, build_priority_city_batch +from web.cache_warmer_service import CacheWarmer, build_default_cache_warmer, build_priority_city_batch def test_priority_city_batch_prefers_local_active_hours_over_night(): @@ -51,6 +51,37 @@ def test_cache_warmer_warms_scan_and_city_panel_without_force_refresh(): assert city_calls == [("alpha day", False)] +def test_default_cache_warmer_enqueues_city_refresh_without_direct_panel_refresh(monkeypatch): + import web.cache_warmer_service as cache_warmer_service + from web.services import city_runtime + + enqueued = [] + + class FakeDB: + @staticmethod + def enqueue_observation_refresh_request(**kwargs): + enqueued.append(kwargs) + return True + + def fail_refresh(*_args, **_kwargs): + raise AssertionError("default cache warmer must not directly refresh city panel") + + monkeypatch.setattr(cache_warmer_service, "_CACHE_WARMER_DB", FakeDB(), raising=False) + monkeypatch.setattr(city_runtime, "_refresh_city_panel_cache", fail_refresh) + + warmer = build_default_cache_warmer() + + assert warmer.city_panel_warmer("shenzhen", force_refresh=False) is True + assert enqueued == [ + { + "city": "shenzhen", + "kind": "panel", + "priority": "normal", + "reason": "cache_warmer", + } + ] + + def test_cache_warmer_skips_work_when_intervals_are_not_due(): now_ts = datetime(2026, 6, 14, 12, 0, tzinfo=timezone.utc).timestamp() scan_calls = [] diff --git a/tests/test_cached_weather_reports.py b/tests/test_cached_weather_reports.py new file mode 100644 index 00000000..a5cc5c1a --- /dev/null +++ b/tests/test_cached_weather_reports.py @@ -0,0 +1,181 @@ +import pytest + + +def test_daily_weather_report_reads_city_cache_without_external_fetch(monkeypatch): + from src.utils import daily_weather_report + + class FailCollector: + def fetch_all_sources(self, *_args, **_kwargs): + raise AssertionError("daily report must not fetch external weather sources") + + class FakeDB: + def __init__(self): + self.enqueued = [] + + def get_city_cache(self, kind, city): + if (kind, city) != ("panel", "beijing"): + return None + return { + "payload": { + "display_name": "Beijing", + "local_date": "2026-06-14", + "current": {"temp": 27.0, "max_so_far": 28.0}, + "deb": {"prediction": 31.5}, + } + } + + def enqueue_observation_refresh_request(self, **kwargs): + self.enqueued.append(kwargs) + return True + + fake_db = FakeDB() + monkeypatch.setattr(daily_weather_report, "_DAILY_REPORT_DB", fake_db, raising=False) + monkeypatch.setattr( + daily_weather_report.httpx, + "get", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("daily report must not scrape CMA from business path") + ), + ) + + data = daily_weather_report._fetch_city_data(FailCollector(), "beijing") + + assert data == { + "city": "beijing", + "name": "北京", + "name_en": "Beijing", + "weather": "?", + "forecast_high": 31.5, + } + assert fake_db.enqueued == [] + + +def test_daily_weather_report_cold_cache_enqueues_refresh_without_fetch(monkeypatch): + from src.utils import daily_weather_report + + class FailCollector: + def fetch_all_sources(self, *_args, **_kwargs): + raise AssertionError("daily report must not fetch external weather sources") + + class FakeDB: + def __init__(self): + self.enqueued = [] + + def get_city_cache(self, _kind, _city): + return None + + def get_canonical_temperature(self, _city): + return None + + def enqueue_observation_refresh_request(self, **kwargs): + self.enqueued.append(kwargs) + return True + + fake_db = FakeDB() + monkeypatch.setattr(daily_weather_report, "_DAILY_REPORT_DB", fake_db, raising=False) + monkeypatch.setattr( + daily_weather_report.httpx, + "get", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("daily report must not scrape CMA from business path") + ), + ) + + assert daily_weather_report._fetch_city_data(FailCollector(), "beijing") is None + assert fake_db.enqueued == [ + { + "city": "beijing", + "kind": "panel", + "priority": "normal", + "reason": "daily_weather_report", + } + ] + + +def test_city_analysis_service_builds_report_from_cached_payload_without_fetch(): + from src.bot.analysis.city_analysis_service import CityAnalysisService + + class FailWeather: + def get_coordinates(self, city_name): + assert city_name == "shanghai" + return {"lat": 31.1434, "lon": 121.8052} + + def fetch_all_sources(self, *_args, **_kwargs): + raise AssertionError("city report must not fetch external weather sources") + + class FakeDB: + def __init__(self): + self.enqueued = [] + + def get_city_cache(self, kind, city): + if (kind, city) != ("panel", "shanghai"): + return None + return { + "payload": { + "local_date": "2026-06-14", + "current": { + "temp": 29.2, + "max_so_far": 30.1, + "observed_at": "2026-06-14T05:00:00+00:00", + }, + "airport_current": { + "temp": 29.2, + "max_so_far": 30.1, + "observation_time": "2026-06-14T05:00:00+00:00", + }, + "deb": {"prediction": 32.4}, + } + } + + def enqueue_observation_refresh_request(self, **kwargs): + self.enqueued.append(kwargs) + return True + + service = CityAnalysisService(weather=FailWeather(), cache_db=FakeDB()) + + report = service.build_city_report("shanghai", 3) + + assert "Shanghai" in report + assert "32.4°C" in report + assert "29.2°C" in report + assert "本次消耗 3 积分" in report + + +def test_city_analysis_service_cold_cache_enqueues_and_asks_to_retry(): + from src.bot.analysis.city_analysis_service import CityAnalysisService + + class FailWeather: + def get_coordinates(self, _city_name): + return {"lat": 31.1434, "lon": 121.8052} + + def fetch_all_sources(self, *_args, **_kwargs): + raise AssertionError("city report must not fetch external weather sources") + + class FakeDB: + def __init__(self): + self.enqueued = [] + + def get_city_cache(self, _kind, _city): + return None + + def get_canonical_temperature(self, _city): + return None + + def enqueue_observation_refresh_request(self, **kwargs): + self.enqueued.append(kwargs) + return True + + fake_db = FakeDB() + service = CityAnalysisService(weather=FailWeather(), cache_db=fake_db) + + with pytest.raises(RuntimeError, match="缓存正在初始化"): + service.build_city_report("shanghai", 3) + + assert fake_db.enqueued == [ + { + "city": "shanghai", + "kind": "panel", + "priority": "high", + "reason": "bot_city_report", + } + ] diff --git a/tests/test_canonical_temperature.py b/tests/test_canonical_temperature.py index 4f60734f..14014fc7 100644 --- a/tests/test_canonical_temperature.py +++ b/tests/test_canonical_temperature.py @@ -89,7 +89,7 @@ def test_refresh_city_panel_cache_persists_canonical_temperature(monkeypatch): def test_city_panel_cold_cache_returns_canonical_latest_without_sync_refresh(monkeypatch): import web.services.city_api as city_api - started = [] + enqueued = [] class FakeDB: def get_city_cache(self, kind, city): @@ -113,8 +113,12 @@ def test_city_panel_cold_cache_returns_canonical_latest_without_sync_refresh(mon "freshness_status": "fresh", "fetched_at": "2026-06-14T01:02:03+00:00", "confidence": 0.92, + } } - } + + def enqueue_observation_refresh_request(self, **kwargs): + enqueued.append(kwargs) + return True def fail_refresh(*_args, **_kwargs): raise AssertionError("cold panel cache must not synchronously refresh when canonical latest exists") @@ -122,11 +126,6 @@ def test_city_panel_cold_cache_returns_canonical_latest_without_sync_refresh(mon monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower()) monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeDB()) monkeypatch.setattr(city_api.legacy_routes, "_refresh_city_panel_cache", fail_refresh) - monkeypatch.setattr( - city_api, - "_start_city_cache_stale_refresh", - lambda city, kind, refresh_fn: started.append((city, kind, refresh_fn)), - ) payload = asyncio.run( city_api.get_city_detail_payload( @@ -140,13 +139,20 @@ def test_city_panel_cold_cache_returns_canonical_latest_without_sync_refresh(mon assert payload["current"]["temp"] == 31.2 assert payload["canonical_temperature"]["source"] == "amsc_awos" assert payload["detail_depth"] == "panel" - assert [(city, kind) for city, kind, _ in started] == [("shanghai", "panel")] + assert enqueued == [ + { + "city": "shanghai", + "kind": "panel", + "priority": "high", + "reason": "canonical_fallback", + } + ] def test_city_full_cold_cache_returns_canonical_latest_without_sync_refresh(monkeypatch): import web.services.city_api as city_api - started = [] + enqueued = [] class FakeDB: def get_city_cache(self, kind, city): @@ -170,28 +176,38 @@ def test_city_full_cold_cache_returns_canonical_latest_without_sync_refresh(monk "freshness_status": "fresh", "fetched_at": "2026-06-14T01:02:03+00:00", "confidence": 0.92, + } } - } + + def enqueue_observation_refresh_request(self, **kwargs): + enqueued.append(kwargs) + return True def fail_refresh(*_args, **_kwargs): raise AssertionError("cold full cache must not synchronously refresh when canonical latest exists") monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeDB()) monkeypatch.setattr(city_api.legacy_routes, "_refresh_city_full_cache", fail_refresh) - monkeypatch.setattr(city_api, "_start_city_full_stale_refresh", lambda city: started.append(city)) payload = asyncio.run(city_api._get_city_full_data("shanghai", force_refresh=False)) assert payload["current"]["temp"] == 31.2 assert payload["canonical_temperature"]["source"] == "amsc_awos" assert payload["detail_depth"] == "full" - assert started == ["shanghai"] + assert enqueued == [ + { + "city": "shanghai", + "kind": "full", + "priority": "high", + "reason": "canonical_fallback", + } + ] def test_city_nearby_and_market_cold_cache_return_canonical_latest_without_sync_refresh(monkeypatch): import web.services.city_api as city_api - started = [] + enqueued = [] requested_kinds = [] class FakeDB: @@ -215,8 +231,12 @@ def test_city_nearby_and_market_cold_cache_return_canonical_latest_without_sync_ "freshness_status": "fresh", "fetched_at": "2026-06-14T01:02:03+00:00", "confidence": 0.92, + } } - } + + def enqueue_observation_refresh_request(self, **kwargs): + enqueued.append(kwargs) + return True def fail_refresh(*_args, **_kwargs): raise AssertionError("cold city cache must not synchronously refresh when canonical latest exists") @@ -226,11 +246,6 @@ def test_city_nearby_and_market_cold_cache_return_canonical_latest_without_sync_ monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeDB()) monkeypatch.setattr(city_api.legacy_routes, "_refresh_city_nearby_cache", fail_refresh) monkeypatch.setattr(city_api.legacy_routes, "_refresh_city_market_cache", fail_refresh) - monkeypatch.setattr( - city_api, - "_start_city_cache_stale_refresh", - lambda city, kind, refresh_fn: started.append((city, kind, refresh_fn)), - ) nearby = asyncio.run( city_api.get_city_detail_payload(object(), "Shanghai", force_refresh=False, depth="nearby") @@ -243,9 +258,19 @@ def test_city_nearby_and_market_cold_cache_return_canonical_latest_without_sync_ assert market["detail_depth"] == "market" assert nearby["current"]["temp"] == 31.2 assert market["current"]["temp"] == 31.2 - assert [(city, kind) for city, kind, _ in started] == [ - ("shanghai", "nearby"), - ("shanghai", "market"), + assert enqueued == [ + { + "city": "shanghai", + "kind": "nearby", + "priority": "high", + "reason": "canonical_fallback", + }, + { + "city": "shanghai", + "kind": "market", + "priority": "high", + "reason": "canonical_fallback", + }, ] assert requested_kinds == [("nearby", "shanghai"), ("market", "shanghai")] diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index f936f86f..aafbd802 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -1716,13 +1716,14 @@ def test_stale_city_detail_uses_cached_full_payload_while_refreshing(monkeypatch assert result["live_temp"] == 20.0 assert build_inputs == [20.0] - assert refresh_calls == 1 + assert refresh_calls == 0 def test_force_refresh_panel_returns_cached_payload_when_refresh_is_slow(monkeypatch): import asyncio refresh_calls = 0 + enqueued = [] class FakeCache: def get_city_cache(self, kind, city): @@ -1736,6 +1737,10 @@ def test_force_refresh_panel_returns_cached_payload_when_refresh_is_slow(monkeyp }, } + def enqueue_observation_refresh_request(self, **kwargs): + enqueued.append(kwargs) + return True + async def fake_run_in_threadpool(fn, *args, **kwargs): if fn is city_api.legacy_routes._refresh_city_panel_cache: await asyncio.sleep(0.05) @@ -1767,13 +1772,22 @@ def test_force_refresh_panel_returns_cached_payload_when_refresh_is_slow(monkeyp assert result["from_cache"] is True assert result["deb"]["prediction"] == 20.0 - assert refresh_calls == 1 + assert refresh_calls == 0 + assert enqueued == [ + { + "city": "paris", + "kind": "panel", + "priority": "high", + "reason": "force_refresh", + } + ] def test_force_refresh_panel_returns_cached_payload_when_refresh_already_running(monkeypatch): import asyncio refresh_calls = 0 + enqueued = [] class FakeCache: def get_city_cache(self, kind, city): @@ -1787,6 +1801,10 @@ def test_force_refresh_panel_returns_cached_payload_when_refresh_already_running }, } + def enqueue_observation_refresh_request(self, **kwargs): + enqueued.append(kwargs) + return True + async def fake_run_in_threadpool(fn, *args, **kwargs): if fn is city_api.legacy_routes._refresh_city_panel_cache: await asyncio.sleep(0.08) @@ -1827,16 +1845,31 @@ def test_force_refresh_panel_returns_cached_payload_when_refresh_already_running first_result, second_result = asyncio.run(run_requests()) - assert first_result["from_cache"] is False + assert first_result["from_cache"] is True assert second_result["from_cache"] is True assert second_result["deb"]["prediction"] == 20.0 - assert refresh_calls == 1 + assert refresh_calls == 0 + assert enqueued == [ + { + "city": "paris", + "kind": "panel", + "priority": "high", + "reason": "force_refresh", + }, + { + "city": "paris", + "kind": "panel", + "priority": "high", + "reason": "force_refresh", + }, + ] def test_stale_panel_returns_cached_payload_while_refreshing(monkeypatch): import asyncio refresh_calls = 0 + enqueued = [] class FakeCache: def get_city_cache(self, kind, city): @@ -1850,6 +1883,10 @@ def test_stale_panel_returns_cached_payload_while_refreshing(monkeypatch): }, } + def enqueue_observation_refresh_request(self, **kwargs): + enqueued.append(kwargs) + return True + async def fake_run_in_threadpool(fn, *args, **kwargs): if fn is city_api.legacy_routes._refresh_city_panel_cache: await asyncio.sleep(0.05) @@ -1883,7 +1920,15 @@ def test_stale_panel_returns_cached_payload_while_refreshing(monkeypatch): assert result["from_cache"] is True assert result["deb"]["prediction"] == 20.0 - assert refresh_calls == 1 + assert refresh_calls == 0 + assert enqueued == [ + { + "city": "paris", + "kind": "panel", + "priority": "high", + "reason": "stale_refresh", + } + ] def test_force_refresh_full_detail_returns_cached_payload_when_refresh_is_slow(monkeypatch): @@ -1891,6 +1936,7 @@ def test_force_refresh_full_detail_returns_cached_payload_when_refresh_is_slow(m refresh_calls = 0 build_inputs = [] + enqueued = [] class FakeCache: def get_city_cache(self, kind, city): @@ -1903,6 +1949,10 @@ def test_force_refresh_full_detail_returns_cached_payload_when_refresh_is_slow(m }, } + def enqueue_observation_refresh_request(self, **kwargs): + enqueued.append(kwargs) + return True + async def fake_run_in_threadpool(fn, *args, **kwargs): if fn is city_api.legacy_routes._refresh_city_full_cache: await asyncio.sleep(0.05) @@ -1948,7 +1998,15 @@ def test_force_refresh_full_detail_returns_cached_payload_when_refresh_is_slow(m assert result["live_temp"] == 20.0 assert build_inputs == [20.0] - assert refresh_calls == 1 + assert refresh_calls == 0 + assert enqueued == [ + { + "city": "paris", + "kind": "full", + "priority": "high", + "reason": "force_refresh", + } + ] def test_force_refresh_cold_city_detail_returns_initializing_without_full_refresh(monkeypatch): diff --git a/web/cache_warmer_service.py b/web/cache_warmer_service.py index eaa8b8e9..8e58981c 100644 --- a/web/cache_warmer_service.py +++ b/web/cache_warmer_service.py @@ -10,6 +10,8 @@ from typing import Any, Callable, Iterable, List, Mapping, Optional, Sequence from loguru import logger +from src.database.db_manager import DBManager + ASIA_CORE_CITIES = ( "hong kong", @@ -62,6 +64,7 @@ SOURCE_PRIORITY_WEIGHT = { "metar": 6, "wunderground": 4, } +_CACHE_WARMER_DB = DBManager() def _env_bool(name: str, default: bool) -> bool: @@ -257,16 +260,25 @@ class CacheWarmer: return completed +def _queue_city_panel_refresh(city: str, *, force_refresh: bool = False) -> bool: + _CACHE_WARMER_DB.enqueue_observation_refresh_request( + city=city, + kind="panel", + priority="normal", + reason="cache_warmer", + ) + return True + + def build_default_cache_warmer() -> CacheWarmer: from web.core import CITIES from web.scan_terminal_service import build_scan_terminal_payload - from web.services.city_runtime import _refresh_city_panel_cache hot_cities = _parse_city_list(os.getenv("POLYWEATHER_WARMER_HOT_CITIES")) return CacheWarmer( city_provider=lambda: CITIES, scan_warmer=build_scan_terminal_payload, - city_panel_warmer=_refresh_city_panel_cache, + city_panel_warmer=_queue_city_panel_refresh, scan_interval_sec=_env_int("POLYWEATHER_WARMER_SCAN_INTERVAL_SEC", 300), city_interval_sec=_env_int("POLYWEATHER_WARMER_CITY_INTERVAL_SEC", 60), city_batch_size=_env_int("POLYWEATHER_WARMER_CITY_BATCH_SIZE", 8), diff --git a/web/services/city_api.py b/web/services/city_api.py index 61d26dea..a2642d1a 100644 --- a/web/services/city_api.py +++ b/web/services/city_api.py @@ -195,15 +195,11 @@ def _request_city_cache_refresh( kind: str, refresh_fn: Callable[[str, bool], Dict[str, Any]], ) -> None: - if _enqueue_collector_refresh_request(city, kind): - return - _start_city_cache_stale_refresh(city, kind, refresh_fn) + _enqueue_collector_refresh_request(city, kind) def _request_city_full_refresh(city: str) -> None: - if _enqueue_collector_refresh_request(city, "full"): - return - _start_city_full_stale_refresh(city) + _enqueue_collector_refresh_request(city, "full") def _build_initializing_city_payload(city: str, *, detail_depth: str) -> Dict[str, Any]: @@ -298,44 +294,13 @@ async def _refresh_city_payload_with_stale_timeout( return canonical_payload return _queue_and_build_initializing_city_payload(city, kind=kind) - task, started = await _get_or_start_city_force_refresh_task(f"{kind}:{city}", refresh_factory) - if not started: - if cached_before_refresh: - logger.warning( - "city force refresh already running city={} kind={}; returning stale cache", - city, - kind, - ) - return await _overlay_cached_wunderground(city, cached_before_refresh) - canonical_payload = await _get_canonical_city_payload(city, detail_depth=kind) - if canonical_payload: - _enqueue_collector_refresh_request(city, kind, reason="force_refresh") - logger.warning( - "city force refresh returning canonical latest while refresh runs city={} kind={}", - city, - kind, - ) - return canonical_payload - timeout_sec = _city_force_refresh_timeout_sec() - try: - return await asyncio.wait_for(asyncio.shield(task), timeout=timeout_sec) - except asyncio.TimeoutError: - cached_payload = await _get_cached_city_payload(city, kind) - if cached_payload: - logger.warning( - "city force refresh timed out city={} kind={} timeout_sec={}; returning stale cache", - city, - kind, - timeout_sec, - ) - return await _overlay_cached_wunderground(city, cached_payload) - return await task - except Exception: - cached_payload = await _get_cached_city_payload(city, kind) - if cached_payload: - logger.warning("city force refresh failed city={} kind={}; returning stale cache", city, kind) - return await _overlay_cached_wunderground(city, cached_payload) - raise + _enqueue_collector_refresh_request(city, kind, reason="force_refresh") + logger.warning( + "city force refresh queued collector refresh city={} kind={}; returning cached payload", + city, + kind, + ) + return await _overlay_cached_wunderground(city, cached_before_refresh) async def _refresh_city_cache_with_stale_timeout( @@ -359,26 +324,7 @@ def _start_city_cache_stale_refresh( cache_kind = str(kind or "").strip().lower() if not normalized or not cache_kind: return - key = f"{cache_kind}:{normalized}" - existing = _CITY_STALE_REFRESH_TASKS.get(key) - if existing is not None and not existing.done(): - return - - async def _run_refresh() -> Dict[str, Any]: - return await run_in_threadpool(refresh_fn, normalized, False) - - task = asyncio.create_task(_run_refresh()) - _CITY_STALE_REFRESH_TASKS[key] = task - - def _cleanup(done: "asyncio.Task[Dict[str, Any]]") -> None: - if _CITY_STALE_REFRESH_TASKS.get(key) is done: - _CITY_STALE_REFRESH_TASKS.pop(key, None) - try: - done.result() - except Exception as exc: # pragma: no cover - defensive background guard - logger.warning("city stale refresh failed city={} kind={}: {}", normalized, cache_kind, exc) - - task.add_done_callback(_cleanup) + _enqueue_collector_refresh_request(normalized, cache_kind, reason="stale_refresh") async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Dict[str, Any]: @@ -500,22 +446,7 @@ def _start_city_full_stale_refresh(city: str) -> None: normalized = str(city or "").strip().lower() if not normalized: return - existing = _CITY_FULL_STALE_REFRESH_TASKS.get(normalized) - if existing is not None and not existing.done(): - return - - task = asyncio.create_task(_refresh_city_full_data(city, False)) - _CITY_FULL_STALE_REFRESH_TASKS[normalized] = task - - def _cleanup(done: "asyncio.Task[Dict[str, Any]]") -> None: - if _CITY_FULL_STALE_REFRESH_TASKS.get(normalized) is done: - _CITY_FULL_STALE_REFRESH_TASKS.pop(normalized, None) - try: - done.result() - except Exception as exc: # pragma: no cover - defensive background guard - logger.warning("city full stale refresh failed city={}: {}", city, exc) - - task.add_done_callback(_cleanup) + _enqueue_collector_refresh_request(normalized, "full", reason="stale_refresh") async def _get_city_full_data(city: str, *, force_refresh: bool) -> Dict[str, Any]: