From 9a0d7ae444813b0e51160498dc557c956ea01062 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Mon, 1 Jun 2026 01:53:54 +0800 Subject: [PATCH] Persist scan terminal cache in Redis --- docker-compose.yml | 2 + tests/test_scan_terminal_modules.py | 60 +++++++++++ web/scan_terminal_cache.py | 160 +++++++++++++++++++++++----- 3 files changed, 195 insertions(+), 27 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 953970d0..8c58caee 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -28,6 +28,7 @@ services: env_file: &id001 - .env environment: + POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0} POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false' POLYWEATHER_SERVICE_ROLE: bot TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC: ${POLYWEATHER_BOT_AIRPORT_PUSH_INTERVAL_SEC:-180} @@ -96,6 +97,7 @@ services: container_name: polyweather_web env_file: *id001 environment: + POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0} POLYWEATHER_SERVICE_ROLE: web healthcheck: interval: 30s diff --git a/tests/test_scan_terminal_modules.py b/tests/test_scan_terminal_modules.py index b906b1c5..2490fb0f 100644 --- a/tests/test_scan_terminal_modules.py +++ b/tests/test_scan_terminal_modules.py @@ -1,4 +1,5 @@ from web.scan_terminal_filters import normalize_scan_terminal_filters +from web import scan_terminal_cache from web.scan_terminal_metar_gate import _apply_metar_gate_to_row from web.scan_terminal_payloads import ( build_failed_scan_terminal_payload, @@ -10,6 +11,65 @@ from web.scan_terminal_city_row import _build_quick_row from web.routers.scan import router as scan_router +class _FakeRedis: + def __init__(self): + self.data = {} + + def get(self, key): + return self.data.get(key) + + def setex(self, key, _ttl, value): + self.data[key] = value + + +def test_scan_terminal_cache_hydrates_success_payload_from_redis(monkeypatch): + fake_redis = _FakeRedis() + monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", "true") + monkeypatch.setattr(scan_terminal_cache, "_get_redis_client", lambda: fake_redis) + scan_terminal_cache._SCAN_TERMINAL_CACHE.clear() + + filters = {"scan_mode": "tradable", "limit": 9} + payload = { + "generated_at": "2026-06-01T00:00:00Z", + "rows": [{"id": "row-1"}], + "summary": {"candidate_total": 1}, + } + + scan_terminal_cache.set_cached_scan_terminal_payload(filters, payload) + scan_terminal_cache._SCAN_TERMINAL_CACHE.clear() + + entry = scan_terminal_cache.get_scan_terminal_cache_entry(filters) + cached = scan_terminal_cache.get_cached_scan_terminal_payload(filters, ttl_sec=3600) + + assert entry["success_payload"]["rows"] == [{"id": "row-1"}] + assert cached["summary"]["candidate_total"] == 1 + + +def test_scan_terminal_failure_state_preserves_redis_success_payload(monkeypatch): + fake_redis = _FakeRedis() + monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", "true") + monkeypatch.setattr(scan_terminal_cache, "_get_redis_client", lambda: fake_redis) + scan_terminal_cache._SCAN_TERMINAL_CACHE.clear() + + filters = {"scan_mode": "tradable", "limit": 9} + scan_terminal_cache.set_cached_scan_terminal_payload( + filters, + { + "generated_at": "2026-06-01T00:00:00Z", + "rows": [{"id": "row-1"}], + }, + ) + scan_terminal_cache._SCAN_TERMINAL_CACHE.clear() + + scan_terminal_cache.set_scan_terminal_failure_state(filters, error_message="timeout") + scan_terminal_cache._SCAN_TERMINAL_CACHE.clear() + + entry = scan_terminal_cache.get_scan_terminal_cache_entry(filters) + + assert entry["success_payload"]["rows"] == [{"id": "row-1"}] + assert entry["last_error"] == "timeout" + + def test_scan_router_does_not_expose_terminal_ai_endpoint(): routes = { getattr(route, "path", None): getattr(route, "methods", set()) diff --git a/web/scan_terminal_cache.py b/web/scan_terminal_cache.py index b7c92151..bb0b486c 100644 --- a/web/scan_terminal_cache.py +++ b/web/scan_terminal_cache.py @@ -1,6 +1,8 @@ from __future__ import annotations import json +import hashlib +import os import threading import time from datetime import datetime @@ -9,39 +11,140 @@ from typing import Any, Dict, Optional _SCAN_TERMINAL_CACHE_LOCK = threading.Lock() _SCAN_TERMINAL_CACHE: Dict[str, Dict[str, Any]] = {} _SCAN_TERMINAL_REFRESHING: set[str] = set() +_SCAN_TERMINAL_REDIS_CLIENT_LOCK = threading.Lock() +_SCAN_TERMINAL_REDIS_CLIENT: Any = None +_SCAN_TERMINAL_REDIS_UNAVAILABLE = False def scan_terminal_cache_key(filters: Dict[str, Any]) -> str: return json.dumps(filters, ensure_ascii=True, sort_keys=True) +def _truthy_env(name: str, *, default: bool = False) -> bool: + value = os.getenv(name) + if value is None: + return default + return str(value).strip().lower() not in ("", "0", "false", "no", "off") + + +def _redis_cache_enabled() -> bool: + return _truthy_env( + "POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", + default=bool(os.getenv("POLYWEATHER_REDIS_URL")), + ) + + +def _redis_cache_ttl_sec() -> int: + try: + value = int(os.getenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_TTL_SEC", "21600")) + except Exception: + value = 21600 + return max(600, min(value, 86400)) + + +def _redis_cache_prefix() -> str: + return os.getenv( + "POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_PREFIX", + "polyweather:scan_terminal:v1:", + ) + + +def _redis_entry_key(cache_key: str) -> str: + digest = hashlib.sha256(cache_key.encode("utf-8")).hexdigest() + return f"{_redis_cache_prefix()}{digest}" + + +def _get_redis_client() -> Any: + global _SCAN_TERMINAL_REDIS_CLIENT, _SCAN_TERMINAL_REDIS_UNAVAILABLE + + if not _redis_cache_enabled() or _SCAN_TERMINAL_REDIS_UNAVAILABLE: + return None + + with _SCAN_TERMINAL_REDIS_CLIENT_LOCK: + if _SCAN_TERMINAL_REDIS_CLIENT is not None: + return _SCAN_TERMINAL_REDIS_CLIENT + try: + import redis # type: ignore + + url = os.getenv("POLYWEATHER_REDIS_URL") or "redis://127.0.0.1:6379/0" + client = redis.Redis.from_url( + url, + socket_timeout=float(os.getenv("POLYWEATHER_REDIS_SOCKET_TIMEOUT_SECONDS", "2")), + socket_connect_timeout=float( + os.getenv("POLYWEATHER_REDIS_SOCKET_CONNECT_TIMEOUT_SECONDS", "1") + ), + health_check_interval=30, + ) + client.ping() + _SCAN_TERMINAL_REDIS_CLIENT = client + return client + except Exception: + _SCAN_TERMINAL_REDIS_UNAVAILABLE = True + return None + + +def _read_redis_cache_entry(cache_key: str) -> Optional[Dict[str, Any]]: + client = _get_redis_client() + if client is None: + return None + try: + raw = client.get(_redis_entry_key(cache_key)) + if not raw: + return None + if isinstance(raw, bytes): + raw = raw.decode("utf-8") + entry = json.loads(str(raw)) + return dict(entry) if isinstance(entry, dict) else None + except Exception: + return None + + +def _write_redis_cache_entry(cache_key: str, entry: Dict[str, Any]) -> None: + client = _get_redis_client() + if client is None: + return + try: + client.setex( + _redis_entry_key(cache_key), + _redis_cache_ttl_sec(), + json.dumps(entry, ensure_ascii=False, separators=(",", ":")), + ) + except Exception: + return + + def get_cached_scan_terminal_payload( filters: Dict[str, Any], *, ttl_sec: int, ) -> Optional[Dict[str, Any]]: - cache_key = scan_terminal_cache_key(filters) now = time.time() - with _SCAN_TERMINAL_CACHE_LOCK: - cached = _SCAN_TERMINAL_CACHE.get(cache_key) - if not cached: - return None - cached_at = float(cached.get("t") or 0.0) - if now - cached_at >= float(ttl_sec): - return None - payload = cached.get("payload") - if not isinstance(payload, dict): - return None - return dict(payload) + cached = get_scan_terminal_cache_entry(filters) + if not cached: + return None + cached_at = float(cached.get("t") or 0.0) + if now - cached_at >= float(ttl_sec): + return None + payload = cached.get("payload") + if not isinstance(payload, dict): + return None + return dict(payload) def get_scan_terminal_cache_entry(filters: Dict[str, Any]) -> Optional[Dict[str, Any]]: cache_key = scan_terminal_cache_key(filters) with _SCAN_TERMINAL_CACHE_LOCK: cached = _SCAN_TERMINAL_CACHE.get(cache_key) - if not isinstance(cached, dict): - return None - return dict(cached) + if isinstance(cached, dict): + return dict(cached) + + redis_entry = _read_redis_cache_entry(cache_key) + if not redis_entry: + return None + + with _SCAN_TERMINAL_CACHE_LOCK: + _SCAN_TERMINAL_CACHE[cache_key] = dict(redis_entry) + return dict(redis_entry) def set_cached_scan_terminal_payload( @@ -51,15 +154,17 @@ def set_cached_scan_terminal_payload( cache_key = scan_terminal_cache_key(filters) existing = get_scan_terminal_cache_entry(filters) or {} now = time.time() + entry = { + "t": now, + "payload": dict(payload), + "success_t": now, + "success_payload": dict(payload), + "last_error": existing.get("last_error"), + "last_failed_at": existing.get("last_failed_at"), + } with _SCAN_TERMINAL_CACHE_LOCK: - _SCAN_TERMINAL_CACHE[cache_key] = { - "t": now, - "payload": dict(payload), - "success_t": now, - "success_payload": dict(payload), - "last_error": existing.get("last_error"), - "last_failed_at": existing.get("last_failed_at"), - } + _SCAN_TERMINAL_CACHE[cache_key] = dict(entry) + _write_redis_cache_entry(cache_key, entry) def set_scan_terminal_failure_state( @@ -68,11 +173,12 @@ def set_scan_terminal_failure_state( error_message: str, ) -> None: cache_key = scan_terminal_cache_key(filters) + existing = get_scan_terminal_cache_entry(filters) or {} + existing["last_error"] = error_message + existing["last_failed_at"] = datetime.utcnow().isoformat() + "Z" with _SCAN_TERMINAL_CACHE_LOCK: - existing = _SCAN_TERMINAL_CACHE.get(cache_key) or {} - existing["last_error"] = error_message - existing["last_failed_at"] = datetime.utcnow().isoformat() + "Z" - _SCAN_TERMINAL_CACHE[cache_key] = existing + _SCAN_TERMINAL_CACHE[cache_key] = dict(existing) + _write_redis_cache_entry(cache_key, existing) def mark_scan_terminal_refreshing(filters: Dict[str, Any]) -> bool: