from __future__ import annotations import os import time from copy import deepcopy from datetime import datetime from typing import Optional from fastapi import APIRouter, BackgroundTasks, HTTPException from loguru import logger from src.analysis.deb_algorithm import load_history from src.database.db_manager import DBManager from src.database.runtime_state import ( DailyRecordRepository, STATE_STORAGE_SQLITE, TruthRecordRepository, # noqa: F401 - compatibility export for ops/truth-history get_state_storage_mode, ) from src.analysis.settlement_rounding import apply_city_settlement from src.data_collection.country_networks import get_country_network_provider # noqa: F401 - compatibility export for transitional routers from src.data_collection.city_registry import ALIASES from src.data_collection.city_time import get_city_utc_offset_seconds # noqa: F401 - compatibility export for transitional routers from src.utils.refresh_policy import OBSERVATION_REFRESH_SEC, SCAN_ROWS_REFRESH_SEC from web.analysis_service import ( _analyze, _analyze_summary, _build_city_detail_payload, # noqa: F401 - compatibility export for tests and transitional routers _build_city_market_scan_payload, _build_city_summary_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, CITY_REGISTRY, # noqa: F401 - compatibility export for tests and transitional routers CITY_RISK_PROFILES, # noqa: F401 - compatibility export for tests and transitional routers PAYMENT_CHECKOUT, # noqa: F401 - compatibility export for tests and transitional routers PaymentCheckoutError, # noqa: F401 - compatibility export for tests and transitional routers SETTLEMENT_SOURCE_LABELS, # noqa: F401 - compatibility export for city list payloads SUPABASE_ENTITLEMENT, # noqa: F401 - compatibility export for tests and transitional routers ConfirmPaymentTxRequest, # noqa: F401 - compatibility export for tests and transitional routers CreatePaymentIntentRequest, # noqa: F401 - compatibility export for tests and transitional routers GrantPointsRequest, # noqa: F401 - compatibility export for tests and transitional routers SubmitPaymentTxRequest, # noqa: F401 - compatibility export for tests and transitional routers WalletChallengeRequest, # noqa: F401 - compatibility export for tests and transitional routers WalletUnbindRequest, # noqa: F401 - compatibility export for tests and transitional routers WalletVerifyRequest, # noqa: F401 - compatibility export for tests and transitional routers _ENTITLEMENT_GUARD_ENABLED, # noqa: F401 - compatibility export for tests and transitional routers _SUPABASE_AUTH_REQUIRED, # noqa: F401 - compatibility export for tests and transitional routers _assert_entitlement, # noqa: F401 - compatibility export for tests and transitional routers _bind_optional_supabase_identity, # noqa: F401 - compatibility export for tests and transitional routers _require_ops_admin, # noqa: F401 - compatibility export for tests and transitional routers _require_supabase_identity, # noqa: F401 - compatibility export for tests and transitional routers _resolve_auth_points, # noqa: F401 - compatibility export for tests and transitional routers _resolve_weekly_profile, # noqa: F401 - compatibility export for tests and transitional routers _sf, _weather, ) router = APIRouter() _CACHE_DB = DBManager() _DEB_RECENT_LOOKBACK = 7 _DEB_RECENT_MIN_SAMPLES = 3 _daily_record_repo = DailyRecordRepository() TRACKABLE_ANALYTICS_EVENTS = { "landing_view", "enter_terminal", "login_start", "signup_success", "trial_created", "payment_start", "payment_success", "degraded_auth_profile", "signup_completed", "dashboard_active", "paywall_feature_clicked", "paywall_viewed", "checkout_started", "checkout_succeeded", } DEFAULT_STATUS_CITIES = [ "ankara", "istanbul", "shanghai", "beijing", "shenzhen", "guangzhou", "qingdao", "wuhan", "chengdu", "chongqing", "hong kong", "taipei", "singapore", "tokyo", "seoul", "busan", "london", "paris", "madrid", ] 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", ] CITY_SUMMARY_CACHE_TTL_SEC = min(SCAN_ROWS_REFRESH_SEC, max(30, int(os.getenv("POLYWEATHER_CITY_SUMMARY_CACHE_TTL_SEC", str(SCAN_ROWS_REFRESH_SEC))))) CITY_PANEL_CACHE_TTL_SEC = min(SCAN_ROWS_REFRESH_SEC, max(30, int(os.getenv("POLYWEATHER_CITY_PANEL_CACHE_TTL_SEC", str(SCAN_ROWS_REFRESH_SEC))))) CITY_NEARBY_CACHE_TTL_SEC = min(SCAN_ROWS_REFRESH_SEC, max(30, int(os.getenv("POLYWEATHER_CITY_NEARBY_CACHE_TTL_SEC", str(SCAN_ROWS_REFRESH_SEC))))) CITY_MARKET_CACHE_TTL_SEC = min(SCAN_ROWS_REFRESH_SEC, max(30, int(os.getenv("POLYWEATHER_CITY_MARKET_CACHE_TTL_SEC", str(SCAN_ROWS_REFRESH_SEC))))) CITY_FULL_CACHE_TTL_SEC = min(OBSERVATION_REFRESH_SEC, max(30, int(os.getenv("POLYWEATHER_CITY_FULL_CACHE_TTL_SEC", str(OBSERVATION_REFRESH_SEC))))) MARKET_SCAN_PAYLOAD_TTL_SEC = max( 5, int(os.getenv("POLYWEATHER_MARKET_SCAN_PAYLOAD_TTL_SEC", "30")), ) 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 _market_analysis_cache_is_fresh(entry: Optional[dict]) -> bool: if not isinstance(entry, dict): return False payload = entry.get("payload") or {} if isinstance(payload, dict): cached_at_ts = float(payload.get("market_analysis_cached_at_ts") or 0.0) if cached_at_ts > 0: return (time.time() - cached_at_ts) < float(CITY_MARKET_CACHE_TTL_SEC) return _city_cache_is_fresh(entry, CITY_MARKET_CACHE_TTL_SEC) def _market_scan_cache_key( data: dict, market_slug: Optional[str] = None, target_date: Optional[str] = None, lite: bool = False, ) -> str: local_date = str(data.get("local_date") or "").strip() requested_date = str(target_date or "").strip() selected_date = requested_date or local_date multi_model_daily = data.get("multi_model_daily") or {} if requested_date and isinstance(multi_model_daily, dict) and requested_date not in multi_model_daily: selected_date = local_date normalized_slug = str(market_slug or "").strip().lower() return f"{selected_date}|{normalized_slug}|lite={1 if lite else 0}" def _attach_market_scan_payload( payload: dict, *, market_slug: Optional[str] = None, target_date: Optional[str] = None, lite: bool = False, ) -> dict: if not isinstance(payload, dict): return payload scan_payload = _build_city_market_scan_payload( payload, market_slug=market_slug, target_date=target_date, lite=lite, ) now_ts = time.time() payload["market_scan_payload"] = scan_payload payload["market_scan_updated_at"] = datetime.now().isoformat() payload["market_scan_updated_at_ts"] = now_ts payload["market_scan_cache_key"] = _market_scan_cache_key( payload, market_slug=market_slug, target_date=target_date, lite=lite, ) return payload def _get_cached_market_scan_payload( payload: dict, *, market_slug: Optional[str] = None, target_date: Optional[str] = None, lite: bool = False, ) -> Optional[dict]: if not isinstance(payload, dict): return None scan_payload = payload.get("market_scan_payload") if not isinstance(scan_payload, dict): return None expected_key = _market_scan_cache_key( payload, market_slug=market_slug, target_date=target_date, lite=lite, ) cached_key = str(payload.get("market_scan_cache_key") or "") if cached_key != expected_key: return None updated_at_ts = float(payload.get("market_scan_updated_at_ts") or 0.0) if updated_at_ts <= 0: return None if (time.time() - updated_at_ts) >= float(MARKET_SCAN_PAYLOAD_TTL_SEC): return None return scan_payload def _refresh_market_scan_payload_from_cached_analysis( city: str, payload: dict, *, market_slug: Optional[str] = None, target_date: Optional[str] = None, lite: bool = False, ) -> dict: _attach_market_scan_payload( payload, market_slug=market_slug, target_date=target_date, lite=lite, ) _CACHE_DB.set_city_cache( "market", city, payload, version="v1", source_fingerprint=f"{city}:market", ) return payload.get("market_scan_payload") or {} 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, 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, 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, detail_mode="market") now_ts = time.time() payload["market_analysis_cached_at"] = datetime.now().isoformat() payload["market_analysis_cached_at_ts"] = now_ts _attach_market_scan_payload(payload) _CACHE_DB.set_city_cache( "market", city, payload, version="v1", source_fingerprint=f"{city}:market", ) return payload def _refresh_city_full_cache(city: str, force_refresh: bool = False) -> dict: payload = _analyze(city, force_refresh=force_refresh, detail_mode="full") _CACHE_DB.set_city_cache( "full", city, payload, version="v1", source_fingerprint=f"{city}:full", ) return payload def _overlay_wunderground_current(payload: dict, wunderground: dict) -> dict: if not isinstance(payload, dict) or not isinstance(wunderground, dict) or not wunderground: return payload next_payload = deepcopy(payload) next_payload["wunderground_current"] = dict(wunderground) official = next_payload.get("official") if isinstance(official, dict): official["wunderground_current"] = dict(wunderground) timeseries = next_payload.get("timeseries") if isinstance(timeseries, dict): timeseries["wunderground_today_obs"] = list(wunderground.get("today_obs") or []) return next_payload def _overlay_latest_wunderground_current(city: str, payload: dict) -> dict: if not isinstance(payload, dict): return payload city_key = str(city or payload.get("name") or payload.get("city") or "").strip().lower() if city_key not in CITIES: return deepcopy(payload) try: utc_offset = int( payload.get("utc_offset_seconds") or (payload.get("overview") or {}).get("utc_offset_seconds") or get_city_utc_offset_seconds(city_key) or 0 ) except Exception: utc_offset = get_city_utc_offset_seconds(city_key) try: latest_wu = _weather.fetch_wunderground_historical( city_key, use_fahrenheit=bool(CITIES[city_key].get("f")), utc_offset=utc_offset, ) except Exception as exc: logger.warning("latest WU overlay failed city={} error={}", city_key, exc) return deepcopy(payload) if not isinstance(latest_wu, dict) or not latest_wu: return deepcopy(payload) return _overlay_wunderground_current(payload, latest_wu) 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", "full"} 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 == "market": _refresh_city_market_cache(normalized_city, force_refresh=force_refresh) else: _refresh_city_full_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 _normalize_city_or_404(name: str) -> str: city = name.lower().strip().replace("-", " ") city = ALIASES.get(city, city) if city not in CITIES: raise HTTPException(404, detail=f"Unknown city: {city}") return city def _normalize_city_list(raw: Optional[str]) -> list[str]: if not raw: return list(DEFAULT_STATUS_CITIES) out: list[str] = [] for part in str(raw).split(","): city = str(part or "").strip().lower().replace("-", " ") if not city: continue city = ALIASES.get(city, city) if city in CITIES and city not in out: out.append(city) 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") def _build_recent_deb_performance_index( history_data: Optional[dict] = None, *, lookback: int = _DEB_RECENT_LOOKBACK, min_samples: int = _DEB_RECENT_MIN_SAMPLES, ) -> dict[str, dict[str, object]]: index: dict[str, dict[str, object]] = {} today = datetime.utcnow().strftime("%Y-%m-%d") settled_by_city: dict[str, list[tuple[str, float, float]]] = {} if isinstance(history_data, dict): for city_name, rows in history_data.items(): if not isinstance(rows, dict): continue settled: list[tuple[str, float, float]] = [] for date_key in sorted(rows.keys(), reverse=True): if date_key >= today: continue record = rows.get(date_key) or {} if not isinstance(record, dict): continue actual = _sf(record.get("actual_high")) deb_prediction = _sf(record.get("deb_prediction")) if actual is None or deb_prediction is None: continue settled.append((date_key, actual, deb_prediction)) if len(settled) >= max(lookback, 1): break settled_by_city[str(city_name).strip().lower()] = settled elif get_state_storage_mode() == STATE_STORAGE_SQLITE: recent_rows = _daily_record_repo.load_recent_settled_rows( before_date=today, per_city_limit=max(lookback, 1), ) for city_name, rows in recent_rows.items(): settled: list[tuple[str, float, float]] = [] for row in rows: actual = _sf(row.get("actual_high")) deb_prediction = _sf(row.get("deb_prediction")) date_key = str(row.get("target_date") or "").strip() if not date_key or actual is None or deb_prediction is None: continue settled.append((date_key, actual, deb_prediction)) settled_by_city[str(city_name).strip().lower()] = settled else: data = load_history(_history_file_path()) if not isinstance(data, dict): return index for city_name, rows in data.items(): if not isinstance(rows, dict): continue settled: list[tuple[str, float, float]] = [] for date_key in sorted(rows.keys(), reverse=True): if date_key >= today: continue record = rows.get(date_key) or {} if not isinstance(record, dict): continue actual = _sf(record.get("actual_high")) deb_prediction = _sf(record.get("deb_prediction")) if actual is None or deb_prediction is None: continue settled.append((date_key, actual, deb_prediction)) if len(settled) >= max(lookback, 1): break settled_by_city[str(city_name).strip().lower()] = settled for city_name, settled in settled_by_city.items(): if not settled: continue hit_count = 0 abs_errors: list[float] = [] for _, actual, deb_prediction in settled: abs_errors.append(abs(deb_prediction - actual)) if apply_city_settlement(city_name, actual) == apply_city_settlement(city_name, deb_prediction): hit_count += 1 sample_count = len(settled) hit_rate = (hit_count / sample_count) if sample_count > 0 else None if sample_count < min_samples: tier = "other" elif hit_rate is not None and hit_rate >= 0.67: tier = "high" elif hit_rate is not None and hit_rate >= 0.34: tier = "medium" else: tier = "low" index[str(city_name).strip().lower()] = { "tier": tier, "sample_count": sample_count, "hit_rate": round(hit_rate, 4) if hit_rate is not None else None, "mae": round(sum(abs_errors) / sample_count, 3) if sample_count > 0 else None, "last_date": settled[0][0] if settled else None, } return index __all__ = [name for name in globals() if not (name.startswith('__') and name.endswith('__'))]