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

@

Browse Source 
将 web/routes.py 拆分为模块化 router + service 架构

    - 新增 web/app_factory.py 集中注册 7 个域名 router
    - 新增 web/routers/ 薄壳路由层（auth/city/system/scan/ops/payments/analytics）
    - 新增 web/services/ 业务函数下沉（每域独立 service 文件）
    - web/routes.py 缩减为 city_runtime 的兼容重导出 facade
    - analysis_service.py/app.py 适配新入口并清理冗余导入

    Scope-risk: LOW — 全量 170 测试通过，router 注册顺序与原路由一致
    Tested: python -m pytest -q (170 passed), ruff check . (All checks passed)
@
This commit is contained in:
2569718930@qq.com
2026-05-14 20:01:26 +08:00
parent a79abc02de
commit 37494a7192
24 changed files with 2855 additions and 2244 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/bot/runtime_coordinator.py
+1 -1
View File
@@ -237,7 +237,7 @@ class StartupCoordinator:
label="Polymarket 钱包异动监听（已停用）",
configured_enabled=False,
started=False,
reason="retired",
reason="retired_replaced_by_market_monitor",
details={
"note": "wallet activity watcher retired",
},
web/analysis_service.py
+27 -430
View File
@@ -1,8 +1,5 @@
from __future__ import annotations
import hashlib
import json
import os
import re
import time as _time
import threading
@@ -10,7 +7,6 @@ from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone, timedelta
from typing import Dict, Any, Optional
import httpx
from fastapi import HTTPException
from loguru import logger
@@ -23,7 +19,6 @@ from web.core import (
CITY_RISK_PROFILES,
SETTLEMENT_SOURCE_LABELS,
_is_excluded_model_name,
_market_layer,
_sf,
_weather,
)
@@ -35,6 +30,19 @@ from src.data_collection.city_time import get_city_utc_offset_seconds
from src.data_collection.nmc_sources import NMC_CITY_REFERENCES
from src.database.runtime_state import IntradayPathSnapshotRepository
from src.models.lgbm_daily_high import predict_lgbm_daily_high
from web.services.groq_commentary import (
build_groq_commentary_context as _groq_context_builder,
clean_commentary_text as _groq_clean_text,
groq_commentary_enabled as _groq_enabled,
maybe_enrich_dynamic_commentary_with_groq as _groq_enrich,
normalize_groq_commentary_payload as _groq_normalize_payload,
request_groq_commentary as _groq_request,
)
from web.services.city_payloads import (
build_city_detail_payload as _city_payload_detail,
build_city_market_scan_payload as _city_payload_market_scan,
build_city_summary_payload as _city_payload_summary,
)
TURKISH_MGM_CITIES = {"ankara", "istanbul"}
_ANALYSIS_CACHE_STATS_LOCK = threading.Lock()
@@ -49,13 +57,6 @@ _ANALYSIS_CACHE_STATS: Dict[str, Any] = {
}
_SUMMARY_CACHE_LOCK = threading.Lock()
_SUMMARY_CACHE: Dict[str, Dict[str, Any]] = {}
_GROQ_COMMENTARY_CACHE_LOCK = threading.Lock()
_GROQ_COMMENTARY_CACHE: Dict[str, Dict[str, Any]] = {}
_GROQ_COMMENTARY_CACHE_TTL_SEC = int(
os.getenv("POLYWEATHER_GROQ_COMMENTARY_CACHE_TTL_SEC", "1800")
)
def _dedupe_forecast_daily(rows: Any) -> list[Dict[str, Any]]:
if not isinstance(rows, list):
return []
@@ -424,206 +425,30 @@ def _set_cached_summary(city: str, payload: Dict[str, Any]) -> None:
def _groq_commentary_enabled() -> bool:
enabled = str(
os.getenv("POLYWEATHER_GROQ_COMMENTARY_ENABLED", "false")
).strip().lower()
api_key = str(os.getenv("GROQ_API_KEY") or "").strip()
return enabled in {"1", "true", "yes", "on"} and bool(api_key)
return _groq_enabled()
def _clean_commentary_text(value: Any, *, limit: int = 240) -> str:
text = str(value or "").strip()
if not text:
return ""
text = re.sub(r"\s+", " ", text)
return text[:limit].strip()
return _groq_clean_text(value, limit=limit)
def _build_groq_commentary_context(result: Dict[str, Any]) -> Dict[str, Any]:
dynamic = result.get("dynamic_commentary") or {}
vertical = result.get("vertical_profile_signal") or {}
taf_signal = ((result.get("taf") or {}).get("signal") or {}) if isinstance(result.get("taf"), dict) else {}
network = result.get("network_lead_signal") or {}
peak = result.get("peak") or {}
current = result.get("current") or {}
airport_primary = result.get("airport_primary") or {}
notes = dynamic.get("notes") if isinstance(dynamic.get("notes"), list) else []
compact_notes = [_clean_commentary_text(item, limit=180) for item in notes]
compact_notes = [item for item in compact_notes if item][:4]
return {
"city": result.get("display_name") or result.get("name"),
"local_date": result.get("local_date"),
"local_time": result.get("local_time"),
"temp_symbol": result.get("temp_symbol"),
"current_temp": current.get("temp"),
"day_high_so_far": current.get("max_so_far"),
"airport_anchor_temp": airport_primary.get("temp"),
"airport_vs_network_delta": result.get("airport_vs_network_delta"),
"peak_hours": peak.get("hours") or [],
"peak_status": peak.get("status"),
"network_lead_status": network.get("status"),
"network_lead_note": _clean_commentary_text(network.get("note"), limit=180),
"rules_summary": _clean_commentary_text(dynamic.get("summary"), limit=260),
"rules_notes": compact_notes,
"upper_air_summary_zh": _clean_commentary_text(vertical.get("summary_zh"), limit=260),
"upper_air_summary_en": _clean_commentary_text(vertical.get("summary_en"), limit=260),
"taf_summary_zh": _clean_commentary_text(taf_signal.get("summary_zh"), limit=220),
"taf_summary_en": _clean_commentary_text(taf_signal.get("summary_en"), limit=220),
"taf_peak_window": _clean_commentary_text(taf_signal.get("peak_window"), limit=80),
}
return _groq_context_builder(result)
def _normalize_groq_commentary_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
def _headline(value: Any, fallback: str) -> str:
text = _clean_commentary_text(value, limit=90)
return text or fallback
def _bullets(value: Any) -> list[str]:
items = value if isinstance(value, list) else []
cleaned = [_clean_commentary_text(item, limit=120) for item in items]
cleaned = [item for item in cleaned if item]
return cleaned[:3]
zh_headline = _headline(payload.get("headline_zh"), "结构信号以现有规则结论为主。")
en_headline = _headline(payload.get("headline_en"), "Structural read stays anchored to the existing rule-based signal.")
zh_bullets = _bullets(payload.get("bullets_zh"))
en_bullets = _bullets(payload.get("bullets_en"))
while len(zh_bullets) < 3:
zh_bullets.append("继续结合当前节奏、边界风险和峰值窗口判断。")
while len(en_bullets) < 3:
en_bullets.append("Keep the read anchored to pace, boundary risk, and the peak window.")
return {
"headline_zh": zh_headline,
"headline_en": en_headline,
"bullets_zh": zh_bullets[:3],
"bullets_en": en_bullets[:3],
"source": "groq",
}
return _groq_normalize_payload(payload)
def _request_groq_commentary(context: Dict[str, Any]) -> Optional[Dict[str, Any]]:
api_key = str(os.getenv("GROQ_API_KEY") or "").strip()
if not api_key:
return None
model = str(os.getenv("POLYWEATHER_GROQ_COMMENTARY_MODEL") or "openai/gpt-oss-20b").strip()
timeout_sec = float(os.getenv("POLYWEATHER_GROQ_COMMENTARY_TIMEOUT_SEC", "8"))
payload = {
"model": model,
"temperature": 0.2,
"max_tokens": 400,
"messages": [
{
"role": "system",
"content": (
"You rewrite weather-market structure commentary. "
"Never invent facts. Use only the provided context. "
"Return concise bilingual output for a dashboard: "
"one headline and exactly three bullets in Chinese, and the same in English. "
"Keep every bullet actionable and short."
),
},
{
"role": "user",
"content": json.dumps(context, ensure_ascii=False),
},
],
"response_format": {
"type": "json_schema",
"json_schema": {
"name": "polyweather_structure_commentary",
"strict": True,
"schema": {
"type": "object",
"additionalProperties": False,
"properties": {
"headline_zh": {"type": "string"},
"bullets_zh": {
"type": "array",
"items": {"type": "string"},
"minItems": 3,
"maxItems": 3,
},
"headline_en": {"type": "string"},
"bullets_en": {
"type": "array",
"items": {"type": "string"},
"minItems": 3,
"maxItems": 3,
},
},
"required": [
"headline_zh",
"bullets_zh",
"headline_en",
"bullets_en",
],
},
},
},
}
with httpx.Client(timeout=timeout_sec) as client:
response = client.post(
"https://api.groq.com/openai/v1/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
json=payload,
)
response.raise_for_status()
body = response.json()
content = (
(((body.get("choices") or [{}])[0]).get("message") or {}).get("content")
if isinstance(body, dict)
else None
)
if not content:
return None
try:
return _normalize_groq_commentary_payload(json.loads(str(content)))
except Exception:
logger.warning("Groq commentary returned non-JSON payload")
return None
return _groq_request(context)
def _maybe_enrich_dynamic_commentary_with_groq(
city: str,
result: Dict[str, Any],
) -> Dict[str, Any]:
dynamic = result.get("dynamic_commentary") or {}
if not _groq_commentary_enabled():
return dynamic
if dynamic.get("headline_zh") and dynamic.get("bullets_zh"):
return dynamic
context = _build_groq_commentary_context(result)
if not context.get("rules_summary") and not context.get("rules_notes"):
return dynamic
cache_key = hashlib.sha256(
json.dumps({"city": city, "context": context}, sort_keys=True, ensure_ascii=False).encode("utf-8")
).hexdigest()
now = _time.time()
with _GROQ_COMMENTARY_CACHE_LOCK:
cached = _GROQ_COMMENTARY_CACHE.get(cache_key)
if cached and now - float(cached.get("t") or 0) < _GROQ_COMMENTARY_CACHE_TTL_SEC:
merged = dict(dynamic)
merged.update(cached.get("payload") or {})
return merged
try:
enriched = _request_groq_commentary(context)
except Exception as exc:
logger.warning("Groq commentary skipped for {}: {}", city, exc)
return dynamic
if not enriched:
return dynamic
with _GROQ_COMMENTARY_CACHE_LOCK:
_GROQ_COMMENTARY_CACHE[cache_key] = {"t": now, "payload": enriched}
merged = dict(dynamic)
merged.update(enriched)
return merged
return _groq_enrich(city, result)
def _interpolate_hourly_value(
@@ -3118,27 +2943,7 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]:
def _build_city_summary_payload(data: Dict[str, Any]) -> Dict[str, Any]:
return {
"name": data.get("name"),
"display_name": data.get("display_name"),
"icao": data.get("risk", {}).get("icao"),
"utc_offset_seconds": data.get("utc_offset_seconds"),
"local_time": data.get("local_time"),
"temp_symbol": data.get("temp_symbol"),
"current": {
"temp": data.get("current", {}).get("temp"),
"obs_time": data.get("current", {}).get("obs_time"),
"settlement_source": data.get("current", {}).get("settlement_source"),
"settlement_source_label": data.get("current", {}).get("settlement_source_label"),
},
"deb": {"prediction": data.get("deb", {}).get("prediction")},
"deviation_monitor": data.get("deviation_monitor") or {},
"risk": {
"level": data.get("risk", {}).get("level"),
"warning": data.get("risk", {}).get("warning"),
},
"updated_at": data.get("updated_at"),
}
return _city_payload_summary(data)
def _build_city_market_scan_payload(
@@ -3148,137 +2953,13 @@ def _build_city_market_scan_payload(
lite: bool = False,
scan_filters: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
city = str(data.get("name") or "").strip().lower()
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 {}
selected_daily = (
multi_model_daily.get(selected_date)
if isinstance(multi_model_daily, dict)
else None
)
if not isinstance(selected_daily, dict):
selected_daily = {}
selected_date = local_date
distribution = selected_daily.get("probabilities")
if not isinstance(distribution, list) or not distribution:
distribution = data.get("probabilities", {}).get("distribution", []) or []
distribution_all = selected_daily.get("probabilities_all")
if not isinstance(distribution_all, list) or not distribution_all:
distribution_all = data.get("probabilities", {}).get("distribution_all", []) or []
if not distribution_all:
distribution_all = distribution
model_map = selected_daily.get("models") or data.get("multi_model") or {}
if not isinstance(model_map, dict):
model_map = {}
anchor_temp = None
anchor_model = None
for model_name, raw_value in model_map.items():
value = _sf(raw_value)
if value is None:
continue
if anchor_temp is None or value > anchor_temp:
anchor_temp = value
anchor_model = str(model_name or "").strip() or None
anchor_temp_c = anchor_temp
temp_symbol = str(data.get("temp_symbol") or "")
if anchor_temp_c is not None and "F" in temp_symbol.upper():
anchor_temp_c = (anchor_temp_c - 32.0) * 5.0 / 9.0
anchor_settlement = apply_city_settlement(city, anchor_temp_c) if anchor_temp_c is not None else None
primary_bucket = None
if isinstance(distribution, list) and distribution:
ranked_buckets = []
temp_symbol_upper = str(temp_symbol or "").upper()
max_primary_bucket_delta = 16.0 if "F" in temp_symbol_upper else 8.0
for idx, row in enumerate(distribution_all):
if not isinstance(row, dict):
continue
bucket_value = _sf(
row.get("temp")
if row.get("temp") is not None
else row.get("value")
if row.get("value") is not None
else row.get("lower")
)
if (
anchor_temp is not None
and bucket_value is not None
and abs(float(bucket_value) - float(anchor_temp)) > max_primary_bucket_delta
):
continue
bucket_prob = _sf(row.get("probability"))
prob_rank = bucket_prob if bucket_prob is not None else -1.0
ranked_buckets.append((-prob_rank, idx, row))
if ranked_buckets:
ranked_buckets.sort(key=lambda x: (x[0], x[1]))
primary_bucket = ranked_buckets[0][2]
elif anchor_temp is None:
primary_bucket = distribution[0]
model_probability = None
if isinstance(primary_bucket, dict) and primary_bucket.get("probability") is not None:
try:
raw_probability = float(primary_bucket.get("probability"))
model_probability = raw_probability / 100.0 if raw_probability > 1.0 else raw_probability
except Exception:
model_probability = None
fallback_sparkline = [
p.get("probability", 0)
for p in distribution_all[:8]
if isinstance(p, dict)
]
current = data.get("current") or {}
selected_deb = selected_daily.get("deb") if isinstance(selected_daily.get("deb"), dict) else {}
current_deb = data.get("deb") if isinstance(data.get("deb"), dict) else {}
scan_context = {
"local_date": data.get("local_date"),
"local_time": data.get("local_time"),
"peak": data.get("peak") or {},
"current_max_so_far": current.get("max_so_far"),
"current_temp": current.get("temp"),
"trend": data.get("trend") or {},
"network_lead_signal": data.get("network_lead_signal") or {},
"models": model_map,
"deb_prediction": selected_deb.get("prediction") or current_deb.get("prediction"),
}
market_scan = _market_layer.build_market_scan(
city=data.get("name"),
target_date=selected_date or data.get("local_date"),
temperature_bucket=primary_bucket if isinstance(primary_bucket, dict) else None,
model_probability=model_probability,
probability_distribution=distribution_all,
temp_symbol=temp_symbol,
fallback_sparkline=fallback_sparkline,
forced_market_slug=market_slug,
include_related_buckets=not lite,
return _city_payload_market_scan(
data,
market_slug=market_slug,
target_date=target_date,
lite=lite,
scan_filters=scan_filters,
scan_context=scan_context,
)
if isinstance(market_scan, dict):
market_scan["anchor_model"] = anchor_model
market_scan["anchor_high"] = anchor_temp
market_scan["anchor_settlement"] = anchor_settlement
market_scan["open_meteo_settlement"] = anchor_settlement
probabilities = data.get("probabilities") or {}
market_scan["probability_engine"] = str(
probabilities.get("engine") or "legacy"
).strip() or "legacy"
market_scan["probability_calibration_mode"] = str(
probabilities.get("calibration_mode") or "legacy"
).strip() or "legacy"
return {
"market_scan": market_scan,
"selected_date": selected_date or data.get("local_date"),
"fetched_at": data.get("updated_at"),
}
def _build_city_detail_payload(
@@ -3286,96 +2967,12 @@ def _build_city_detail_payload(
market_slug: Optional[str] = None,
target_date: Optional[str] = None,
) -> Dict[str, Any]:
market_payload = _build_city_market_scan_payload(
return _city_payload_detail(
data,
market_slug=market_slug,
target_date=target_date,
)
market_scan = market_payload.get("market_scan")
return {
"city": data.get("name"),
"fetched_at": data.get("updated_at"),
"overview": {
"name": data.get("name"),
"display_name": data.get("display_name"),
"icao": data.get("risk", {}).get("icao"),
"airport": data.get("risk", {}).get("airport"),
"lat": data.get("lat"),
"lon": data.get("lon"),
"local_time": data.get("local_time"),
"local_date": data.get("local_date"),
"temp_symbol": data.get("temp_symbol"),
"current_temp": data.get("current", {}).get("temp"),
"settlement_source": data.get("current", {}).get("settlement_source"),
"settlement_source_label": data.get("current", {}).get("settlement_source_label"),
"settlement_station": data.get("settlement_station") or {},
"deb_prediction": data.get("deb", {}).get("prediction"),
"risk_level": data.get("risk", {}).get("level"),
"risk_warning": data.get("risk", {}).get("warning"),
"updated_at": data.get("updated_at"),
},
"official": {
"available": bool(data.get("current", {}).get("temp") is not None),
"metar": {
"observation_time": data.get("airport_current", {}).get("obs_time"),
"obs_age_min": data.get("airport_current", {}).get("obs_age_min"),
"report_time": data.get("airport_current", {}).get("report_time"),
"receipt_time": data.get("airport_current", {}).get("receipt_time"),
"raw_metar": data.get("airport_current", {}).get("raw_metar"),
"current": data.get("airport_current") or {},
},
"taf": data.get("taf") or {},
"weather_gov": {},
"mgm": data.get("mgm") or {},
"mgm_nearby": data.get("mgm_nearby") or [],
"nearby_source": data.get("nearby_source") or ("mgm" if str(data.get("name") or "").lower() in TURKISH_MGM_CITIES else "metar_cluster"),
"airport_primary": data.get("airport_primary") or {},
"airport_primary_today_obs": data.get("airport_primary_today_obs") or [],
"official_nearby": data.get("official_nearby") or [],
"official_network_source": data.get("official_network_source"),
"official_network_status": data.get("official_network_status") or {},
"network_lead_signal": data.get("network_lead_signal") or {},
"network_spread_signal": data.get("network_spread_signal") or {},
"center_station_candidate": data.get("center_station_candidate"),
"airport_vs_network_delta": data.get("airport_vs_network_delta"),
},
"timeseries": {
"metar_recent_obs": data.get("metar_recent_obs") or [],
"metar_today_obs": data.get("metar_today_obs") or [],
"settlement_today_obs": data.get("settlement_today_obs") or [],
"hourly": data.get("hourly") or {},
"mgm_hourly": (data.get("mgm") or {}).get("hourly", []),
"forecast_daily": (data.get("forecast") or {}).get("daily", []),
},
"models": {
k: v
for k, v in (data.get("multi_model") or {}).items()
if not _is_excluded_model_name(k)
},
"deb": data.get("deb") or {},
"multi_model_daily": data.get("multi_model_daily") or {},
"probabilities": data.get("probabilities") or {"mu": None, "distribution": []},
"dynamic_commentary": data.get("dynamic_commentary") or {"summary": "", "notes": []},
"intraday_meteorology": data.get("intraday_meteorology")
or _build_intraday_meteorology(data),
"vertical_profile_signal": data.get("vertical_profile_signal") or {},
"taf": data.get("taf") or {},
"market_scan": market_scan,
"risk": data.get("risk"),
"settlement_station": data.get("settlement_station") or {},
"airport_primary": data.get("airport_primary") or {},
"official_nearby": data.get("official_nearby") or [],
"official_network_source": data.get("official_network_source"),
"official_network_status": data.get("official_network_status") or {},
"network_lead_signal": data.get("network_lead_signal") or {},
"network_spread_signal": data.get("network_spread_signal") or {},
"center_station_candidate": data.get("center_station_candidate"),
"airport_vs_network_delta": data.get("airport_vs_network_delta"),
"airport_current": data.get("airport_current") or {},
"nearby_source": data.get("nearby_source") or ("mgm" if str(data.get("name") or "").lower() in TURKISH_MGM_CITIES else "metar_cluster"),
"ai_analysis": data.get("ai_analysis") or "",
"errors": {},
}
# ──────────────────────────────────────────────────────────
web/app.py
+2 -3
View File
@@ -20,10 +20,9 @@ from web.analysis_service import ( # noqa: E402
_build_city_detail_payload,
_build_city_summary_payload,
)
from web.core import app # noqa: E402
from web.routes import router # noqa: E402
from web.app_factory import create_app # noqa: E402
app.include_router(router)
app = create_app()
__all__ = [
"app",
web/app_factory.py
+35
View File
@@ -0,0 +1,35 @@
"""Application assembly for the PolyWeather FastAPI backend.
This module centralizes router registration while preserving the existing
``web.core.app`` singleton and middleware setup during the transition toward a
more modular backend structure.
"""
from fastapi import FastAPI
from web.core import app as core_app
from web.routers.analytics import router as analytics_router
from web.routers.city import router as city_router
from web.routers.auth import router as auth_router
from web.routers.ops import router as ops_router
from web.routers.payments import router as payments_router
from web.routers.scan import router as scan_router
from web.routers.system import router as system_router
from web.routes import router as legacy_router
_ROUTES_REGISTERED_FLAG = "_polyweather_routes_registered"
def create_app() -> FastAPI:
"""Return the configured FastAPI app with routers registered once."""
if not bool(getattr(core_app.state, _ROUTES_REGISTERED_FLAG, False)):
core_app.include_router(system_router)
core_app.include_router(city_router)
core_app.include_router(auth_router)
core_app.include_router(analytics_router)
core_app.include_router(scan_router)
core_app.include_router(payments_router)
core_app.include_router(ops_router)
core_app.include_router(legacy_router)
setattr(core_app.state, _ROUTES_REGISTERED_FLAG, True)
return core_app
web/routers/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Router package for grouped FastAPI route modules."""
web/routers/analytics.py
+13
View File
@@ -0,0 +1,13 @@
"""Analytics event API routes."""
from fastapi import APIRouter, Request
from web.core import AnalyticsEventRequest
from web.services.analytics_api import track_analytics_event
router = APIRouter(tags=["analytics"])
@router.post("/api/analytics/events")
async def analytics_track(request: Request, body: AnalyticsEventRequest):
return track_analytics_event(request, body)
web/routers/auth.py
+12
View File
@@ -0,0 +1,12 @@
"""Authentication API routes."""
from fastapi import APIRouter, Request
from web.services.auth_api import get_auth_me_payload
router = APIRouter(tags=["auth"])
@router.get("/api/auth/me")
async def auth_me(request: Request):
return get_auth_me_payload(request)
web/routers/city.py
+103
View File
@@ -0,0 +1,103 @@
"""City and city-analysis API routes."""
from typing import Optional
from fastapi import APIRouter, BackgroundTasks, Request
from web.services.city_api import (
get_city_detail_aggregate_payload,
get_city_detail_payload,
get_city_history_payload,
get_city_market_scan_payload,
get_city_summary_payload,
list_cities_payload,
)
router = APIRouter(tags=["city"])
@router.get("/api/cities")
async def list_cities(request: Request):
return await list_cities_payload(request)
@router.get("/api/city/{name}")
async def city_detail(
request: Request,
background_tasks: BackgroundTasks,
name: str,
force_refresh: bool = False,
depth: str = "panel",
):
return await get_city_detail_payload(
request,
name,
force_refresh=force_refresh,
depth=depth,
)
@router.get("/api/history/{name}")
async def city_history(
request: Request,
background_tasks: BackgroundTasks,
name: str,
include_records: bool = False,
):
return await get_city_history_payload(
request,
name,
include_records=include_records,
)
@router.get("/api/city/{name}/summary")
async def city_summary(
request: Request,
background_tasks: BackgroundTasks,
name: str,
force_refresh: bool = False,
):
return await get_city_summary_payload(
request,
name,
force_refresh=force_refresh,
)
@router.get("/api/city/{name}/detail")
async def city_detail_aggregate(
request: Request,
name: str,
force_refresh: bool = False,
market_slug: Optional[str] = None,
target_date: Optional[str] = None,
):
return await get_city_detail_aggregate_payload(
request,
name,
force_refresh=force_refresh,
market_slug=market_slug,
target_date=target_date,
)
@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,
target_date: Optional[str] = None,
lite: bool = False,
):
return await get_city_market_scan_payload(
request,
background_tasks,
name,
force_refresh=force_refresh,
market_slug=market_slug,
target_date=target_date,
lite=lite,
)
web/routers/ops.py
+79
View File
@@ -0,0 +1,79 @@
"""Operations/admin API routes."""
from fastapi import APIRouter, Request
from web.core import GrantPointsRequest
from web.services.ops_api import (
get_ops_analytics_funnel,
get_ops_truth_history,
get_ops_weekly_leaderboard,
grant_ops_points,
list_ops_memberships,
list_ops_payment_incidents,
resolve_ops_payment_incident,
search_ops_users,
)
router = APIRouter(tags=["ops"])
@router.get("/api/ops/users")
async def ops_search_users(request: Request, q: str = "", limit: int = 20):
return search_ops_users(request, q=q, limit=limit)
@router.get("/api/ops/leaderboard/weekly")
async def ops_weekly_leaderboard(request: Request, limit: int = 20):
return get_ops_weekly_leaderboard(request, limit=limit)
@router.get("/api/ops/memberships")
async def ops_memberships(request: Request, limit: int = 200):
return list_ops_memberships(request, limit=limit)
@router.get("/api/ops/payments/incidents")
async def ops_payment_incidents(
request: Request,
limit: int = 50,
reason: str = "",
include_resolved: bool = False,
):
return list_ops_payment_incidents(
request,
limit=limit,
reason=reason,
include_resolved=include_resolved,
)
@router.post("/api/ops/payments/incidents/{event_id}/resolve")
async def ops_resolve_payment_incident(request: Request, event_id: int):
return resolve_ops_payment_incident(request, event_id)
@router.post("/api/ops/users/grant-points")
async def ops_grant_points(request: Request, body: GrantPointsRequest):
return grant_ops_points(request, body)
@router.get("/api/ops/analytics/funnel")
async def ops_analytics_funnel(request: Request, days: int = 30):
return get_ops_analytics_funnel(request, days=days)
@router.get("/api/ops/truth-history")
async def ops_truth_history(
request: Request,
city: str = "",
date_from: str = "",
date_to: str = "",
limit: int = 200,
):
return get_ops_truth_history(
request,
city=city,
date_from=date_from,
date_to=date_to,
limit=limit,
)
web/routers/payments.py
+90
View File
@@ -0,0 +1,90 @@
"""Payment API routes."""
from fastapi import APIRouter, Request
from web.core import (
ConfirmPaymentTxRequest,
CreatePaymentIntentRequest,
SubmitPaymentTxRequest,
WalletChallengeRequest,
WalletUnbindRequest,
WalletVerifyRequest,
)
from web.services.payment_api import (
confirm_payment_tx as confirm_payment_tx_service,
create_payment_intent as create_payment_intent_service,
create_payment_wallet_challenge,
get_payment_config,
get_payment_intent,
get_payment_runtime,
list_payment_wallets,
reconcile_latest_payment,
submit_payment_tx as submit_payment_tx_service,
unbind_payment_wallet,
verify_payment_wallet,
)
router = APIRouter(tags=["payments"])
@router.get("/api/payments/config")
async def payment_config(request: Request):
return get_payment_config(request)
@router.get("/api/payments/runtime")
async def payment_runtime(request: Request):
return get_payment_runtime(request)
@router.get("/api/payments/wallets")
async def payment_wallets(request: Request):
return list_payment_wallets(request)
@router.delete("/api/payments/wallets")
async def payment_wallet_unbind(request: Request, body: WalletUnbindRequest):
return unbind_payment_wallet(request, body)
@router.post("/api/payments/wallets/challenge")
async def payment_wallet_challenge(request: Request, body: WalletChallengeRequest):
return create_payment_wallet_challenge(request, body)
@router.post("/api/payments/wallets/verify")
async def payment_wallet_verify(request: Request, body: WalletVerifyRequest):
return verify_payment_wallet(request, body)
@router.post("/api/payments/intents")
async def payment_create_intent(request: Request, body: CreatePaymentIntentRequest):
return create_payment_intent_service(request, body)
@router.get("/api/payments/intents/{intent_id}")
async def payment_get_intent(request: Request, intent_id: str):
return get_payment_intent(request, intent_id)
@router.post("/api/payments/intents/{intent_id}/submit")
async def payment_submit_tx(
request: Request,
intent_id: str,
body: SubmitPaymentTxRequest,
):
return submit_payment_tx_service(request, intent_id, body)
@router.post("/api/payments/intents/{intent_id}/confirm")
async def payment_confirm_tx(
request: Request,
intent_id: str,
body: ConfirmPaymentTxRequest,
):
return confirm_payment_tx_service(request, intent_id, body)
@router.post("/api/payments/reconcile-latest")
async def payment_reconcile_latest(request: Request):
return reconcile_latest_payment(request)
web/routers/scan.py
+56
View File
@@ -0,0 +1,56 @@
"""Market scan and scan AI API routes."""
from fastapi import APIRouter, Request
from web.services.scan_api import (
get_scan_city_ai_forecast_payload,
get_scan_city_ai_stream_response,
get_scan_terminal_ai_payload,
get_scan_terminal_payload,
)
router = APIRouter(tags=["scan"])
@router.get("/api/scan/terminal")
async def scan_terminal(
request: Request,
scan_mode: str = "tradable",
min_price: float = 0.05,
max_price: float = 0.95,
min_edge_pct: float = 2.0,
min_liquidity: float = 500.0,
high_liquidity_only: bool = False,
market_type: str = "maxtemp",
time_range: str = "today",
limit: int = 25,
force_refresh: bool = False,
):
return await get_scan_terminal_payload(
request,
scan_mode=scan_mode,
min_price=min_price,
max_price=max_price,
min_edge_pct=min_edge_pct,
min_liquidity=min_liquidity,
high_liquidity_only=high_liquidity_only,
market_type=market_type,
time_range=time_range,
limit=limit,
force_refresh=force_refresh,
)
@router.post("/api/scan/terminal/ai")
async def scan_terminal_ai(request: Request):
return await get_scan_terminal_ai_payload(request)
@router.post("/api/scan/terminal/ai-city")
async def scan_terminal_ai_city(request: Request):
return await get_scan_city_ai_forecast_payload(request)
@router.post("/api/scan/terminal/ai-city/stream")
async def scan_terminal_ai_city_stream(request: Request):
return await get_scan_city_ai_stream_response(request)
web/routers/system.py
+63
View File
@@ -0,0 +1,63 @@
"""System and observability API routes for PolyWeather."""
from typing import Optional
from fastapi import APIRouter, BackgroundTasks, Request
from fastapi.responses import PlainTextResponse
from web.services.system_api import (
get_health_payload,
get_prometheus_metrics_response,
get_system_cache_status,
get_system_status_payload,
run_system_prewarm,
run_system_priority_warm,
)
router = APIRouter(tags=["system"])
@router.get("/healthz")
async def healthz():
return get_health_payload()
@router.get("/api/system/status")
async def system_status():
return await get_system_status_payload()
@router.post("/api/system/prewarm")
async def system_prewarm(
request: Request,
cities: Optional[str] = None,
force_refresh: bool = False,
include_detail: bool = False,
include_market: bool = False,
):
return run_system_prewarm(
request,
cities=cities,
force_refresh=force_refresh,
include_detail=include_detail,
include_market=include_market,
)
@router.get("/api/system/cache-status")
async def system_cache_status(request: Request, cities: Optional[str] = None):
return get_system_cache_status(request, cities=cities)
@router.post("/api/system/priority-warm")
async def system_priority_warm(
request: Request,
background_tasks: BackgroundTasks,
timezone: Optional[str] = None,
):
return run_system_priority_warm(request, background_tasks, timezone=timezone)
@router.get("/metrics", response_class=PlainTextResponse)
async def metrics():
return get_prometheus_metrics_response()
web/routes.py
+10 -1810
View File
File diff suppressed because it is too large Load Diff
web/services/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Service/helper modules used by the FastAPI web layer."""
web/services/analytics_api.py
+35
View File
@@ -0,0 +1,35 @@
"""Analytics API service functions."""
from __future__ import annotations
from typing import Any, Dict
from fastapi import HTTPException, Request
from src.database.db_manager import DBManager
from web.core import AnalyticsEventRequest
import web.routes as legacy_routes
def track_analytics_event(request: Request, body: AnalyticsEventRequest) -> Dict[str, Any]:
legacy_routes._bind_optional_supabase_identity(request)
event_type = str(body.event_type or "").strip().lower()
if event_type not in legacy_routes.TRACKABLE_ANALYTICS_EVENTS:
raise HTTPException(status_code=400, detail="unsupported_event_type")
payload = body.payload if isinstance(body.payload, dict) else {}
normalized_payload = {
key: value
for key, value in payload.items()
if isinstance(key, str) and len(key) <= 64
}
db = DBManager()
db.append_app_analytics_event(
event_type,
normalized_payload,
user_id=getattr(request.state, "auth_user_id", None),
client_id=body.client_id,
session_id=body.session_id,
)
return {"ok": True}
web/services/auth_api.py
+108
View File
@@ -0,0 +1,108 @@
"""Authentication API service functions."""
from __future__ import annotations
from typing import Any, Dict
from fastapi import Request
import web.routes as legacy_routes
def get_auth_me_payload(request: Request) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
legacy_routes._bind_optional_supabase_identity(request)
user_id = getattr(request.state, "auth_user_id", None)
subscription_required = bool(
legacy_routes.SUPABASE_ENTITLEMENT.enabled
and legacy_routes.SUPABASE_ENTITLEMENT.require_subscription
)
subscription_active = None
subscription_plan_code = None
subscription_starts_at = None
subscription_expires_at = None
subscription_total_expires_at = None
subscription_queued_days = 0
subscription_queued_count = 0
if legacy_routes.SUPABASE_ENTITLEMENT.enabled and user_id:
try:
latest_subscription = legacy_routes.SUPABASE_ENTITLEMENT.ensure_signup_trial(
user_id,
created_at=getattr(request.state, "auth_created_at", None),
)
if not latest_subscription:
latest_subscription = (
legacy_routes.SUPABASE_ENTITLEMENT.get_latest_active_subscription(
user_id,
respect_requirement=False,
)
)
latest_known_subscription = latest_subscription
if not latest_known_subscription:
latest_known_subscription = (
legacy_routes.SUPABASE_ENTITLEMENT.get_latest_subscription_any_status(
user_id
)
)
subscription_window = legacy_routes.SUPABASE_ENTITLEMENT.get_subscription_window(
user_id,
respect_requirement=False,
)
subscription_active = bool(latest_subscription)
if isinstance(latest_subscription, dict):
subscription_plan_code = latest_subscription.get("plan_code")
subscription_starts_at = latest_subscription.get("starts_at")
subscription_expires_at = latest_subscription.get("expires_at")
elif isinstance(latest_known_subscription, dict):
subscription_plan_code = latest_known_subscription.get("plan_code")
subscription_starts_at = latest_known_subscription.get("starts_at")
subscription_expires_at = latest_known_subscription.get("expires_at")
if isinstance(subscription_window, dict):
subscription_total_expires_at = subscription_window.get("total_expires_at")
subscription_queued_days = int(subscription_window.get("queued_days") or 0)
subscription_queued_count = int(subscription_window.get("queued_count") or 0)
except Exception:
subscription_active = None
subscription_plan_code = None
subscription_starts_at = None
subscription_expires_at = None
subscription_total_expires_at = None
subscription_queued_days = 0
subscription_queued_count = 0
points = legacy_routes._resolve_auth_points(request)
weekly_profile = legacy_routes._resolve_weekly_profile(request)
return {
"authenticated": bool(user_id),
"user_id": user_id,
"email": getattr(request.state, "auth_email", None),
"points": points,
"weekly_points": weekly_profile["weekly_points"],
"weekly_rank": weekly_profile["weekly_rank"],
"entitlement_mode": (
"supabase_required"
if legacy_routes.SUPABASE_ENTITLEMENT.enabled
and legacy_routes._SUPABASE_AUTH_REQUIRED
else "supabase_optional"
if legacy_routes.SUPABASE_ENTITLEMENT.enabled
else "legacy_token"
if legacy_routes._ENTITLEMENT_GUARD_ENABLED
else "disabled"
),
"auth_required": bool(
legacy_routes.SUPABASE_ENTITLEMENT.enabled
and legacy_routes._SUPABASE_AUTH_REQUIRED
),
"subscription_required": subscription_required,
"subscription_active": subscription_active,
"subscription_plan_code": subscription_plan_code,
"subscription_starts_at": subscription_starts_at,
"subscription_expires_at": subscription_expires_at,
"subscription_total_expires_at": subscription_total_expires_at,
"subscription_queued_days": subscription_queued_days,
"subscription_queued_count": subscription_queued_count,
}
web/services/city_api.py
+228
View File
@@ -0,0 +1,228 @@
"""City API service functions used by the city router."""
from __future__ import annotations
from typing import Any, Dict, Optional
from fastapi import BackgroundTasks, HTTPException, Request
from fastapi.concurrency import run_in_threadpool
from loguru import logger
import web.routes as legacy_routes
def _build_cities_payload() -> Dict[str, Any]:
out = []
deb_recent_index = legacy_routes._build_recent_deb_performance_index()
for name, info in legacy_routes.CITIES.items():
risk = legacy_routes.CITY_RISK_PROFILES.get(name, {})
city_meta = legacy_routes.CITY_REGISTRY.get(name, {}) or {}
deb_recent = deb_recent_index.get(name, {})
settlement_source = str(info.get("settlement_source") or "metar").strip().lower() or "metar"
provider = legacy_routes.get_country_network_provider(name)
out.append(
{
"name": name,
"display_name": str(city_meta.get("display_name") or city_meta.get("name") or name.title()),
"lat": info["lat"],
"lon": info["lon"],
"utc_offset_seconds": legacy_routes.get_city_utc_offset_seconds(name),
"risk_level": risk.get("risk_level", "low"),
"risk_emoji": risk.get("risk_emoji", "🟢"),
"airport": risk.get("airport_name", ""),
"icao": risk.get("icao", ""),
"temp_unit": "fahrenheit" if info["f"] else "celsius",
"is_major": city_meta.get("is_major", True),
"settlement_source": settlement_source,
"settlement_source_label": legacy_routes.SETTLEMENT_SOURCE_LABELS.get(
settlement_source,
settlement_source.upper(),
),
"settlement_station_code": city_meta.get("settlement_station_code") or city_meta.get("icao"),
"settlement_station_label": city_meta.get("settlement_station_label") or city_meta.get("airport_name"),
"network_provider": provider.provider_code,
"network_provider_label": provider.provider_label,
"deb_recent_tier": deb_recent.get("tier", "other"),
"deb_recent_hit_rate": deb_recent.get("hit_rate"),
"deb_recent_sample_count": deb_recent.get("sample_count", 0),
"deb_recent_mae": deb_recent.get("mae"),
"deb_recent_last_date": deb_recent.get("last_date"),
}
)
return {"cities": out}
async def list_cities_payload(_request: Request) -> Dict[str, Any]:
try:
return await run_in_threadpool(_build_cities_payload)
except Exception as exc:
logger.error(f"Error in list_cities: {exc}")
raise HTTPException(status_code=500, detail=str(exc)) from exc
async def get_city_detail_payload(
request: Request,
name: str,
*,
force_refresh: bool = False,
depth: str = "panel",
) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
city = legacy_routes._normalize_city_or_404(name)
normalized_depth = str(depth or "panel").strip().lower()
if normalized_depth == "full":
detail_mode = "full"
elif normalized_depth == "market":
detail_mode = "market"
elif normalized_depth == "nearby":
detail_mode = "nearby"
else:
detail_mode = "panel"
if detail_mode == "panel":
if force_refresh:
return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, True)
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "panel", city)
if cached_entry:
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_PANEL_CACHE_TTL_SEC):
return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, False)
return cached_entry.get("payload") or {}
return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, False)
if detail_mode == "nearby":
if force_refresh:
return await run_in_threadpool(legacy_routes._refresh_city_nearby_cache, city, True)
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "nearby", city)
if cached_entry:
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_NEARBY_CACHE_TTL_SEC):
return await run_in_threadpool(legacy_routes._refresh_city_nearby_cache, city, False)
return cached_entry.get("payload") or {}
return await run_in_threadpool(legacy_routes._refresh_city_nearby_cache, city, False)
if detail_mode == "market":
if force_refresh:
return await run_in_threadpool(legacy_routes._refresh_city_market_cache, city, True)
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "market", city)
if cached_entry:
if not legacy_routes._market_analysis_cache_is_fresh(cached_entry):
return await run_in_threadpool(legacy_routes._refresh_city_market_cache, city, False)
return cached_entry.get("payload") or {}
return await run_in_threadpool(legacy_routes._refresh_city_market_cache, city, False)
return await run_in_threadpool(legacy_routes._analyze, city, force_refresh, False, detail_mode)
async def get_city_history_payload(
request: Request,
name: str,
*,
include_records: bool = False,
) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
city = legacy_routes._normalize_city_or_404(name)
if include_records:
return await run_in_threadpool(legacy_routes._build_city_history_payload, city, True)
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "history_preview", city)
if cached_entry:
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_HISTORY_PREVIEW_CACHE_TTL_SEC):
return await run_in_threadpool(legacy_routes._refresh_city_history_preview_cache, city)
return cached_entry.get("payload") or {}
return await run_in_threadpool(legacy_routes._refresh_city_history_preview_cache, city)
async def get_city_summary_payload(
_request: Request,
name: str,
*,
force_refresh: bool = False,
) -> Dict[str, Any]:
city = legacy_routes._normalize_city_or_404(name)
if force_refresh:
return await run_in_threadpool(legacy_routes._refresh_city_summary_cache, city, True)
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "summary", city)
if cached_entry:
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_SUMMARY_CACHE_TTL_SEC):
return await run_in_threadpool(legacy_routes._refresh_city_summary_cache, city, False)
return cached_entry.get("payload") or {}
return await run_in_threadpool(legacy_routes._refresh_city_summary_cache, city, False)
async def get_city_detail_aggregate_payload(
request: Request,
name: str,
*,
force_refresh: bool = False,
market_slug: Optional[str] = None,
target_date: Optional[str] = None,
) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
city = legacy_routes._normalize_city_or_404(name)
data = await run_in_threadpool(legacy_routes._analyze, city, force_refresh, True)
return await run_in_threadpool(
legacy_routes._build_city_detail_payload,
data,
market_slug,
target_date,
)
async def get_city_market_scan_payload(
request: Request,
background_tasks: BackgroundTasks,
name: str,
*,
force_refresh: bool = False,
market_slug: Optional[str] = None,
target_date: Optional[str] = None,
lite: bool = False,
) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
city = legacy_routes._normalize_city_or_404(name)
if force_refresh:
data = await run_in_threadpool(legacy_routes._refresh_city_market_cache, city, True)
cached_scan = legacy_routes._get_cached_market_scan_payload(
data,
market_slug=market_slug,
target_date=target_date,
lite=lite,
)
if cached_scan is not None:
return cached_scan
else:
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "market", city)
if cached_entry:
data = cached_entry.get("payload") or {}
cached_scan = legacy_routes._get_cached_market_scan_payload(
data,
market_slug=market_slug,
target_date=target_date,
lite=lite,
)
if cached_scan is not None:
if not legacy_routes._market_analysis_cache_is_fresh(cached_entry):
legacy_routes._schedule_cache_refresh(background_tasks, kind="market", city=city)
return cached_scan
if legacy_routes._market_analysis_cache_is_fresh(cached_entry):
return await run_in_threadpool(
legacy_routes._refresh_market_scan_payload_from_cached_analysis,
city,
data,
market_slug=market_slug,
target_date=target_date,
lite=lite,
)
else:
legacy_routes._schedule_cache_refresh(background_tasks, kind="market", city=city)
else:
data = await run_in_threadpool(legacy_routes._refresh_city_market_cache, city, False)
cached_scan = legacy_routes._get_cached_market_scan_payload(
data,
market_slug=market_slug,
target_date=target_date,
lite=lite,
)
if cached_scan is not None:
return cached_scan
return await run_in_threadpool(
legacy_routes._build_city_market_scan_payload,
data,
market_slug,
target_date,
lite,
)
web/services/city_payloads.py
+277
View File
@@ -0,0 +1,277 @@
"""City payload builders for API-facing response shapes."""
from __future__ import annotations
from typing import Any, Dict, Optional
from src.analysis.settlement_rounding import apply_city_settlement
from web.core import _is_excluded_model_name, _market_layer, _sf
TURKISH_MGM_CITIES = {"ankara", "istanbul"}
def build_city_summary_payload(data: Dict[str, Any]) -> Dict[str, Any]:
return {
"name": data.get("name"),
"display_name": data.get("display_name"),
"icao": data.get("risk", {}).get("icao"),
"utc_offset_seconds": data.get("utc_offset_seconds"),
"local_time": data.get("local_time"),
"temp_symbol": data.get("temp_symbol"),
"current": {
"temp": data.get("current", {}).get("temp"),
"obs_time": data.get("current", {}).get("obs_time"),
"settlement_source": data.get("current", {}).get("settlement_source"),
"settlement_source_label": data.get("current", {}).get("settlement_source_label"),
},
"deb": {"prediction": data.get("deb", {}).get("prediction")},
"deviation_monitor": data.get("deviation_monitor") or {},
"risk": {
"level": data.get("risk", {}).get("level"),
"warning": data.get("risk", {}).get("warning"),
},
"updated_at": data.get("updated_at"),
}
def build_city_market_scan_payload(
data: Dict[str, Any],
market_slug: Optional[str] = None,
target_date: Optional[str] = None,
lite: bool = False,
scan_filters: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
city = str(data.get("name") or "").strip().lower()
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 {}
selected_daily = (
multi_model_daily.get(selected_date)
if isinstance(multi_model_daily, dict)
else None
)
if not isinstance(selected_daily, dict):
selected_daily = {}
selected_date = local_date
distribution = selected_daily.get("probabilities")
if not isinstance(distribution, list) or not distribution:
distribution = data.get("probabilities", {}).get("distribution", []) or []
distribution_all = selected_daily.get("probabilities_all")
if not isinstance(distribution_all, list) or not distribution_all:
distribution_all = data.get("probabilities", {}).get("distribution_all", []) or []
if not distribution_all:
distribution_all = distribution
model_map = selected_daily.get("models") or data.get("multi_model") or {}
if not isinstance(model_map, dict):
model_map = {}
anchor_temp = None
anchor_model = None
for model_name, raw_value in model_map.items():
value = _sf(raw_value)
if value is None:
continue
if anchor_temp is None or value > anchor_temp:
anchor_temp = value
anchor_model = str(model_name or "").strip() or None
anchor_temp_c = anchor_temp
temp_symbol = str(data.get("temp_symbol") or "")
if anchor_temp_c is not None and "F" in temp_symbol.upper():
anchor_temp_c = (anchor_temp_c - 32.0) * 5.0 / 9.0
anchor_settlement = apply_city_settlement(city, anchor_temp_c) if anchor_temp_c is not None else None
primary_bucket = None
if isinstance(distribution, list) and distribution:
ranked_buckets = []
temp_symbol_upper = str(temp_symbol or "").upper()
max_primary_bucket_delta = 16.0 if "F" in temp_symbol_upper else 8.0
for idx, row in enumerate(distribution_all):
if not isinstance(row, dict):
continue
bucket_value = _sf(
row.get("temp")
if row.get("temp") is not None
else row.get("value")
if row.get("value") is not None
else row.get("lower")
)
if (
anchor_temp is not None
and bucket_value is not None
and abs(float(bucket_value) - float(anchor_temp)) > max_primary_bucket_delta
):
continue
bucket_prob = _sf(row.get("probability"))
prob_rank = bucket_prob if bucket_prob is not None else -1.0
ranked_buckets.append((-prob_rank, idx, row))
if ranked_buckets:
ranked_buckets.sort(key=lambda x: (x[0], x[1]))
primary_bucket = ranked_buckets[0][2]
elif anchor_temp is None:
primary_bucket = distribution[0]
model_probability = None
if isinstance(primary_bucket, dict) and primary_bucket.get("probability") is not None:
try:
raw_probability = float(primary_bucket.get("probability"))
model_probability = raw_probability / 100.0 if raw_probability > 1.0 else raw_probability
except Exception:
model_probability = None
fallback_sparkline = [
p.get("probability", 0)
for p in distribution_all[:8]
if isinstance(p, dict)
]
current = data.get("current") or {}
selected_deb = selected_daily.get("deb") if isinstance(selected_daily.get("deb"), dict) else {}
current_deb = data.get("deb") if isinstance(data.get("deb"), dict) else {}
scan_context = {
"local_date": data.get("local_date"),
"local_time": data.get("local_time"),
"peak": data.get("peak") or {},
"current_max_so_far": current.get("max_so_far"),
"current_temp": current.get("temp"),
"trend": data.get("trend") or {},
"network_lead_signal": data.get("network_lead_signal") or {},
"models": model_map,
"deb_prediction": selected_deb.get("prediction") or current_deb.get("prediction"),
}
market_scan = _market_layer.build_market_scan(
city=data.get("name"),
target_date=selected_date or data.get("local_date"),
temperature_bucket=primary_bucket if isinstance(primary_bucket, dict) else None,
model_probability=model_probability,
probability_distribution=distribution_all,
temp_symbol=temp_symbol,
fallback_sparkline=fallback_sparkline,
forced_market_slug=market_slug,
include_related_buckets=not lite,
scan_filters=scan_filters,
scan_context=scan_context,
)
if isinstance(market_scan, dict):
market_scan["anchor_model"] = anchor_model
market_scan["anchor_high"] = anchor_temp
market_scan["anchor_settlement"] = anchor_settlement
market_scan["open_meteo_settlement"] = anchor_settlement
probabilities = data.get("probabilities") or {}
market_scan["probability_engine"] = str(
probabilities.get("engine") or "legacy"
).strip() or "legacy"
market_scan["probability_calibration_mode"] = str(
probabilities.get("calibration_mode") or "legacy"
).strip() or "legacy"
return {
"market_scan": market_scan,
"selected_date": selected_date or data.get("local_date"),
"fetched_at": data.get("updated_at"),
}
def build_city_detail_payload(
data: Dict[str, Any],
market_slug: Optional[str] = None,
target_date: Optional[str] = None,
) -> Dict[str, Any]:
market_payload = build_city_market_scan_payload(
data,
market_slug=market_slug,
target_date=target_date,
)
market_scan = market_payload.get("market_scan")
return {
"city": data.get("name"),
"fetched_at": data.get("updated_at"),
"overview": {
"name": data.get("name"),
"display_name": data.get("display_name"),
"icao": data.get("risk", {}).get("icao"),
"airport": data.get("risk", {}).get("airport"),
"lat": data.get("lat"),
"lon": data.get("lon"),
"local_time": data.get("local_time"),
"local_date": data.get("local_date"),
"temp_symbol": data.get("temp_symbol"),
"current_temp": data.get("current", {}).get("temp"),
"settlement_source": data.get("current", {}).get("settlement_source"),
"settlement_source_label": data.get("current", {}).get("settlement_source_label"),
"settlement_station": data.get("settlement_station") or {},
"deb_prediction": data.get("deb", {}).get("prediction"),
"risk_level": data.get("risk", {}).get("level"),
"risk_warning": data.get("risk", {}).get("warning"),
"updated_at": data.get("updated_at"),
},
"official": {
"available": bool(data.get("current", {}).get("temp") is not None),
"metar": {
"observation_time": data.get("airport_current", {}).get("obs_time"),
"obs_age_min": data.get("airport_current", {}).get("obs_age_min"),
"report_time": data.get("airport_current", {}).get("report_time"),
"receipt_time": data.get("airport_current", {}).get("receipt_time"),
"raw_metar": data.get("airport_current", {}).get("raw_metar"),
"current": data.get("airport_current") or {},
},
"taf": data.get("taf") or {},
"weather_gov": {},
"mgm": data.get("mgm") or {},
"mgm_nearby": data.get("mgm_nearby") or [],
"nearby_source": data.get("nearby_source") or ("mgm" if str(data.get("name") or "").lower() in TURKISH_MGM_CITIES else "metar_cluster"),
"airport_primary": data.get("airport_primary") or {},
"airport_primary_today_obs": data.get("airport_primary_today_obs") or [],
"official_nearby": data.get("official_nearby") or [],
"official_network_source": data.get("official_network_source"),
"official_network_status": data.get("official_network_status") or {},
"network_lead_signal": data.get("network_lead_signal") or {},
"network_spread_signal": data.get("network_spread_signal") or {},
"center_station_candidate": data.get("center_station_candidate"),
"airport_vs_network_delta": data.get("airport_vs_network_delta"),
},
"timeseries": {
"metar_recent_obs": data.get("metar_recent_obs") or [],
"metar_today_obs": data.get("metar_today_obs") or [],
"settlement_today_obs": data.get("settlement_today_obs") or [],
"hourly": data.get("hourly") or {},
"mgm_hourly": (data.get("mgm") or {}).get("hourly", []),
"forecast_daily": (data.get("forecast") or {}).get("daily", []),
},
"models": {
k: v
for k, v in (data.get("multi_model") or {}).items()
if not _is_excluded_model_name(k)
},
"deb": data.get("deb") or {},
"multi_model_daily": data.get("multi_model_daily") or {},
"probabilities": data.get("probabilities") or {"mu": None, "distribution": []},
"dynamic_commentary": data.get("dynamic_commentary") or {"summary": "", "notes": []},
"intraday_meteorology": data.get("intraday_meteorology")
or _build_intraday_meteorology(data),
"vertical_profile_signal": data.get("vertical_profile_signal") or {},
"taf": data.get("taf") or {},
"market_scan": market_scan,
"risk": data.get("risk"),
"settlement_station": data.get("settlement_station") or {},
"airport_primary": data.get("airport_primary") or {},
"official_nearby": data.get("official_nearby") or [],
"official_network_source": data.get("official_network_source"),
"official_network_status": data.get("official_network_status") or {},
"network_lead_signal": data.get("network_lead_signal") or {},
"network_spread_signal": data.get("network_spread_signal") or {},
"center_station_candidate": data.get("center_station_candidate"),
"airport_vs_network_delta": data.get("airport_vs_network_delta"),
"airport_current": data.get("airport_current") or {},
"nearby_source": data.get("nearby_source") or ("mgm" if str(data.get("name") or "").lower() in TURKISH_MGM_CITIES else "metar_cluster"),
"ai_analysis": data.get("ai_analysis") or "",
"errors": {},
}
def _build_intraday_meteorology(data: Dict[str, Any]) -> Dict[str, Any]:
from web.analysis_service import _build_intraday_meteorology as build_intraday
return build_intraday(data)
web/services/city_runtime.py
+804
View File
@@ -0,0 +1,804 @@
from __future__ import annotations
import os
import time
from datetime import datetime, timedelta
from typing import Optional
from fastapi import APIRouter, BackgroundTasks, HTTPException
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 (
DailyRecordRepository,
STATE_STORAGE_SQLITE,
TrainingFeatureRecordRepository,
TruthRecordRepository,
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 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_city_ai_forecast_payload, # noqa: F401 - compatibility export for tests and transitional routers
build_scan_terminal_ai_payload, # noqa: F401 - compatibility export for tests and transitional routers
build_scan_terminal_payload, # noqa: F401 - compatibility export for tests and transitional routers
stream_scan_city_ai_forecast_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,
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,
_is_excluded_model_name,
)
router = APIRouter()
_CACHE_DB = DBManager()
_DEB_RECENT_LOOKBACK = 7
_DEB_RECENT_MIN_SAMPLES = 3
_daily_record_repo = DailyRecordRepository()
_truth_record_repo = TruthRecordRepository()
_training_feature_repo = TrainingFeatureRecordRepository()
TRACKABLE_ANALYTICS_EVENTS = {
"signup_completed",
"dashboard_active",
"paywall_feature_clicked",
"paywall_viewed",
"checkout_started",
"checkout_succeeded",
}
DEFAULT_PREWARM_CITIES = [
"ankara",
"istanbul",
"shanghai",
"beijing",
"shenzhen",
"guangzhou",
"qingdao",
"wuhan",
"chengdu",
"chongqing",
"hong kong",
"taipei",
"singapore",
"tokyo",
"seoul",
"busan",
"london",
"paris",
"madrid",
]
HISTORY_PREVIEW_DAY_LIMIT = 21
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 = max(30, int(os.getenv("POLYWEATHER_CITY_SUMMARY_CACHE_TTL_SEC", "1800")))
CITY_PANEL_CACHE_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CITY_PANEL_CACHE_TTL_SEC", "1800")))
CITY_NEARBY_CACHE_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CITY_NEARBY_CACHE_TTL_SEC", "1800")))
CITY_MARKET_CACHE_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CITY_MARKET_CACHE_TTL_SEC", "1800")))
MARKET_SCAN_PAYLOAD_TTL_SEC = max(
5,
int(os.getenv("POLYWEATHER_MARKET_SCAN_PAYLOAD_TTL_SEC", "30")),
)
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 _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, 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")
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 _build_history_model_reference(
*,
forecasts: dict,
actual: object,
deb: object,
) -> dict:
"""Expose the archived model snapshot as reference evidence, not truth."""
actual_value = _sf(actual)
deb_value = _sf(deb)
entries = []
for model_name, model_value in (forecasts or {}).items():
if _is_excluded_model_name(str(model_name)):
continue
value = _sf(model_value)
if value is None:
continue
error = abs(value - actual_value) if actual_value is not None else None
entries.append(
{
"model": str(model_name),
"value": round(value, 1),
"error": round(error, 1) if error is not None else None,
"participates_in_deb": True,
}
)
entries.sort(
key=lambda row: (
row["error"] is None,
row["error"] if row["error"] is not None else 999,
row["model"],
)
)
deb_error = abs(deb_value - actual_value) if deb_value is not None and actual_value is not None else None
return {
"available": bool(entries),
"truth_layer": "settlement_actual",
"reference_layer": "archived_model_snapshot",
"deb": {
"value": round(deb_value, 1) if deb_value is not None else None,
"error": round(deb_error, 1) if deb_error is not None else None,
},
"models": entries,
"model_count": len(entries),
}
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,
)
model_reference = _build_history_model_reference(
forecasts=forecasts,
actual=act,
deb=deb,
)
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,
"model_reference": model_reference,
"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]:
raw = str(value or "").strip()
if not raw:
return None
try:
return datetime.fromisoformat(raw.replace("Z", "+00:00"))
except Exception:
return None
def _build_peak_minus_12h_reference(
*,
actual_high: object,
snapshots: list[dict],
) -> dict:
actual = _sf(actual_high)
if actual is None or not snapshots:
return {}
tolerance = 0.11
normalized = []
for row in snapshots:
if not isinstance(row, dict):
continue
dt = _parse_snapshot_dt(row.get("timestamp"))
if dt is None:
continue
normalized.append(
{
"dt": dt,
"max_so_far": _sf(row.get("max_so_far")),
"deb_prediction": _sf(row.get("deb_prediction")),
}
)
if not normalized:
return {}
peak_row = next(
(
row
for row in normalized
if row["max_so_far"] is not None and row["max_so_far"] >= actual - tolerance
),
None,
)
if peak_row is None:
return {}
peak_dt = peak_row["dt"]
anchor_dt = peak_dt - timedelta(hours=12)
anchor_row = None
for row in normalized:
if row["dt"] <= anchor_dt and row["deb_prediction"] is not None:
anchor_row = row
elif row["dt"] > anchor_dt:
break
peak_time = peak_dt.strftime("%H:%M")
result = {
"actual_peak_time": peak_time,
}
if anchor_row and anchor_row["deb_prediction"] is not None:
deb_value = float(anchor_row["deb_prediction"])
result.update(
{
"deb_at_peak_minus_12h": deb_value,
"deb_at_peak_minus_12h_time": anchor_row["dt"].strftime("%H:%M"),
"deb_at_peak_minus_12h_error": round(deb_value - actual, 1),
}
)
return result
def _merge_missing_history_forecasts_from_snapshots(
forecasts: dict,
snapshots: list[dict],
) -> dict:
merged = dict(forecasts or {})
if not snapshots:
return merged
fallback_values: dict[str, Optional[float]] = {}
for row in snapshots:
if not isinstance(row, dict):
continue
multi_model = row.get("multi_model") or {}
if not isinstance(multi_model, dict):
continue
for model_name, model_value in multi_model.items():
model_key = str(model_name or "").strip()
if not model_key or _is_excluded_model_name(model_key):
continue
parsed = _sf(model_value)
if parsed is not None:
fallback_values[model_key] = parsed
for model_name, model_value in fallback_values.items():
existing = _sf(merged.get(model_name))
if existing is None:
merged[model_name] = model_value
return merged
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_PREWARM_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('__'))]
web/services/groq_commentary.py
+223
View File
@@ -0,0 +1,223 @@
"""Groq-backed bilingual commentary enrichment."""
from __future__ import annotations
import hashlib
import json
import os
import re
import threading
import time
from typing import Any, Dict, Optional
import httpx
from loguru import logger
_GROQ_COMMENTARY_CACHE_LOCK = threading.Lock()
_GROQ_COMMENTARY_CACHE: Dict[str, Dict[str, Any]] = {}
_GROQ_COMMENTARY_CACHE_TTL_SEC = int(
os.getenv("POLYWEATHER_GROQ_COMMENTARY_CACHE_TTL_SEC", "1800")
)
def groq_commentary_enabled() -> bool:
enabled = str(
os.getenv("POLYWEATHER_GROQ_COMMENTARY_ENABLED", "false")
).strip().lower()
api_key = str(os.getenv("GROQ_API_KEY") or "").strip()
return enabled in {"1", "true", "yes", "on"} and bool(api_key)
def clean_commentary_text(value: Any, *, limit: int = 240) -> str:
text = str(value or "").strip()
if not text:
return ""
text = re.sub(r"\s+", " ", text)
return text[:limit].strip()
def build_groq_commentary_context(result: Dict[str, Any]) -> Dict[str, Any]:
dynamic = result.get("dynamic_commentary") or {}
vertical = result.get("vertical_profile_signal") or {}
taf_signal = ((result.get("taf") or {}).get("signal") or {}) if isinstance(result.get("taf"), dict) else {}
network = result.get("network_lead_signal") or {}
peak = result.get("peak") or {}
current = result.get("current") or {}
airport_primary = result.get("airport_primary") or {}
notes = dynamic.get("notes") if isinstance(dynamic.get("notes"), list) else []
compact_notes = [clean_commentary_text(item, limit=180) for item in notes]
compact_notes = [item for item in compact_notes if item][:4]
return {
"city": result.get("display_name") or result.get("name"),
"local_date": result.get("local_date"),
"local_time": result.get("local_time"),
"temp_symbol": result.get("temp_symbol"),
"current_temp": current.get("temp"),
"day_high_so_far": current.get("max_so_far"),
"airport_anchor_temp": airport_primary.get("temp"),
"airport_vs_network_delta": result.get("airport_vs_network_delta"),
"peak_hours": peak.get("hours") or [],
"peak_status": peak.get("status"),
"network_lead_status": network.get("status"),
"network_lead_note": clean_commentary_text(network.get("note"), limit=180),
"rules_summary": clean_commentary_text(dynamic.get("summary"), limit=260),
"rules_notes": compact_notes,
"upper_air_summary_zh": clean_commentary_text(vertical.get("summary_zh"), limit=260),
"upper_air_summary_en": clean_commentary_text(vertical.get("summary_en"), limit=260),
"taf_summary_zh": clean_commentary_text(taf_signal.get("summary_zh"), limit=220),
"taf_summary_en": clean_commentary_text(taf_signal.get("summary_en"), limit=220),
"taf_peak_window": clean_commentary_text(taf_signal.get("peak_window"), limit=80),
}
def normalize_groq_commentary_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
def _headline(value: Any, fallback: str) -> str:
text = clean_commentary_text(value, limit=90)
return text or fallback
def _bullets(value: Any) -> list[str]:
items = value if isinstance(value, list) else []
cleaned = [clean_commentary_text(item, limit=120) for item in items]
cleaned = [item for item in cleaned if item]
return cleaned[:3]
zh_headline = _headline(payload.get("headline_zh"), "结构信号以现有规则结论为主。")
en_headline = _headline(payload.get("headline_en"), "Structural read stays anchored to the existing rule-based signal.")
zh_bullets = _bullets(payload.get("bullets_zh"))
en_bullets = _bullets(payload.get("bullets_en"))
while len(zh_bullets) < 3:
zh_bullets.append("继续结合当前节奏、边界风险和峰值窗口判断。")
while len(en_bullets) < 3:
en_bullets.append("Keep the read anchored to pace, boundary risk, and the peak window.")
return {
"headline_zh": zh_headline,
"headline_en": en_headline,
"bullets_zh": zh_bullets[:3],
"bullets_en": en_bullets[:3],
"source": "groq",
}
def request_groq_commentary(context: Dict[str, Any]) -> Optional[Dict[str, Any]]:
api_key = str(os.getenv("GROQ_API_KEY") or "").strip()
if not api_key:
return None
model = str(os.getenv("POLYWEATHER_GROQ_COMMENTARY_MODEL") or "openai/gpt-oss-20b").strip()
timeout_sec = float(os.getenv("POLYWEATHER_GROQ_COMMENTARY_TIMEOUT_SEC", "8"))
payload = {
"model": model,
"temperature": 0.2,
"max_tokens": 400,
"messages": [
{
"role": "system",
"content": (
"You rewrite weather-market structure commentary. "
"Never invent facts. Use only the provided context. "
"Return concise bilingual output for a dashboard: "
"one headline and exactly three bullets in Chinese, and the same in English. "
"Keep every bullet actionable and short."
),
},
{
"role": "user",
"content": json.dumps(context, ensure_ascii=False),
},
],
"response_format": {
"type": "json_schema",
"json_schema": {
"name": "polyweather_structure_commentary",
"strict": True,
"schema": {
"type": "object",
"additionalProperties": False,
"properties": {
"headline_zh": {"type": "string"},
"bullets_zh": {
"type": "array",
"items": {"type": "string"},
"minItems": 3,
"maxItems": 3,
},
"headline_en": {"type": "string"},
"bullets_en": {
"type": "array",
"items": {"type": "string"},
"minItems": 3,
"maxItems": 3,
},
},
"required": [
"headline_zh",
"bullets_zh",
"headline_en",
"bullets_en",
],
},
},
},
}
with httpx.Client(timeout=timeout_sec) as client:
response = client.post(
"https://api.groq.com/openai/v1/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
json=payload,
)
response.raise_for_status()
body = response.json()
content = (
(((body.get("choices") or [{}])[0]).get("message") or {}).get("content")
if isinstance(body, dict)
else None
)
if not content:
return None
try:
return normalize_groq_commentary_payload(json.loads(str(content)))
except Exception:
logger.warning("Groq commentary returned non-JSON payload")
return None
def maybe_enrich_dynamic_commentary_with_groq(
city: str,
result: Dict[str, Any],
) -> Dict[str, Any]:
dynamic = result.get("dynamic_commentary") or {}
if not groq_commentary_enabled():
return dynamic
if dynamic.get("headline_zh") and dynamic.get("bullets_zh"):
return dynamic
context = build_groq_commentary_context(result)
if not context.get("rules_summary") and not context.get("rules_notes"):
return dynamic
cache_key = hashlib.sha256(
json.dumps({"city": city, "context": context}, sort_keys=True, ensure_ascii=False).encode("utf-8")
).hexdigest()
now = time.time()
with _GROQ_COMMENTARY_CACHE_LOCK:
cached = _GROQ_COMMENTARY_CACHE.get(cache_key)
if cached and now - float(cached.get("t") or 0) < _GROQ_COMMENTARY_CACHE_TTL_SEC:
merged = dict(dynamic)
merged.update(cached.get("payload") or {})
return merged
try:
enriched = request_groq_commentary(context)
except Exception as exc:
logger.warning("Groq commentary skipped for {}: {}", city, exc)
return dynamic
if not enriched:
return dynamic
with _GROQ_COMMENTARY_CACHE_LOCK:
_GROQ_COMMENTARY_CACHE[cache_key] = {"t": now, "payload": enriched}
merged = dict(dynamic)
merged.update(enriched)
return merged
web/services/ops_api.py
+228
View File
@@ -0,0 +1,228 @@
"""Operations/admin API service functions."""
from __future__ import annotations
from typing import Any, Dict
from fastapi import HTTPException, Request
from src.database.db_manager import DBManager
from web.core import GrantPointsRequest
import web.routes as legacy_routes
def _require_ops(request: Request) -> Dict[str, Any] | None:
legacy_routes._assert_entitlement(request)
return legacy_routes._require_ops_admin(request)
def search_ops_users(request: Request, q: str = "", limit: int = 20) -> Dict[str, Any]:
_require_ops(request)
db = DBManager()
return {"users": db.search_users(q, limit=limit)}
def get_ops_weekly_leaderboard(request: Request, limit: int = 20) -> Dict[str, Any]:
_require_ops(request)
db = DBManager()
return {"leaderboard": db.get_weekly_leaderboard(limit=limit)}
def list_ops_memberships(request: Request, limit: int = 200) -> Dict[str, Any]:
_require_ops(request)
db = DBManager()
if getattr(legacy_routes.PAYMENT_CHECKOUT, "enabled", False):
try:
legacy_routes.PAYMENT_CHECKOUT.reconcile_recent_intents(
limit=min(max(int(limit or 200), 20), 200)
)
except Exception:
pass
subscriptions = legacy_routes.SUPABASE_ENTITLEMENT.list_active_subscriptions(limit=limit)
subscription_user_ids = [str(item.get("user_id") or "") for item in subscriptions]
user_map = db.get_users_by_supabase_user_ids(subscription_user_ids)
unresolved_user_ids = [
user_id
for user_id in subscription_user_ids
if str(user_id or "").strip().lower()
and not str(
(user_map.get(str(user_id).strip().lower(), {}) or {}).get("supabase_email") or ""
).strip()
]
auth_user_map = legacy_routes.SUPABASE_ENTITLEMENT.get_auth_users(unresolved_user_ids)
deduped: dict[str, dict] = {}
for item in subscriptions:
user_id = str(item.get("user_id") or "").strip().lower()
local_user = user_map.get(user_id, {})
auth_user = auth_user_map.get(user_id, {})
subscription_window = legacy_routes.SUPABASE_ENTITLEMENT.get_subscription_window(
user_id,
respect_requirement=False,
bypass_cache=True,
)
current_expires_at = item.get("expires_at")
total_expires_at = (
subscription_window.get("total_expires_at")
if isinstance(subscription_window, dict)
else None
)
queued_days = (
int(subscription_window.get("queued_days") or 0)
if isinstance(subscription_window, dict)
else 0
)
queued_count = (
int(subscription_window.get("queued_count") or 0)
if isinstance(subscription_window, dict)
else 0
)
row = {
"user_id": user_id,
"email": str(auth_user.get("email") or local_user.get("supabase_email") or ""),
"telegram_id": local_user.get("telegram_id"),
"username": local_user.get("username"),
"registered_at": local_user.get("created_at") or auth_user.get("created_at"),
"plan_code": item.get("plan_code"),
"starts_at": item.get("starts_at"),
"current_expires_at": current_expires_at,
"total_expires_at": total_expires_at or current_expires_at,
"expires_at": total_expires_at or current_expires_at,
"queued_days": queued_days,
"queued_count": queued_count,
}
existing = deduped.get(user_id)
existing_expires = str(existing.get("expires_at") or "") if existing else ""
current_expires = str(row.get("expires_at") or "")
if existing is None or current_expires > existing_expires:
deduped[user_id] = row
rows = sorted(
deduped.values(),
key=lambda item: str(item.get("expires_at") or ""),
)
return {"memberships": rows}
def list_ops_payment_incidents(
request: Request,
limit: int = 50,
reason: str = "",
include_resolved: bool = False,
) -> Dict[str, Any]:
_require_ops(request)
db = DBManager()
incidents = db.list_payment_audit_events(
limit=max(1, min(int(limit or 50), 200)),
event_type="payment_intent_failed",
)
normalized_reason = str(reason or "").strip().lower()
filtered = []
for item in incidents:
payload = item.get("payload") if isinstance(item, dict) else {}
payload = payload if isinstance(payload, dict) else {}
item_reason = str(payload.get("reason") or "").strip().lower()
resolved_at = str(payload.get("resolved_at") or "").strip()
if normalized_reason and item_reason != normalized_reason:
continue
if not include_resolved and resolved_at:
continue
filtered.append(item)
return {"incidents": filtered}
def resolve_ops_payment_incident(request: Request, event_id: int) -> Dict[str, Any]:
admin = _require_ops(request) or {}
db = DBManager()
resolved = db.mark_payment_audit_event_resolved(event_id, str(admin.get("email") or ""))
if not resolved:
raise HTTPException(status_code=404, detail="payment_incident_not_found")
return {"ok": True, "incident": resolved}
def grant_ops_points(request: Request, body: GrantPointsRequest) -> Dict[str, Any]:
admin = _require_ops(request) or {}
db = DBManager()
result = db.grant_points_by_supabase_email(body.email, body.points)
result["operator_email"] = admin.get("email")
if not result.get("ok"):
reason = str(result.get("reason") or "grant_points_failed")
status_code = 404 if reason == "user_not_found" else 400
raise HTTPException(status_code=status_code, detail=result)
return result
def get_ops_analytics_funnel(request: Request, days: int = 30) -> Dict[str, Any]:
_require_ops(request)
db = DBManager()
return db.get_app_analytics_funnel_summary(days=days)
def get_ops_truth_history(
request: Request,
city: str = "",
date_from: str = "",
date_to: str = "",
limit: int = 200,
) -> Dict[str, Any]:
_require_ops(request)
truth_history = legacy_routes.TruthRecordRepository().load_all()
normalized_city = str(city or "").strip().lower()
normalized_from = str(date_from or "").strip()
normalized_to = str(date_to or "").strip()
max_limit = max(1, min(int(limit or 200), 1000))
rows = []
for row_city, by_date in truth_history.items():
if normalized_city and row_city != normalized_city:
continue
if not isinstance(by_date, dict):
continue
for target_date, payload in by_date.items():
if normalized_from and str(target_date) < normalized_from:
continue
if normalized_to and str(target_date) > normalized_to:
continue
if not isinstance(payload, dict):
continue
rows.append(
{
"city": row_city,
"display_name": str(
(legacy_routes.CITY_REGISTRY.get(row_city) or {}).get("name") or row_city
),
"target_date": str(target_date),
"actual_high": payload.get("actual_high"),
"settlement_source": payload.get("settlement_source"),
"settlement_station_code": payload.get("settlement_station_code"),
"settlement_station_label": payload.get("settlement_station_label"),
"truth_version": payload.get("truth_version"),
"updated_by": payload.get("updated_by"),
"truth_updated_at": payload.get("truth_updated_at"),
"is_final": payload.get("is_final"),
}
)
rows.sort(key=lambda item: (str(item["target_date"]), str(item["city"])), reverse=True)
filtered_count = len(rows)
rows = rows[:max_limit]
available_cities = [
{
"city": city_id,
"name": str(info.get("name") or city_id),
}
for city_id, info in sorted(
legacy_routes.CITY_REGISTRY.items(),
key=lambda item: str(item[1].get("name") or item[0]),
)
]
return {
"items": rows,
"available_cities": available_cities,
"filters": {
"city": normalized_city or None,
"date_from": normalized_from or None,
"date_to": normalized_to or None,
"limit": max_limit,
},
"filtered_count": filtered_count,
}
web/services/payment_api.py
+179
View File
@@ -0,0 +1,179 @@
"""Payment API service functions used by the payments router."""
from __future__ import annotations
from typing import Any, Dict
from fastapi import HTTPException, Request
from src.database.db_manager import DBManager
from web.core import (
ConfirmPaymentTxRequest,
CreatePaymentIntentRequest,
SubmitPaymentTxRequest,
WalletChallengeRequest,
WalletUnbindRequest,
WalletVerifyRequest,
)
import web.routes as legacy_routes
def _raise_payment_error(exc: Exception) -> None:
raise HTTPException(
status_code=getattr(exc, "status_code", 400),
detail=getattr(exc, "detail", str(exc)),
) from exc
def _require_payment_identity(request: Request) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
return legacy_routes._require_supabase_identity(request)
def get_payment_config(request: Request) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
try:
return legacy_routes.PAYMENT_CHECKOUT.get_config_payload()
except legacy_routes.PaymentCheckoutError as exc:
_raise_payment_error(exc)
def get_payment_runtime(request: Request) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
try:
db = DBManager()
return {
"checkout": legacy_routes.PAYMENT_CHECKOUT.get_config_payload(),
"rpc": legacy_routes.PAYMENT_CHECKOUT.get_rpc_runtime_status(),
"event_loop_state": db.get_payment_runtime_state("payment_event_loop") or {},
"recent_audit_events": db.list_payment_audit_events(limit=20),
}
except legacy_routes.PaymentCheckoutError as exc:
_raise_payment_error(exc)
def list_payment_wallets(request: Request) -> Dict[str, Any]:
identity = _require_payment_identity(request)
try:
wallets = legacy_routes.PAYMENT_CHECKOUT.list_wallets(identity["user_id"])
return {
"wallets": [wallet.__dict__ for wallet in wallets],
"chain_id": legacy_routes.PAYMENT_CHECKOUT.chain_id,
}
except legacy_routes.PaymentCheckoutError as exc:
_raise_payment_error(exc)
def unbind_payment_wallet(request: Request, body: WalletUnbindRequest) -> Dict[str, Any]:
identity = _require_payment_identity(request)
try:
return legacy_routes.PAYMENT_CHECKOUT.unbind_wallet(
user_id=identity["user_id"],
address=body.address,
)
except legacy_routes.PaymentCheckoutError as exc:
_raise_payment_error(exc)
def create_payment_wallet_challenge(
request: Request,
body: WalletChallengeRequest,
) -> Dict[str, Any]:
identity = _require_payment_identity(request)
try:
return legacy_routes.PAYMENT_CHECKOUT.create_wallet_challenge(
user_id=identity["user_id"],
address=body.address,
)
except legacy_routes.PaymentCheckoutError as exc:
_raise_payment_error(exc)
def verify_payment_wallet(
request: Request,
body: WalletVerifyRequest,
) -> Dict[str, Any]:
identity = _require_payment_identity(request)
try:
bound = legacy_routes.PAYMENT_CHECKOUT.verify_wallet_binding(
user_id=identity["user_id"],
address=body.address,
nonce=body.nonce,
signature=body.signature,
)
return {"wallet": bound.__dict__}
except legacy_routes.PaymentCheckoutError as exc:
_raise_payment_error(exc)
def create_payment_intent(
request: Request,
body: CreatePaymentIntentRequest,
) -> Dict[str, Any]:
identity = _require_payment_identity(request)
try:
return legacy_routes.PAYMENT_CHECKOUT.create_intent(
user_id=identity["user_id"],
plan_code=body.plan_code,
payment_mode=body.payment_mode,
allowed_wallet=body.allowed_wallet,
token_address=body.token_address,
use_points=body.use_points,
points_to_consume=body.points_to_consume,
metadata=body.metadata,
)
except legacy_routes.PaymentCheckoutError as exc:
_raise_payment_error(exc)
def get_payment_intent(request: Request, intent_id: str) -> Dict[str, Any]:
identity = _require_payment_identity(request)
try:
intent = legacy_routes.PAYMENT_CHECKOUT.get_intent(
user_id=identity["user_id"],
intent_id=intent_id,
)
return {"intent": intent.__dict__}
except legacy_routes.PaymentCheckoutError as exc:
_raise_payment_error(exc)
def submit_payment_tx(
request: Request,
intent_id: str,
body: SubmitPaymentTxRequest,
) -> Dict[str, Any]:
identity = _require_payment_identity(request)
try:
return legacy_routes.PAYMENT_CHECKOUT.submit_intent_tx(
user_id=identity["user_id"],
intent_id=intent_id,
tx_hash=body.tx_hash,
from_address=body.from_address,
)
except legacy_routes.PaymentCheckoutError as exc:
_raise_payment_error(exc)
def confirm_payment_tx(
request: Request,
intent_id: str,
body: ConfirmPaymentTxRequest,
) -> Dict[str, Any]:
identity = _require_payment_identity(request)
try:
return legacy_routes.PAYMENT_CHECKOUT.confirm_intent_tx(
user_id=identity["user_id"],
intent_id=intent_id,
tx_hash=body.tx_hash,
)
except legacy_routes.PaymentCheckoutError as exc:
_raise_payment_error(exc)
def reconcile_latest_payment(request: Request) -> Dict[str, Any]:
identity = _require_payment_identity(request)
try:
return legacy_routes.PAYMENT_CHECKOUT.reconcile_latest_intent(identity["user_id"])
except legacy_routes.PaymentCheckoutError as exc:
_raise_payment_error(exc)
web/services/scan_api.py
+111
View File
@@ -0,0 +1,111 @@
"""Market scan API service functions."""
from __future__ import annotations
from typing import Any, Dict
from fastapi import HTTPException, Request
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import StreamingResponse
import web.routes as legacy_routes
def _boolish(value: Any) -> bool:
return str(value or "false").lower() in {"1", "true", "yes", "on"}
async def _json_body_or_empty(request: Request) -> Dict[str, Any]:
try:
body = await request.json()
except Exception:
body = {}
if not isinstance(body, dict):
raise HTTPException(status_code=400, detail="Invalid JSON body")
return body
def _extract_required_city(body: Dict[str, Any]) -> str:
city = str(body.get("city") or "").strip()
if not city:
raise HTTPException(status_code=400, detail="city is required")
return city
async def get_scan_terminal_payload(
request: Request,
*,
scan_mode: str = "tradable",
min_price: float = 0.05,
max_price: float = 0.95,
min_edge_pct: float = 2.0,
min_liquidity: float = 500.0,
high_liquidity_only: bool = False,
market_type: str = "maxtemp",
time_range: str = "today",
limit: int = 25,
force_refresh: bool = False,
) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
filters = {
"scan_mode": scan_mode,
"min_price": min_price,
"max_price": max_price,
"min_edge_pct": min_edge_pct,
"min_liquidity": min_liquidity,
"high_liquidity_only": high_liquidity_only,
"market_type": market_type,
"time_range": time_range,
"limit": limit,
}
return await run_in_threadpool(
legacy_routes.build_scan_terminal_payload,
filters,
force_refresh=force_refresh,
)
async def get_scan_terminal_ai_payload(request: Request) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
body = await _json_body_or_empty(request)
filters = body.get("filters") if isinstance(body.get("filters"), dict) else {}
snapshot_id = str(body.get("snapshot_id") or "").strip() or None
return await run_in_threadpool(
legacy_routes.build_scan_terminal_ai_payload,
filters,
snapshot_id=snapshot_id,
)
async def get_scan_city_ai_forecast_payload(request: Request) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
body = await _json_body_or_empty(request)
city = _extract_required_city(body)
force_refresh = _boolish(body.get("force_refresh"))
locale = str(body.get("locale") or "zh-CN").strip()
return await run_in_threadpool(
legacy_routes.build_scan_city_ai_forecast_payload,
city,
force_refresh=force_refresh,
locale=locale,
)
async def get_scan_city_ai_stream_response(request: Request) -> StreamingResponse:
legacy_routes._assert_entitlement(request)
body = await _json_body_or_empty(request)
city = _extract_required_city(body)
force_refresh = _boolish(body.get("force_refresh"))
locale = str(body.get("locale") or "zh-CN").strip()
return StreamingResponse(
legacy_routes.stream_scan_city_ai_forecast_payload(
city,
force_refresh=force_refresh,
locale=locale,
),
media_type="text/event-stream",
headers={
"Cache-Control": "no-store",
"X-Accel-Buffering": "no",
},
)
web/services/system_api.py
+169
View File
@@ -0,0 +1,169 @@
"""System and observability API service functions."""
from __future__ import annotations
import time
from typing import Any, Dict, Optional
from fastapi import BackgroundTasks, HTTPException, Request
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import PlainTextResponse
from loguru import logger
from src.utils.metrics import export_prometheus_metrics
from web.core import build_health_payload, build_system_status_payload
import web.routes as legacy_routes
def get_health_payload() -> Dict[str, Any]:
payload = build_health_payload()
if payload.get("status") != "ok":
raise HTTPException(status_code=503, detail=payload)
return payload
async def get_system_status_payload() -> Dict[str, Any]:
return await run_in_threadpool(build_system_status_payload)
def run_system_prewarm(
request: Request,
*,
cities: Optional[str] = None,
force_refresh: bool = False,
include_detail: bool = False,
include_market: bool = False,
) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
selected = legacy_routes._normalize_city_list(cities)
if not selected:
raise HTTPException(status_code=400, detail="No valid cities to prewarm")
started = time.perf_counter()
warmed: list[dict[str, object]] = []
failed: list[dict[str, object]] = []
summary_ok = 0
detail_ok = 0
market_ok = 0
for city in selected:
city_started = time.perf_counter()
try:
legacy_routes._refresh_city_summary_cache(city, force_refresh=force_refresh)
entry: dict[str, object] = {
"city": city,
"summary": True,
"duration_ms": round((time.perf_counter() - city_started) * 1000.0, 1),
}
summary_ok += 1
if include_detail:
legacy_routes._refresh_city_panel_cache(city, force_refresh=force_refresh)
entry["detail"] = True
detail_ok += 1
if include_market:
entry["market"] = False
warmed.append(entry)
except Exception as exc:
failed.append(
{
"city": city,
"error": str(exc),
"duration_ms": round((time.perf_counter() - city_started) * 1000.0, 1),
}
)
total_ms = round((time.perf_counter() - started) * 1000.0, 1)
logger.info(
"system prewarm finished count={} failed={} force_refresh={} include_detail={} include_market={} duration_ms={}",
len(warmed),
len(failed),
force_refresh,
include_detail,
include_market,
total_ms,
)
return {
"ok": len(failed) == 0,
"cities": selected,
"warmed": warmed,
"failed": failed,
"summary_ok": summary_ok,
"panel_ok": detail_ok,
"detail_ok": detail_ok,
"market_ok": market_ok,
"failed_count": len(failed),
"duration_ms": total_ms,
}
def get_system_cache_status(request: Request, cities: Optional[str] = None) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
selected = legacy_routes._normalize_city_list(cities)
if not selected:
selected = list(legacy_routes.DEFAULT_PREWARM_CITIES)
kinds = {
"summary": legacy_routes.CITY_SUMMARY_CACHE_TTL_SEC,
"panel": legacy_routes.CITY_PANEL_CACHE_TTL_SEC,
"nearby": legacy_routes.CITY_NEARBY_CACHE_TTL_SEC,
"market": legacy_routes.CITY_MARKET_CACHE_TTL_SEC,
"history_preview": legacy_routes.CITY_HISTORY_PREVIEW_CACHE_TTL_SEC,
}
items = []
for city in selected:
row = {"city": city}
for kind, ttl_sec in kinds.items():
entry = legacy_routes._CACHE_DB.get_city_cache(kind, city)
row[kind] = {
"exists": bool(entry),
"fresh": legacy_routes._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}
def run_system_priority_warm(
request: Request,
background_tasks: BackgroundTasks,
timezone: Optional[str] = None,
) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request)
batches = legacy_routes._select_priority_city_batches(timezone)
primary = list(batches.get("primary") or [])
secondary = list(batches.get("secondary") or [])
def _runner() -> None:
for city in primary:
try:
legacy_routes._refresh_city_summary_cache(city, force_refresh=False)
legacy_routes._refresh_city_panel_cache(city, force_refresh=False)
legacy_routes._refresh_city_nearby_cache(city, force_refresh=False)
legacy_routes._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:
legacy_routes._refresh_city_summary_cache(city, force_refresh=False)
legacy_routes._refresh_city_panel_cache(city, force_refresh=False)
except Exception as exc:
logger.warning("priority warm secondary failed city={} timezone={}: {}", city, timezone, exc)
background_tasks.add_task(_runner)
return {
"ok": True,
"region": batches.get("region"),
"timezone": batches.get("timezone"),
"primary": primary,
"secondary": secondary,
}
def get_prometheus_metrics_response() -> PlainTextResponse:
return PlainTextResponse(
export_prometheus_metrics(),
media_type="text/plain; version=0.0.4; charset=utf-8",
)