gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reject stale scan terminal panel data

Browse Source
This commit is contained in:
2569718930@qq.com
2026-06-30 11:46:53 +08:00
parent f9b7b49690
commit ad9b33f957
2 changed files with 151 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_scan_terminal_modules.py
+112 -2
View File
@@ -1,3 +1,6 @@
import time
from datetime import datetime, timedelta, timezone
from web.scan_terminal_filters import normalize_scan_terminal_filters
from web import scan_terminal_cache
from web import scan_terminal_service
@@ -18,6 +21,10 @@ from web.routers.scan import router as scan_router
from web.scan_terminal_service import _scan_terminal_prewarm_filters
def _local_date_for_offset(offset_seconds):
return (datetime.now(timezone.utc) + timedelta(seconds=offset_seconds)).strftime("%Y-%m-%d")
class _FakeRedis:
def __init__(self):
self.data = {}
@@ -257,7 +264,7 @@ def test_scan_terminal_prewarm_queues_city_refresh_without_analyze(monkeypatch):
def test_scan_city_terminal_rows_reuses_persisted_panel_cache(monkeypatch):
payload = {
"display_name": "Paris",
"local_date": "2026-06-10",
"local_date": _local_date_for_offset(3600),
"local_time": "12:00",
"current": {"temp": 18.0},
"risk": {},
@@ -270,7 +277,7 @@ def test_scan_city_terminal_rows_reuses_persisted_panel_cache(monkeypatch):
@staticmethod
def get_city_cache(kind, city):
assert (kind, city) == ("panel", "paris")
return {"payload": payload}
return {"payload": payload, "updated_at_ts": time.time()}
monkeypatch.setattr(scan_terminal_city_row, "_PANEL_CACHE_DB", _Cache())
monkeypatch.setattr(
@@ -291,6 +298,109 @@ 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_rejects_expired_panel_cache(monkeypatch):
enqueued = []
payload = {
"display_name": "Paris",
"local_date": _local_date_for_offset(3600),
"local_time": "12:00",
"current": {"temp": 18.0},
"risk": {},
"deb": {"prediction": 20.0},
"probabilities": {},
"multi_model_daily": {},
}
class _Cache:
@staticmethod
def get_city_cache(kind, city):
assert (kind, city) == ("panel", "paris")
return {"payload": payload, "updated_at_ts": 1}
@staticmethod
def get_canonical_temperature(city):
assert city == "paris"
return None
@staticmethod
def enqueue_observation_refresh_request(**kwargs):
enqueued.append(kwargs)
return True
monkeypatch.setattr(scan_terminal_city_row, "_PANEL_CACHE_DB", _Cache())
result = scan_terminal_city_row._scan_city_terminal_rows(
"paris",
{"market_type": "maxtemp"},
force_refresh=False,
)
assert result["city"] == "paris"
assert result["rows"] == []
assert enqueued == [
{
"city": "paris",
"kind": "panel",
"priority": "high",
"reason": "scan_terminal_stale_panel",
}
]
def test_scan_city_terminal_rows_rejects_wrong_local_date_panel_cache(monkeypatch):
enqueued = []
payload = {
"display_name": "Paris",
"local_date": "2000-01-01",
"local_time": "12:00",
"current": {"temp": 18.0},
"risk": {},
"deb": {"prediction": 20.0},
"probabilities": {},
"multi_model_daily": {
"2000-01-01": {
"deb": {"prediction": 20.0},
"models": {"ECMWF": 19.0},
}
},
}
class _Cache:
@staticmethod
def get_city_cache(kind, city):
assert (kind, city) == ("panel", "paris")
return {"payload": payload, "updated_at_ts": time.time()}
@staticmethod
def get_canonical_temperature(city):
assert city == "paris"
return None
@staticmethod
def enqueue_observation_refresh_request(**kwargs):
enqueued.append(kwargs)
return True
monkeypatch.setattr(scan_terminal_city_row, "_PANEL_CACHE_DB", _Cache())
result = scan_terminal_city_row._scan_city_terminal_rows(
"paris",
{"market_type": "maxtemp"},
force_refresh=False,
)
assert result["city"] == "paris"
assert result["rows"] == []
assert enqueued == [
{
"city": "paris",
"kind": "panel",
"priority": "high",
"reason": "scan_terminal_stale_panel_date",
}
]
def test_scan_city_terminal_rows_uses_canonical_without_analyze(monkeypatch):
enqueued = []
web/scan_terminal_city_row.py
+39 -3
View File
@@ -2,10 +2,12 @@ from __future__ import annotations
import hashlib
import re
from datetime import datetime, timedelta
import time
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional
from src.database.db_manager import DBManager
from src.utils.refresh_policy import SCAN_ROWS_REFRESH_SEC
from web.core import CITIES, _sf as _safe_float
from web.scan_terminal_filters import (
market_region_from_tz_offset as _market_region_from_tz_offset,
@@ -19,6 +21,7 @@ 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.
SCAN_PANEL_CACHE_MAX_AGE_SEC = max(300, int(SCAN_ROWS_REFRESH_SEC) * 3)
def _compact_runway_plate_history_for_scan(raw_history: Any) -> Dict[str, List[Dict[str, Any]]]:
@@ -47,12 +50,44 @@ def _enqueue_scan_terminal_refresh(city: str, *, reason: str) -> None:
return
def _city_local_date(city: str, utc_offset_seconds: Optional[int] = None) -> str:
city_meta = CITIES.get(city) or {}
offset = utc_offset_seconds
if offset is None:
offset = _safe_int(city_meta.get("tz"), 0)
try:
offset = int(offset or 0)
except Exception:
offset = 0
return (datetime.now(timezone.utc) + timedelta(seconds=offset)).strftime("%Y-%m-%d")
def _panel_cache_stale_reason(city: str, cached_entry: Dict[str, Any], payload: Dict[str, Any]) -> Optional[str]:
updated_at_ts = _safe_float(cached_entry.get("updated_at_ts"))
if updated_at_ts is None or time.time() - updated_at_ts > SCAN_PANEL_CACHE_MAX_AGE_SEC:
return "scan_terminal_stale_panel"
tz_offset = payload.get("utc_offset_seconds")
if tz_offset is None:
tz_offset = (CITIES.get(city) or {}).get("tz")
expected_date = _city_local_date(city, _safe_int(tz_offset, 0))
payload_date = str(payload.get("local_date") or "").strip()
if payload_date and payload_date != expected_date:
return "scan_terminal_stale_panel_date"
return None
def _load_scan_panel_payload(city: str, *, force_refresh: bool) -> Optional[Dict[str, Any]]:
refresh_already_queued = False
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
stale_reason = _panel_cache_stale_reason(city, cached_entry, cached_payload)
if stale_reason is None:
return cached_payload
_enqueue_scan_terminal_refresh(city, reason=stale_reason)
refresh_already_queued = True
canonical_getter = getattr(_PANEL_CACHE_DB, "get_canonical_temperature", None)
canonical_entry = canonical_getter(city) if callable(canonical_getter) else None
@@ -69,7 +104,8 @@ def _load_scan_panel_payload(city: str, *, force_refresh: bool) -> Optional[Dict
_enqueue_scan_terminal_refresh(city, reason="scan_terminal_canonical_fallback")
return payload
_enqueue_scan_terminal_refresh(city, reason="scan_terminal_cold_start")
if not refresh_already_queued:
_enqueue_scan_terminal_refresh(city, reason="scan_terminal_cold_start")
return None