Stop legacy city refreshes from fetching sources

2026-06-14 22:51:15 +08:00
parent a9e9eed2f1
commit 7fb82c5eba
4 changed files with 247 additions and 47 deletions
@@ -99,7 +99,7 @@ def test_refresh_city_panel_cache_persists_canonical_temperature(monkeypatch):
        lambda city, force_refresh=False, detail_mode="panel": _sample_panel_payload(),
    )

-    payload = city_runtime._refresh_city_panel_cache("shanghai")
+    payload = city_runtime._refresh_city_panel_cache("shanghai", allow_external_fetch=True)

    assert payload["canonical_temperature"]["value"] == 31.2
    assert payload["canonical_temperature"]["source"] == "amsc_awos"
@@ -109,6 +109,46 @@ def test_refresh_city_panel_cache_persists_canonical_temperature(monkeypatch):
    assert writes["canonical"][1]["observed_at"] == "2026-06-14T01:01:00+00:00"


+def test_refresh_city_panel_cache_defaults_to_queue_without_sync_analyze(monkeypatch):
+    import web.services.city_runtime as city_runtime
+
+    enqueued = []
+
+    class FakeDB:
+        def get_city_cache(self, kind, city):
+            assert kind == "panel"
+            assert city == "shanghai"
+            return None
+
+        def get_canonical_temperature(self, city):
+            assert city == "shanghai"
+            return None
+
+        def enqueue_observation_refresh_request(self, **kwargs):
+            enqueued.append(kwargs)
+            return True
+
+    def fail_analyze(*_args, **_kwargs):
+        raise AssertionError("business cache refresh must not call _analyze by default")
+
+    monkeypatch.setattr(city_runtime, "_CACHE_DB", FakeDB())
+    monkeypatch.setattr(city_runtime, "_analyze", fail_analyze)
+
+    payload = city_runtime._refresh_city_panel_cache("shanghai")
+
+    assert payload["status"] == "initializing"
+    assert payload["stale_reason"] == "collector_refresh_queued"
+    assert payload["current"]["observation_status"] == "initializing"
+    assert enqueued == [
+        {
+            "city": "shanghai",
+            "kind": "panel",
+            "priority": "high",
+            "reason": "cold_start",
+        }
+    ]
+
+
 def test_city_panel_cold_cache_returns_canonical_latest_without_sync_refresh(monkeypatch):
    import web.services.city_api as city_api

@@ -53,7 +53,7 @@ def test_city_payloads_expose_wunderground_current():
    assert detail["timeseries"]["wunderground_today_obs"] == [{"time": "13:30", "temp": 26}]


-def test_api_payload_overlays_latest_wunderground_state(monkeypatch):
+def test_api_payload_overlay_uses_cached_wunderground_state_without_fetch(monkeypatch):
    stale_payload = {
        "name": "guangzhou",
        "temp_symbol": "°C",
@@ -78,26 +78,21 @@ def test_api_payload_overlays_latest_wunderground_state(monkeypatch):
        },
    }

-    def fake_fetch(city: str, *, use_fahrenheit: bool, utc_offset: int):
-        assert city == "guangzhou"
-        assert use_fahrenheit is False
-        assert utc_offset == 28800
-        return {
-            "source": "wunderground_historical",
-            "station_code": "ZGGG",
-            "temp": 38,
-            "max_so_far": 38,
-            "today_obs": [{"time": "15:17", "temp": 38}],
-        }
+    fetch_calls = []

-    monkeypatch.setattr(city_runtime._weather, "fetch_wunderground_historical", fake_fetch)
+    def fail_fetch(*args, **kwargs):
+        fetch_calls.append((args, kwargs))
+        raise AssertionError("city API overlay must not fetch Wunderground directly")
+
+    monkeypatch.setattr(city_runtime._weather, "fetch_wunderground_historical", fail_fetch)

    overlay = getattr(city_runtime, "_overlay_latest_wunderground_current", None)
-    assert callable(overlay), "city API must overlay cached payloads with latest WU state"
+    assert callable(overlay), "city API must preserve cached WU state without direct fetch"
    payload = overlay("guangzhou", stale_payload)

-    assert payload["wunderground_current"]["temp"] == 38
-    assert payload["wunderground_current"]["max_so_far"] == 38
-    assert payload["official"]["wunderground_current"]["max_so_far"] == 38
-    assert payload["timeseries"]["wunderground_today_obs"] == [{"time": "15:17", "temp": 38}]
+    assert payload["wunderground_current"]["temp"] == 36
+    assert payload["wunderground_current"]["max_so_far"] == 36
+    assert payload["official"]["wunderground_current"]["max_so_far"] == 36
+    assert payload["timeseries"]["wunderground_today_obs"] == [{"time": "14:00", "temp": 36}]
    assert stale_payload["wunderground_current"]["max_so_far"] == 36
+    assert fetch_calls == []
@@ -763,7 +763,7 @@ async def get_city_detail_payload(
            _request_city_cache_refresh(city, "market")
            return canonical_payload
        return _queue_and_build_initializing_city_payload(city, kind="market")
-    return await run_in_threadpool(legacy_routes._analyze, city, force_refresh, False, detail_mode)
+    return _queue_and_build_initializing_city_payload(city, kind=detail_mode)


 async def get_city_summary_payload(
@@ -32,6 +32,7 @@ from web.analysis_service import (
 )
 from web.services.canonical_temperature import (
    attach_canonical_temperature,
+    build_city_weather_from_canonical,
    store_canonical_temperature_from_payload,
 )
 from web.scan_terminal_service import build_scan_terminal_payload  # noqa: F401 - compatibility export for tests and transitional routers
@@ -59,7 +60,7 @@ from web.core import (
    _resolve_auth_points,  # noqa: F401 - compatibility export for tests and transitional routers
    _resolve_weekly_profile,  # noqa: F401 - compatibility export for tests and transitional routers
    _sf,
-    _weather,
+    _weather,  # noqa: F401 - compatibility export for tests and transitional routers
 )

 router = APIRouter()
@@ -287,7 +288,155 @@ def _attach_and_store_canonical_temperature(city: str, payload: dict) -> dict:
    return payload


-def _refresh_city_summary_cache(city: str, force_refresh: bool = False) -> dict:
+def _enqueue_city_observation_refresh(city: str, kind: str, *, reason: str) -> bool:
+    try:
+        enqueue = getattr(_CACHE_DB, "enqueue_observation_refresh_request", None)
+        if not callable(enqueue):
+            return False
+        return bool(
+            enqueue(
+                city=str(city or "").strip().lower(),
+                kind=kind,
+                priority="high",
+                reason=reason,
+            )
+        )
+    except Exception as exc:
+        logger.debug("city cache collector enqueue skipped city={} kind={}: {}", city, kind, exc)
+        return False
+
+
+def _cached_city_payload(kind: str, city: str) -> dict:
+    try:
+        entry = _CACHE_DB.get_city_cache(kind, city)
+    except Exception as exc:
+        logger.debug("city cache read skipped city={} kind={}: {}", city, kind, exc)
+        return {}
+    if not isinstance(entry, dict):
+        return {}
+    payload = entry.get("payload")
+    return deepcopy(payload) if isinstance(payload, dict) else {}
+
+
+def _canonical_city_payload(city: str, *, detail_depth: str) -> dict:
+    try:
+        row = _CACHE_DB.get_canonical_temperature(city)
+    except Exception as exc:
+        logger.debug("canonical city cache read skipped city={}: {}", city, exc)
+        return {}
+    if not isinstance(row, dict):
+        return {}
+    canonical = row.get("payload") or row
+    if not isinstance(canonical, dict):
+        return {}
+    payload = build_city_weather_from_canonical(city, canonical)
+    if not isinstance(payload, dict) or not payload:
+        return {}
+
+    city_meta = CITY_REGISTRY.get(city, {}) or {}
+    city_info = CITIES.get(city, {}) or {}
+    risk = CITY_RISK_PROFILES.get(city, {}) or {}
+    payload.update(
+        {
+            "city": city,
+            "detail_depth": detail_depth,
+            "display_name": str(city_meta.get("display_name") or city_meta.get("name") or city.title()),
+            "lat": city_info.get("lat"),
+            "lon": city_info.get("lon"),
+            "temp_symbol": canonical.get("temp_symbol") or payload.get("temp_symbol") or ("\u00b0F" if city_info.get("f") else "\u00b0C"),
+            "risk": {
+                "level": risk.get("risk_level", "low"),
+                "emoji": risk.get("risk_emoji", ""),
+                "airport": risk.get("airport_name", ""),
+                "icao": risk.get("icao", ""),
+                "distance_km": risk.get("distance_km", 0),
+                "warning": risk.get("warning", ""),
+            },
+            "probabilities": payload.get("probabilities") or {"mu": None, "distribution": []},
+        }
+    )
+    return payload
+
+
+def _initializing_city_payload(city: str, *, detail_depth: str) -> dict:
+    city_meta = CITY_REGISTRY.get(city, {}) or {}
+    city_info = CITIES.get(city, {}) or {}
+    risk = CITY_RISK_PROFILES.get(city, {}) or {}
+    return {
+        "city": city,
+        "name": city,
+        "display_name": str(city_meta.get("display_name") or city_meta.get("name") or city.title()),
+        "detail_depth": detail_depth,
+        "status": "initializing",
+        "stale": True,
+        "stale_reason": "collector_refresh_queued",
+        "lat": city_info.get("lat"),
+        "lon": city_info.get("lon"),
+        "temp_symbol": "\u00b0F" if city_info.get("f") else "\u00b0C",
+        "risk": {
+            "level": risk.get("risk_level", "low"),
+            "emoji": risk.get("risk_emoji", ""),
+            "airport": risk.get("airport_name", ""),
+            "icao": risk.get("icao", ""),
+            "distance_km": risk.get("distance_km", 0),
+            "warning": risk.get("warning", ""),
+        },
+        "current": {
+            "temp": None,
+            "source_code": None,
+            "settlement_source": None,
+            "settlement_source_label": None,
+            "obs_time": None,
+            "freshness": {
+                "freshness_status": "missing",
+                "freshness_reason": "collector_refresh_queued",
+            },
+            "observation_status": "initializing",
+        },
+        "airport_current": {},
+        "airport_primary": {},
+        "canonical_temperature": None,
+        "deb": {"prediction": None},
+        "probabilities": {"mu": None, "distribution": []},
+        "hourly": {"times": [], "temps": []},
+        "multi_model_daily": {},
+    }
+
+
+def _queued_city_cache_payload(city: str, kind: str, *, force_refresh: bool = False) -> dict:
+    normalized = str(city or "").strip().lower()
+    cached = _cached_city_payload(kind, normalized)
+    if cached:
+        if force_refresh:
+            _enqueue_city_observation_refresh(normalized, kind, reason="force_refresh")
+        return cached
+
+    canonical_payload = _canonical_city_payload(normalized, detail_depth=kind)
+    if canonical_payload:
+        _enqueue_city_observation_refresh(
+            normalized,
+            kind,
+            reason="force_refresh" if force_refresh else "canonical_fallback",
+        )
+        return canonical_payload
+
+    _enqueue_city_observation_refresh(
+        normalized,
+        kind,
+        reason="force_refresh" if force_refresh else "cold_start",
+    )
+    return _initializing_city_payload(normalized, detail_depth=kind)
+
+
+def _refresh_city_summary_cache(
+    city: str,
+    force_refresh: bool = False,
+    *,
+    allow_external_fetch: bool = False,
+) -> dict:
+    if not allow_external_fetch:
+        return _queued_city_cache_payload(city, "summary", force_refresh=force_refresh)
+
    data = _analyze_summary(city, force_refresh=force_refresh)
    _attach_and_store_canonical_temperature(city, data)
    payload = _build_city_summary_payload(data)
@@ -303,7 +452,15 @@ def _refresh_city_summary_cache(city: str, force_refresh: bool = False) -> dict:
    return payload


-def _refresh_city_panel_cache(city: str, force_refresh: bool = False) -> dict:
+def _refresh_city_panel_cache(
+    city: str,
+    force_refresh: bool = False,
+    *,
+    allow_external_fetch: bool = False,
+) -> dict:
+    if not allow_external_fetch:
+        return _queued_city_cache_payload(city, "panel", force_refresh=force_refresh)
+
    payload = _analyze(city, force_refresh=force_refresh, detail_mode="panel")
    _attach_and_store_canonical_temperature(city, payload)
    _CACHE_DB.set_city_cache(
@@ -316,7 +473,15 @@ def _refresh_city_panel_cache(city: str, force_refresh: bool = False) -> dict:
    return payload


-def _refresh_city_nearby_cache(city: str, force_refresh: bool = False) -> dict:
+def _refresh_city_nearby_cache(
+    city: str,
+    force_refresh: bool = False,
+    *,
+    allow_external_fetch: bool = False,
+) -> dict:
+    if not allow_external_fetch:
+        return _queued_city_cache_payload(city, "nearby", force_refresh=force_refresh)
+
    payload = _analyze(city, force_refresh=force_refresh, detail_mode="nearby")
    _attach_and_store_canonical_temperature(city, payload)
    _CACHE_DB.set_city_cache(
@@ -329,7 +494,15 @@ def _refresh_city_nearby_cache(city: str, force_refresh: bool = False) -> dict:
    return payload


-def _refresh_city_market_cache(city: str, force_refresh: bool = False) -> dict:
+def _refresh_city_market_cache(
+    city: str,
+    force_refresh: bool = False,
+    *,
+    allow_external_fetch: bool = False,
+) -> dict:
+    if not allow_external_fetch:
+        return _queued_city_cache_payload(city, "market", force_refresh=force_refresh)
+
    payload = _analyze(city, force_refresh=force_refresh, detail_mode="market")
    _attach_and_store_canonical_temperature(city, payload)
    now_ts = time.time()
@@ -346,7 +519,15 @@ def _refresh_city_market_cache(city: str, force_refresh: bool = False) -> dict:
    return payload


-def _refresh_city_full_cache(city: str, force_refresh: bool = False) -> dict:
+def _refresh_city_full_cache(
+    city: str,
+    force_refresh: bool = False,
+    *,
+    allow_external_fetch: bool = False,
+) -> dict:
+    if not allow_external_fetch:
+        return _queued_city_cache_payload(city, "full", force_refresh=force_refresh)
+
    payload = _analyze(city, force_refresh=force_refresh, detail_mode="full")
    _attach_and_store_canonical_temperature(city, payload)
    _CACHE_DB.set_city_cache(
@@ -379,27 +560,11 @@ def _overlay_wunderground_current(payload: dict, wunderground: dict) -> dict:
 def _overlay_latest_wunderground_current(city: str, payload: dict) -> dict:
    if not isinstance(payload, dict):
        return payload
-    city_key = str(city or payload.get("name") or payload.get("city") or "").strip().lower()
-    if city_key not in CITIES:
-        return deepcopy(payload)
-    try:
-        utc_offset = int(
-            payload.get("utc_offset_seconds")
-            or (payload.get("overview") or {}).get("utc_offset_seconds")
-            or get_city_utc_offset_seconds(city_key)
-            or 0
-        )
-    except Exception:
-        utc_offset = get_city_utc_offset_seconds(city_key)
-    try:
-        latest_wu = _weather.fetch_wunderground_historical(
-            city_key,
-            use_fahrenheit=bool(CITIES[city_key].get("f")),
-            utc_offset=utc_offset,
-        )
-    except Exception as exc:
-        logger.warning("latest WU overlay failed city={} error={}", city_key, exc)
-        return deepcopy(payload)
+    latest_wu = payload.get("wunderground_current")
+    if not isinstance(latest_wu, dict) or not latest_wu:
+        official = payload.get("official")
+        if isinstance(official, dict):
+            latest_wu = official.get("wunderground_current")
    if not isinstance(latest_wu, dict) or not latest_wu:
        return deepcopy(payload)
    return _overlay_wunderground_current(payload, latest_wu)