From c0e530789398001611eb70120a022bfa9e652fc6 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Thu, 16 Apr 2026 15:05:14 +0800 Subject: [PATCH] Add cached dashboard prewarm hints for priority cities --- frontend/hooks/useDashboardStore.tsx | 19 + frontend/lib/dashboard-client.ts | 19 + src/database/db_manager.py | 203 ++++++++++ src/utils/prewarm_dashboard.py | 4 +- web/routes.py | 550 +++++++++++++++++++++------ 5 files changed, 672 insertions(+), 123 deletions(-) diff --git a/frontend/hooks/useDashboardStore.tsx b/frontend/hooks/useDashboardStore.tsx index ca49c106..3cccc13e 100644 --- a/frontend/hooks/useDashboardStore.tsx +++ b/frontend/hooks/useDashboardStore.tsx @@ -618,6 +618,25 @@ export function DashboardStoreProvider({ void loadCities(); }, []); + useEffect(() => { + if (!cities.length) return; + if (typeof window === "undefined") return; + const schedule = () => dashboardClient.sendPriorityWarmHint(); + const idleCallback = (window as Window & { + requestIdleCallback?: (callback: () => void, options?: { timeout: number }) => number; + }).requestIdleCallback; + if (typeof idleCallback === "function") { + const id = idleCallback(schedule, { timeout: 3000 }); + return () => { + if (typeof window.cancelIdleCallback === "function") { + window.cancelIdleCallback(id); + } + }; + } + const timer = window.setTimeout(schedule, 2000); + return () => window.clearTimeout(timer); + }, [cities.length]); + useEffect(() => { void refreshProAccess(); }, []); diff --git a/frontend/lib/dashboard-client.ts b/frontend/lib/dashboard-client.ts index 614d9ff6..694b5267 100644 --- a/frontend/lib/dashboard-client.ts +++ b/frontend/lib/dashboard-client.ts @@ -12,6 +12,7 @@ const CACHE_TTL_MS = 5 * 60 * 1000; const pendingCityDetailRequests = new Map>(); const pendingHistoryRequests = new Map>(); const pendingCitySummaryRequests = new Map>(); +const PRIORITY_WARM_SESSION_KEY = "polyWeather_priority_warm_v1"; type CityCacheMeta = { cachedAt: number; @@ -142,6 +143,24 @@ export const dashboardClient = { return data.cities || []; }, + sendPriorityWarmHint(timezone?: string | null) { + if (!isClient()) return; + const tz = String( + timezone || Intl.DateTimeFormat().resolvedOptions().timeZone || "", + ).trim(); + if (!tz) return; + const cacheKey = `${PRIORITY_WARM_SESSION_KEY}:${tz}`; + if (window.sessionStorage.getItem(cacheKey)) return; + window.sessionStorage.setItem(cacheKey, "1"); + const params = new URLSearchParams({ timezone: tz }); + void fetch(`/api/system/priority-warm?${params.toString()}`, { + method: "POST", + headers: { Accept: "application/json" }, + cache: "no-store", + keepalive: true, + }).catch(() => {}); + }, + async getCitySummary(cityName: string, options?: { force?: boolean }) { const force = options?.force ?? false; const requestKey = `${cityName}::${force ? "force" : "cached"}`; diff --git a/src/database/db_manager.py b/src/database/db_manager.py index aa3f2c49..b8bb67c0 100644 --- a/src/database/db_manager.py +++ b/src/database/db_manager.py @@ -182,6 +182,64 @@ class DBManager: updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS city_summary_cache ( + city TEXT PRIMARY KEY, + payload_json TEXT NOT NULL, + updated_at TEXT NOT NULL, + updated_at_ts REAL NOT NULL, + version TEXT, + source_fingerprint TEXT + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS city_panel_cache ( + city TEXT PRIMARY KEY, + payload_json TEXT NOT NULL, + updated_at TEXT NOT NULL, + updated_at_ts REAL NOT NULL, + version TEXT, + source_fingerprint TEXT + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS city_nearby_cache ( + city TEXT PRIMARY KEY, + payload_json TEXT NOT NULL, + updated_at TEXT NOT NULL, + updated_at_ts REAL NOT NULL, + version TEXT, + source_fingerprint TEXT + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS city_market_cache ( + city TEXT PRIMARY KEY, + payload_json TEXT NOT NULL, + updated_at TEXT NOT NULL, + updated_at_ts REAL NOT NULL, + version TEXT, + source_fingerprint TEXT + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS city_history_preview_cache ( + city TEXT PRIMARY KEY, + payload_json TEXT NOT NULL, + updated_at TEXT NOT NULL, + updated_at_ts REAL NOT NULL, + version TEXT, + source_fingerprint TEXT + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS cache_refresh_locks ( + cache_key TEXT PRIMARY KEY, + locked_until_ts REAL NOT NULL, + owner TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + """) conn.execute(""" CREATE TABLE IF NOT EXISTS payment_audit_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -247,6 +305,151 @@ class DBManager: conn.commit() logger.info(f"Database initialized successfully path={self.db_path}") + def _cache_table_name(self, kind: str) -> Optional[str]: + normalized = str(kind or "").strip().lower() + if normalized == "summary": + return "city_summary_cache" + if normalized == "panel": + return "city_panel_cache" + if normalized == "nearby": + return "city_nearby_cache" + if normalized == "market": + return "city_market_cache" + if normalized == "history_preview": + return "city_history_preview_cache" + return None + + def get_city_cache(self, kind: str, city: str) -> Optional[Dict[str, Any]]: + table = self._cache_table_name(kind) + normalized_city = str(city or "").strip().lower() + if not table or not normalized_city: + return None + with self._get_connection() as conn: + conn.row_factory = sqlite3.Row + row = conn.execute( + f""" + SELECT city, payload_json, updated_at, updated_at_ts, version, source_fingerprint + FROM {table} + 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, + "updated_at": str(row["updated_at"] or ""), + "updated_at_ts": float(row["updated_at_ts"] or 0.0), + "version": str(row["version"] or ""), + "source_fingerprint": str(row["source_fingerprint"] or ""), + } + + def set_city_cache( + self, + kind: str, + city: str, + payload: Dict[str, Any], + *, + version: str = "v1", + source_fingerprint: Optional[str] = None, + ) -> None: + table = self._cache_table_name(kind) + normalized_city = str(city or "").strip().lower() + if not table or not normalized_city or not isinstance(payload, dict): + return + now = datetime.now().isoformat() + now_ts = datetime.now().timestamp() + with self._get_connection() as conn: + conn.execute( + f""" + INSERT INTO {table} ( + city, + payload_json, + updated_at, + updated_at_ts, + version, + source_fingerprint + ) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(city) DO UPDATE SET + payload_json = excluded.payload_json, + updated_at = excluded.updated_at, + updated_at_ts = excluded.updated_at_ts, + version = excluded.version, + source_fingerprint = excluded.source_fingerprint + """, + ( + normalized_city, + json.dumps(payload, ensure_ascii=False), + now, + now_ts, + str(version or "v1"), + str(source_fingerprint or ""), + ), + ) + conn.commit() + + def acquire_cache_refresh_lock( + self, + cache_key: str, + *, + ttl_sec: int = 120, + owner: Optional[str] = None, + ) -> Optional[str]: + normalized_key = str(cache_key or "").strip().lower() + if not normalized_key: + return None + lock_owner = str(owner or "").strip() or hashlib.sha1( + f"{normalized_key}:{datetime.now().timestamp()}".encode("utf-8") + ).hexdigest()[:12] + now_ts = datetime.now().timestamp() + locked_until_ts = now_ts + max(15, int(ttl_sec or 120)) + updated_at = datetime.now().isoformat() + with self._get_connection() as conn: + cursor = conn.execute( + """ + INSERT INTO cache_refresh_locks (cache_key, locked_until_ts, owner, updated_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(cache_key) DO UPDATE SET + locked_until_ts = excluded.locked_until_ts, + owner = excluded.owner, + updated_at = excluded.updated_at + WHERE cache_refresh_locks.locked_until_ts < ? + """, + ( + normalized_key, + locked_until_ts, + lock_owner, + updated_at, + now_ts, + ), + ) + conn.commit() + return lock_owner if int(cursor.rowcount or 0) > 0 else None + + def release_cache_refresh_lock(self, cache_key: str, owner: str) -> None: + normalized_key = str(cache_key or "").strip().lower() + normalized_owner = str(owner or "").strip() + if not normalized_key or not normalized_owner: + return + with self._get_connection() as conn: + conn.execute( + """ + DELETE FROM cache_refresh_locks + WHERE cache_key = ? AND owner = ? + """, + (normalized_key, normalized_owner), + ) + conn.commit() + def get_payment_runtime_state(self, state_key: str) -> Optional[Dict[str, Any]]: key = str(state_key or "").strip() if not key: diff --git a/src/utils/prewarm_dashboard.py b/src/utils/prewarm_dashboard.py index 27f92944..65cf46d2 100644 --- a/src/utils/prewarm_dashboard.py +++ b/src/utils/prewarm_dashboard.py @@ -208,7 +208,7 @@ def get_prewarm_runtime_summary() -> Dict[str, Any]: "interval_sec": interval_sec, "jitter_sec": jitter_sec, "include_detail": _truthy_env("POLYWEATHER_PREWARM_INCLUDE_DETAIL", True), - "include_market": _truthy_env("POLYWEATHER_PREWARM_INCLUDE_MARKET", True), + "include_market": _truthy_env("POLYWEATHER_PREWARM_INCLUDE_MARKET", False), "force_refresh": _truthy_env("POLYWEATHER_PREWARM_FORCE_REFRESH", False), "thread_alive": bool(_WORKER_THREAD and _WORKER_THREAD.is_alive()) or shared_alive, "heartbeat_age_sec": None if heartbeat_age_sec is None else round(heartbeat_age_sec, 2), @@ -380,7 +380,7 @@ def start_prewarm_worker_thread() -> Optional[threading.Thread]: jitter_sec = int(os.getenv("POLYWEATHER_PREWARM_JITTER_SEC", "20")) force_refresh = str(os.getenv("POLYWEATHER_PREWARM_FORCE_REFRESH") or "").strip().lower() in {"1", "true", "yes", "on"} include_detail = str(os.getenv("POLYWEATHER_PREWARM_INCLUDE_DETAIL", "true")).strip().lower() in {"1", "true", "yes", "on"} - include_market = str(os.getenv("POLYWEATHER_PREWARM_INCLUDE_MARKET", "true")).strip().lower() in {"1", "true", "yes", "on"} + include_market = str(os.getenv("POLYWEATHER_PREWARM_INCLUDE_MARKET", "false")).strip().lower() in {"1", "true", "yes", "on"} thread = threading.Thread( target=run_worker_loop, diff --git a/web/routes.py b/web/routes.py index 3ecc211e..99798202 100644 --- a/web/routes.py +++ b/web/routes.py @@ -6,12 +6,13 @@ from datetime import datetime, timedelta from typing import Optional from fastapi.concurrency import run_in_threadpool -from fastapi import APIRouter, HTTPException, Request +from fastapi import APIRouter, BackgroundTasks, HTTPException, Request from fastapi.responses import PlainTextResponse from loguru import logger from src.analysis.deb_algorithm import load_history from src.analysis.probability_snapshot_archive import load_snapshot_rows_for_day +from src.database.db_manager import DBManager from src.database.runtime_state import TrainingFeatureRecordRepository, TruthRecordRepository from src.analysis.settlement_rounding import apply_city_settlement from src.data_collection.country_networks import get_country_network_provider @@ -55,6 +56,7 @@ from web.core import ( ) router = APIRouter() +_CACHE_DB = DBManager() _DEB_RECENT_LOOKBACK = 7 _DEB_RECENT_MIN_SAMPLES = 3 @@ -91,6 +93,278 @@ DEFAULT_PREWARM_CITIES = [ "madrid", ] HISTORY_PREVIEW_DAY_LIMIT = 21 +ASIA_CORE_CITIES = [ + "hong kong", + "taipei", + "tokyo", + "seoul", + "busan", + "shanghai", + "beijing", + "guangzhou", + "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", +] +CITY_SUMMARY_CACHE_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CITY_SUMMARY_CACHE_TTL_SEC", "180"))) +CITY_PANEL_CACHE_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CITY_PANEL_CACHE_TTL_SEC", "300"))) +CITY_NEARBY_CACHE_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CITY_NEARBY_CACHE_TTL_SEC", "480"))) +CITY_MARKET_CACHE_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CITY_MARKET_CACHE_TTL_SEC", "900"))) +CITY_HISTORY_PREVIEW_CACHE_TTL_SEC = max( + 60, + int(os.getenv("POLYWEATHER_CITY_HISTORY_PREVIEW_CACHE_TTL_SEC", "1800")), +) +CACHE_REFRESH_LOCK_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CACHE_REFRESH_LOCK_TTL_SEC", "120"))) + + +def _city_cache_is_fresh(entry: Optional[dict], ttl_sec: int) -> bool: + if not isinstance(entry, dict): + return False + updated_at_ts = float(entry.get("updated_at_ts") or 0.0) + if updated_at_ts <= 0: + return False + return (time.time() - updated_at_ts) < float(ttl_sec) + + +def _refresh_city_summary_cache(city: str, force_refresh: bool = False) -> dict: + data = _analyze_summary(city, force_refresh=force_refresh) + payload = _build_city_summary_payload(data) + _CACHE_DB.set_city_cache( + "summary", + city, + payload, + version="v1", + source_fingerprint=f"{city}:summary", + ) + return payload + + +def _refresh_city_panel_cache(city: str, force_refresh: bool = False) -> dict: + payload = _analyze(city, force_refresh=force_refresh, include_llm_commentary=False, detail_mode="panel") + _CACHE_DB.set_city_cache( + "panel", + city, + payload, + version="v1", + source_fingerprint=f"{city}:panel", + ) + return payload + + +def _refresh_city_nearby_cache(city: str, force_refresh: bool = False) -> dict: + payload = _analyze(city, force_refresh=force_refresh, include_llm_commentary=False, detail_mode="nearby") + _CACHE_DB.set_city_cache( + "nearby", + city, + payload, + version="v1", + source_fingerprint=f"{city}:nearby", + ) + return payload + + +def _refresh_city_market_cache(city: str, force_refresh: bool = False) -> dict: + payload = _analyze(city, force_refresh=force_refresh, include_llm_commentary=False, detail_mode="market") + _CACHE_DB.set_city_cache( + "market", + city, + payload, + version="v1", + source_fingerprint=f"{city}:market", + ) + return payload + + +def _build_city_history_payload(city: str, include_records: bool = False) -> dict: + source = str(CITIES.get(city, {}).get("settlement_source") or "metar").strip().lower() + truth_rows = _truth_record_repo.load_city(city) + feature_rows = _training_feature_repo.load_city(city) + + if not truth_rows and not feature_rows: + project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + history_file = os.path.join(project_root, "data", "daily_records.json") + data = load_history(history_file) + city_data = data.get(city, {}) if isinstance(data.get(city, {}), dict) else {} + else: + all_dates = sorted(set(truth_rows.keys()) | set(feature_rows.keys())) + city_data = {} + for day in all_dates: + record: dict[str, object] = {} + truth = truth_rows.get(day) or {} + features = feature_rows.get(day) or {} + if truth.get("actual_high") is not None: + record["actual_high"] = truth.get("actual_high") + record["settlement_source"] = truth.get("settlement_source") + record["settlement_station_code"] = truth.get("settlement_station_code") + record["settlement_station_label"] = truth.get("settlement_station_label") + record["truth_version"] = truth.get("truth_version") + record["updated_by"] = truth.get("updated_by") + record["truth_updated_at"] = truth.get("truth_updated_at") + if isinstance(features, dict): + if features.get("deb_prediction") is not None: + record["deb_prediction"] = features.get("deb_prediction") + if features.get("mu") is not None: + record["mu"] = features.get("mu") + if isinstance(features.get("forecasts"), dict): + record["forecasts"] = features.get("forecasts") + city_data[day] = record + + if not city_data: + return { + "history": [], + "mode": "full" if include_records else "preview", + "has_more": False, + "full_count": 0, + "preview_count": 0, + "settlement_source": source, + "settlement_source_label": SETTLEMENT_SOURCE_LABELS.get(source, source.upper()), + } + + all_days = sorted(city_data.keys()) + selected_days = all_days if include_records else all_days[-HISTORY_PREVIEW_DAY_LIMIT:] + out = [] + for day in selected_days: + rec = city_data.get(day, {}) + if not isinstance(rec, dict): + rec = {} + + act = rec.get("actual_high") + deb = rec.get("deb_prediction") + mu = rec.get("mu") + snapshots = load_snapshot_rows_for_day(city, day) + peak_ref = _build_peak_minus_12h_reference( + actual_high=act, + snapshots=snapshots, + ) + forecasts_raw = rec.get("forecasts", {}) or {} + forecasts = {} + if isinstance(forecasts_raw, dict): + for model_name, model_value in forecasts_raw.items(): + if _is_excluded_model_name(str(model_name)): + continue + fv = _sf(model_value) + forecasts[str(model_name)] = fv if fv is not None else None + forecasts = _merge_missing_history_forecasts_from_snapshots( + forecasts, + snapshots, + ) + mgm = forecasts.get("MGM") + out.append( + { + "date": day, + "actual": float(act) if act is not None else None, + "deb": float(deb) if deb is not None else None, + "mu": float(mu) if mu is not None else None, + "mgm": float(mgm) if mgm is not None else None, + "forecasts": forecasts, + "settlement_source": rec.get("settlement_source"), + "settlement_station_code": rec.get("settlement_station_code"), + "settlement_station_label": rec.get("settlement_station_label"), + "truth_version": rec.get("truth_version"), + "updated_by": rec.get("updated_by"), + "truth_updated_at": rec.get("truth_updated_at"), + "actual_peak_time": peak_ref.get("actual_peak_time"), + "deb_at_peak_minus_12h": peak_ref.get("deb_at_peak_minus_12h"), + "deb_at_peak_minus_12h_time": peak_ref.get("deb_at_peak_minus_12h_time"), + "deb_at_peak_minus_12h_error": peak_ref.get("deb_at_peak_minus_12h_error"), + } + ) + + return { + "history": out, + "mode": "full" if include_records else "preview", + "has_more": len(all_days) > len(selected_days), + "full_count": len(all_days), + "preview_count": len(out), + "settlement_source": source, + "settlement_source_label": SETTLEMENT_SOURCE_LABELS.get(source, source.upper()), + } + + +def _refresh_city_history_preview_cache(city: str) -> dict: + payload = _build_city_history_payload(city, include_records=False) + _CACHE_DB.set_city_cache( + "history_preview", + city, + payload, + version="v1", + source_fingerprint=f"{city}:history_preview", + ) + return payload + + +def _schedule_cache_refresh( + background_tasks: BackgroundTasks, + *, + kind: str, + city: str, + force_refresh: bool = False, +) -> bool: + normalized_kind = str(kind or "").strip().lower() + normalized_city = str(city or "").strip().lower() + if normalized_kind not in {"summary", "panel", "nearby", "market", "history_preview"} or not normalized_city: + return False + cache_key = f"city:{normalized_kind}:{normalized_city}" + owner = _CACHE_DB.acquire_cache_refresh_lock( + cache_key, + ttl_sec=CACHE_REFRESH_LOCK_TTL_SEC, + ) + if not owner: + return False + + def _runner() -> None: + try: + if normalized_kind == "summary": + _refresh_city_summary_cache(normalized_city, force_refresh=force_refresh) + elif normalized_kind == "panel": + _refresh_city_panel_cache(normalized_city, force_refresh=force_refresh) + elif normalized_kind == "nearby": + _refresh_city_nearby_cache(normalized_city, force_refresh=force_refresh) + elif normalized_kind == "history_preview": + _refresh_city_history_preview_cache(normalized_city) + else: + _refresh_city_market_cache(normalized_city, force_refresh=force_refresh) + except Exception as exc: + logger.warning( + "cache refresh failed kind={} city={} force_refresh={}: {}", + normalized_kind, + normalized_city, + force_refresh, + exc, + ) + finally: + _CACHE_DB.release_cache_refresh_lock(cache_key, owner) + + background_tasks.add_task(_runner) + return True def _parse_snapshot_dt(value: object) -> Optional[datetime]: @@ -218,6 +492,33 @@ def _normalize_city_list(raw: Optional[str]) -> list[str]: return out +def _select_priority_city_batches(client_timezone: Optional[str]) -> dict[str, object]: + tz = str(client_timezone or "").strip() + normalized = tz.lower() + if normalized.startswith("america/"): + primary = list(US_CORE_CITIES) + secondary = [] + region = "america" + elif normalized.startswith("europe/"): + primary = list(EUROPE_CORE_CITIES) + secondary = list(ASIA_CORE_CITIES) + region = "europe" + elif normalized.startswith("asia/") or normalized.startswith("australia/") or normalized.startswith("pacific/"): + primary = list(ASIA_CORE_CITIES) + secondary = list(EUROPE_CORE_CITIES) + region = "asia" + else: + primary = list(ASIA_CORE_CITIES) + secondary = list(EUROPE_CORE_CITIES) + region = "default" + return { + "region": region, + "timezone": tz or None, + "primary": primary, + "secondary": secondary, + } + + def _history_file_path() -> str: project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) return os.path.join(project_root, "data", "daily_records.json") @@ -317,7 +618,7 @@ async def system_prewarm( for city in selected: city_started = time.perf_counter() try: - data = _analyze(city, force_refresh=force_refresh) + _refresh_city_summary_cache(city, force_refresh=force_refresh) entry: dict[str, object] = { "city": city, "summary": True, @@ -325,20 +626,11 @@ async def system_prewarm( } summary_ok += 1 if include_detail: - _build_city_summary_payload(data) - _build_city_detail_payload( - data, - target_date=str(data.get("local_date") or "").strip() or None, - ) + _refresh_city_panel_cache(city, force_refresh=force_refresh) entry["detail"] = True detail_ok += 1 if include_market: - _build_city_detail_payload( - data, - target_date=str(data.get("local_date") or "").strip() or None, - ) - entry["market"] = True - market_ok += 1 + entry["market"] = False warmed.append(entry) except Exception as exc: failed.append( @@ -365,6 +657,7 @@ async def system_prewarm( "warmed": warmed, "failed": failed, "summary_ok": summary_ok, + "panel_ok": detail_ok, "detail_ok": detail_ok, "market_ok": market_ok, "failed_count": len(failed), @@ -431,6 +724,7 @@ async def list_cities(request: Request): @router.get("/api/city/{name}") async def city_detail( request: Request, + background_tasks: BackgroundTasks, name: str, force_refresh: bool = False, depth: str = "panel", @@ -446,128 +740,121 @@ async def city_detail( detail_mode = "nearby" else: detail_mode = "panel" + if detail_mode == "panel": + if force_refresh: + return await run_in_threadpool(_refresh_city_panel_cache, city, True) + cached_entry = await run_in_threadpool(_CACHE_DB.get_city_cache, "panel", city) + if cached_entry: + if not _city_cache_is_fresh(cached_entry, CITY_PANEL_CACHE_TTL_SEC): + _schedule_cache_refresh(background_tasks, kind="panel", city=city) + return cached_entry.get("payload") or {} + return await run_in_threadpool(_refresh_city_panel_cache, city, False) + if detail_mode == "nearby": + if force_refresh: + return await run_in_threadpool(_refresh_city_nearby_cache, city, True) + cached_entry = await run_in_threadpool(_CACHE_DB.get_city_cache, "nearby", city) + if cached_entry: + if not _city_cache_is_fresh(cached_entry, CITY_NEARBY_CACHE_TTL_SEC): + _schedule_cache_refresh(background_tasks, kind="nearby", city=city) + return cached_entry.get("payload") or {} + return await run_in_threadpool(_refresh_city_nearby_cache, city, False) + if detail_mode == "market": + if force_refresh: + return await run_in_threadpool(_refresh_city_market_cache, city, True) + cached_entry = await run_in_threadpool(_CACHE_DB.get_city_cache, "market", city) + if cached_entry: + if not _city_cache_is_fresh(cached_entry, CITY_MARKET_CACHE_TTL_SEC): + _schedule_cache_refresh(background_tasks, kind="market", city=city) + return cached_entry.get("payload") or {} + return await run_in_threadpool(_refresh_city_market_cache, city, False) return await run_in_threadpool(_analyze, city, force_refresh, False, detail_mode) @router.get("/api/history/{name}") async def city_history( request: Request, + background_tasks: BackgroundTasks, name: str, include_records: bool = False, ): _assert_entitlement(request) city = _normalize_city_or_404(name) + if include_records: + return await run_in_threadpool(_build_city_history_payload, city, True) + cached_entry = await run_in_threadpool(_CACHE_DB.get_city_cache, "history_preview", city) + if cached_entry: + if not _city_cache_is_fresh(cached_entry, CITY_HISTORY_PREVIEW_CACHE_TTL_SEC): + _schedule_cache_refresh(background_tasks, kind="history_preview", city=city) + return cached_entry.get("payload") or {} + return await run_in_threadpool(_refresh_city_history_preview_cache, city) - def _build_history_payload(): - source = str(CITIES.get(city, {}).get("settlement_source") or "metar").strip().lower() - truth_rows = _truth_record_repo.load_city(city) - feature_rows = _training_feature_repo.load_city(city) - if not truth_rows and not feature_rows: - project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) - history_file = os.path.join(project_root, "data", "daily_records.json") - data = load_history(history_file) - city_data = data.get(city, {}) if isinstance(data.get(city, {}), dict) else {} - else: - all_dates = sorted(set(truth_rows.keys()) | set(feature_rows.keys())) - city_data = {} - for day in all_dates: - record: dict[str, object] = {} - truth = truth_rows.get(day) or {} - features = feature_rows.get(day) or {} - if truth.get("actual_high") is not None: - record["actual_high"] = truth.get("actual_high") - record["settlement_source"] = truth.get("settlement_source") - record["settlement_station_code"] = truth.get("settlement_station_code") - record["settlement_station_label"] = truth.get("settlement_station_label") - record["truth_version"] = truth.get("truth_version") - record["updated_by"] = truth.get("updated_by") - record["truth_updated_at"] = truth.get("truth_updated_at") - if isinstance(features, dict): - if features.get("deb_prediction") is not None: - record["deb_prediction"] = features.get("deb_prediction") - if features.get("mu") is not None: - record["mu"] = features.get("mu") - if isinstance(features.get("forecasts"), dict): - record["forecasts"] = features.get("forecasts") - city_data[day] = record - - if not city_data: - return { - "history": [], - "mode": "full" if include_records else "preview", - "has_more": False, - "full_count": 0, - "preview_count": 0, - "settlement_source": source, - "settlement_source_label": SETTLEMENT_SOURCE_LABELS.get(source, source.upper()), +@router.get("/api/system/cache-status") +async def system_cache_status(request: Request, cities: Optional[str] = None): + _assert_entitlement(request) + selected = _normalize_city_list(cities) + if not selected: + selected = list(DEFAULT_PREWARM_CITIES) + kinds = { + "summary": CITY_SUMMARY_CACHE_TTL_SEC, + "panel": CITY_PANEL_CACHE_TTL_SEC, + "nearby": CITY_NEARBY_CACHE_TTL_SEC, + "market": CITY_MARKET_CACHE_TTL_SEC, + "history_preview": CITY_HISTORY_PREVIEW_CACHE_TTL_SEC, + } + items = [] + for city in selected: + row = {"city": city} + for kind, ttl_sec in kinds.items(): + entry = _CACHE_DB.get_city_cache(kind, city) + row[kind] = { + "exists": bool(entry), + "fresh": _city_cache_is_fresh(entry, ttl_sec), + "updated_at": entry.get("updated_at") if entry else None, + "age_sec": round(max(0.0, time.time() - float(entry.get("updated_at_ts") or 0.0)), 1) + if entry + else None, + "ttl_sec": ttl_sec, } + items.append(row) + return {"cities": items} - all_days = sorted(city_data.keys()) - selected_days = ( - all_days - if include_records - else all_days[-HISTORY_PREVIEW_DAY_LIMIT:] - ) - out = [] - for day in selected_days: - rec = city_data.get(day, {}) - if not isinstance(rec, dict): - rec = {} - act = rec.get("actual_high") - deb = rec.get("deb_prediction") - mu = rec.get("mu") - snapshots = load_snapshot_rows_for_day(city, day) - peak_ref = _build_peak_minus_12h_reference( - actual_high=act, - snapshots=snapshots, - ) - forecasts_raw = rec.get("forecasts", {}) or {} - forecasts = {} - if isinstance(forecasts_raw, dict): - for model_name, model_value in forecasts_raw.items(): - if _is_excluded_model_name(str(model_name)): - continue - fv = _sf(model_value) - forecasts[str(model_name)] = fv if fv is not None else None - forecasts = _merge_missing_history_forecasts_from_snapshots( - forecasts, - snapshots, - ) - mgm = forecasts.get("MGM") - out.append( - { - "date": day, - "actual": float(act) if act is not None else None, - "deb": float(deb) if deb is not None else None, - "mu": float(mu) if mu is not None else None, - "mgm": float(mgm) if mgm is not None else None, - "forecasts": forecasts, - "settlement_source": rec.get("settlement_source"), - "settlement_station_code": rec.get("settlement_station_code"), - "settlement_station_label": rec.get("settlement_station_label"), - "truth_version": rec.get("truth_version"), - "updated_by": rec.get("updated_by"), - "truth_updated_at": rec.get("truth_updated_at"), - "actual_peak_time": peak_ref.get("actual_peak_time"), - "deb_at_peak_minus_12h": peak_ref.get("deb_at_peak_minus_12h"), - "deb_at_peak_minus_12h_time": peak_ref.get("deb_at_peak_minus_12h_time"), - "deb_at_peak_minus_12h_error": peak_ref.get("deb_at_peak_minus_12h_error"), - } - ) +@router.post("/api/system/priority-warm") +async def system_priority_warm( + request: Request, + background_tasks: BackgroundTasks, + timezone: Optional[str] = None, +): + _assert_entitlement(request) + batches = _select_priority_city_batches(timezone) + primary = list(batches.get("primary") or []) + secondary = list(batches.get("secondary") or []) - return { - "history": out, - "mode": "full" if include_records else "preview", - "has_more": len(all_days) > len(selected_days), - "full_count": len(all_days), - "preview_count": len(out), - "settlement_source": source, - "settlement_source_label": SETTLEMENT_SOURCE_LABELS.get(source, source.upper()), - } + def _runner() -> None: + for city in primary: + try: + _refresh_city_summary_cache(city, force_refresh=False) + _refresh_city_panel_cache(city, force_refresh=False) + _refresh_city_nearby_cache(city, force_refresh=False) + _refresh_city_market_cache(city, force_refresh=False) + except Exception as exc: + logger.warning("priority warm primary failed city={} timezone={}: {}", city, timezone, exc) + for city in secondary: + try: + _refresh_city_summary_cache(city, force_refresh=False) + _refresh_city_panel_cache(city, force_refresh=False) + except Exception as exc: + logger.warning("priority warm secondary failed city={} timezone={}: {}", city, timezone, exc) - return await run_in_threadpool(_build_history_payload) + background_tasks.add_task(_runner) + return { + "ok": True, + "region": batches.get("region"), + "timezone": batches.get("timezone"), + "primary": primary, + "secondary": secondary, + } @router.get("/api/auth/me") @@ -1063,10 +1350,21 @@ async def payment_reconcile_latest(request: Request): @router.get("/api/city/{name}/summary") -async def city_summary(request: Request, name: str, force_refresh: bool = False): +async def city_summary( + request: Request, + background_tasks: BackgroundTasks, + name: str, + force_refresh: bool = False, +): city = _normalize_city_or_404(name) - data = await run_in_threadpool(_analyze_summary, city, force_refresh) - return await run_in_threadpool(_build_city_summary_payload, data) + if force_refresh: + return await run_in_threadpool(_refresh_city_summary_cache, city, True) + cached_entry = await run_in_threadpool(_CACHE_DB.get_city_cache, "summary", city) + if cached_entry: + if not _city_cache_is_fresh(cached_entry, CITY_SUMMARY_CACHE_TTL_SEC): + _schedule_cache_refresh(background_tasks, kind="summary", city=city) + return cached_entry.get("payload") or {} + return await run_in_threadpool(_refresh_city_summary_cache, city, False) @router.get("/api/city/{name}/detail") @@ -1091,6 +1389,7 @@ async def city_detail_aggregate( @router.get("/api/city/{name}/market-scan") async def city_market_scan( request: Request, + background_tasks: BackgroundTasks, name: str, force_refresh: bool = False, market_slug: Optional[str] = None, @@ -1098,7 +1397,16 @@ async def city_market_scan( ): _assert_entitlement(request) city = _normalize_city_or_404(name) - data = await run_in_threadpool(_analyze, city, force_refresh, False, "market") + if force_refresh: + data = await run_in_threadpool(_refresh_city_market_cache, city, True) + else: + cached_entry = await run_in_threadpool(_CACHE_DB.get_city_cache, "market", city) + if cached_entry: + if not _city_cache_is_fresh(cached_entry, CITY_MARKET_CACHE_TTL_SEC): + _schedule_cache_refresh(background_tasks, kind="market", city=city) + data = cached_entry.get("payload") or {} + else: + data = await run_in_threadpool(_refresh_city_market_cache, city, False) return await run_in_threadpool( _build_city_market_scan_payload, data,