From e111031119ab93b48d2dd32c8f5f2743041e3265 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Sun, 14 Jun 2026 17:53:24 +0800 Subject: [PATCH] Decouple live weather reads from source collection --- src/database/db_manager.py | 541 ++++++++++++++++++++++++++ src/utils/telegram_push.py | 35 +- tests/test_canonical_temperature.py | 445 +++++++++++++++++++++ tests/test_city_realtime_stream.py | 51 +++ tests/test_observation_collector.py | 163 ++++++++ tests/test_scan_terminal_modules.py | 143 +++++++ tests/test_telegram_hashtags.py | 64 ++- tests/test_web_observability.py | 98 +++-- web/analysis_service.py | 5 + web/observation_collector_service.py | 175 ++++++++- web/scan_terminal_city_row.py | 62 ++- web/scan_terminal_service.py | 19 +- web/services/canonical_temperature.py | 214 ++++++++++ web/services/city_api.py | 208 +++++++++- web/services/city_realtime_stream.py | 84 ++-- web/services/city_runtime.py | 22 ++ web/services/dashboard_init_api.py | 15 +- web/services/system_api.py | 21 +- 18 files changed, 2248 insertions(+), 117 deletions(-) create mode 100644 tests/test_canonical_temperature.py create mode 100644 tests/test_city_realtime_stream.py create mode 100644 web/services/canonical_temperature.py diff --git a/src/database/db_manager.py b/src/database/db_manager.py index 83d77c2a..96e9892e 100644 --- a/src/database/db_manager.py +++ b/src/database/db_manager.py @@ -524,6 +524,111 @@ class DBManager: source_fingerprint TEXT ) """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS canonical_temperature_latest ( + city TEXT PRIMARY KEY, + payload_json TEXT NOT NULL, + value REAL, + source TEXT, + source_role TEXT, + observed_at TEXT, + fetched_at TEXT, + freshness_sec INTEGER, + freshness_status TEXT, + confidence REAL, + explanation TEXT, + updated_at TEXT NOT NULL, + updated_at_ts REAL NOT NULL + ) + """) + conn.execute( + """ + CREATE INDEX IF NOT EXISTS idx_canonical_temperature_latest_updated + ON canonical_temperature_latest(updated_at_ts DESC) + """ + ) + conn.execute(""" + CREATE TABLE IF NOT EXISTS raw_observation_store ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT NOT NULL, + city TEXT NOT NULL, + station_code TEXT NOT NULL DEFAULT '', + station_name TEXT NOT NULL DEFAULT '', + runway TEXT NOT NULL DEFAULT '', + value REAL, + value_unit TEXT NOT NULL DEFAULT '', + observed_at TEXT, + fetched_at TEXT NOT NULL, + source_latency_sec REAL, + status TEXT NOT NULL DEFAULT 'ok', + error_count INTEGER NOT NULL DEFAULT 0, + last_success_at TEXT, + payload_json TEXT NOT NULL, + created_at_ts REAL NOT NULL + ) + """) + conn.execute( + """ + CREATE INDEX IF NOT EXISTS idx_raw_observation_store_source_city_time + ON raw_observation_store(source, city, observed_at DESC, fetched_at DESC) + """ + ) + conn.execute(""" + CREATE TABLE IF NOT EXISTS raw_observation_latest ( + source TEXT NOT NULL, + city TEXT NOT NULL, + station_code TEXT NOT NULL DEFAULT '', + station_name TEXT NOT NULL DEFAULT '', + runway TEXT NOT NULL DEFAULT '', + value REAL, + value_unit TEXT NOT NULL DEFAULT '', + observed_at TEXT, + fetched_at TEXT NOT NULL, + source_latency_sec REAL, + status TEXT NOT NULL DEFAULT 'ok', + error_count INTEGER NOT NULL DEFAULT 0, + last_success_at TEXT, + payload_json TEXT NOT NULL, + updated_at_ts REAL NOT NULL, + PRIMARY KEY (source, city, station_code, runway) + ) + """) + conn.execute( + """ + CREATE INDEX IF NOT EXISTS idx_raw_observation_latest_city_source + ON raw_observation_latest(city, source, updated_at_ts DESC) + """ + ) + conn.execute(""" + CREATE TABLE IF NOT EXISTS observation_refresh_requests ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + city TEXT NOT NULL, + kind TEXT NOT NULL DEFAULT '', + source TEXT NOT NULL DEFAULT '', + priority TEXT NOT NULL DEFAULT 'normal', + reason TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'pending', + attempts INTEGER NOT NULL DEFAULT 0, + owner TEXT NOT NULL DEFAULT '', + requested_at TEXT NOT NULL, + requested_at_ts REAL NOT NULL, + claimed_at_ts REAL, + completed_at_ts REAL, + last_error TEXT NOT NULL DEFAULT '' + ) + """) + conn.execute( + """ + CREATE INDEX IF NOT EXISTS idx_observation_refresh_requests_status + ON observation_refresh_requests(status, priority, requested_at_ts) + """ + ) + conn.execute( + """ + CREATE INDEX IF NOT EXISTS idx_observation_refresh_requests_city + ON observation_refresh_requests(city, kind, status) + """ + ) conn.execute(""" CREATE TABLE IF NOT EXISTS cache_refresh_locks ( cache_key TEXT PRIMARY KEY, @@ -727,6 +832,24 @@ class DBManager: conn.commit() logger.info(f"Database initialized successfully path={self.db_path}") + @staticmethod + def _float_or_none(value: Any) -> Optional[float]: + try: + if value is None or value == "": + return None + return float(value) + except (TypeError, ValueError): + return None + + @staticmethod + def _int_or_none(value: Any) -> Optional[int]: + try: + if value is None or value == "": + return None + return int(float(value)) + except (TypeError, ValueError): + return None + def _cache_table_name(self, kind: str) -> Optional[str]: normalized = str(kind or "").strip().lower() if normalized == "summary": @@ -819,6 +942,424 @@ class DBManager: ) conn.commit() + def get_canonical_temperature(self, city: str) -> Optional[Dict[str, Any]]: + normalized_city = str(city or "").strip().lower() + if not normalized_city: + return None + with self._get_connection() as conn: + conn.row_factory = sqlite3.Row + row = conn.execute( + """ + SELECT city, payload_json, value, source, source_role, observed_at, + fetched_at, freshness_sec, freshness_status, confidence, + explanation, updated_at, updated_at_ts + FROM canonical_temperature_latest + WHERE city = ? + LIMIT 1 + """, + (normalized_city,), + ).fetchone() + if not row: + return None + try: + payload = json.loads(str(row["payload_json"] or "{}")) + except Exception: + return None + if not isinstance(payload, dict): + return None + return { + "city": str(row["city"] or normalized_city), + "payload": payload, + "value": self._float_or_none(row["value"]), + "source": str(row["source"] or ""), + "source_role": str(row["source_role"] or ""), + "observed_at": str(row["observed_at"] or ""), + "fetched_at": str(row["fetched_at"] or ""), + "freshness_sec": self._int_or_none(row["freshness_sec"]), + "freshness_status": str(row["freshness_status"] or ""), + "confidence": self._float_or_none(row["confidence"]), + "explanation": str(row["explanation"] or ""), + "updated_at": str(row["updated_at"] or ""), + "updated_at_ts": float(row["updated_at_ts"] or 0.0), + } + + def set_canonical_temperature(self, city: str, payload: Dict[str, Any]) -> None: + normalized_city = str(city or "").strip().lower() + if not normalized_city or not isinstance(payload, dict): + return + value = self._float_or_none(payload.get("value")) + source = str(payload.get("source") or "").strip().lower() + source_role = str(payload.get("source_role") or "").strip().lower() + observed_at = str(payload.get("observed_at") or "").strip() + fetched_at = str(payload.get("fetched_at") or "").strip() + freshness_sec = self._int_or_none(payload.get("freshness_sec")) + freshness_status = str(payload.get("freshness_status") or "").strip().lower() + confidence = self._float_or_none(payload.get("confidence")) + explanation = str(payload.get("explanation") or "").strip() + now_dt = datetime.now() + now = now_dt.isoformat() + now_ts = now_dt.timestamp() + with self._get_connection() as conn: + conn.execute( + """ + INSERT INTO canonical_temperature_latest ( + city, payload_json, value, source, source_role, observed_at, + fetched_at, freshness_sec, freshness_status, confidence, + explanation, updated_at, updated_at_ts + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(city) DO UPDATE SET + payload_json = excluded.payload_json, + value = excluded.value, + source = excluded.source, + source_role = excluded.source_role, + observed_at = excluded.observed_at, + fetched_at = excluded.fetched_at, + freshness_sec = excluded.freshness_sec, + freshness_status = excluded.freshness_status, + confidence = excluded.confidence, + explanation = excluded.explanation, + updated_at = excluded.updated_at, + updated_at_ts = excluded.updated_at_ts + """, + ( + normalized_city, + json.dumps({**payload, "city": normalized_city}, ensure_ascii=False), + value, + source, + source_role, + observed_at, + fetched_at, + freshness_sec, + freshness_status, + confidence, + explanation, + now, + now_ts, + ), + ) + conn.commit() + + def append_raw_observation( + self, + *, + source: str, + city: str, + value: Any = None, + observed_at: str = "", + fetched_at: str = "", + station_code: str = "", + station_name: str = "", + runway: str = "", + value_unit: str = "", + source_latency_sec: Any = None, + status: str = "ok", + error_count: int = 0, + last_success_at: str = "", + payload: Optional[Dict[str, Any]] = None, + ) -> None: + normalized_source = str(source or "").strip().lower() + normalized_city = str(city or "").strip().lower() + if not normalized_source or not normalized_city: + return + safe_station_code = str(station_code or "").strip().upper() + safe_station_name = str(station_name or "").strip() + safe_runway = str(runway or "").strip().upper() + safe_observed_at = str(observed_at or "").strip() + now_dt = datetime.now() + safe_fetched_at = str(fetched_at or now_dt.isoformat()).strip() + safe_status = str(status or "ok").strip().lower() or "ok" + value_float = self._float_or_none(value) + latency_float = self._float_or_none(source_latency_sec) + payload_json = json.dumps(payload or {}, ensure_ascii=False) + created_at_ts = now_dt.timestamp() + success_at = str(last_success_at or (safe_fetched_at if safe_status == "ok" else "")).strip() + with self._get_connection() as conn: + conn.execute( + """ + INSERT INTO raw_observation_store ( + source, city, station_code, station_name, runway, value, + value_unit, observed_at, fetched_at, source_latency_sec, + status, error_count, last_success_at, payload_json, created_at_ts + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + normalized_source, + normalized_city, + safe_station_code, + safe_station_name, + safe_runway, + value_float, + str(value_unit or "").strip(), + safe_observed_at, + safe_fetched_at, + latency_float, + safe_status, + max(0, int(error_count or 0)), + success_at, + payload_json, + created_at_ts, + ), + ) + conn.execute( + """ + INSERT INTO raw_observation_latest ( + source, city, station_code, station_name, runway, value, + value_unit, observed_at, fetched_at, source_latency_sec, + status, error_count, last_success_at, payload_json, updated_at_ts + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(source, city, station_code, runway) DO UPDATE SET + station_name = excluded.station_name, + value = excluded.value, + value_unit = excluded.value_unit, + observed_at = excluded.observed_at, + fetched_at = excluded.fetched_at, + source_latency_sec = excluded.source_latency_sec, + status = excluded.status, + error_count = excluded.error_count, + last_success_at = excluded.last_success_at, + payload_json = excluded.payload_json, + updated_at_ts = excluded.updated_at_ts + """, + ( + normalized_source, + normalized_city, + safe_station_code, + safe_station_name, + safe_runway, + value_float, + str(value_unit or "").strip(), + safe_observed_at, + safe_fetched_at, + latency_float, + safe_status, + max(0, int(error_count or 0)), + success_at, + payload_json, + created_at_ts, + ), + ) + conn.commit() + + def get_latest_raw_observation( + self, + source: str, + city: str, + *, + station_code: str = "", + runway: str = "", + ) -> Optional[Dict[str, Any]]: + normalized_source = str(source or "").strip().lower() + normalized_city = str(city or "").strip().lower() + if not normalized_source or not normalized_city: + return None + filters = ["source = ?", "city = ?"] + params: List[Any] = [normalized_source, normalized_city] + if station_code: + filters.append("station_code = ?") + params.append(str(station_code or "").strip().upper()) + if runway: + filters.append("runway = ?") + params.append(str(runway or "").strip().upper()) + with self._get_connection() as conn: + conn.row_factory = sqlite3.Row + row = conn.execute( + f""" + SELECT * + FROM raw_observation_latest + WHERE {' AND '.join(filters)} + ORDER BY updated_at_ts DESC + LIMIT 1 + """, + params, + ).fetchone() + if not row: + return None + try: + payload = json.loads(str(row["payload_json"] or "{}")) + except Exception: + payload = {} + if not isinstance(payload, dict): + payload = {} + return { + "source": str(row["source"] or ""), + "city": str(row["city"] or ""), + "station_code": str(row["station_code"] or ""), + "station_name": str(row["station_name"] or ""), + "runway": str(row["runway"] or ""), + "value": self._float_or_none(row["value"]), + "value_unit": str(row["value_unit"] or ""), + "observed_at": str(row["observed_at"] or ""), + "fetched_at": str(row["fetched_at"] or ""), + "source_latency_sec": self._float_or_none(row["source_latency_sec"]), + "status": str(row["status"] or ""), + "error_count": int(row["error_count"] or 0), + "last_success_at": str(row["last_success_at"] or ""), + "payload": payload, + "updated_at_ts": float(row["updated_at_ts"] or 0.0), + } + + def enqueue_observation_refresh_request( + self, + *, + city: str, + kind: str = "", + source: str = "", + priority: str = "normal", + reason: str = "", + ) -> bool: + normalized_city = str(city or "").strip().lower() + normalized_kind = str(kind or "").strip().lower() + normalized_source = str(source or "").strip().lower() + normalized_priority = str(priority or "normal").strip().lower() + if normalized_priority not in {"high", "normal", "low"}: + normalized_priority = "normal" + if not normalized_city: + return False + priority_rank = {"low": 0, "normal": 1, "high": 2} + now_dt = datetime.now() + now = now_dt.isoformat() + now_ts = now_dt.timestamp() + with self._get_connection() as conn: + existing = conn.execute( + """ + SELECT id, priority + FROM observation_refresh_requests + WHERE city = ? AND source = ? AND status IN ('pending', 'claimed') + ORDER BY requested_at_ts DESC + LIMIT 1 + """, + (normalized_city, normalized_source), + ).fetchone() + if existing: + existing_priority = str(existing[1] or "normal").strip().lower() + if priority_rank.get(existing_priority, 1) > priority_rank[normalized_priority]: + normalized_priority = existing_priority + conn.execute( + """ + UPDATE observation_refresh_requests + SET kind = ?, priority = ?, reason = ?, requested_at = ?, requested_at_ts = ? + WHERE id = ? + """, + ( + normalized_kind, + normalized_priority, + str(reason or "").strip(), + now, + now_ts, + int(existing[0]), + ), + ) + else: + conn.execute( + """ + INSERT INTO observation_refresh_requests ( + city, kind, source, priority, reason, status, + requested_at, requested_at_ts + ) + VALUES (?, ?, ?, ?, ?, 'pending', ?, ?) + """, + ( + normalized_city, + normalized_kind, + normalized_source, + normalized_priority, + str(reason or "").strip(), + now, + now_ts, + ), + ) + conn.commit() + return True + + def claim_observation_refresh_requests( + self, + *, + limit: int = 20, + owner: str = "", + now_ts: Optional[float] = None, + ) -> List[Dict[str, Any]]: + safe_limit = max(1, min(int(limit or 20), 200)) + safe_owner = str(owner or "").strip() or secrets.token_hex(6) + claim_ts = float(now_ts if now_ts is not None else datetime.now().timestamp()) + with self._get_connection() as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute( + """ + SELECT * + FROM observation_refresh_requests + WHERE status = 'pending' + ORDER BY + CASE priority WHEN 'high' THEN 0 WHEN 'normal' THEN 1 ELSE 2 END, + requested_at_ts ASC + LIMIT ? + """, + (safe_limit,), + ).fetchall() + ids = [int(row["id"]) for row in rows] + if ids: + placeholders = ",".join("?" for _ in ids) + conn.execute( + f""" + UPDATE observation_refresh_requests + SET status = 'claimed', + owner = ?, + attempts = attempts + 1, + claimed_at_ts = ? + WHERE id IN ({placeholders}) + """, + [safe_owner, claim_ts, *ids], + ) + conn.commit() + return [ + { + "id": int(row["id"]), + "city": str(row["city"] or ""), + "kind": str(row["kind"] or ""), + "source": str(row["source"] or ""), + "priority": str(row["priority"] or ""), + "reason": str(row["reason"] or ""), + "status": "claimed", + "attempts": int(row["attempts"] or 0) + 1, + "owner": safe_owner, + "requested_at": str(row["requested_at"] or ""), + "requested_at_ts": float(row["requested_at_ts"] or 0.0), + "claimed_at_ts": claim_ts, + "last_error": str(row["last_error"] or ""), + } + for row in rows + ] + + def mark_observation_refresh_request_done( + self, + request_id: int, + *, + status: str = "done", + error: str = "", + ) -> None: + safe_status = str(status or "done").strip().lower() + if safe_status not in {"done", "failed"}: + safe_status = "done" + with self._get_connection() as conn: + conn.execute( + """ + UPDATE observation_refresh_requests + SET status = ?, + completed_at_ts = ?, + last_error = ? + WHERE id = ? + """, + ( + safe_status, + datetime.now().timestamp(), + str(error or "").strip(), + int(request_id), + ), + ) + conn.commit() + def acquire_cache_refresh_lock( self, cache_key: str, diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index e4c53636..22fb5eb8 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -27,6 +27,7 @@ from src.utils.telegram_i18n import ( 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. @@ -1504,9 +1505,11 @@ def _cached_payload_observation_epoch(payload: Dict[str, Any]) -> Optional[int]: 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 = [ @@ -1574,19 +1577,37 @@ def _read_cached_airport_city_weather(city: str, max_age_sec: Optional[int] = No return None +def _read_canonical_airport_city_weather(city: str) -> Optional[Dict[str, Any]]: + normalized_city = (city or "").strip().lower() + if not normalized_city: + return None + try: + db = 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 + return build_city_weather_from_canonical(normalized_city, canonical) + + def _load_airport_city_weather_for_push(city: str) -> Dict[str, Any]: cached = _read_cached_airport_city_weather(city) if cached is not None: return cached - from web.app import _analyze # lazy import - only the bot process needs it + canonical = _read_canonical_airport_city_weather(city) + if canonical is not None: + return canonical - return _analyze( - city, - force_refresh=False, - force_refresh_observations_only=False, - detail_mode="panel", - ) + raise RuntimeError(f"no cached city weather for airport push city={city}") # Per-city temperature window threshold (°C below DEB predicted high) diff --git a/tests/test_canonical_temperature.py b/tests/test_canonical_temperature.py new file mode 100644 index 00000000..4f60734f --- /dev/null +++ b/tests/test_canonical_temperature.py @@ -0,0 +1,445 @@ +from __future__ import annotations + +import asyncio +from types import SimpleNamespace + + +def _sample_panel_payload() -> dict: + return { + "name": "shanghai", + "temp_symbol": "°C", + "updated_at": "2026-06-14T01:02:03+00:00", + "current": { + "temp": 31.2, + "source_code": "amsc_awos", + "settlement_source": "amsc_awos", + "settlement_source_label": "AMSC AWOS runway-point air temperature", + "station_code": "ZSPD", + "station_name": "Shanghai Pudong", + "observed_at": "2026-06-14T01:01:00+00:00", + "obs_time": "09:01", + "freshness": { + "freshness_status": "fresh", + "age_sec": 63, + "observed_at": "2026-06-14T01:01:00+00:00", + "observed_at_local": "09:01", + }, + }, + } + + +def test_db_manager_stores_canonical_temperature_latest(tmp_path): + from src.database.db_manager import DBManager + + db = DBManager(str(tmp_path / "polyweather.db")) + canonical = { + "city": "shanghai", + "value": 31.2, + "source": "amsc_awos", + "source_role": "settlement_proxy", + "observed_at": "2026-06-14T01:01:00+00:00", + "fetched_at": "2026-06-14T01:02:03+00:00", + "freshness_sec": 63, + "freshness_status": "fresh", + "confidence": 0.92, + "explanation": "AMSC AWOS runway-point air temperature updated 63s ago.", + } + + db.set_canonical_temperature("Shanghai", canonical) + + row = db.get_canonical_temperature("shanghai") + assert row is not None + assert row["city"] == "shanghai" + assert row["value"] == 31.2 + assert row["source"] == "amsc_awos" + assert row["source_role"] == "settlement_proxy" + assert row["freshness_status"] == "fresh" + assert row["payload"]["observed_at"] == "2026-06-14T01:01:00+00:00" + + +def test_refresh_city_panel_cache_persists_canonical_temperature(monkeypatch): + import web.services.city_runtime as city_runtime + + writes = {} + + class FakeDB: + def set_city_cache(self, kind, city, payload, **_kwargs): + writes["city_cache"] = (kind, city, payload) + + def set_canonical_temperature(self, city, payload): + writes["canonical"] = (city, payload) + + monkeypatch.setattr(city_runtime, "_CACHE_DB", FakeDB()) + monkeypatch.setattr( + city_runtime, + "_analyze", + lambda city, force_refresh=False, detail_mode="panel": _sample_panel_payload(), + ) + + payload = city_runtime._refresh_city_panel_cache("shanghai") + + assert payload["canonical_temperature"]["value"] == 31.2 + assert payload["canonical_temperature"]["source"] == "amsc_awos" + assert payload["canonical_temperature"]["freshness_sec"] == 63 + assert payload["canonical_temperature"]["source_role"] == "settlement_proxy" + assert writes["canonical"][0] == "shanghai" + assert writes["canonical"][1]["observed_at"] == "2026-06-14T01:01:00+00:00" + + +def test_city_panel_cold_cache_returns_canonical_latest_without_sync_refresh(monkeypatch): + import web.services.city_api as city_api + + started = [] + + 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 { + "payload": { + "city": "shanghai", + "value": 31.2, + "temp_symbol": "°C", + "source": "amsc_awos", + "source_label": "AMSC AWOS runway-point air temperature", + "source_role": "settlement_proxy", + "observed_at": "2026-06-14T01:01:00+00:00", + "observed_at_local": "09:01", + "freshness_sec": 63, + "freshness_status": "fresh", + "fetched_at": "2026-06-14T01:02:03+00:00", + "confidence": 0.92, + } + } + + def fail_refresh(*_args, **_kwargs): + raise AssertionError("cold panel cache must not synchronously refresh when canonical latest exists") + + 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( + object(), + "Shanghai", + force_refresh=False, + depth="panel", + ) + ) + + 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")] + + +def test_city_full_cold_cache_returns_canonical_latest_without_sync_refresh(monkeypatch): + import web.services.city_api as city_api + + started = [] + + class FakeDB: + def get_city_cache(self, kind, city): + assert kind == "full" + assert city == "shanghai" + return None + + def get_canonical_temperature(self, city): + assert city == "shanghai" + return { + "payload": { + "city": "shanghai", + "value": 31.2, + "temp_symbol": "°C", + "source": "amsc_awos", + "source_label": "AMSC AWOS runway-point air temperature", + "source_role": "settlement_proxy", + "observed_at": "2026-06-14T01:01:00+00:00", + "observed_at_local": "09:01", + "freshness_sec": 63, + "freshness_status": "fresh", + "fetched_at": "2026-06-14T01:02:03+00:00", + "confidence": 0.92, + } + } + + 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"] + + +def test_city_nearby_and_market_cold_cache_return_canonical_latest_without_sync_refresh(monkeypatch): + import web.services.city_api as city_api + + started = [] + requested_kinds = [] + + class FakeDB: + def get_city_cache(self, kind, city): + requested_kinds.append((kind, city)) + return None + + def get_canonical_temperature(self, city): + assert city == "shanghai" + return { + "payload": { + "city": "shanghai", + "value": 31.2, + "temp_symbol": "°C", + "source": "amsc_awos", + "source_label": "AMSC AWOS runway-point air temperature", + "source_role": "settlement_proxy", + "observed_at": "2026-06-14T01:01:00+00:00", + "observed_at_local": "09:01", + "freshness_sec": 63, + "freshness_status": "fresh", + "fetched_at": "2026-06-14T01:02:03+00:00", + "confidence": 0.92, + } + } + + def fail_refresh(*_args, **_kwargs): + raise AssertionError("cold city cache must not synchronously refresh when canonical latest exists") + + monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower()) + monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None) + 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") + ) + market = asyncio.run( + city_api.get_city_detail_payload(object(), "Shanghai", force_refresh=False, depth="market") + ) + + assert nearby["detail_depth"] == "nearby" + 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 requested_kinds == [("nearby", "shanghai"), ("market", "shanghai")] + + +def test_force_refresh_panel_returns_canonical_latest_without_waiting_for_sync_refresh(monkeypatch): + import web.services.city_api as city_api + + 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 { + "payload": { + "city": "shanghai", + "value": 31.2, + "temp_symbol": "°C", + "source": "amsc_awos", + "source_label": "AMSC AWOS runway-point air temperature", + "source_role": "settlement_proxy", + "observed_at": "2026-06-14T01:01:00+00:00", + "observed_at_local": "09:01", + "freshness_sec": 63, + "freshness_status": "fresh", + "fetched_at": "2026-06-14T01:02:03+00:00", + "confidence": 0.92, + } + } + + 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) + + payload = asyncio.run( + city_api.get_city_detail_payload( + object(), + "Shanghai", + force_refresh=True, + depth="panel", + ) + ) + + assert payload["current"]["temp"] == 31.2 + assert payload["canonical_temperature"]["source"] == "amsc_awos" + assert payload["detail_depth"] == "panel" + + +def test_city_panel_canonical_fallback_enqueues_collector_refresh_when_queue_exists(monkeypatch): + import web.services.city_api as city_api + + 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 { + "payload": { + "city": "shanghai", + "value": 31.2, + "temp_symbol": "°C", + "source": "amsc_awos", + "source_label": "AMSC AWOS runway-point air temperature", + "source_role": "settlement_proxy", + "observed_at": "2026-06-14T01:01:00+00:00", + "observed_at_local": "09:01", + "freshness_sec": 63, + "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_start(*_args, **_kwargs): + raise AssertionError("canonical fallback should enqueue collector refresh instead of direct stale refresh") + + 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, "_start_city_cache_stale_refresh", fail_start) + + payload = asyncio.run( + city_api.get_city_detail_payload( + object(), + "Shanghai", + force_refresh=False, + depth="panel", + ) + ) + + assert payload["current"]["temp"] == 31.2 + assert enqueued == [ + { + "city": "shanghai", + "kind": "panel", + "priority": "high", + "reason": "canonical_fallback", + } + ] + + +def test_city_panel_cold_start_without_canonical_returns_initializing_payload(monkeypatch): + import web.services.city_api as city_api + + 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_refresh(*_args, **_kwargs): + raise AssertionError("cold start without canonical must not synchronously refresh") + + 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) + + payload = asyncio.run( + city_api.get_city_detail_payload( + object(), + "Shanghai", + force_refresh=False, + depth="panel", + ) + ) + + assert payload["name"] == "shanghai" + assert payload["detail_depth"] == "panel" + assert payload["status"] == "initializing" + assert payload["current"]["temp"] is None + assert enqueued == [ + { + "city": "shanghai", + "kind": "panel", + "priority": "high", + "reason": "cold_start", + } + ] + + +def test_dashboard_init_uses_safe_city_detail_path_without_direct_refresh(monkeypatch): + import web.services.dashboard_init_api as dashboard_init_api + + calls = [] + + class FakeRequest: + headers = {} + state = SimpleNamespace() + + class FakeDB: + def get_city_cache(self, kind, city): + assert kind == "panel" + assert city == "shanghai" + return None + + def fail_refresh(*_args, **_kwargs): + raise AssertionError("dashboard init must not directly refresh city panel cache") + + async def fake_get_city_detail_payload(request, name, *, force_refresh=False, depth="panel"): + calls.append((name, force_refresh, depth)) + return {"name": name, "status": "initializing"} + + monkeypatch.setattr(dashboard_init_api.legacy_routes, "_bind_optional_supabase_identity", lambda request: None) + monkeypatch.setattr(dashboard_init_api, "_build_cities_payload", lambda: {"cities": []}) + monkeypatch.setattr(dashboard_init_api, "_resolve_default_city", lambda request: "shanghai") + monkeypatch.setattr(dashboard_init_api.legacy_routes, "_CACHE_DB", FakeDB()) + monkeypatch.setattr(dashboard_init_api.legacy_routes, "_refresh_city_panel_cache", fail_refresh) + monkeypatch.setattr(dashboard_init_api, "get_city_detail_payload", fake_get_city_detail_payload, raising=False) + + payload = asyncio.run(dashboard_init_api.build_dashboard_init_payload(FakeRequest())) + + assert payload["default_city_detail"] == {"name": "shanghai", "status": "initializing"} + assert calls == [("shanghai", False, "panel")] diff --git a/tests/test_city_realtime_stream.py b/tests/test_city_realtime_stream.py new file mode 100644 index 00000000..6c544742 --- /dev/null +++ b/tests/test_city_realtime_stream.py @@ -0,0 +1,51 @@ +from __future__ import annotations + + +def test_realtime_stream_reads_canonical_latest_without_analysis(monkeypatch): + import web.services.city_realtime_stream as stream + + stream._STREAM_BUFFERS.clear() + + class FakeDB: + def get_canonical_temperature(self, city): + assert city == "shanghai" + return { + "payload": { + "city": "shanghai", + "value": 30.5, + "source": "amsc_awos", + "source_label": "AMSC AWOS runway-point air temperature", + "observed_at": "2026-06-14T04:00:00+00:00", + "freshness_status": "fresh", + }, + } + + def get_city_cache(self, kind, city): + assert city == "shanghai" + if kind != "panel": + return None + return { + "payload": { + "probabilities": { + "distribution": [ + {"temp": 30.0, "label": "30°C"}, + {"temp": 32.0, "label": "32°C"}, + ] + } + } + } + + def fail_analyze(*_args, **_kwargs): + raise AssertionError("realtime stream must read cache/latest instead of _analyze") + + monkeypatch.setattr(stream, "_CACHE_DB", FakeDB(), raising=False) + monkeypatch.setattr(stream, "_analyze", fail_analyze) + + payload = stream.get_realtime_stream_payload("shanghai") + + assert payload["points"][-1] == { + "timestamp": "2026-06-14T04:00:00+00:00", + "temp": 30.5, + "source": "amsc_awos", + } + assert [row["threshold_c"] for row in payload["thresholds"]] == [30.0, 32.0] diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index d746f9b2..04e943b5 100644 --- a/tests/test_observation_collector.py +++ b/tests/test_observation_collector.py @@ -135,6 +135,169 @@ def test_observation_collector_run_due_once_refreshes_panel_cache(): assert refreshed == ["qingdao", "qingdao"] +def test_raw_observation_store_records_latest_observation(tmp_path): + from src.database.db_manager import DBManager + + db = DBManager(str(tmp_path / "polyweather.db")) + + db.append_raw_observation( + source="amsc_awos", + city="Qingdao", + value=24.0, + observed_at="2026-06-14T01:00:00+00:00", + fetched_at="2026-06-14T01:01:00+00:00", + station_code="ZSQD", + station_name="Qingdao Jiaodong", + status="ok", + payload={"temp_c": 24.0}, + ) + + latest = db.get_latest_raw_observation("amsc_awos", "qingdao") + + assert latest is not None + assert latest["source"] == "amsc_awos" + assert latest["city"] == "qingdao" + assert latest["value"] == 24.0 + assert latest["observed_at"] == "2026-06-14T01:00:00+00:00" + assert latest["fetched_at"] == "2026-06-14T01:01:00+00:00" + assert latest["station_code"] == "ZSQD" + assert latest["status"] == "ok" + assert latest["payload"]["temp_c"] == 24.0 + + +def test_observation_refresh_request_queue_claims_pending_requests(tmp_path): + from src.database.db_manager import DBManager + + db = DBManager(str(tmp_path / "polyweather.db")) + + assert db.enqueue_observation_refresh_request( + city="Shanghai", + kind="panel", + priority="high", + reason="cold_canonical_fallback", + ) + + claimed = db.claim_observation_refresh_requests(limit=5, owner="collector-1", now_ts=1000.0) + + assert len(claimed) == 1 + request = claimed[0] + assert request["city"] == "shanghai" + assert request["kind"] == "panel" + assert request["priority"] == "high" + assert request["reason"] == "cold_canonical_fallback" + + db.mark_observation_refresh_request_done(request["id"], status="done") + + assert db.claim_observation_refresh_requests(limit=5, owner="collector-1", now_ts=1001.0) == [] + + +def test_observation_refresh_request_queue_coalesces_city_source_across_kinds(tmp_path): + from src.database.db_manager import DBManager + + db = DBManager(str(tmp_path / "polyweather.db")) + + assert db.enqueue_observation_refresh_request( + city="Shanghai", + kind="panel", + priority="high", + reason="panel_cold_start", + ) + assert db.enqueue_observation_refresh_request( + city="shanghai", + kind="full", + priority="normal", + reason="chart_cold_start", + ) + + claimed = db.claim_observation_refresh_requests(limit=5, owner="collector-1", now_ts=1000.0) + + assert len(claimed) == 1 + request = claimed[0] + assert request["city"] == "shanghai" + assert request["kind"] == "full" + assert request["priority"] == "high" + assert request["reason"] == "chart_cold_start" + + +def test_observation_collector_writes_raw_observation_store(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", + "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 + + latest = db.get_latest_raw_observation("amsc_awos", "qingdao") + assert latest is not None + assert latest["value"] == 24.0 + assert latest["observed_at"] == "2026-06-14T01:00:00+00:00" + assert latest["station_code"] == "ZSQD" + + +def test_observation_collector_consumes_refresh_request_queue(tmp_path): + from src.database.db_manager import DBManager + from web.observation_collector_service import ( + ObservationCollector, + ObservationSourceProfile, + ) + + db = DBManager(str(tmp_path / "polyweather.db")) + db.enqueue_observation_refresh_request( + city="qingdao", + kind="panel", + priority="high", + reason="canonical_fallback", + ) + calls = [] + + class FakeWeather: + def _uses_fahrenheit(self, city): + return False + + def _attach_china_amsc_awos_data(self, results, city, use_fahrenheit): + calls.append(city) + results["amos"] = { + "source": "amsc_awos", + "temp_c": 24.0, + "observation_time": "2026-06-14T01:00:00+00:00", + "icao": "ZSQD", + } + + collector = ObservationCollector( + weather=FakeWeather(), + profiles=[ObservationSourceProfile("amsc_awos", ("qingdao",), 180)], + observation_store=db, + async_cache_refresh=False, + ) + + assert collector.run_due_once(now_ts=100.0) == 1 + assert calls == ["qingdao"] + assert db.claim_observation_refresh_requests(limit=5, owner="test", now_ts=101.0) == [] + + def test_observation_collector_cache_refresh_does_not_block_source_polling(): from web.observation_collector_service import ( ObservationCollector, diff --git a/tests/test_scan_terminal_modules.py b/tests/test_scan_terminal_modules.py index 4a2ac38b..d08d185e 100644 --- a/tests/test_scan_terminal_modules.py +++ b/tests/test_scan_terminal_modules.py @@ -1,5 +1,6 @@ from web.scan_terminal_filters import normalize_scan_terminal_filters from web import scan_terminal_cache +from web import scan_terminal_service from web.scan_terminal_metar_gate import _apply_metar_gate_to_row from web.scan_terminal_payloads import ( build_failed_scan_terminal_payload, @@ -82,6 +83,36 @@ def test_scan_terminal_prewarm_covers_default_api_limit(): assert 180 in limits +def test_scan_terminal_prewarm_queues_city_refresh_without_analyze(monkeypatch): + enqueued = [] + + class _DB: + @staticmethod + def enqueue_observation_refresh_request(**kwargs): + enqueued.append(kwargs) + return True + + monkeypatch.setattr(scan_terminal_service, "_SCAN_PREWARM_DB", _DB(), raising=False) + monkeypatch.setattr( + scan_terminal_service, + "_analyze", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("scan terminal prewarm must not fetch external sources") + ), + raising=False, + ) + + assert scan_terminal_service._queue_scan_terminal_city_prewarm("shenzhen") == "shenzhen" + assert enqueued == [ + { + "city": "shenzhen", + "kind": "panel", + "priority": "normal", + "reason": "scan_terminal_prewarm", + } + ] + + def test_scan_city_terminal_rows_reuses_persisted_panel_cache(monkeypatch): payload = { "display_name": "Paris", @@ -119,6 +150,118 @@ def test_scan_city_terminal_rows_reuses_persisted_panel_cache(monkeypatch): assert result["rows"][0]["current_temp"] == 18.0 +def test_scan_city_terminal_rows_uses_canonical_without_analyze(monkeypatch): + enqueued = [] + + class _Cache: + @staticmethod + def get_city_cache(kind, city): + assert (kind, city) == ("panel", "qingdao") + return None + + @staticmethod + def get_canonical_temperature(city): + assert city == "qingdao" + return { + "payload": { + "city": "qingdao", + "value": 22.5, + "temp_symbol": "°C", + "source": "amsc_awos", + "source_label": "AMSC AWOS", + "source_role": "settlement_proxy", + "observed_at": "2026-06-01T08:00:00Z", + "fetched_at": "2026-06-01T08:00:30Z", + "freshness_sec": 30, + "freshness_status": "fresh", + "confidence": 0.92, + } + } + + @staticmethod + def enqueue_observation_refresh_request(**kwargs): + enqueued.append(kwargs) + return True + + monkeypatch.setattr(scan_terminal_city_row, "_PANEL_CACHE_DB", _Cache()) + monkeypatch.setattr( + scan_terminal_city_row, + "_analyze", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("scan terminal must not fetch external sources") + ), + ) + + result = scan_terminal_city_row._scan_city_terminal_rows( + "qingdao", + {"market_type": "maxtemp"}, + force_refresh=False, + ) + + assert result["city"] == "qingdao" + assert result["candidate_total"] == 1 + assert result["rows"][0]["current_temp"] == 22.5 + assert result["rows"][0]["temp_symbol"] == "°C" + assert enqueued == [ + { + "city": "qingdao", + "kind": "panel", + "priority": "high", + "reason": "scan_terminal_canonical_fallback", + } + ] + + +def test_scan_city_terminal_rows_cold_start_only_enqueues(monkeypatch): + enqueued = [] + + class _Cache: + @staticmethod + def get_city_cache(kind, city): + assert (kind, city) == ("panel", "seoul") + return None + + @staticmethod + def get_canonical_temperature(city): + assert city == "seoul" + return None + + @staticmethod + def enqueue_observation_refresh_request(**kwargs): + enqueued.append(kwargs) + return True + + monkeypatch.setattr(scan_terminal_city_row, "_PANEL_CACHE_DB", _Cache()) + monkeypatch.setattr( + scan_terminal_city_row, + "_analyze", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("scan terminal must not fetch external sources") + ), + ) + + result = scan_terminal_city_row._scan_city_terminal_rows( + "seoul", + {"market_type": "maxtemp"}, + force_refresh=False, + ) + + assert result == { + "city": "seoul", + "rows": [], + "candidate_total": 0, + "primary_scores": [], + } + assert enqueued == [ + { + "city": "seoul", + "kind": "panel", + "priority": "high", + "reason": "scan_terminal_cold_start", + } + ] + + def test_scan_router_does_not_expose_terminal_ai_endpoint(): routes = { getattr(route, "path", None): getattr(route, "methods", set()) diff --git a/tests/test_telegram_hashtags.py b/tests/test_telegram_hashtags.py index a9e512a0..d38cd4f7 100644 --- a/tests/test_telegram_hashtags.py +++ b/tests/test_telegram_hashtags.py @@ -1,3 +1,5 @@ +import pytest + from src.utils.telegram_push import ( HIGH_FREQ_AIRPORT_CITIES, HIGH_FREQ_AIRPORT_ICAO, @@ -294,36 +296,64 @@ def test_airport_push_prefers_cache_with_newest_observation(monkeypatch): assert city_weather["airport_primary"]["temp"] == 31.4 -def test_airport_push_fallback_analysis_does_not_force_observation_refresh(monkeypatch): +def test_airport_push_without_cache_uses_canonical_latest_not_analysis(monkeypatch): import src.utils.telegram_push as telegram_push import web.app as web_app - calls = [] - class FakeDB: def get_city_cache(self, kind, city): return None - def fake_analyze(city, force_refresh=False, force_refresh_observations_only=False, detail_mode="full", **_kwargs): - calls.append((city, force_refresh, force_refresh_observations_only, detail_mode)) - return { - "local_time": "12:00", - "current": {"temp": 31.0}, - "deb": {"prediction": 29.0}, - "airport_current": {"max_so_far": 30.0, "max_temp_time": "11:50", "obs_time": "12:00"}, - "mgm_nearby": [ - {"icao": "ZSQD", "temp": 31.0, "obs_time": "2026-05-17T04:00:00Z"}, - ], - } + def get_canonical_temperature(self, city): + assert city == "qingdao" + return { + "payload": { + "city": "qingdao", + "value": 31.0, + "source": "amsc_awos", + "source_label": "AMSC AWOS runway-point air temperature", + "source_role": "settlement_proxy", + "observed_at": "2026-06-14T04:00:00+00:00", + "observed_at_local": "12:00", + "freshness_sec": 45, + "freshness_status": "fresh", + "fetched_at": "2026-06-14T04:00:45+00:00", + "confidence": 0.92, + }, + } + + def fail_analyze(*_args, **_kwargs): + raise AssertionError("Telegram push must not call _analyze when canonical latest exists") monkeypatch.setattr(telegram_push, "DBManager", lambda: FakeDB()) - monkeypatch.setattr(web_app, "_analyze", fake_analyze) + monkeypatch.setattr(web_app, "_analyze", fail_analyze) city_weather = telegram_push._load_airport_city_weather_for_push("qingdao") assert city_weather["current"]["temp"] == 31.0 - assert ("qingdao", False, False, "panel") in calls - assert not any(city == "qingdao" and force_obs for city, _force, force_obs, _mode in calls) + assert city_weather["airport_primary"]["obs_time"] == "2026-06-14T04:00:00+00:00" + assert city_weather["current"]["freshness"]["freshness_status"] == "fresh" + + +def test_airport_push_without_cache_or_canonical_skips_analysis(monkeypatch): + import src.utils.telegram_push as telegram_push + import web.app as web_app + + class FakeDB: + def get_city_cache(self, kind, city): + return None + + def get_canonical_temperature(self, city): + return None + + def fail_analyze(*_args, **_kwargs): + raise AssertionError("Telegram push must not call _analyze on cold cache") + + monkeypatch.setattr(telegram_push, "DBManager", lambda: FakeDB()) + monkeypatch.setattr(web_app, "_analyze", fail_analyze) + + with pytest.raises(RuntimeError, match="no cached city weather"): + telegram_push._load_airport_city_weather_for_push("qingdao") def test_airport_push_uses_stale_cache_before_fallback_analysis(monkeypatch): diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index 9d3d21c4..f936f86f 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -129,6 +129,69 @@ def test_system_cache_status_requires_ops_admin(monkeypatch): assert response.status_code in {401, 403, 503} +def test_system_priority_warm_enqueues_collector_refresh_without_direct_refresh(monkeypatch): + from web.services import system_api + + tasks = [] + enqueued = [] + + class _BackgroundTasks: + @staticmethod + def add_task(fn, *args, **kwargs): + tasks.append((fn, args, kwargs)) + + class _Cache: + @staticmethod + def enqueue_observation_refresh_request(**kwargs): + enqueued.append(kwargs) + return True + + def fail_refresh(*_args, **_kwargs): + raise AssertionError("priority warm must enqueue collector refreshes") + + monkeypatch.setattr(system_api.legacy_routes, "_assert_entitlement", lambda request: None) + monkeypatch.setattr( + system_api.legacy_routes, + "_select_priority_city_batches", + lambda timezone: { + "region": "asia", + "timezone": timezone, + "primary": ["shenzhen"], + "secondary": ["seoul"], + }, + ) + monkeypatch.setattr(system_api.legacy_routes, "_CACHE_DB", _Cache()) + monkeypatch.setattr(system_api.legacy_routes, "_refresh_city_summary_cache", fail_refresh) + monkeypatch.setattr(system_api.legacy_routes, "_refresh_city_panel_cache", fail_refresh) + monkeypatch.setattr(system_api.legacy_routes, "_refresh_city_nearby_cache", fail_refresh) + monkeypatch.setattr(system_api.legacy_routes, "_refresh_city_market_cache", fail_refresh) + monkeypatch.setattr(system_api.legacy_routes, "_refresh_city_full_cache", fail_refresh) + + payload = system_api.run_system_priority_warm( + object(), + _BackgroundTasks(), + timezone="Asia/Shanghai", + ) + tasks[0][0](*tasks[0][1], **tasks[0][2]) + + assert payload["primary"] == ["shenzhen"] + assert payload["secondary"] == ["seoul"] + assert enqueued == [ + { + "city": "shenzhen", + "kind": "panel", + "priority": "high", + "reason": "system_priority_warm", + }, + { + "city": "seoul", + "kind": "panel", + "priority": "normal", + "reason": "system_priority_warm", + }, + ] + + def test_standard_growth_funnel_events_are_trackable(): assert { "landing_view", @@ -1531,7 +1594,7 @@ def test_concurrent_city_detail_batch_requests_share_inflight_response(monkeypat assert build_calls == 1 -def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypatch): +def test_concurrent_cold_city_detail_requests_return_initializing_without_full_refresh(monkeypatch): import asyncio refresh_calls = 0 @@ -1587,8 +1650,9 @@ def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypat results = asyncio.run(run_two_requests()) assert [item["city"] for item in results] == ["paris", "paris"] - assert refresh_calls == 1 - assert build_calls == 1 + assert [item["status"] for item in results] == ["initializing", "initializing"] + assert refresh_calls == 0 + assert build_calls == 0 def test_stale_city_detail_uses_cached_full_payload_while_refreshing(monkeypatch): @@ -1887,22 +1951,11 @@ def test_force_refresh_full_detail_returns_cached_payload_when_refresh_is_slow(m assert refresh_calls == 1 -def test_force_refresh_invalidates_short_city_detail_payload_cache(monkeypatch): +def test_force_refresh_cold_city_detail_returns_initializing_without_full_refresh(monkeypatch): import asyncio build_calls = 0 - refreshed_payloads = [ - { - "city": "paris", - "local_date": "2026-05-30", - "hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]}, - }, - { - "city": "paris", - "local_date": "2026-05-30", - "hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [21.0]}, - }, - ] + refresh_calls = 0 class FakeCache: def get_city_cache(self, kind, city): @@ -1913,9 +1966,9 @@ def test_force_refresh_invalidates_short_city_detail_payload_cache(monkeypatch): return fn(*args, **kwargs) def refresh_full(city, force_refresh): - assert city == "paris" - assert refreshed_payloads - return refreshed_payloads.pop(0) + nonlocal refresh_calls + refresh_calls += 1 + raise AssertionError("cold force refresh must not synchronously refresh full city data") def build_detail(data, market_slug, target_date, resolution): nonlocal build_calls @@ -1947,9 +2000,10 @@ def test_force_refresh_invalidates_short_city_detail_payload_cache(monkeypatch): ), ) - assert first["live_temp"] == 20.0 - assert second["live_temp"] == 21.0 - assert build_calls == 2 + assert first["status"] == "initializing" + assert second["status"] == "initializing" + assert refresh_calls == 0 + assert build_calls == 0 def test_payment_runtime_endpoint_returns_shape(): diff --git a/web/analysis_service.py b/web/analysis_service.py index 6ac6dc58..f0ba30a6 100644 --- a/web/analysis_service.py +++ b/web/analysis_service.py @@ -1820,6 +1820,8 @@ def _analyze( "settlement_source_label": current_source_label, "station_code": current_station_code, "station_name": current_station_name, + "observed_at": current_obs_raw, + "observed_at_local": obs_time_str, "obs_time": obs_time_str, "obs_age_min": None if use_settlement_current else metar_age_min, "freshness": current_freshness, @@ -2316,8 +2318,11 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]: "max_so_far": _sf(display_settlement_max), "max_temp_time": max_temp_time, "wu_settlement": _sf(wu_settle), + "source_code": current_source, "settlement_source": current_source, "settlement_source_label": current_source_label, + "observed_at": obs_t or None, + "observed_at_local": obs_time_str or None, "obs_time": obs_time_str or None, "obs_age_min": obs_age_min, "observation_status": "live" if cur_temp is not None else "missing", diff --git a/web/observation_collector_service.py b/web/observation_collector_service.py index 49e5a95f..e0d837fb 100644 --- a/web/observation_collector_service.py +++ b/web/observation_collector_service.py @@ -15,6 +15,7 @@ from src.data_collection.amos_station_sources import AMOS_AIRPORT_CODES from src.data_collection.amsc_awos_sources import AMSC_AWOS_AIRPORTS 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 @@ -54,6 +55,7 @@ class ObservationCollector: profiles: Sequence[ObservationSourceProfile], cache_refresher: Optional[Callable[[str], Any]] = None, status_recorder: Optional[ObservationCollectorStatusRepository] = None, + observation_store: Optional[Any] = None, async_cache_refresh: Optional[bool] = None, cache_refresh_workers: Optional[int] = None, ) -> None: @@ -61,6 +63,7 @@ class ObservationCollector: self.profiles = list(profiles) self.cache_refresher = cache_refresher self.status_recorder = status_recorder + self.observation_store = observation_store self._last_run_ts: dict[tuple[str, str], float] = {} self._lock = threading.Lock() self._cache_refresh_lock = threading.Lock() @@ -89,7 +92,7 @@ class ObservationCollector: def run_due_once(self, *, now_ts: Optional[float] = None) -> int: now = float(time.time() if now_ts is None else now_ts) - due: List[tuple[ObservationSourceProfile, str]] = [] + due: List[tuple[ObservationSourceProfile, str, Optional[int]]] = [] with self._lock: for profile in self.profiles: interval = max(1, int(profile.interval_sec or 60)) @@ -98,10 +101,12 @@ class ObservationCollector: last_ts = float(self._last_run_ts.get(key) or 0.0) if now - last_ts >= interval: self._last_run_ts[key] = now - due.append((profile, city)) + due.append((profile, city, None)) + + due.extend(self._claim_due_refresh_requests(now)) completed = 0 - for profile, city in due: + for profile, city, request_id in due: started_wall = time.time() started_ts = now if now_ts is not None else started_wall ok = False @@ -132,8 +137,71 @@ class ObservationCollector: ok=ok, error=error, ) + self._mark_refresh_request_done(request_id, ok=ok, error=error) return completed + def _claim_due_refresh_requests(self, now: float) -> List[tuple[ObservationSourceProfile, str, Optional[int]]]: + store = self.observation_store + claimer = getattr(store, "claim_observation_refresh_requests", None) + if not callable(claimer): + return [] + try: + requests = claimer(limit=50, owner="observation_collector", now_ts=now) + except Exception as exc: + logger.debug("observation refresh request claim skipped: {}", exc) + return [] + due: List[tuple[ObservationSourceProfile, str, Optional[int]]] = [] + for request in requests or []: + if not isinstance(request, dict): + continue + city = str(request.get("city") or "").strip().lower() + requested_source = str(request.get("source") or "").strip().lower() + request_id = int(request.get("id") or 0) or None + matched = False + rate_limited = False + for profile in self.profiles: + profile_source = str(profile.source or "").strip().lower() + if requested_source and requested_source != profile_source: + continue + if city not in set(profile.cities): + continue + key = (profile.source, city) + interval = max(1, int(profile.interval_sec or 60)) + with self._lock: + last_ts = float(self._last_run_ts.get(key) or 0.0) + if last_ts and now - last_ts < interval: + rate_limited = True + continue + self._last_run_ts[key] = now + due.append((profile, city, request_id)) + matched = True + if not matched: + reason = "rate_limited" if rate_limited else "no_matching_profile" + self._mark_refresh_request_done(request_id, ok=False, error=reason) + return due + + def _mark_refresh_request_done( + self, + request_id: Optional[int], + *, + ok: bool, + error: Optional[str], + ) -> None: + if not request_id: + return + store = self.observation_store + marker = getattr(store, "mark_observation_refresh_request_done", None) + if not callable(marker): + return + try: + marker( + request_id, + status="done" if ok else "failed", + error="" if ok else str(error or "collection_failed"), + ) + except Exception as exc: + logger.debug("observation refresh request completion skipped id={}: {}", request_id, exc) + def _record_source_status( self, *, @@ -187,7 +255,104 @@ class ObservationCollector: else: logger.debug("observation collector skipped unknown source={}", normalized_source) return False - return bool(results) + ok = bool(results) + if ok: + self._store_raw_observations(normalized_source, normalized_city, results) + return ok + + @staticmethod + def _observation_value(row: dict[str, Any]) -> Optional[float]: + for key in ("temp_c", "temperature_c", "temp", "value"): + try: + value = row.get(key) + if value is not None and value != "": + return float(value) + except (TypeError, ValueError): + continue + current = row.get("current") + if isinstance(current, dict): + try: + value = current.get("temp") + if value is not None and value != "": + return float(value) + except (TypeError, ValueError): + return None + return None + + @staticmethod + def _observation_time(row: dict[str, Any]) -> str: + for key in ("observation_time", "observed_at", "obs_time", "time_utc", "time"): + value = str(row.get(key) or "").strip() + if value: + return value + return "" + + @staticmethod + def _station_code(row: dict[str, Any]) -> str: + for key in ("station_code", "icao", "istNo", "station_id", "code"): + value = str(row.get(key) or "").strip() + if value: + return value + return "" + + @staticmethod + def _station_name(row: dict[str, Any]) -> str: + for key in ("station_name", "station_label", "name", "label"): + value = str(row.get(key) or "").strip() + if value: + return value + return "" + + def _iter_raw_observation_rows( + self, + source: str, + results: dict[str, Any], + ) -> Iterable[dict[str, Any]]: + for value in (results or {}).values(): + if isinstance(value, dict): + yield value + continue + if isinstance(value, list): + for item in value: + if isinstance(item, dict): + yield item + + def _store_raw_observations(self, source: str, city: str, results: dict[str, Any]) -> None: + store = self.observation_store + writer = getattr(store, "append_raw_observation", None) + if not callable(writer): + return + fetched_at = time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime()) + wrote = 0 + for row in self._iter_raw_observation_rows(source, results): + value = self._observation_value(row) + if value is None: + continue + payload_source = str(row.get("source") or row.get("source_code") or source).strip().lower() + try: + writer( + source=payload_source or source, + city=city, + value=value, + observed_at=self._observation_time(row), + fetched_at=fetched_at, + station_code=self._station_code(row), + station_name=self._station_name(row), + runway=str(row.get("runway") or "").strip(), + value_unit=str(row.get("unit") or row.get("temp_unit") or "c").strip().lower(), + status="ok", + payload=dict(row), + ) + wrote += 1 + except Exception as exc: + logger.debug( + "raw observation store write skipped source={} city={}: {}", + source, + city, + exc, + ) + if wrote: + logger.debug("raw observations stored source={} city={} count={}", source, city, wrote) def _refresh_city_cache(self, city: str) -> None: if not callable(self.cache_refresher): @@ -276,6 +441,7 @@ def start_observation_collector_loop( cache_refresher: Optional[Callable[[str], Any]] = None, profiles: Optional[Sequence[ObservationSourceProfile]] = None, status_recorder: Optional[ObservationCollectorStatusRepository] = None, + observation_store: Optional[Any] = None, ) -> Optional[threading.Thread]: if not _env_bool("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED", True): return None @@ -290,6 +456,7 @@ def start_observation_collector_loop( profiles=selected_profiles, cache_refresher=cache_refresher, status_recorder=status_recorder or ObservationCollectorStatusRepository(), + observation_store=observation_store or DBManager(), ) global _COLLECTOR_THREAD diff --git a/web/scan_terminal_city_row.py b/web/scan_terminal_city_row.py index 3a0f95a4..a1e080e8 100644 --- a/web/scan_terminal_city_row.py +++ b/web/scan_terminal_city_row.py @@ -7,17 +7,18 @@ from typing import Any, Dict, List, Optional from src.database.db_manager import DBManager from web.core import CITIES, _sf as _safe_float -from web.analysis_service import _analyze from web.scan_terminal_filters import ( market_region_from_tz_offset as _market_region_from_tz_offset, safe_int as _safe_int, ) +from web.services.canonical_temperature import build_city_weather_from_canonical from web.services.city_payloads import aggregate_runway_history SCAN_ROW_RUNWAY_HISTORY_RESOLUTION = "10m" SCAN_ROW_MAX_RUNWAY_POINTS = 144 _PANEL_CACHE_DB = DBManager() +_analyze = None # compatibility hook for tests that assert scan terminal stays cache-only. def _compact_runway_plate_history_for_scan(raw_history: Any) -> Dict[str, List[Dict[str, Any]]]: @@ -31,6 +32,47 @@ def _compact_runway_plate_history_for_scan(raw_history: Any) -> Dict[str, List[D } +def _enqueue_scan_terminal_refresh(city: str, *, reason: str) -> None: + enqueue = getattr(_PANEL_CACHE_DB, "enqueue_observation_refresh_request", None) + if not callable(enqueue): + return + try: + enqueue( + city=city, + kind="panel", + priority="high", + reason=reason, + ) + except Exception: + return + + +def _load_scan_panel_payload(city: str, *, force_refresh: bool) -> Optional[Dict[str, Any]]: + if not force_refresh: + cached_entry = _PANEL_CACHE_DB.get_city_cache("panel", city) + cached_payload = cached_entry.get("payload") if isinstance(cached_entry, dict) else None + if isinstance(cached_payload, dict): + return cached_payload + + canonical_getter = getattr(_PANEL_CACHE_DB, "get_canonical_temperature", None) + canonical_entry = canonical_getter(city) 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 + ) + payload = build_city_weather_from_canonical(city, canonical) if isinstance(canonical, dict) else None + if payload: + city_meta = CITIES.get(city) or {} + payload.setdefault("display_name", city_meta.get("name") or city_meta.get("display_name") or city) + payload.setdefault("temp_symbol", canonical.get("temp_symbol") or "°C") + _enqueue_scan_terminal_refresh(city, reason="scan_terminal_canonical_fallback") + return payload + + _enqueue_scan_terminal_refresh(city, reason="scan_terminal_cold_start") + return None + + def _resolve_time_range_dates(data: Dict[str, Any], time_range: str) -> List[str]: local_date = str(data.get("local_date") or "").strip() multi_model_daily = data.get("multi_model_daily") or {} @@ -173,18 +215,14 @@ def _scan_city_terminal_rows_quick( ) -> Dict[str, Any]: """Fast path that returns cached analysis rows only — returns a single row per city with cached analysis data (Obs, DEB, probabilities) but no market prices.""" - data: Dict[str, Any] = {} - if not force_refresh: - cached_entry = _PANEL_CACHE_DB.get_city_cache("panel", city) - cached_payload = cached_entry.get("payload") if isinstance(cached_entry, dict) else None - if isinstance(cached_payload, dict): - data = cached_payload + data = _load_scan_panel_payload(city, force_refresh=force_refresh) if not data: - data = _analyze( - city, - force_refresh=force_refresh, - detail_mode="panel", - ) + return { + "city": city, + "rows": [], + "candidate_total": 0, + "primary_scores": [], + } row = _build_quick_row(city=city, data=data) return { "city": city, diff --git a/web/scan_terminal_service.py b/web/scan_terminal_service.py index 430d210e..7eb3dd32 100644 --- a/web/scan_terminal_service.py +++ b/web/scan_terminal_service.py @@ -11,7 +11,7 @@ from typing import Any, Dict, List, Optional from loguru import logger -from web.analysis_service import _analyze +from src.database.db_manager import DBManager from web.core import CITIES from web.services.scan_terminal_config import ( SCAN_TERMINAL_BUILD_TIMEOUT_SEC, @@ -43,6 +43,7 @@ from web.scan_terminal_ranker import build_ranked_scan_terminal_result _SCAN_TERMINAL_INFLIGHT_BUILD_LOCK = threading.Lock() _SCAN_TERMINAL_INFLIGHT_BUILDS: Dict[str, Future] = {} +_SCAN_PREWARM_DB = DBManager() SCAN_TERMINAL_STALE_SUCCESS_MAX_AGE_SEC = max( SCAN_TERMINAL_PAYLOAD_TTL_SEC * 2, 600, @@ -460,10 +461,20 @@ def _warm_scan_terminal_payloads() -> int: return ok +def _queue_scan_terminal_city_prewarm(city: str) -> str: + _SCAN_PREWARM_DB.enqueue_observation_refresh_request( + city=city, + kind="panel", + priority="normal", + reason="scan_terminal_prewarm", + ) + return city + + def start_scan_terminal_prewarm() -> None: """Warm analysis caches for all cities at startup so the first terminal - scan returns quickly instead of forcing every city through a cold - _analyze() path. + scan returns quickly without forcing every city through a cold external + source refresh path. Runs once per process. Safe to call from any thread and at any point during the server lifecycle. @@ -483,7 +494,7 @@ def start_scan_terminal_prewarm() -> None: def _warm_one(city: str) -> str: try: - _analyze(city, force_refresh=False, detail_mode="panel") + _queue_scan_terminal_city_prewarm(city) except Exception: pass return city diff --git a/web/services/canonical_temperature.py b/web/services/canonical_temperature.py new file mode 100644 index 00000000..21411e0b --- /dev/null +++ b/web/services/canonical_temperature.py @@ -0,0 +1,214 @@ +from __future__ import annotations + +from copy import deepcopy +from datetime import datetime, timezone +from typing import Any, Dict, Optional + + +_SETTLEMENT_PROXY_SOURCES = {"amsc_awos", "amos", "runway", "awos"} +_SETTLEMENT_OFFICIAL_SOURCES = {"hko", "cwa", "noaa", "wunderground", "mgm", "knmi", "ims"} +_AIRPORT_OFFICIAL_SOURCES = {"metar", "madis_hfmetar", "aeroweb"} + + +def _to_float(value: Any) -> Optional[float]: + try: + if value is None or value == "": + return None + return float(value) + except (TypeError, ValueError): + return None + + +def _to_int(value: Any) -> Optional[int]: + try: + if value is None or value == "": + return None + return int(float(value)) + except (TypeError, ValueError): + return None + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +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: + return "settlement_official" + if normalized in _AIRPORT_OFFICIAL_SOURCES: + return "airport_official" + if "model" in normalized or normalized in {"deb", "open_meteo"}: + return "model_blend" + return "fallback" + + +def _confidence(source_role: str, freshness_status: str) -> float: + base = { + "settlement_proxy": 0.92, + "settlement_official": 0.9, + "airport_official": 0.78, + "model_blend": 0.55, + "fallback": 0.42, + }.get(source_role, 0.42) + penalty = { + "fresh": 0.0, + "expected_wait": 0.03, + "delayed": 0.1, + "stale": 0.18, + "expired": 0.28, + "missing": 0.35, + "unknown": 0.12, + }.get(str(freshness_status or "").strip().lower(), 0.12) + return round(max(0.05, min(0.99, base - penalty)), 2) + + +def build_canonical_temperature( + city: str, + payload: Dict[str, Any], + *, + fetched_at: Optional[str] = None, +) -> Optional[Dict[str, Any]]: + if not isinstance(payload, dict): + return None + current = payload.get("current") or {} + if not isinstance(current, dict): + return None + value = _to_float(current.get("temp")) + if value is None: + return None + + freshness = current.get("freshness") or {} + if not isinstance(freshness, dict): + freshness = {} + source = str( + current.get("source_code") + or current.get("settlement_source") + or current.get("source") + or freshness.get("source_code") + or "" + ).strip().lower() + source_label = str( + current.get("settlement_source_label") + or current.get("source_label") + or freshness.get("source_label") + or source.upper() + ).strip() + observed_at = str( + current.get("observed_at") + or current.get("observation_time") + or freshness.get("observed_at") + or "" + ).strip() + observed_at_local = str( + current.get("observed_at_local") + or freshness.get("observed_at_local") + or current.get("obs_time") + or "" + ).strip() + freshness_sec = _to_int(freshness.get("age_sec")) + if freshness_sec is None: + age_min = _to_float(current.get("obs_age_min")) + if age_min is not None: + freshness_sec = int(age_min * 60) + freshness_status = str( + freshness.get("freshness_status") + or current.get("observation_status") + or "unknown" + ).strip().lower() + role = _source_role(source) + canonical = { + "city": str(city or payload.get("name") or payload.get("city") or "").strip().lower(), + "value": round(value, 2), + "temp_symbol": str(payload.get("temp_symbol") or "°C"), + "source": source, + "source_label": source_label, + "source_role": role, + "station_code": current.get("station_code"), + "station_name": current.get("station_name"), + "observed_at": observed_at or None, + "observed_at_local": observed_at_local or None, + "fetched_at": str(fetched_at or payload.get("updated_at") or _now_iso()), + "freshness_sec": freshness_sec, + "freshness_status": freshness_status, + "confidence": _confidence(role, freshness_status), + } + age_text = f" updated {freshness_sec}s ago" if freshness_sec is not None else "" + canonical["explanation"] = f"{source_label or source or 'Source'}{age_text}." + return canonical + + +def attach_canonical_temperature( + payload: Dict[str, Any], + *, + city: Optional[str] = None, +) -> Dict[str, Any]: + if not isinstance(payload, dict): + return payload + next_payload = payload + canonical = build_canonical_temperature( + city or str(payload.get("name") or payload.get("city") or ""), + next_payload, + ) + if canonical: + next_payload["canonical_temperature"] = canonical + return next_payload + + +def store_canonical_temperature_from_payload(db: Any, city: str, payload: Dict[str, Any]) -> Optional[Dict[str, Any]]: + canonical = build_canonical_temperature(city, payload) + if not canonical: + return None + setter = getattr(db, "set_canonical_temperature", None) + if callable(setter): + setter(city, canonical) + return canonical + + +def build_city_weather_from_canonical(city: str, canonical: Dict[str, Any]) -> Optional[Dict[str, Any]]: + if not isinstance(canonical, dict): + return None + value = _to_float(canonical.get("value")) + if value is None: + return None + observed_at = str(canonical.get("observed_at") or "").strip() + observed_at_local = str(canonical.get("observed_at_local") or "").strip() + source = str(canonical.get("source") or "").strip().lower() + source_label = str(canonical.get("source_label") or source.upper()).strip() + freshness = { + "freshness_status": canonical.get("freshness_status") or "unknown", + "age_sec": canonical.get("freshness_sec"), + "observed_at": observed_at or None, + "observed_at_local": observed_at_local or None, + } + current = { + "temp": value, + "source_code": source, + "settlement_source": source, + "settlement_source_label": source_label, + "observed_at": observed_at or None, + "observed_at_local": observed_at_local or None, + "obs_time": observed_at_local or observed_at, + "freshness": freshness, + "observation_status": "live", + } + airport_primary = { + "temp": value, + "source_code": source, + "source_label": source_label, + "obs_time": observed_at or observed_at_local, + "freshness": freshness, + } + payload = { + "name": str(city or canonical.get("city") or "").strip().lower(), + "temp_symbol": str(canonical.get("temp_symbol") or "°C"), + "current": current, + "airport_primary": airport_primary, + "airport_current": deepcopy(airport_primary), + "canonical_temperature": dict(canonical), + "deb": {"prediction": canonical.get("deb_prediction")}, + "updated_at": canonical.get("fetched_at"), + } + return payload diff --git a/web/services/city_api.py b/web/services/city_api.py index 9ab10589..61d26dea 100644 --- a/web/services/city_api.py +++ b/web/services/city_api.py @@ -15,6 +15,7 @@ from loguru import logger import web.routes as legacy_routes from web.analysis_service import _runway_history_temp_for_city +from web.services.canonical_temperature import build_city_weather_from_canonical from web.services.request_timing import ServerTimingRecorder _RECENT_DEB_CACHE: Optional[Dict[str, Dict[str, object]]] = None @@ -129,6 +130,132 @@ async def _get_cached_city_payload(city: str, kind: str) -> Dict[str, Any]: return payload if isinstance(payload, dict) else {} +async def _get_canonical_city_payload(city: str, *, detail_depth: str = "panel") -> Dict[str, Any]: + try: + row = await run_in_threadpool(legacy_routes._CACHE_DB.get_canonical_temperature, city) + except Exception: + 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 = legacy_routes.CITY_REGISTRY.get(city, {}) or {} + city_info = legacy_routes.CITIES.get(city, {}) or {} + risk = legacy_routes.CITY_RISK_PROFILES.get(city, {}) or {} + payload.update( + { + "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 ("°F" if city_info.get("f") else "°C"), + "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": {"mu": None, "distribution": []}, + } + ) + return payload + + +def _enqueue_collector_refresh_request( + city: str, + kind: str, + *, + reason: str = "canonical_fallback", +) -> bool: + try: + enqueue = getattr(legacy_routes._CACHE_DB, "enqueue_observation_refresh_request", None) + if not callable(enqueue): + return False + return bool( + enqueue( + city=city, + kind=kind, + priority="high", + reason=reason, + ) + ) + except Exception as exc: + logger.debug("collector refresh enqueue failed city={} kind={}: {}", city, kind, exc) + return False + + +def _request_city_cache_refresh( + city: str, + 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) + + +def _request_city_full_refresh(city: str) -> None: + if _enqueue_collector_refresh_request(city, "full"): + return + _start_city_full_stale_refresh(city) + + +def _build_initializing_city_payload(city: str, *, detail_depth: str) -> Dict[str, Any]: + city_meta = legacy_routes.CITY_REGISTRY.get(city, {}) or {} + city_info = legacy_routes.CITIES.get(city, {}) or {} + risk = legacy_routes.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": "°F" if city_info.get("f") else "°C", + "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 _queue_and_build_initializing_city_payload(city: str, *, kind: str) -> Dict[str, Any]: + _enqueue_collector_refresh_request(city, kind, reason="cold_start") + 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]]], @@ -158,16 +285,37 @@ async def _refresh_city_payload_with_stale_timeout( 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: + 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 without sync refresh city={} kind={}", + city, + kind, + ) + 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: - cached_payload = await _get_cached_city_payload(city, kind) - if cached_payload: + 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_payload) + 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) @@ -384,9 +532,17 @@ async def _get_city_full_data(city: str, *, force_refresh: bool) -> Dict[str, An if payload: _start_city_full_stale_refresh(city) return await _overlay_cached_wunderground(city, payload) - return await _refresh_city_full_data(city, False) + canonical_payload = await _get_canonical_city_payload(city, detail_depth="full") + if canonical_payload: + _request_city_full_refresh(city) + return canonical_payload + return _queue_and_build_initializing_city_payload(city, kind="full") return await _overlay_cached_wunderground(city, payload) - return await _refresh_city_full_data(city, False) + canonical_payload = await _get_canonical_city_payload(city, detail_depth="full") + if canonical_payload: + _request_city_full_refresh(city) + return canonical_payload + return _queue_and_build_initializing_city_payload(city, kind="full") async def _get_city_chart_data(city: str, *, force_refresh: bool) -> Dict[str, Any]: @@ -714,9 +870,13 @@ async def get_city_detail_payload( if payload: _start_city_cache_stale_refresh(city, "panel", legacy_routes._refresh_city_panel_cache) return await _overlay_cached_wunderground(city, payload) - return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, False) + return _queue_and_build_initializing_city_payload(city, kind="panel") return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) - return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, False) + 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) + return canonical_payload + return _queue_and_build_initializing_city_payload(city, kind="panel") if detail_mode == "nearby": if force_refresh: return await _refresh_city_cache_with_stale_timeout( @@ -731,9 +891,17 @@ async def get_city_detail_payload( if payload: _start_city_cache_stale_refresh(city, "nearby", legacy_routes._refresh_city_nearby_cache) return await _overlay_cached_wunderground(city, payload) - return await run_in_threadpool(legacy_routes._refresh_city_nearby_cache, city, False) + 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) + return canonical_payload + return _queue_and_build_initializing_city_payload(city, kind="nearby") return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) - return await run_in_threadpool(legacy_routes._refresh_city_nearby_cache, city, False) + 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) + return canonical_payload + return _queue_and_build_initializing_city_payload(city, kind="nearby") if detail_mode == "market": if force_refresh: return await _refresh_city_cache_with_stale_timeout( @@ -748,9 +916,17 @@ async def get_city_detail_payload( if payload: _start_city_cache_stale_refresh(city, "market", legacy_routes._refresh_city_market_cache) return await _overlay_cached_wunderground(city, payload) - return await run_in_threadpool(legacy_routes._refresh_city_market_cache, city, False) + 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) + return canonical_payload + return _queue_and_build_initializing_city_payload(city, kind="market") return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) - return await run_in_threadpool(legacy_routes._refresh_city_market_cache, city, False) + 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) + 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) @@ -774,9 +950,13 @@ async def get_city_summary_payload( if payload: _start_city_cache_stale_refresh(city, "summary", legacy_routes._refresh_city_summary_cache) return await _overlay_cached_wunderground(city, payload) - return await run_in_threadpool(legacy_routes._refresh_city_summary_cache, city, False) + return _queue_and_build_initializing_city_payload(city, kind="summary") return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) - return await run_in_threadpool(legacy_routes._refresh_city_summary_cache, city, False) + 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) + return canonical_payload + return _queue_and_build_initializing_city_payload(city, kind="summary") async def get_city_detail_aggregate_payload( @@ -803,6 +983,8 @@ async def get_city_detail_aggregate_payload( "full_data", lambda: _get_city_full_data(city, force_refresh=force_refresh), ) + if isinstance(data, dict) and data.get("status") == "initializing": + return data return await timer.measure_async( "detail_payload", diff --git a/web/services/city_realtime_stream.py b/web/services/city_realtime_stream.py index dd4b7417..b785b07a 100644 --- a/web/services/city_realtime_stream.py +++ b/web/services/city_realtime_stream.py @@ -1,9 +1,8 @@ """Lightweight realtime temperature stream for scrolling chart. -Maintains per-city deque buffers (max 1440 points) fed by _analyze() -refreshes. The /api/city/{name}/realtime-stream endpoint reads from -these buffers and returns a simple {points, thresholds} payload that -the frontend RealtimeScrollChart polls every 30 seconds. +Maintains per-city deque buffers (max 1440 points) fed by cached latest +observations. The polling endpoint must not trigger external weather +fetches; collectors and cache refreshers feed DB state. """ from __future__ import annotations @@ -13,12 +12,14 @@ import threading import time from typing import Any, Dict, List, Optional -from web.analysis_service import _analyze +from src.database.db_manager import DBManager # Per-city ring buffers: city_name → deque of {timestamp, temp, source} _STREAM_BUFFERS: Dict[str, collections.deque] = {} _BUFFER_LOCK = threading.Lock() _MAXLEN = 1440 +_CACHE_DB = DBManager() +_analyze = None # compatibility placeholder for older tests/monkeypatches def _best_temp(data: Dict[str, Any]) -> Optional[float]: @@ -76,19 +77,63 @@ def _extract_thresholds(data: Dict[str, Any]) -> List[Dict[str, Any]]: return thresholds +def _cached_city_payload(city: str) -> Dict[str, Any]: + normalized_city = str(city or "").strip().lower() + if not normalized_city: + return {} + for kind in ("panel", "full"): + try: + entry = _CACHE_DB.get_city_cache(kind, normalized_city) + except Exception: + continue + if not isinstance(entry, dict): + continue + payload = entry.get("payload") or {} + if isinstance(payload, dict) and payload: + return payload + return {} + + +def _latest_canonical_point(city: str) -> Optional[Dict[str, Any]]: + normalized_city = str(city or "").strip().lower() + if not normalized_city: + return None + try: + row = _CACHE_DB.get_canonical_temperature(normalized_city) + except Exception: + return None + if not isinstance(row, dict): + return None + canonical = row.get("payload") or row + if not isinstance(canonical, dict): + return None + try: + temp = float(canonical.get("value")) + except (TypeError, ValueError): + return None + timestamp = str( + canonical.get("observed_at") + or canonical.get("observed_at_local") + or canonical.get("fetched_at") + or time.strftime("%H:%M:%S") + ) + source = str(canonical.get("source") or "canonical") + return {"timestamp": timestamp, "temp": round(temp, 1), "source": source} + + def capture_sample(city: str) -> None: """Record one sample for *city* into its ring buffer.""" - try: - data = _analyze(city, force_refresh=False, detail_mode="panel") - except Exception: - return - - temp = _best_temp(data) - if temp is None: - return - - ts = time.strftime("%H:%M:%S") - point = {"timestamp": ts, "temp": round(temp, 1), "source": "metar"} + point = _latest_canonical_point(city) + if point is None: + data = _cached_city_payload(city) + temp = _best_temp(data) + if temp is None: + return + current = data.get("current") if isinstance(data, dict) else {} + source = "cache" + if isinstance(current, dict): + source = str(current.get("source_code") or current.get("settlement_source") or source) + point = {"timestamp": time.strftime("%H:%M:%S"), "temp": round(temp, 1), "source": source} with _BUFFER_LOCK: buf = _STREAM_BUFFERS.get(city) @@ -107,11 +152,6 @@ def get_realtime_stream_payload(city: str) -> Dict[str, Any]: buf = _STREAM_BUFFERS.get(city) points = list(buf) if buf else [] - # Build thresholds from cached analysis - try: - data = _analyze(city, force_refresh=False, detail_mode="panel") - thresholds = _extract_thresholds(data) - except Exception: - thresholds = [] + thresholds = _extract_thresholds(_cached_city_payload(city)) return {"points": points, "thresholds": thresholds} diff --git a/web/services/city_runtime.py b/web/services/city_runtime.py index b4d1b9ce..966d52bc 100644 --- a/web/services/city_runtime.py +++ b/web/services/city_runtime.py @@ -30,6 +30,10 @@ from web.analysis_service import ( _build_city_market_scan_payload, _build_city_summary_payload, ) +from web.services.canonical_temperature import ( + attach_canonical_temperature, + store_canonical_temperature_from_payload, +) from web.scan_terminal_service import build_scan_terminal_payload # noqa: F401 - compatibility export for tests and transitional routers from web.core import ( CITIES, @@ -275,9 +279,23 @@ def _refresh_market_scan_payload_from_cached_analysis( return payload.get("market_scan_payload") or {} +def _attach_and_store_canonical_temperature(city: str, payload: dict) -> dict: + if not isinstance(payload, dict): + return payload + attach_canonical_temperature(payload, city=city) + try: + store_canonical_temperature_from_payload(_CACHE_DB, city, payload) + except Exception as exc: + logger.debug("canonical temperature store skipped city={}: {}", city, exc) + return payload + + def _refresh_city_summary_cache(city: str, force_refresh: bool = False) -> dict: data = _analyze_summary(city, force_refresh=force_refresh) + _attach_and_store_canonical_temperature(city, data) payload = _build_city_summary_payload(data) + if data.get("canonical_temperature"): + payload["canonical_temperature"] = data["canonical_temperature"] _CACHE_DB.set_city_cache( "summary", city, @@ -290,6 +308,7 @@ def _refresh_city_summary_cache(city: str, force_refresh: bool = False) -> dict: def _refresh_city_panel_cache(city: str, force_refresh: bool = False) -> dict: payload = _analyze(city, force_refresh=force_refresh, detail_mode="panel") + _attach_and_store_canonical_temperature(city, payload) _CACHE_DB.set_city_cache( "panel", city, @@ -302,6 +321,7 @@ def _refresh_city_panel_cache(city: str, force_refresh: bool = False) -> dict: def _refresh_city_nearby_cache(city: str, force_refresh: bool = False) -> dict: payload = _analyze(city, force_refresh=force_refresh, detail_mode="nearby") + _attach_and_store_canonical_temperature(city, payload) _CACHE_DB.set_city_cache( "nearby", city, @@ -314,6 +334,7 @@ def _refresh_city_nearby_cache(city: str, force_refresh: bool = False) -> dict: def _refresh_city_market_cache(city: str, force_refresh: bool = False) -> dict: payload = _analyze(city, force_refresh=force_refresh, detail_mode="market") + _attach_and_store_canonical_temperature(city, payload) now_ts = time.time() payload["market_analysis_cached_at"] = datetime.now().isoformat() payload["market_analysis_cached_at_ts"] = now_ts @@ -330,6 +351,7 @@ def _refresh_city_market_cache(city: str, force_refresh: bool = False) -> dict: def _refresh_city_full_cache(city: str, force_refresh: bool = False) -> dict: payload = _analyze(city, force_refresh=force_refresh, detail_mode="full") + _attach_and_store_canonical_temperature(city, payload) _CACHE_DB.set_city_cache( "full", city, diff --git a/web/services/dashboard_init_api.py b/web/services/dashboard_init_api.py index 5842d53d..9a051b7a 100644 --- a/web/services/dashboard_init_api.py +++ b/web/services/dashboard_init_api.py @@ -10,7 +10,7 @@ from fastapi.concurrency import run_in_threadpool from loguru import logger import web.routes as legacy_routes -from web.services.city_api import _build_cities_payload +from web.services.city_api import _build_cities_payload, get_city_detail_payload def _resolve_default_city(request: Request) -> Optional[str]: @@ -37,15 +37,12 @@ async def build_dashboard_init_payload(request: Request) -> Dict[str, Any]: detail_payload: Optional[Dict[str, Any]] = None if default_city: try: - cached_entry = await run_in_threadpool( - legacy_routes._CACHE_DB.get_city_cache, "panel", default_city, + detail_payload = await get_city_detail_payload( + request, + default_city, + force_refresh=False, + depth="panel", ) - if cached_entry: - detail_payload = cached_entry.get("payload") or {} - else: - detail_payload = await run_in_threadpool( - legacy_routes._refresh_city_panel_cache, default_city, False, - ) except Exception as exc: logger.warning( "dashboard_init default_city={} panel failed: {}", default_city, exc diff --git a/web/services/system_api.py b/web/services/system_api.py index 334798bc..1afe9d76 100644 --- a/web/services/system_api.py +++ b/web/services/system_api.py @@ -92,20 +92,27 @@ def run_system_priority_warm( primary = list(batches.get("primary") or []) secondary = list(batches.get("secondary") or []) + def _queue_city_refresh(city: str, *, priority: str) -> None: + enqueue = getattr(legacy_routes._CACHE_DB, "enqueue_observation_refresh_request", None) + if not callable(enqueue): + logger.warning("priority warm queue unavailable city={} timezone={}", city, timezone) + return + enqueue( + city=city, + kind="panel", + priority=priority, + reason="system_priority_warm", + ) + def _runner() -> None: for city in primary: try: - legacy_routes._refresh_city_summary_cache(city, force_refresh=False) - legacy_routes._refresh_city_panel_cache(city, force_refresh=False) - legacy_routes._refresh_city_nearby_cache(city, force_refresh=False) - legacy_routes._refresh_city_market_cache(city, force_refresh=False) - legacy_routes._refresh_city_full_cache(city, force_refresh=False) + _queue_city_refresh(city, priority="high") except Exception as exc: logger.warning("priority warm primary failed city={} timezone={}: {}", city, timezone, exc) for city in secondary: try: - legacy_routes._refresh_city_summary_cache(city, force_refresh=False) - legacy_routes._refresh_city_panel_cache(city, force_refresh=False) + _queue_city_refresh(city, priority="normal") except Exception as exc: logger.warning("priority warm secondary failed city={} timezone={}: {}", city, timezone, exc)