From 25408797f1cf18e2d52ddb1889bc5c7f17d2995d Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Sun, 14 Jun 2026 04:23:15 +0800 Subject: [PATCH] Add production cache warmer --- deploy.sh | 36 +++ docker-compose.yml | 60 +++- .../__tests__/deployStability.test.ts | 9 +- tests/test_cache_warmer_service.py | 75 +++++ tests/test_deployment_runtime_config.py | 34 ++- web/cache_warmer_service.py | 282 ++++++++++++++++++ web/cache_warmer_worker.py | 53 ++++ 7 files changed, 539 insertions(+), 10 deletions(-) create mode 100644 tests/test_cache_warmer_service.py create mode 100644 web/cache_warmer_service.py create mode 100644 web/cache_warmer_worker.py diff --git a/deploy.sh b/deploy.sh index 9a69fd57..da1c2d87 100644 --- a/deploy.sh +++ b/deploy.sh @@ -207,6 +207,37 @@ warm_public_route() { return 0 } +wait_for_scan_terminal_ready() { + local name="$1" + local url="$2" + local timeout="${3:-35}" + local attempts="${4:-8}" + local delay="${5:-5}" + local output="" + local compact="" + local status="" + + for i in $(seq 1 "$attempts"); do + if output=$(curl -fsS --max-time "$timeout" "$url" 2>&1); then + compact="$(printf '%s' "$output" | tr -d '\n\r\t ')" + if printf '%s' "$compact" | grep -q '"status":"ready"'; then + echo "✅ $name ready after attempt $i/$attempts" + return 0 + fi + status="$(printf '%s' "$compact" | sed -n 's/.*"status":"\([^"]*\)".*/\1/p' | head -n 1)" + echo " $name not ready attempt $i/$attempts status=${status:-unknown}" + else + echo " $name request failed attempt $i/$attempts ($output)" + fi + if [ "$i" != "$attempts" ]; then + sleep "$delay" + fi + done + + echo "❌ $name did not return status=ready" + return 1 +} + read_env_file_value() { local key="$1" if [ ! -f ".env" ]; then @@ -278,12 +309,16 @@ fi echo "Updating observation collector..." compose_up_retry "observation collector" -d --no-deps polyweather_collector +echo "Updating cache warmer..." +compose_up_retry "cache warmer" -d --no-deps polyweather_warmer + echo "Updating frontend..." compose_up_retry "frontend" -d --no-deps polyweather_frontend echo "Waiting for frontend..." wait_for_local_service "frontend root" "http://127.0.0.1:3001/" 5 40 2 || FAILED_FRONTEND=1 wait_for_local_service "frontend terminal" "http://127.0.0.1:3001/terminal" 10 20 2 || FAILED_FRONTEND=1 +wait_for_scan_terminal_ready "scan terminal snapshot" "http://127.0.0.1:3001/api/scan/terminal" 35 8 5 || FAILED_FRONTEND=1 FAILED_FRONTEND="${FAILED_FRONTEND:-0}" if [ "$FAILED_FRONTEND" = "1" ]; then echo "❌ Frontend did not become healthy" @@ -292,6 +327,7 @@ if [ "$FAILED_FRONTEND" = "1" ]; then fi warm_public_route "terminal" "https://polyweather.top/terminal" 20 4 3 +warm_public_route "scan terminal" "https://polyweather.top/api/scan/terminal" 35 3 2 warm_public_route "auth snapshot" "https://polyweather.top/api/auth/me?prefer_snapshot=1" 10 3 2 warm_public_route "local cities recent stats" "http://127.0.0.1:8000/api/cities?refresh_deb_recent=1" 15 2 2 warm_public_route "cities" "https://polyweather.top/api/cities" 20 3 2 diff --git a/docker-compose.yml b/docker-compose.yml index 5285026c..9863dd2b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -2,7 +2,7 @@ services: polyweather_redis: image: redis:7-alpine container_name: polyweather_redis - command: redis-server --appendonly yes --maxmemory 128mb --maxmemory-policy noeviction + command: redis-server --appendonly yes --maxmemory ${POLYWEATHER_REDIS_MAXMEMORY:-512mb} --maxmemory-policy noeviction restart: unless-stopped healthcheck: interval: 10s @@ -100,9 +100,13 @@ services: env_file: *id001 environment: POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0} + POLYWEATHER_EVENT_STORE: ${POLYWEATHER_EVENT_STORE:-redis} + POLYWEATHER_REDIS_REQUIRED: ${POLYWEATHER_REDIS_REQUIRED:-true} + POLYWEATHER_REDIS_STREAM_MAXLEN: ${POLYWEATHER_REDIS_STREAM_MAXLEN:-100000} + POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED: ${POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED:-true} POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: '' - POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY:-1} - POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY:-1} + POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY:-2} + POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY:-2} POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS:-3000} POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS:-8000} POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC:-60} @@ -114,8 +118,8 @@ services: POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300} POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC:-30} POLYWEATHER_SCAN_TERMINAL_BUILD_TIMEOUT_SEC: '30' - POLYWEATHER_SCAN_TERMINAL_MAX_WORKERS: ${POLYWEATHER_SCAN_TERMINAL_MAX_WORKERS:-1} - POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false' + POLYWEATHER_SCAN_TERMINAL_MAX_WORKERS: ${POLYWEATHER_SCAN_TERMINAL_MAX_WORKERS:-4} + POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: ${POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED:-false} POLYWEATHER_SERVICE_ROLE: web UVICORN_WORKERS: ${UVICORN_WORKERS:-2} healthcheck: @@ -164,6 +168,7 @@ services: POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC:-15} POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300} POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC:-30} + POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS: ${POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS:-2} POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0} POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false' POLYWEATHER_SERVICE_ROLE: collector @@ -186,6 +191,51 @@ services: volumes: - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/var/lib/polyweather - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/app/data + polyweather_warmer: + command: python -m web.cache_warmer_worker + container_name: polyweather_warmer + depends_on: + polyweather_redis: + condition: service_healthy + polyweather_web: + condition: service_started + logging: + driver: "json-file" + options: + max-size: "50m" + max-file: "3" + cpus: ${POLYWEATHER_WARMER_CPUS:-0.75} + env_file: *id001 + environment: + POLYWEATHER_EVENT_STORE: ${POLYWEATHER_EVENT_STORE:-redis} + POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false' + POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0} + POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false' + POLYWEATHER_SERVICE_ROLE: warmer + POLYWEATHER_WARMER_CITY_BATCH_SIZE: ${POLYWEATHER_WARMER_CITY_BATCH_SIZE:-8} + POLYWEATHER_WARMER_CITY_INTERVAL_SEC: ${POLYWEATHER_WARMER_CITY_INTERVAL_SEC:-60} + POLYWEATHER_WARMER_ENABLED: ${POLYWEATHER_WARMER_ENABLED:-true} + POLYWEATHER_WARMER_SCAN_INTERVAL_SEC: ${POLYWEATHER_WARMER_SCAN_INTERVAL_SEC:-300} + POLYWEATHER_WARMER_TICK_SEC: ${POLYWEATHER_WARMER_TICK_SEC:-60} + healthcheck: + interval: 60s + retries: 3 + test: + - CMD + - python + - -c + - import sqlite3; c=sqlite3.connect('/var/lib/polyweather/polyweather.db'); + c.execute('SELECT 1'); c.close() + timeout: 10s + image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest} + mem_limit: ${POLYWEATHER_WARMER_MEM_LIMIT:-768m} + memswap_limit: ${POLYWEATHER_WARMER_MEMSWAP_LIMIT:-1g} + pids_limit: 256 + restart: unless-stopped + user: ${UID:-1000}:${GID:-1000} + volumes: + - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/var/lib/polyweather + - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/app/data x-polyweather-base: env_file: *id001 image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest} diff --git a/frontend/components/dashboard/scan-terminal/__tests__/deployStability.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/deployStability.test.ts index 608632dd..087f68ff 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/deployStability.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/deployStability.test.ts @@ -28,11 +28,18 @@ export function runTests() { deployScript.includes("http://127.0.0.1:3001/terminal"), "deploy script must wait for the local frontend before relying on Cloudflare/public smoke checks", ); + assert( + deployScript.includes("wait_for_scan_terminal_ready") && + deployScript.includes("http://127.0.0.1:3001/api/scan/terminal") && + deployScript.includes('"status":"ready"'), + "deploy script must wait for the scan terminal snapshot to be ready so first users do not see initialization payloads", + ); assert( deployScript.includes("warm_public_route") && deployScript.includes("https://polyweather.top/terminal") && + deployScript.includes("https://polyweather.top/api/scan/terminal") && deployScript.includes("https://polyweather.top/api/auth/me?prefer_snapshot=1"), - "deploy script must warm terminal and auth snapshot routes after container replacement", + "deploy script must warm terminal, scan terminal, and auth snapshot routes after container replacement", ); assert( deployScript.includes("validate_frontend_api_base_url") && diff --git a/tests/test_cache_warmer_service.py b/tests/test_cache_warmer_service.py new file mode 100644 index 00000000..2babdbbf --- /dev/null +++ b/tests/test_cache_warmer_service.py @@ -0,0 +1,75 @@ +from datetime import datetime, timezone + +from web.cache_warmer_service import CacheWarmer, build_priority_city_batch + + +def test_priority_city_batch_prefers_local_active_hours_over_night(): + cities = { + "alpha day": {"tz": 0}, + "beta night": {"tz": 12 * 3600}, + "gamma morning": {"tz": -4 * 3600}, + } + + selected = build_priority_city_batch( + cities, + now_utc=datetime(2026, 6, 14, 12, 0, tzinfo=timezone.utc), + batch_size=2, + ) + + assert selected == ["alpha day", "gamma morning"] + + +def test_cache_warmer_warms_scan_and_city_panel_without_force_refresh(): + now_ts = datetime(2026, 6, 14, 12, 0, tzinfo=timezone.utc).timestamp() + scan_calls = [] + city_calls = [] + + def scan_warmer(filters, *, force_refresh): + scan_calls.append((filters, force_refresh)) + return {"status": "ready"} + + def city_panel_warmer(city, *, force_refresh): + city_calls.append((city, force_refresh)) + return {"city": city} + + warmer = CacheWarmer( + city_provider=lambda: { + "alpha day": {"tz": 0}, + "beta night": {"tz": 12 * 3600}, + }, + scan_warmer=scan_warmer, + city_panel_warmer=city_panel_warmer, + scan_interval_sec=300, + city_interval_sec=60, + city_batch_size=1, + ) + + completed = warmer.run_due_once(now_ts=now_ts) + + assert completed == 2 + assert scan_calls == [({}, False)] + assert city_calls == [("alpha day", False)] + + +def test_cache_warmer_skips_work_when_intervals_are_not_due(): + now_ts = datetime(2026, 6, 14, 12, 0, tzinfo=timezone.utc).timestamp() + scan_calls = [] + city_calls = [] + warmer = CacheWarmer( + city_provider=lambda: {"alpha day": {"tz": 0}}, + scan_warmer=lambda filters, *, force_refresh: scan_calls.append(True), + city_panel_warmer=lambda city, *, force_refresh: city_calls.append(city), + scan_interval_sec=300, + city_interval_sec=60, + city_batch_size=1, + ) + + warmer.run_due_once(now_ts=now_ts) + scan_calls.clear() + city_calls.clear() + + completed = warmer.run_due_once(now_ts=now_ts + 20) + + assert completed == 0 + assert scan_calls == [] + assert city_calls == [] diff --git a/tests/test_deployment_runtime_config.py b/tests/test_deployment_runtime_config.py index 3700ac81..530b2461 100644 --- a/tests/test_deployment_runtime_config.py +++ b/tests/test_deployment_runtime_config.py @@ -83,6 +83,10 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services(): 1, )[0] collector_block = compose.split(" polyweather_collector:", 1)[1].split( + "\n polyweather_warmer:", + 1, + )[0] + warmer_block = compose.split(" polyweather_warmer:", 1)[1].split( "\nx-polyweather-base:", 1, )[0] @@ -90,16 +94,24 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services(): assert "POLYWEATHER_SERVICE_ROLE: web" in compose assert "POLYWEATHER_SERVICE_ROLE: bot" in compose assert "POLYWEATHER_SERVICE_ROLE: collector" in collector_block + assert "POLYWEATHER_SERVICE_ROLE: warmer" in warmer_block + assert "redis-server --appendonly yes --maxmemory ${POLYWEATHER_REDIS_MAXMEMORY:-512mb} --maxmemory-policy noeviction" in compose assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in bot_block - assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in web_block + assert "POLYWEATHER_EVENT_STORE: ${POLYWEATHER_EVENT_STORE:-redis}" in web_block + assert "POLYWEATHER_REDIS_REQUIRED: ${POLYWEATHER_REDIS_REQUIRED:-true}" in web_block + assert "POLYWEATHER_REDIS_STREAM_MAXLEN: ${POLYWEATHER_REDIS_STREAM_MAXLEN:-100000}" in web_block + assert "POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED: ${POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED:-true}" in web_block + assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: ${POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED:-false}" in web_block assert "POLYWEATHER_SCAN_TERMINAL_BUILD_TIMEOUT_SEC: '30'" in web_block - assert "POLYWEATHER_SCAN_TERMINAL_MAX_WORKERS: ${POLYWEATHER_SCAN_TERMINAL_MAX_WORKERS:-1}" in web_block + assert "POLYWEATHER_SCAN_TERMINAL_MAX_WORKERS: ${POLYWEATHER_SCAN_TERMINAL_MAX_WORKERS:-4}" in web_block assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in collector_block + assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in warmer_block assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in bot_block assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in web_block assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'true'" in collector_block - assert "POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY:-1}" in web_block - assert "POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY:-1}" in web_block + assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in warmer_block + assert "POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY:-2}" in web_block + assert "POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY:-2}" in web_block assert "POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS:-3000}" in web_block assert "POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS:-8000}" in web_block assert "UVICORN_WORKERS: ${UVICORN_WORKERS:-2}" in web_block @@ -111,7 +123,15 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services(): in collector_block ) assert "command: python -m web.observation_collector_worker" in collector_block + assert "command: python -m web.cache_warmer_worker" in warmer_block assert "POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-60}" in collector_block + assert "POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS: ${POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS:-2}" in collector_block + assert "POLYWEATHER_WARMER_ENABLED: ${POLYWEATHER_WARMER_ENABLED:-true}" in warmer_block + assert "POLYWEATHER_WARMER_TICK_SEC: ${POLYWEATHER_WARMER_TICK_SEC:-60}" in warmer_block + assert "POLYWEATHER_WARMER_SCAN_INTERVAL_SEC: ${POLYWEATHER_WARMER_SCAN_INTERVAL_SEC:-300}" in warmer_block + assert "POLYWEATHER_WARMER_CITY_INTERVAL_SEC: ${POLYWEATHER_WARMER_CITY_INTERVAL_SEC:-60}" in warmer_block + assert "POLYWEATHER_WARMER_CITY_BATCH_SIZE: ${POLYWEATHER_WARMER_CITY_BATCH_SIZE:-8}" in warmer_block + assert "cpus: ${POLYWEATHER_WARMER_CPUS:-0.75}" in warmer_block assert "TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC: ${POLYWEATHER_BOT_AIRPORT_PUSH_INTERVAL_SEC:-60}" in bot_block assert "POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300}" in collector_block @@ -186,8 +206,13 @@ def test_deploy_script_retries_startup_smoke_checks(): script = (ROOT / "deploy.sh").read_text(encoding="utf-8") assert "smoke_check()" in script + assert "wait_for_scan_terminal_ready()" in script + assert '"status":"ready"' in script + assert 'wait_for_scan_terminal_ready "scan terminal snapshot" "http://127.0.0.1:3001/api/scan/terminal"' in script + assert script.index("wait_for_scan_terminal_ready") < script.index("run_public_smoke_checks") assert 'smoke_check "healthz" "https://api.polyweather.top/healthz" 15 3 5' in script assert 'warm_public_route "local cities recent stats" "http://127.0.0.1:8000/api/cities?refresh_deb_recent=1"' in script + assert 'warm_public_route "scan terminal" "https://polyweather.top/api/scan/terminal"' in script assert 'smoke_check "local cities" "http://127.0.0.1:8000/api/cities" 10 6 3' not in script assert 'smoke_check "frontend cities" "https://polyweather.top/api/cities" 20 5 5' in script assert 'smoke_check "frontend" "https://www.polyweather.top/" 15 3 5' in script @@ -211,6 +236,7 @@ def test_deploy_script_retries_compose_recreate_races(): assert "removal of container .* is already in progress" in script assert 'compose_up_retry "backend services" -d --no-deps polyweather_web polyweather' in script assert 'compose_up_retry "observation collector" -d --no-deps polyweather_collector' in script + assert 'compose_up_retry "cache warmer" -d --no-deps polyweather_warmer' in script assert 'compose_up_retry "frontend" -d --no-deps polyweather_frontend' in script diff --git a/web/cache_warmer_service.py b/web/cache_warmer_service.py new file mode 100644 index 00000000..eaa8b8e9 --- /dev/null +++ b/web/cache_warmer_service.py @@ -0,0 +1,282 @@ +"""Low-priority cache warmer for user-facing aggregate payloads.""" + +from __future__ import annotations + +import os +import time +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any, Callable, Iterable, List, Mapping, Optional, Sequence + +from loguru import logger + + +ASIA_CORE_CITIES = ( + "hong kong", + "taipei", + "tokyo", + "seoul", + "busan", + "shanghai", + "beijing", + "guangzhou", + "qingdao", + "shenzhen", + "chongqing", + "chengdu", + "singapore", + "kuala lumpur", + "jakarta", +) +EUROPE_CORE_CITIES = ( + "istanbul", + "ankara", + "moscow", + "tel aviv", + "london", + "paris", + "madrid", + "milan", + "warsaw", + "amsterdam", + "helsinki", +) +US_CORE_CITIES = ( + "new york", + "los angeles", + "san francisco", + "austin", + "houston", + "chicago", + "dallas", + "miami", + "atlanta", + "seattle", +) +DEFAULT_HOT_CITIES = ASIA_CORE_CITIES + EUROPE_CORE_CITIES + US_CORE_CITIES +SOURCE_PRIORITY_WEIGHT = { + "hko": 10, + "cwa": 9, + "noaa": 8, + "mgm": 8, + "metar": 6, + "wunderground": 4, +} + + +def _env_bool(name: str, default: bool) -> bool: + raw = os.getenv(name) + if raw is None: + return default + return raw.strip().lower() in {"1", "true", "yes", "on"} + + +def _env_int(name: str, default: int) -> int: + raw = os.getenv(name) + if raw is None: + return default + try: + return int(raw) + except Exception: + return default + + +def _normalize_city_name(city: str) -> str: + return str(city or "").strip().lower().replace("-", " ") + + +def _parse_city_list(raw: Optional[str]) -> Sequence[str]: + if not raw: + return () + out: List[str] = [] + for part in raw.split(","): + city = _normalize_city_name(part) + if city and city not in out: + out.append(city) + return tuple(out) + + +def _coerce_utc_offset_seconds(meta: Mapping[str, Any]) -> int: + for key in ("tz", "tz_offset", "utc_offset_seconds"): + raw = meta.get(key) + if raw is None: + continue + try: + value = float(raw) + except Exception: + continue + if abs(value) <= 24: + value *= 3600 + return int(value) + return 0 + + +def _local_hour(meta: Mapping[str, Any], now_utc: datetime) -> int: + offset = _coerce_utc_offset_seconds(meta) + return int((now_utc + timedelta(seconds=offset)).hour) + + +@dataclass(frozen=True) +class WarmCandidate: + city: str + score: int + local_hour: int + settlement_source: str + + +def rank_priority_cities( + cities: Mapping[str, Mapping[str, Any]], + *, + now_utc: Optional[datetime] = None, + hot_cities: Optional[Iterable[str]] = None, +) -> List[WarmCandidate]: + """Rank cities by local activity window and source usefulness.""" + + now = now_utc or datetime.now(timezone.utc) + if now.tzinfo is None: + now = now.replace(tzinfo=timezone.utc) + hot_set = { + _normalize_city_name(city) + for city in (DEFAULT_HOT_CITIES if hot_cities is None else hot_cities) + } + ranked: List[WarmCandidate] = [] + for raw_city, raw_meta in cities.items(): + city = _normalize_city_name(raw_city) + if not city: + continue + meta = raw_meta or {} + hour = _local_hour(meta, now) + source = str(meta.get("settlement_source") or "metar").strip().lower() + score = SOURCE_PRIORITY_WEIGHT.get(source, 3) + if city in hot_set: + score += 30 + if 9 <= hour < 18: + score += 40 + elif 6 <= hour < 22: + score += 25 + else: + score += 2 + ranked.append( + WarmCandidate( + city=city, + score=score, + local_hour=hour, + settlement_source=source, + ) + ) + ranked.sort(key=lambda item: (-item.score, item.local_hour, item.city)) + return ranked + + +def build_priority_city_batch( + cities: Mapping[str, Mapping[str, Any]], + *, + now_utc: Optional[datetime] = None, + batch_size: int = 8, + hot_cities: Optional[Iterable[str]] = None, +) -> List[str]: + ranked = rank_priority_cities(cities, now_utc=now_utc, hot_cities=hot_cities) + limit = max(0, int(batch_size or 0)) + return [candidate.city for candidate in ranked[:limit]] + + +class CacheWarmer: + def __init__( + self, + *, + city_provider: Callable[[], Mapping[str, Mapping[str, Any]]], + scan_warmer: Callable[..., Any], + city_panel_warmer: Callable[..., Any], + scan_interval_sec: int = 300, + city_interval_sec: int = 60, + city_batch_size: int = 8, + hot_cities: Optional[Iterable[str]] = None, + ) -> None: + self.city_provider = city_provider + self.scan_warmer = scan_warmer + self.city_panel_warmer = city_panel_warmer + self.scan_interval_sec = max(60, int(scan_interval_sec or 300)) + self.city_interval_sec = max(30, int(city_interval_sec or 60)) + self.city_batch_size = max(1, min(32, int(city_batch_size or 8))) + self.hot_cities = tuple(hot_cities or DEFAULT_HOT_CITIES) + self._last_scan_ts = 0.0 + self._last_city_ts = 0.0 + self._city_cursor = 0 + + def run_due_once(self, *, now_ts: Optional[float] = None) -> int: + now = float(time.time() if now_ts is None else now_ts) + completed = 0 + if now - self._last_scan_ts >= self.scan_interval_sec: + self._last_scan_ts = now + if self._warm_scan(): + completed += 1 + if now - self._last_city_ts >= self.city_interval_sec: + self._last_city_ts = now + completed += self._warm_city_batch(now_ts=now) + return completed + + def _warm_scan(self) -> bool: + try: + self.scan_warmer({}, force_refresh=False) + return True + except Exception as exc: + logger.warning("cache warmer scan payload failed: {}", exc) + return False + + def _warm_city_batch(self, *, now_ts: float) -> int: + try: + cities = self.city_provider() or {} + except Exception as exc: + logger.warning("cache warmer city provider failed: {}", exc) + return 0 + ranked = rank_priority_cities( + cities, + now_utc=datetime.fromtimestamp(now_ts, timezone.utc), + hot_cities=self.hot_cities, + ) + if not ranked: + return 0 + start = self._city_cursor % len(ranked) + self._city_cursor += self.city_batch_size + selected = ranked[start : start + self.city_batch_size] + if len(selected) < self.city_batch_size: + selected.extend(ranked[: self.city_batch_size - len(selected)]) + + completed = 0 + for candidate in selected: + try: + self.city_panel_warmer(candidate.city, force_refresh=False) + completed += 1 + except Exception as exc: + logger.warning( + "cache warmer city panel failed city={} source={}: {}", + candidate.city, + candidate.settlement_source, + exc, + ) + return completed + + +def build_default_cache_warmer() -> CacheWarmer: + from web.core import CITIES + from web.scan_terminal_service import build_scan_terminal_payload + from web.services.city_runtime import _refresh_city_panel_cache + + hot_cities = _parse_city_list(os.getenv("POLYWEATHER_WARMER_HOT_CITIES")) + return CacheWarmer( + city_provider=lambda: CITIES, + scan_warmer=build_scan_terminal_payload, + city_panel_warmer=_refresh_city_panel_cache, + scan_interval_sec=_env_int("POLYWEATHER_WARMER_SCAN_INTERVAL_SEC", 300), + city_interval_sec=_env_int("POLYWEATHER_WARMER_CITY_INTERVAL_SEC", 60), + city_batch_size=_env_int("POLYWEATHER_WARMER_CITY_BATCH_SIZE", 8), + hot_cities=hot_cities or DEFAULT_HOT_CITIES, + ) + + +def warmer_enabled() -> bool: + return _env_bool("POLYWEATHER_WARMER_ENABLED", True) + + +def warmer_tick_sec() -> int: + return max(10, _env_int("POLYWEATHER_WARMER_TICK_SEC", 60)) diff --git a/web/cache_warmer_worker.py b/web/cache_warmer_worker.py new file mode 100644 index 00000000..b47acc60 --- /dev/null +++ b/web/cache_warmer_worker.py @@ -0,0 +1,53 @@ +"""Standalone cache warmer process for production aggregate payloads.""" + +from __future__ import annotations + +import signal +import threading +from types import FrameType +from typing import Optional + +from loguru import logger + +from web.cache_warmer_service import ( + build_default_cache_warmer, + warmer_enabled, + warmer_tick_sec, +) + +_STOP_EVENT = threading.Event() + + +def _handle_stop_signal(signum: int, _frame: Optional[FrameType]) -> None: + logger.info("cache warmer worker stopping signal={}", signum) + _STOP_EVENT.set() + + +def main() -> None: + signal.signal(signal.SIGINT, _handle_stop_signal) + signal.signal(signal.SIGTERM, _handle_stop_signal) + + if not warmer_enabled(): + logger.warning("cache warmer worker disabled; idling") + while not _STOP_EVENT.wait(3600): + pass + return + + warmer = build_default_cache_warmer() + tick_sec = warmer_tick_sec() + logger.info( + "cache warmer worker started tick={}s city_batch={}", + tick_sec, + warmer.city_batch_size, + ) + while not _STOP_EVENT.is_set(): + try: + completed = warmer.run_due_once() + logger.debug("cache warmer run completed work_items={}", completed) + except Exception as exc: + logger.exception("cache warmer loop failed: {}", exc) + _STOP_EVENT.wait(tick_sec) + + +if __name__ == "__main__": + main()