diff --git a/tests/test_canonical_temperature.py b/tests/test_canonical_temperature.py index 14014fc7..b83f0840 100644 --- a/tests/test_canonical_temperature.py +++ b/tests/test_canonical_temperature.py @@ -28,6 +28,29 @@ def _sample_panel_payload() -> dict: } +def test_canonical_temperature_roles_understand_adapter_source_names(): + from web.services.canonical_temperature import build_canonical_temperature + + def role_for(source): + canonical = build_canonical_temperature( + "hong kong", + { + "current": { + "temp": 28.0, + "source_code": source, + "source_label": source, + "freshness": {"freshness_status": "fresh"}, + }, + }, + ) + assert canonical is not None + return canonical["source_role"] + + assert role_for("hko_obs") == "settlement_official" + assert role_for("cowin_obs") == "settlement_official" + assert role_for("madis_hfmetar") == "airport_official" + + def test_db_manager_stores_canonical_temperature_latest(tmp_path): from src.database.db_manager import DBManager @@ -306,7 +329,6 @@ def test_force_refresh_panel_returns_canonical_latest_without_waiting_for_sync_r def fail_refresh(*_args, **_kwargs): raise AssertionError("force refresh must not block on sync refresh when canonical latest exists") - city_api._CITY_FORCE_REFRESH_INFLIGHT.clear() 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) diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index 04e943b5..3f1e4944 100644 --- a/tests/test_observation_collector.py +++ b/tests/test_observation_collector.py @@ -2,6 +2,7 @@ from concurrent.futures import ThreadPoolExecutor, TimeoutError import sqlite3 import threading import time +from types import SimpleNamespace def test_observation_source_gate_shares_inflight_and_cooldown(monkeypatch): @@ -93,14 +94,13 @@ def test_observation_collector_profiles_match_source_cadence(): assert SOURCE_CADENCE_SECONDS["amsc_awos"] == 180 -def test_observation_collector_run_due_once_refreshes_panel_cache(): +def test_observation_collector_run_due_once_collects_without_panel_cache_refresh(): from web.observation_collector_service import ( ObservationCollector, ObservationSourceProfile, ) calls = [] - refreshed = [] class FakeWeather: def _uses_fahrenheit(self, city): @@ -119,20 +119,17 @@ def test_observation_collector_run_due_once_refreshes_panel_cache(): interval_sec=180, ) ], - cache_refresher=lambda city: refreshed.append(city), async_cache_refresh=False, ) assert collector.run_due_once(now_ts=1000.0) == 1 assert calls == [("qingdao", False)] - assert refreshed == ["qingdao"] assert collector.run_due_once(now_ts=1100.0) == 0 assert calls == [("qingdao", False)] assert collector.run_due_once(now_ts=1180.0) == 1 assert calls == [("qingdao", False), ("qingdao", False)] - assert refreshed == ["qingdao", "qingdao"] def test_raw_observation_store_records_latest_observation(tmp_path): @@ -257,6 +254,46 @@ def test_observation_collector_writes_raw_observation_store(tmp_path): assert latest["station_code"] == "ZSQD" +def test_observation_collector_writes_canonical_latest_from_source(tmp_path): + from src.database.db_manager import DBManager + from web.observation_collector_service import ( + ObservationCollector, + ObservationSourceProfile, + ) + + db = DBManager(str(tmp_path / "polyweather.db")) + + class FakeWeather: + def _uses_fahrenheit(self, city): + return False + + def _attach_china_amsc_awos_data(self, results, city, use_fahrenheit): + results["amos"] = { + "source": "amsc_awos", + "source_label": "AMSC AWOS", + "temp_c": 24.0, + "observation_time": "2026-06-14T01:00:00+00:00", + "icao": "ZSQD", + "station_label": "Qingdao Jiaodong", + } + + collector = ObservationCollector( + weather=FakeWeather(), + profiles=[ObservationSourceProfile("amsc_awos", ("qingdao",), 180)], + observation_store=db, + async_cache_refresh=False, + ) + + assert collector.run_due_once(now_ts=1000.0) == 1 + + canonical = db.get_canonical_temperature("qingdao") + assert canonical is not None + assert canonical["payload"]["value"] == 24.0 + assert canonical["payload"]["source"] == "amsc_awos" + assert canonical["payload"]["source_role"] == "settlement_proxy" + assert canonical["payload"]["observed_at"] == "2026-06-14T01:00:00+00:00" + + def test_observation_collector_consumes_refresh_request_queue(tmp_path): from src.database.db_manager import DBManager from web.observation_collector_service import ( @@ -486,6 +523,38 @@ def test_observation_collector_worker_entrypoint_exists(): assert callable(observation_collector_worker.main) +def test_observation_collector_worker_does_not_bind_panel_cache_refresher(monkeypatch): + from web import observation_collector_worker + + captured = {} + + def fake_start_observation_collector_loop(**kwargs): + captured.update(kwargs) + return SimpleNamespace(name="observation-collector") + + class StopEvent: + @staticmethod + def wait(_timeout): + return True + + @staticmethod + def set(): + return None + + monkeypatch.setattr(observation_collector_worker.signal, "signal", lambda *_args, **_kwargs: None) + monkeypatch.setattr( + observation_collector_worker, + "start_observation_collector_loop", + fake_start_observation_collector_loop, + ) + monkeypatch.setattr(observation_collector_worker, "_STOP_EVENT", StopEvent()) + + observation_collector_worker.main() + + assert captured["weather"] is observation_collector_worker._weather + assert captured.get("cache_refresher") is None + + def test_ephemeral_observation_log_writes_skip_sqlite_lock(monkeypatch, tmp_path): from src.database.db_manager import DBManager diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index aafbd802..2ecc0c7d 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -1693,7 +1693,6 @@ def test_stale_city_detail_uses_cached_full_payload_while_refreshing(monkeypatch "resolution": resolution, } - city_api._CITY_FULL_REFRESH_INFLIGHT.clear() city_api._CITY_DETAIL_PAYLOAD_CACHE.clear() city_api._CITY_DETAIL_PAYLOAD_CACHE_TS.clear() city_api._CITY_DETAIL_PAYLOAD_INFLIGHT.clear() @@ -1815,8 +1814,6 @@ def test_force_refresh_panel_returns_cached_payload_when_refresh_already_running refresh_calls += 1 return {"name": city, "deb": {"prediction": 21.0}, "from_cache": False} - city_api._CITY_FORCE_REFRESH_INFLIGHT.clear() - monkeypatch.setenv("POLYWEATHER_CITY_FORCE_REFRESH_TIMEOUT_SEC", "0.5") monkeypatch.setattr(city_api, "run_in_threadpool", fake_run_in_threadpool) monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower()) @@ -1897,8 +1894,6 @@ def test_stale_panel_returns_cached_payload_while_refreshing(monkeypatch): refresh_calls += 1 return {"name": city, "deb": {"prediction": 21.0}, "from_cache": False} - city_api._CITY_STALE_REFRESH_TASKS.clear() - monkeypatch.setattr(city_api, "run_in_threadpool", fake_run_in_threadpool) monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower()) monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeCache()) @@ -1970,7 +1965,6 @@ def test_force_refresh_full_detail_returns_cached_payload_when_refresh_is_slow(m build_inputs.append(data["hourly"]["temps"][0]) return {"city": data["city"], "live_temp": data["hourly"]["temps"][0]} - city_api._CITY_FULL_REFRESH_INFLIGHT.clear() city_api._CITY_DETAIL_PAYLOAD_CACHE.clear() city_api._CITY_DETAIL_PAYLOAD_CACHE_TS.clear() city_api._CITY_DETAIL_PAYLOAD_INFLIGHT.clear() diff --git a/web/app_factory.py b/web/app_factory.py index 180d6488..4451b782 100644 --- a/web/app_factory.py +++ b/web/app_factory.py @@ -69,12 +69,10 @@ def create_app() -> FastAPI: and not bool(getattr(core_app.state, _OBSERVATION_COLLECTOR_STARTED_FLAG, False)) ): from web.core import _weather - from web.services.city_runtime import _refresh_city_panel_cache from web.observation_collector_service import start_observation_collector_loop thread = start_observation_collector_loop( weather=_weather, - cache_refresher=lambda city: _refresh_city_panel_cache(city, force_refresh=False), ) setattr(core_app.state, _OBSERVATION_COLLECTOR_STARTED_FLAG, bool(thread)) return core_app diff --git a/web/observation_collector_service.py b/web/observation_collector_service.py index e0d837fb..fc9da4b7 100644 --- a/web/observation_collector_service.py +++ b/web/observation_collector_service.py @@ -17,6 +17,7 @@ from src.data_collection.city_registry import CITY_REGISTRY from src.data_collection.hko_obs_sources import HKO_STATIONS from src.database.db_manager import DBManager from src.database.runtime_state import ObservationCollectorStatusRepository +from web.services.canonical_temperature import build_canonical_temperature def _env_bool(name: str, default: bool) -> bool: @@ -303,6 +304,59 @@ class ObservationCollector: return value return "" + @staticmethod + def _source_label(row: dict[str, Any], source: str) -> str: + for key in ("source_label", "label", "source_name"): + value = str(row.get(key) or "").strip() + if value: + return value + return str(source or "").replace("_", " ").upper() + + def _store_canonical_temperature_from_observation( + self, + *, + city: str, + source: str, + row: dict[str, Any], + value: float, + observed_at: str, + fetched_at: str, + ) -> None: + setter = getattr(self.observation_store, "set_canonical_temperature", None) + if not callable(setter): + return + value_unit = str(row.get("unit") or row.get("temp_unit") or "c").strip().lower() + payload = { + "name": city, + "temp_symbol": "°F" if value_unit.startswith("f") else "°C", + "updated_at": fetched_at, + "current": { + "temp": value, + "source_code": source, + "source_label": self._source_label(row, source), + "settlement_source": source, + "settlement_source_label": self._source_label(row, source), + "station_code": self._station_code(row), + "station_name": self._station_name(row), + "observed_at": observed_at or None, + "observed_at_local": row.get("observation_time_local"), + "obs_time": row.get("observation_time_local") or observed_at, + "freshness": { + "freshness_status": "fresh", + "observed_at": observed_at or None, + "observed_at_local": row.get("observation_time_local"), + }, + "observation_status": "live", + }, + } + canonical = build_canonical_temperature(city, payload, fetched_at=fetched_at) + if not canonical: + return + try: + setter(city, canonical) + except Exception as exc: + logger.debug("canonical temperature write skipped source={} city={}: {}", source, city, exc) + def _iter_raw_observation_rows( self, source: str, @@ -330,11 +384,12 @@ class ObservationCollector: continue payload_source = str(row.get("source") or row.get("source_code") or source).strip().lower() try: + observed_at = self._observation_time(row) writer( source=payload_source or source, city=city, value=value, - observed_at=self._observation_time(row), + observed_at=observed_at, fetched_at=fetched_at, station_code=self._station_code(row), station_name=self._station_name(row), @@ -343,6 +398,14 @@ class ObservationCollector: status="ok", payload=dict(row), ) + self._store_canonical_temperature_from_observation( + city=city, + source=payload_source or source, + row=row, + value=value, + observed_at=observed_at, + fetched_at=fetched_at, + ) wrote += 1 except Exception as exc: logger.debug( diff --git a/web/observation_collector_worker.py b/web/observation_collector_worker.py index 8b84ed8e..1d2e8f91 100644 --- a/web/observation_collector_worker.py +++ b/web/observation_collector_worker.py @@ -16,7 +16,6 @@ from loguru import logger from web.core import _weather from web.observation_collector_service import start_observation_collector_loop -from web.services.city_runtime import _refresh_city_panel_cache _STOP_EVENT = threading.Event() @@ -32,7 +31,6 @@ def main() -> None: thread = start_observation_collector_loop( weather=_weather, - cache_refresher=lambda city: _refresh_city_panel_cache(city, force_refresh=False), ) if thread is None: logger.warning("observation collector worker started with collector disabled") diff --git a/web/services/canonical_temperature.py b/web/services/canonical_temperature.py index 21411e0b..4b805197 100644 --- a/web/services/canonical_temperature.py +++ b/web/services/canonical_temperature.py @@ -36,9 +36,14 @@ def _source_role(source: str) -> str: normalized = str(source or "").strip().lower() if normalized in _SETTLEMENT_PROXY_SOURCES or "amsc" in normalized or "runway" in normalized: return "settlement_proxy" - if normalized in _SETTLEMENT_OFFICIAL_SOURCES: + if ( + normalized in _SETTLEMENT_OFFICIAL_SOURCES + or "hko" in normalized + or "cowin" in normalized + or "cwa" in normalized + ): return "settlement_official" - if normalized in _AIRPORT_OFFICIAL_SOURCES: + if normalized in _AIRPORT_OFFICIAL_SOURCES or "metar" in normalized or "madis" in normalized: return "airport_official" if "model" in normalized or normalized in {"deb", "open_meteo"}: return "model_blend" diff --git a/web/services/city_api.py b/web/services/city_api.py index a2642d1a..8f992a41 100644 --- a/web/services/city_api.py +++ b/web/services/city_api.py @@ -7,7 +7,7 @@ import asyncio import threading import time from copy import deepcopy -from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple +from typing import Any, Callable, Dict, List, Optional, Tuple from fastapi import HTTPException, Request from fastapi.concurrency import run_in_threadpool @@ -26,12 +26,6 @@ _RECENT_DEB_CACHE_TTL_SEC = max( 60, int(os.getenv("POLYWEATHER_CITIES_DEB_RECENT_CACHE_TTL_SEC", "300") or "300"), ) -_CITY_FULL_REFRESH_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {} -_CITY_FULL_STALE_REFRESH_TASKS: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {} -_CITY_FULL_REFRESH_LOCK = asyncio.Lock() -_CITY_FORCE_REFRESH_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {} -_CITY_FORCE_REFRESH_LOCK = asyncio.Lock() -_CITY_STALE_REFRESH_TASKS: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {} CityDetailPayloadCacheKey = Tuple[str, str, str, str, str, int] CityChartDetailPayloadCacheKey = Tuple[str, str, str, int] CityDetailBatchResponseCacheKey = Tuple[Tuple[str, ...], bool, str, str, str, str] @@ -71,14 +65,6 @@ def _city_detail_batch_response_cache_ttl() -> float: return max(0.0, min(30.0, value)) -def _city_force_refresh_timeout_sec() -> float: - try: - value = float(os.getenv("POLYWEATHER_CITY_FORCE_REFRESH_TIMEOUT_SEC", "8") or "8") - except ValueError: - value = 8.0 - return max(0.01, min(30.0, value)) - - def _city_chart_optional_overlay_timeout_sec() -> float: try: timeout_ms = int( @@ -193,7 +179,6 @@ def _enqueue_collector_refresh_request( def _request_city_cache_refresh( city: str, kind: str, - refresh_fn: Callable[[str, bool], Dict[str, Any]], ) -> None: _enqueue_collector_refresh_request(city, kind) @@ -252,34 +237,9 @@ def _queue_and_build_initializing_city_payload(city: str, *, kind: str) -> Dict[ return _build_initializing_city_payload(city, detail_depth=kind) -async def _get_or_start_city_force_refresh_task( - key: str, - refresh_factory: Callable[[], Awaitable[Dict[str, Any]]], -) -> Tuple["asyncio.Task[Dict[str, Any]]", bool]: - async with _CITY_FORCE_REFRESH_LOCK: - task = _CITY_FORCE_REFRESH_INFLIGHT.get(key) - started = False - if task is None or task.done(): - task = asyncio.create_task(refresh_factory()) - _CITY_FORCE_REFRESH_INFLIGHT[key] = task - started = True - - def _cleanup(done: "asyncio.Task[Dict[str, Any]]") -> None: - if _CITY_FORCE_REFRESH_INFLIGHT.get(key) is done: - _CITY_FORCE_REFRESH_INFLIGHT.pop(key, None) - try: - done.result() - except Exception as exc: # pragma: no cover - defensive background guard - logger.warning("city force refresh failed key={}: {}", key, exc) - - task.add_done_callback(_cleanup) - return task, started - - async def _refresh_city_payload_with_stale_timeout( city: str, kind: str, - refresh_factory: Callable[[], Awaitable[Dict[str, Any]]], ) -> Dict[str, Any]: cached_before_refresh = await _get_cached_city_payload(city, kind) if not cached_before_refresh: @@ -306,19 +266,16 @@ async def _refresh_city_payload_with_stale_timeout( async def _refresh_city_cache_with_stale_timeout( city: str, kind: str, - refresh_fn: Callable[[str, bool], Dict[str, Any]], ) -> Dict[str, Any]: return await _refresh_city_payload_with_stale_timeout( city, kind, - lambda: run_in_threadpool(refresh_fn, city, True), ) def _start_city_cache_stale_refresh( city: str, kind: str, - refresh_fn: Callable[[str, bool], Dict[str, Any]], ) -> None: normalized = str(city or "").strip().lower() cache_kind = str(kind or "").strip().lower() @@ -394,54 +351,6 @@ def _overlay_cached_runway_history_from_db(city: str, payload: Dict[str, Any]) - return next_payload -async def _refresh_city_full_cache_singleflight(city: str, force_refresh: bool) -> Dict[str, Any]: - key = f"{city}:{bool(force_refresh)}" - async with _CITY_FULL_REFRESH_LOCK: - task = _CITY_FULL_REFRESH_INFLIGHT.get(key) - if task is None: - async def _run_refresh() -> Dict[str, Any]: - try: - return await run_in_threadpool( - legacy_routes._refresh_city_full_cache, - city, - force_refresh, - ) - finally: - await _invalidate_city_detail_payload_cache(city) - - task = asyncio.create_task(_run_refresh()) - _CITY_FULL_REFRESH_INFLIGHT[key] = task - try: - return await task - finally: - if task.done(): - async with _CITY_FULL_REFRESH_LOCK: - if _CITY_FULL_REFRESH_INFLIGHT.get(key) is task: - _CITY_FULL_REFRESH_INFLIGHT.pop(key, None) - - -async def _invalidate_city_detail_payload_cache(city: str) -> None: - normalized = str(city or "").strip().lower() - if not normalized: - return - async with _CITY_DETAIL_PAYLOAD_LOCK: - _CITY_DETAIL_PAYLOAD_EPOCH[normalized] = _CITY_DETAIL_PAYLOAD_EPOCH.get(normalized, 0) + 1 - old_keys = [key for key in _CITY_DETAIL_PAYLOAD_CACHE if key[0] == normalized] - for key in old_keys: - _CITY_DETAIL_PAYLOAD_CACHE.pop(key, None) - _CITY_DETAIL_PAYLOAD_CACHE_TS.pop(key, None) - async with _CITY_CHART_DETAIL_PAYLOAD_LOCK: - old_chart_keys = [key for key in _CITY_CHART_DETAIL_PAYLOAD_CACHE if key[0] == normalized] - for key in old_chart_keys: - _CITY_CHART_DETAIL_PAYLOAD_CACHE.pop(key, None) - _CITY_CHART_DETAIL_PAYLOAD_CACHE_TS.pop(key, None) - - -async def _refresh_city_full_data(city: str, force_refresh: bool) -> Dict[str, Any]: - await _invalidate_city_detail_payload_cache(city) - return await _refresh_city_full_cache_singleflight(city, force_refresh) - - def _start_city_full_stale_refresh(city: str) -> None: normalized = str(city or "").strip().lower() if not normalized: @@ -454,7 +363,6 @@ async def _get_city_full_data(city: str, *, force_refresh: bool) -> Dict[str, An return await _refresh_city_payload_with_stale_timeout( city, "full", - lambda: _refresh_city_full_data(city, True), ) cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city) if cached_entry: @@ -792,20 +700,19 @@ async def get_city_detail_payload( return await _refresh_city_cache_with_stale_timeout( city, "panel", - legacy_routes._refresh_city_panel_cache, ) cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "panel", city) if cached_entry: if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_PANEL_CACHE_TTL_SEC): payload = cached_entry.get("payload") or {} if payload: - _start_city_cache_stale_refresh(city, "panel", legacy_routes._refresh_city_panel_cache) + _start_city_cache_stale_refresh(city, "panel") return await _overlay_cached_wunderground(city, payload) return _queue_and_build_initializing_city_payload(city, kind="panel") return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) canonical_payload = await _get_canonical_city_payload(city, detail_depth="panel") if canonical_payload: - _request_city_cache_refresh(city, "panel", legacy_routes._refresh_city_panel_cache) + _request_city_cache_refresh(city, "panel") return canonical_payload return _queue_and_build_initializing_city_payload(city, kind="panel") if detail_mode == "nearby": @@ -813,24 +720,23 @@ async def get_city_detail_payload( return await _refresh_city_cache_with_stale_timeout( city, "nearby", - legacy_routes._refresh_city_nearby_cache, ) cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "nearby", city) if cached_entry: if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_NEARBY_CACHE_TTL_SEC): payload = cached_entry.get("payload") or {} if payload: - _start_city_cache_stale_refresh(city, "nearby", legacy_routes._refresh_city_nearby_cache) + _start_city_cache_stale_refresh(city, "nearby") return await _overlay_cached_wunderground(city, payload) canonical_payload = await _get_canonical_city_payload(city, detail_depth="nearby") if canonical_payload: - _request_city_cache_refresh(city, "nearby", legacy_routes._refresh_city_nearby_cache) + _request_city_cache_refresh(city, "nearby") return canonical_payload return _queue_and_build_initializing_city_payload(city, kind="nearby") return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) canonical_payload = await _get_canonical_city_payload(city, detail_depth="nearby") if canonical_payload: - _request_city_cache_refresh(city, "nearby", legacy_routes._refresh_city_nearby_cache) + _request_city_cache_refresh(city, "nearby") return canonical_payload return _queue_and_build_initializing_city_payload(city, kind="nearby") if detail_mode == "market": @@ -838,24 +744,23 @@ async def get_city_detail_payload( return await _refresh_city_cache_with_stale_timeout( city, "market", - legacy_routes._refresh_city_market_cache, ) cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "market", city) if cached_entry: if not legacy_routes._market_analysis_cache_is_fresh(cached_entry): payload = cached_entry.get("payload") or {} if payload: - _start_city_cache_stale_refresh(city, "market", legacy_routes._refresh_city_market_cache) + _start_city_cache_stale_refresh(city, "market") return await _overlay_cached_wunderground(city, payload) canonical_payload = await _get_canonical_city_payload(city, detail_depth="market") if canonical_payload: - _request_city_cache_refresh(city, "market", legacy_routes._refresh_city_market_cache) + _request_city_cache_refresh(city, "market") return canonical_payload return _queue_and_build_initializing_city_payload(city, kind="market") return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) canonical_payload = await _get_canonical_city_payload(city, detail_depth="market") if canonical_payload: - _request_city_cache_refresh(city, "market", legacy_routes._refresh_city_market_cache) + _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) @@ -872,20 +777,19 @@ async def get_city_summary_payload( return await _refresh_city_cache_with_stale_timeout( city, "summary", - legacy_routes._refresh_city_summary_cache, ) cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "summary", city) if cached_entry: if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_SUMMARY_CACHE_TTL_SEC): payload = cached_entry.get("payload") or {} if payload: - _start_city_cache_stale_refresh(city, "summary", legacy_routes._refresh_city_summary_cache) + _start_city_cache_stale_refresh(city, "summary") return await _overlay_cached_wunderground(city, payload) return _queue_and_build_initializing_city_payload(city, kind="summary") return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) canonical_payload = await _get_canonical_city_payload(city, detail_depth="summary") if canonical_payload: - _request_city_cache_refresh(city, "summary", legacy_routes._refresh_city_summary_cache) + _request_city_cache_refresh(city, "summary") return canonical_payload return _queue_and_build_initializing_city_payload(city, kind="summary") diff --git a/web/services/city_runtime.py b/web/services/city_runtime.py index 966d52bc..b5b13ddb 100644 --- a/web/services/city_runtime.py +++ b/web/services/city_runtime.py @@ -6,7 +6,7 @@ from copy import deepcopy from datetime import datetime from typing import Optional -from fastapi import APIRouter, BackgroundTasks, HTTPException +from fastapi import APIRouter, HTTPException from loguru import logger from src.analysis.deb_algorithm import load_history @@ -159,9 +159,6 @@ MARKET_SCAN_PAYLOAD_TTL_SEC = max( 5, int(os.getenv("POLYWEATHER_MARKET_SCAN_PAYLOAD_TTL_SEC", "30")), ) -CACHE_REFRESH_LOCK_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CACHE_REFRESH_LOCK_TTL_SEC", "120"))) - - def _city_cache_is_fresh(entry: Optional[dict], ttl_sec: int) -> bool: if not isinstance(entry, dict): return False @@ -407,55 +404,6 @@ def _overlay_latest_wunderground_current(city: str, payload: dict) -> dict: return deepcopy(payload) return _overlay_wunderground_current(payload, latest_wu) - - -def _schedule_cache_refresh( - background_tasks: BackgroundTasks, - *, - kind: str, - city: str, - force_refresh: bool = False, -) -> bool: - normalized_kind = str(kind or "").strip().lower() - normalized_city = str(city or "").strip().lower() - if normalized_kind not in {"summary", "panel", "nearby", "market", "full"} or not normalized_city: - return False - cache_key = f"city:{normalized_kind}:{normalized_city}" - owner = _CACHE_DB.acquire_cache_refresh_lock( - cache_key, - ttl_sec=CACHE_REFRESH_LOCK_TTL_SEC, - ) - if not owner: - return False - - def _runner() -> None: - try: - if normalized_kind == "summary": - _refresh_city_summary_cache(normalized_city, force_refresh=force_refresh) - elif normalized_kind == "panel": - _refresh_city_panel_cache(normalized_city, force_refresh=force_refresh) - elif normalized_kind == "nearby": - _refresh_city_nearby_cache(normalized_city, force_refresh=force_refresh) - elif normalized_kind == "market": - _refresh_city_market_cache(normalized_city, force_refresh=force_refresh) - else: - _refresh_city_full_cache(normalized_city, force_refresh=force_refresh) - except Exception as exc: - logger.warning( - "cache refresh failed kind={} city={} force_refresh={}: {}", - normalized_kind, - normalized_city, - force_refresh, - exc, - ) - finally: - _CACHE_DB.release_cache_refresh_lock(cache_key, owner) - - background_tasks.add_task(_runner) - return True - - - def _normalize_city_or_404(name: str) -> str: city = name.lower().strip().replace("-", " ") city = ALIASES.get(city, city)