架构重构:拆分 core/ops/DBManager,统一 SQLite 锁,DEB 改进,新增注意力模型

- web/core.py 858→236行,拆出 schemas/middleware/auth/diagnostics
- ops_api.py 2876→4个 domain 模块 (users/payments/health/config)
- DBManager Supabase HTTP 调用提取到 SupabaseAdminClient
- 新增 LockedSQLiteConnection 统一多进程读写锁
- 新增 WeatherCacheManager 替代 12 个独立缓存字典
- METAR 缓存迁移至统一缓存管理器
- analysis_service 提取 _build_intraday_meteorology 到独立模块
- DEB 改进:偏差惩罚、分歧回退、自适应 lookback (MAE ↓12.6%)
- 新增 PyTorch 注意力模型 deb_attention.py (数据积累后启用)
- 新增 torch 到 requirements.lock
This commit is contained in:
2569718930@qq.com
2026-06-16 02:00:22 +08:00
parent ed0447f408
commit dca4f2d618
26 changed files with 5010 additions and 3974 deletions
+1 -1
View File
@@ -6,7 +6,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
PolyWeather Pro — a paid institutional weather-intelligence terminal. 50 monitored cities with real-time METAR/AMOS/MADIS observations, DEB multi-model temperature blending, Mu probability calibration, and intraday bias correction. Pure meteorological decision workspace; no market/price layer. Next.js 15 + React 19 frontend (Docker / VPS, behind Cloudflare + Nginx), FastAPI backend (VPS), Telegram bot.
**Business model**: Paid-only, $10/month, no free tier, no trial. Landing page is public; `/terminal` requires login + active subscription.
**Business model**: Paid-only, 29.9 USDC/month or 79.9 USDC/quarter, referral first month 20 USDC. New users get a one-time 3-day trial. Landing page is public; `/terminal` requires login + active subscription.
## Environment & Preferences
+4
View File
@@ -109,6 +109,10 @@ multidict==6.7.1
netcdf4==1.7.4
# via -r requirements.txt
numpy==2.4.6
torch @ https://download.pytorch.org/whl/cpu/torch-2.5.1%2Bcpu-cp311-cp311-manylinux_2_28_x86_64.whl ; python_version >= "3.11"
typing_extensions==4.15.0
# via torch
# via
# -r requirements.txt
# cftime
+47 -6
View File
@@ -1159,11 +1159,24 @@ def calculate_dynamic_weight_components(
city_data = data[city_name]
sorted_dates = sorted(city_data.keys(), reverse=True)
today_str = datetime.now().strftime("%Y-%m-%d")
available_days = sum(
1 for d in sorted_dates
if d != today_str and city_data[d].get("actual_high") is not None
)
# ── 改进3: 自适应 lookback — 数据多的城市用更多历史 ──
adaptive_lookback = min(14, max(7, available_days // 4))
effective_lookback = max(lookback_days, adaptive_lookback) if lookback_days <= 7 else lookback_days
# ── 改进1: 偏差惩罚 — per-model signed bias ──
model_biases: dict = {model: 0.0 for model in forecasts.keys()}
bias_samples: dict = {model: 0 for model in forecasts.keys()}
errors: dict = {model: [] for model in forecasts.keys()}
days_used = 0
for date_str in sorted_dates:
if date_str == datetime.now().strftime("%Y-%m-%d"):
if date_str == today_str:
continue
record = city_data[date_str]
@@ -1174,8 +1187,6 @@ def calculate_dynamic_weight_components(
if actual is None:
continue
decay_weight = decay_factor ** days_used
for model in forecasts.keys():
if model in past_forecasts and past_forecasts[model] is not None:
try:
@@ -1183,6 +1194,10 @@ def calculate_dynamic_weight_components(
av = float(actual)
except (TypeError, ValueError):
continue
# Track signed error for bias
model_biases[model] += (pv - av) # positive = model overpredicts
bias_samples[model] += 1
# Track absolute error for MAE
daily_error = abs(pv - av)
h_err = (
past_hourly_error.get(model)
@@ -1190,15 +1205,17 @@ def calculate_dynamic_weight_components(
else None
)
blended_error = _blend_mae(daily_error, h_err)
decay_weight = decay_factor ** days_used
errors[model].append((blended_error, decay_weight))
days_used += 1
if days_used >= lookback_days:
if days_used >= effective_lookback:
break
if days_used < 2:
return _equal_weight_result(f"等权平均(由于仅{days_used}天历史)", days_used)
# Compute weighted MAEs
maes = {}
for model, err_weighted in errors.items():
if err_weighted:
@@ -1210,8 +1227,14 @@ def calculate_dynamic_weight_components(
else:
maes[model] = 2.0
# Finalize per-model biases — only trust when enough samples
for model in forecasts.keys():
n = bias_samples.get(model, 0)
model_biases[model] = model_biases[model] / n if n >= 5 else 0.0
# ── 改进1: 偏差惩罚进逆误差 ──
inverse_errors = {
m: 1.0 / (mae + 0.1)
m: 1.0 / (mae + abs(model_biases[m]) * 0.5 + 0.1)
for m, mae in maes.items()
if forecasts.get(m) is not None
}
@@ -1229,12 +1252,30 @@ def calculate_dynamic_weight_components(
}
weights = {m: inv / total_inv for m, inv in inverse_errors.items()}
# ── 改进2: 分歧感知的权重回退 ──
forecast_values = [v for v in forecasts.values() if v is not None]
if len(forecast_values) >= 2:
spread = max(forecast_values) - min(forecast_values)
if spread > 3.0:
trust_factor = 3.0 / spread # 分歧>3°F时才回退
n = len(weights)
weights = {
m: w * trust_factor + (1.0 - trust_factor) / n
for m, w in weights.items()
}
blended_high = sum(forecasts[m] * weights[m] for m in weights)
sorted_models = sorted(weights.items(), key=lambda x: x[1], reverse=True)
weight_str_parts = []
for m, w in sorted_models[:3]:
weight_str_parts.append(f"{m}({w * 100:.0f}%,MAE:{maes[m]:.1f}°)")
parts = [f"{m}({w * 100:.0f}%"]
if maes.get(m):
parts.append(f"MAE:{maes[m]:.1f}°")
if abs(model_biases.get(m, 0.0)) >= 0.3:
parts.append(f"bias:{model_biases[m]:+.1f}")
weight_str_parts.append(",".join(parts) + ")")
if dedup_note:
weight_str_parts.append(dedup_note)
+276
View File
@@ -0,0 +1,276 @@
"""PyTorch learnable attention ensemble for DEB.
Tiny MLP (~5K params) that learns per-city model weighting from
cross-city training data. Trains in seconds on CPU, <1ms inference.
Usage:
python -m src.analysis.deb_attention --train
python -m src.analysis.deb_attention --eval
"""
from __future__ import annotations
import json
import math
import os
import random
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
import torch
import torch.nn as nn
import torch.nn.functional as F
def _load_training_data(history, min_models=3):
samples = []
for city, days in (history or {}).items():
for date_str, record in days.items():
forecasts = record.get("forecasts", {})
actual = record.get("actual_high")
if actual is None:
continue
valid = {m: float(v) for m, v in forecasts.items() if v is not None}
if len(valid) < min_models:
continue
samples.append({"city": city, "forecasts": valid, "actual": float(actual), "date": date_str})
return samples
def _build_vocabs(samples, min_model_freq=2):
from collections import Counter
mf = Counter[str]()
cities = set()
for s in samples:
cities.add(s["city"])
for m in s["forecasts"]:
mf[m] += 1
mv = {m: i for i, m in enumerate(sorted([m for m, c in mf.items() if c >= min_model_freq], key=lambda m: (-mf[m], m)))}
cv = {c: i for i, c in enumerate(sorted(cities))}
return mv, cv
class DebAttention(nn.Module):
"""Learnable attention: per-model scorer + city bias correction.
Input: per-model forecast values + city id
Output: weighted prediction (learned attention weights) + city bias
"""
def __init__(self, n_models, n_cities, d_hidden=32, dropout=0.1):
super().__init__()
self.n_models = n_models
self.n_cities = n_cities
# City embedding for context
self.city_emb = nn.Embedding(n_cities + 1, 16, padding_idx=0)
# Per-model scorer: [fc_norm, city_emb] → attention score
self.scorer = nn.Sequential(
nn.Linear(3 + 16, d_hidden), nn.GELU(), nn.Dropout(dropout),
nn.Linear(d_hidden, d_hidden // 2), nn.GELU(),
nn.Linear(d_hidden // 2, 1),
)
# City-level bias
self.city_bias = nn.Embedding(n_cities + 1, 1, padding_idx=0)
self.register_buffer("fc_mean", torch.tensor(0.0))
self.register_buffer("fc_std", torch.tensor(10.0))
def forward(self, forecasts, city_idx, mask):
"""(B, M), (B,), (B, M) → (B,), (B, M)"""
batch, max_m = forecasts.shape
fc_norm = (forecasts - self.fc_mean) / self.fc_std.clamp_min(0.1)
# Consensus: trimmed mean of valid forecasts
fc_masked = forecasts * mask.float()
n_valid = mask.sum(dim=1, keepdim=True).clamp_min(1)
consensus = fc_masked.sum(dim=1) / n_valid.squeeze() # (B,)
# Deviation from consensus
dev = fc_norm - consensus.unsqueeze(1) * mask.float() / self.fc_std.clamp_min(0.1)
# Feature: [fc_norm, dev_from_consensus, is_extreme]
is_extreme = (dev.abs() > 1.5).float() # flag if >1.5 std away
feats = torch.stack([fc_norm, dev, is_extreme], dim=-1) # (B, M, 3)
city = self.city_emb(city_idx.clamp(0, self.n_cities)) # (B, 16)
city_tiled = city.unsqueeze(1).expand(-1, max_m, -1) # (B, M, 16)
feats = torch.cat([feats, city_tiled], dim=-1) # (B, M, 19)
scores = self.scorer(feats).squeeze(-1) # (B, M)
scores = scores.masked_fill(~mask, float("-inf"))
attn = F.softmax(scores, dim=-1)
pred = (forecasts * attn).sum(dim=-1)
bias = self.city_bias(city_idx.clamp(0, self.n_cities)).squeeze(-1)
pred = pred + bias * 0.3
return pred, attn
def train_model(history, *, epochs=200, batch_size=128, lr=0.002, patience=30, checkpoint_dir="data", seed=42):
torch.manual_seed(seed)
random.seed(seed)
samples = _load_training_data(history)
if len(samples) < 50:
raise ValueError(f"Need >=50 samples, got {len(samples)}")
mv, cv = _build_vocabs(samples)
max_models = len(mv) + 4
samples.sort(key=lambda s: s["date"])
split = int(len(samples) * 0.8)
train_s = samples[:split]
val_s = samples[split:]
all_fc = [v for s in train_s for v in s["forecasts"].values()]
fc_mean = sum(all_fc) / len(all_fc)
fc_std = max(1.0, math.sqrt(sum((v - fc_mean) ** 2 for v in all_fc) / len(all_fc)))
def precompute(s_list):
X_fc, X_cid, X_mask, Y = [], [], [], []
for s in s_list:
fc_list = [s["forecasts"][m] for m in sorted(s["forecasts"]) if m in mv]
n = len(fc_list)
X_fc.append(fc_list + [0.0] * (max_models - n))
X_cid.append(cv.get(s["city"], 0))
X_mask.append([1.0] * n + [0.0] * (max_models - n))
Y.append(s["actual"])
return (
torch.tensor(X_fc, dtype=torch.float32),
torch.tensor(X_cid, dtype=torch.long),
torch.tensor(X_mask, dtype=torch.bool),
torch.tensor(Y, dtype=torch.float32),
)
train_X = precompute(train_s)
val_X = precompute(val_s)
train_loader = torch.utils.data.DataLoader(torch.utils.data.TensorDataset(*train_X), batch_size=batch_size, shuffle=True)
val_loader = torch.utils.data.DataLoader(torch.utils.data.TensorDataset(*val_X), batch_size=batch_size * 2, shuffle=False)
model = DebAttention(max_models, len(cv) + 1)
model.fc_mean.fill_(fc_mean)
model.fc_std.fill_(fc_std)
opt = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=1e-5)
sched = torch.optim.lr_scheduler.ReduceLROnPlateau(opt, mode="min", factor=0.5, patience=10)
best_val = float("inf")
best_state = None
no_imp = 0
print(f"Training: {len(train_s)} train, {len(val_s)} val, models={len(mv)}, cities={len(cv)}")
for epoch in range(epochs):
model.train()
tr_loss = 0.0
for fc, cid, mask, y in train_loader:
pred, _ = model(fc, cid, mask)
loss = F.mse_loss(pred, y)
opt.zero_grad()
loss.backward()
opt.step()
tr_loss += loss.item() * fc.size(0)
tr_loss /= len(train_X[0])
model.eval()
val_loss = 0.0
with torch.no_grad():
for fc, cid, mask, y in val_loader:
pred, _ = model(fc, cid, mask)
loss = F.mse_loss(pred, y)
val_loss += loss.item() * fc.size(0)
val_loss /= len(val_X[0])
sched.step(val_loss)
if val_loss < best_val:
best_val = val_loss
best_state = {k: v.cpu().clone() for k, v in model.state_dict().items()}
no_imp = 0
else:
no_imp += 1
if epoch % 20 == 0 or epoch == epochs - 1:
print(f" epoch {epoch:3d} tr={math.sqrt(tr_loss):.3f} val={math.sqrt(val_loss):.3f} lr={opt.param_groups[0]['lr']:.1e}")
if no_imp >= patience:
print(f" early stop {epoch}")
break
if best_state:
model.load_state_dict(best_state)
os.makedirs(checkpoint_dir, exist_ok=True)
ckpt = os.path.join(checkpoint_dir, "deb_attention.pt")
torch.save({"state": model.state_dict(), "mv": mv, "cv": cv, "fc_mean": fc_mean, "fc_std": fc_std, "max_m": max_models}, ckpt)
print(f" saved {ckpt}")
return model, mv, cv
def load_model(path):
ckpt = torch.load(path, map_location="cpu", weights_only=False)
model = DebAttention(ckpt["max_m"], len(ckpt["cv"]) + 1)
model.load_state_dict(ckpt["state"])
model.fc_mean.fill_(ckpt["fc_mean"])
model.fc_std.fill_(ckpt["fc_std"])
model.eval()
return model, ckpt["mv"], ckpt["cv"]
def predict(model, forecasts, city, mv, cv):
model.eval()
fc_list = [forecasts[m] for m in sorted(forecasts) if m in mv]
if not fc_list:
return None, {}
n = len(fc_list)
max_m = model.n_models
fc_t = torch.zeros(max_m)
mask_t = torch.zeros(max_m, dtype=torch.bool)
fc_t[:n] = torch.tensor(fc_list)
mask_t[:n] = True
cid = cv.get(city, 0)
with torch.no_grad():
pred, attn = model(fc_t.unsqueeze(0), torch.tensor([cid]), mask_t.unsqueeze(0))
pred = round(pred.item(), 1)
models = [m for m in sorted(forecasts) if m in mv]
weights = {models[i]: attn[0, i].item() for i in range(n)}
total = sum(weights.values())
if total > 0:
weights = {m: w / total for m, w in weights.items()}
return pred, weights
if __name__ == "__main__":
import sys, statistics
from src.analysis.deb_algorithm import load_history as lh, get_deb_accuracy, calculate_dynamic_weights
from src.data_collection.city_registry import CITY_REGISTRY
root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
hp = os.path.join(root, "data", "daily_records.json")
history = lh(hp)
if "--train" in sys.argv:
m, mv, cv = train_model(history)
print("Done.")
elif "--eval" in sys.argv:
ckpt = os.path.join(root, "data", "deb_attention.pt")
if not os.path.exists(ckpt):
print("No checkpoint. Run --train first.")
sys.exit(1)
m, mv, cv = load_model(ckpt)
base, attn_errs = [], []
for city in [c for c in CITY_REGISTRY if c in history]:
cd = history.get(city, {})
for d in sorted(cd.keys(), reverse=True)[:5]:
rec = cd[d]; fc = rec.get("forecasts", {}); actual = rec.get("actual_high")
if not fc or actual is None: continue
try:
bp, _ = calculate_dynamic_weights(city, fc)
if bp: base.append(abs(bp - float(actual)))
except: pass
try:
ap, _ = predict(m, fc, city, mv, cv)
if ap: attn_errs.append(abs(ap - float(actual)))
except: pass
if base and attn_errs:
print(f"Baseline MAE: {statistics.mean(base):.3f}")
print(f"Attention MAE: {statistics.mean(attn_errs):.3f}")
else:
print("python -m src.analysis.deb_attention --train | --eval")
+149
View File
@@ -0,0 +1,149 @@
"""Thin Supabase Admin HTTP client — extracted from DBManager.
Handles URL construction, auth headers, and the actual HTTP PATCH calls
to Supabase REST / Admin APIs. Cache-cooldown and local SQLite lookups
remain in DBManager.
"""
from __future__ import annotations
import os
from typing import Any, Dict, Optional
import requests
from loguru import logger
class SupabaseAdminClient:
"""Stateless HTTP client for Supabase Management API calls."""
def __init__(self) -> None:
self._supabase_url = str(os.getenv("SUPABASE_URL") or "").strip().rstrip("/")
self._service_role_key = str(
os.getenv("SUPABASE_SERVICE_ROLE_KEY") or ""
).strip()
@property
def configured(self) -> bool:
return bool(self._supabase_url and self._service_role_key)
# ------------------------------------------------------------------
# URL builders
# ------------------------------------------------------------------
def profiles_endpoint(self) -> str:
if not self._supabase_url:
return ""
return f"{self._supabase_url}/rest/v1/profiles"
def admin_users_endpoint(self) -> str:
if not self._supabase_url:
return ""
return f"{self._supabase_url}/auth/v1/admin/users"
# ------------------------------------------------------------------
# Headers
# ------------------------------------------------------------------
def _service_headers(self, *, prefer: str = "return=minimal") -> Dict[str, str]:
return {
"apikey": self._service_role_key,
"Authorization": f"Bearer {self._service_role_key}",
"Accept": "application/json",
"Content-Type": "application/json",
"Prefer": prefer,
}
# ------------------------------------------------------------------
# Operations
# ------------------------------------------------------------------
def patch_user_metadata(
self,
supabase_user_id: str,
metadata: Dict[str, Any],
*,
timeout: float = 8.0,
) -> bool:
"""PATCH /auth/v1/admin/users/{id} with user_metadata."""
if not self.configured:
return False
endpoint = self.admin_users_endpoint()
if not endpoint:
return False
try:
resp = requests.patch(
f"{endpoint}/{supabase_user_id}",
json={"user_metadata": metadata},
headers=self._service_headers(),
timeout=timeout,
)
ok = resp.status_code in (200, 204)
if not ok:
logger.warning(
"supabase admin patch user_metadata failed "
"user_id={} status={} body={}",
supabase_user_id,
resp.status_code,
(resp.text or "")[:200],
)
return ok
except Exception as exc:
logger.warning(
"supabase admin patch user_metadata error user_id={}: {}",
supabase_user_id,
exc,
)
return False
def patch_profile(
self,
supabase_user_id: str,
fields: Dict[str, Any],
*,
timeout: float = 8.0,
) -> bool:
"""PATCH /rest/v1/profiles?id=eq.{user_id} with telegram fields."""
if not self.configured:
return False
endpoint = self.profiles_endpoint()
if not endpoint:
return False
try:
resp = requests.patch(
endpoint,
params={"id": f"eq.{supabase_user_id}"},
json=fields,
headers=self._service_headers(),
timeout=timeout,
)
ok = resp.status_code in (200, 204)
if not ok:
logger.warning(
"supabase profiles patch failed "
"user_id={} status={} body={}",
supabase_user_id,
resp.status_code,
(resp.text or "")[:240],
)
return ok
except Exception as exc:
logger.warning(
"supabase profiles patch error user_id={}: {}",
supabase_user_id,
exc,
)
return False
# Module-level singleton — mirrors the pattern used by DBManager.
_supabase_admin_client: Optional[SupabaseAdminClient] = None
def get_supabase_admin_client() -> SupabaseAdminClient:
global _supabase_admin_client
if _supabase_admin_client is None:
_supabase_admin_client = SupabaseAdminClient()
return _supabase_admin_client
+11 -14
View File
@@ -97,12 +97,11 @@ class MetarSourceMixin:
cache_key = f"{icao}:{utc_offset}:{use_fahrenheit}"
now_ts = time.time()
cache_ttl_sec = self._metar_cache_ttl_for_city(city, icao)
with self._metar_cache_lock:
cached = self._metar_cache.get(cache_key)
if cached and now_ts - cached["t"] < cache_ttl_sec:
logger.debug(f"METAR cache hit {icao} age={int(now_ts - cached['t'])}s")
record_source_call("metar", "current", "cache_hit", (time.perf_counter() - started) * 1000.0)
return cached["d"]
cached = self.cache.get_ttl("metar", cache_key, cache_ttl_sec)
if cached is not None:
logger.debug(f"METAR cache hit {icao} age=...")
record_source_call("metar", "current", "cache_hit", (time.perf_counter() - started) * 1000.0)
return cached
try:
url = "https://aviationweather.gov/api/data/metar"
@@ -331,19 +330,17 @@ class MetarSourceMixin:
)
else:
logger.info(f"✈️ METAR {icao}: temperature unavailable (obs: {obs_time})")
with self._metar_cache_lock:
self._metar_cache[cache_key] = {"d": result, "t": now_ts}
self.cache.set_ttl("metar", cache_key, result)
record_source_call("metar", "current", "success", (time.perf_counter() - started) * 1000.0)
return result
except httpx.HTTPError as exc:
logger.error(f"METAR 请求失败 ({icao}): {exc}")
with self._metar_cache_lock:
stale = self._metar_cache.get(cache_key)
if stale:
logger.warning(f"METAR {icao} 请求失败,使用缓存回退")
record_source_call("metar", "current", "stale_cache", (time.perf_counter() - started) * 1000.0)
return stale["d"]
stale = self.cache.get("metar", cache_key)
if stale:
logger.warning(f"METAR {icao} 请求失败,使用缓存回退")
record_source_call("metar", "current", "stale_cache", (time.perf_counter() - started) * 1000.0)
return stale.get("d") if isinstance(stale, dict) else None
record_source_call("metar", "current", "error", (time.perf_counter() - started) * 1000.0)
return None
except (KeyError, IndexError, TypeError) as exc:
+120
View File
@@ -0,0 +1,120 @@
"""Thread-safe TTL cache manager for weather data collector sources.
Replaces the per-source cache dictionaries and locks scattered across
WeatherDataCollector Mixins with a single unified store.
"""
from __future__ import annotations
import threading
import time
from typing import Any, Callable, Dict, Optional
class WeatherCacheManager:
"""Key-value store with per-entry TTL and automatic trimming."""
def __init__(self, trim_interval_writes: int = 200) -> None:
self._lock = threading.Lock()
self._stores: Dict[str, Dict[str, Dict[str, Any]]] = {}
self._write_count = 0
self._trim_interval = max(1, int(trim_interval_writes))
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def get(self, store: str, key: str) -> Optional[Dict[str, Any]]:
with self._lock:
entries = self._stores.get(store)
if entries is None:
return None
return entries.get(key)
def set(self, store: str, key: str, value: Dict[str, Any]) -> None:
with self._lock:
if store not in self._stores:
self._stores[store] = {}
self._stores[store][key] = value
self._write_count += 1
self._maybe_trim()
def get_ttl(
self,
store: str,
key: str,
ttl_sec: float,
) -> Optional[Dict[str, Any]]:
"""Return the cached value if it exists and is younger than *ttl_sec*."""
now = time.time()
with self._lock:
entries = self._stores.get(store)
if entries is None:
return None
entry = entries.get(key)
if entry is None:
return None
cached_ts = float(entry.get("_ts") or 0)
if now - cached_ts > ttl_sec:
return None
return entry.get("d")
def set_ttl(self, store: str, key: str, value: Dict[str, Any]) -> None:
"""Store a value with an automatic timestamp for TTL lookups."""
self.set(store, key, {"_ts": time.time(), "d": value})
def store_size(self, store: str) -> int:
with self._lock:
entries = self._stores.get(store)
return len(entries) if entries else 0
# ------------------------------------------------------------------
# Trimming
# ------------------------------------------------------------------
def _maybe_trim(self) -> None:
if self._write_count % self._trim_interval != 0:
return
self.trim_stale()
def trim_stale(
self,
ttl_map: Optional[Dict[str, float]] = None,
) -> None:
"""Remove entries older than their store's TTL.
*ttl_map* maps store name → ttl_sec. Stores not in the map are skipped.
"""
if not ttl_map:
return
now = time.time()
with self._lock:
for store_name, ttl_sec in ttl_map.items():
entries = self._stores.get(store_name)
if not entries:
continue
stale = [
key
for key, entry in entries.items()
if now - float(entry.get("_ts") or 0) > ttl_sec
]
for key in stale:
del entries[key]
def clear_store(self, store: str) -> None:
with self._lock:
self._stores.pop(store, None)
# Module-level singleton for the default cache used by WeatherDataCollector.
_weather_cache: Optional[WeatherCacheManager] = None
_weather_cache_lock = threading.Lock()
def get_weather_cache() -> WeatherCacheManager:
global _weather_cache
if _weather_cache is None:
with _weather_cache_lock:
if _weather_cache is None:
_weather_cache = WeatherCacheManager()
return _weather_cache
+15 -7
View File
@@ -212,8 +212,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
os.getenv("METAR_FAST_CACHE_TTL_SEC", str(OBSERVATION_REFRESH_SEC))
)
self.metar_fast_cache_ttl_sec = min(self.metar_fast_cache_ttl_sec, OBSERVATION_REFRESH_SEC)
self._metar_cache: Dict[str, Dict] = {}
self._metar_cache_lock = threading.Lock()
# METAR cache migrated to self.cache (WeatherCacheManager)
self.taf_cache_ttl_sec = int(
os.getenv("TAF_CACHE_TTL_SEC", "900")
)
@@ -300,6 +299,11 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
self._CACHE_TRIM_EVERY_N_WRITES = int(
os.getenv("POLYWEATHER_CACHE_TRIM_INTERVAL", "200")
)
# Unified cache manager — new sources should use this instead of per-source dicts.
from src.data_collection.weather_cache import WeatherCacheManager
self.cache = WeatherCacheManager(
trim_interval_writes=self._CACHE_TRIM_EVERY_N_WRITES,
)
# 设置代理
proxy = config.get("proxy")
@@ -344,7 +348,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
(self._open_meteo_cache, self._open_meteo_cache_lock, 3600.0),
(self._ensemble_cache, self._ensemble_cache_lock, 7200.0),
(self._multi_model_cache, self._multi_model_cache_lock, 7200.0),
(self._metar_cache, self._metar_cache_lock, float(self.metar_cache_ttl_sec * 2)),
# metar cache migrated to self.cache
(self._taf_cache, self._taf_cache_lock, float(self.taf_cache_ttl_sec * 2)),
(self._jma_cache, self._jma_cache_lock, float(self.jma_cache_ttl_sec * 2)),
(self._settlement_cache, self._settlement_cache_lock, float(self.settlement_cache_ttl_sec * 2)),
@@ -367,6 +371,8 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
with lock:
for key in stale:
cache.pop(key, None)
# Unified cache stores
self.cache.trim_stale({"metar": float(self.metar_cache_ttl_sec * 2)})
def _post_temperature_patch_payload(
self,
@@ -960,10 +966,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
icao = self.get_icao_code(city)
if icao:
prefix = f"{icao}:"
with self._metar_cache_lock:
for key in list(self._metar_cache.keys()):
if key.startswith(prefix):
self._metar_cache.pop(key, None)
# Clear matching entries from unified cache
with self.cache._lock:
entries = self.cache._stores.get("metar", {})
stale = [k for k in entries if k.startswith(prefix)]
for key in stale:
del entries[key]
normalized = str(city or "").strip().lower()
with self._jma_cache_lock:
self._jma_cache.pop(f"{normalized}:{use_fahrenheit}", None)
+16 -66
View File
@@ -14,6 +14,7 @@ import requests
from loguru import logger
from src.database.sqlite_connection import connect_sqlite
from src.auth.supabase_admin_client import get_supabase_admin_client
class DBManager:
@@ -54,28 +55,16 @@ class DBManager:
self._initialized_paths.add(cache_key)
def _supabase_profiles_endpoint(self) -> str:
supabase_url = str(os.getenv("SUPABASE_URL") or "").strip().rstrip("/")
if not supabase_url:
return ""
return f"{supabase_url}/rest/v1/profiles"
return get_supabase_admin_client().profiles_endpoint()
def _supabase_service_headers(self) -> Dict[str, str]:
service_role_key = str(os.getenv("SUPABASE_SERVICE_ROLE_KEY") or "").strip()
if not service_role_key:
client = get_supabase_admin_client()
if not client.configured:
return {}
return {
"apikey": service_role_key,
"Authorization": f"Bearer {service_role_key}",
"Accept": "application/json",
"Content-Type": "application/json",
"Prefer": "return=minimal",
}
return client._service_headers()
def _supabase_admin_users_endpoint(self) -> str:
supabase_url = str(os.getenv("SUPABASE_URL") or "").strip().rstrip("/")
if not supabase_url:
return ""
return f"{supabase_url}/auth/v1/admin/users"
return get_supabase_admin_client().admin_users_endpoint()
def _profile_sync_min_interval_sec(self) -> float:
raw = str(
@@ -249,35 +238,16 @@ class DBManager:
):
return False
try:
resp = requests.patch(
f"{endpoint}/{supabase_user_id}",
json={"user_metadata": {"points": points}},
headers={**headers, "Prefer": "return=minimal"},
timeout=8,
)
if resp.status_code not in (200, 204):
logger.warning(
"supabase points sync failed tg={} suid={} status={} body={}",
telegram_id,
supabase_user_id,
resp.status_code,
(resp.text or "")[:200],
)
return False
admin = get_supabase_admin_client()
if not admin.configured:
return False
ok = admin.patch_user_metadata(supabase_user_id, {"points": points})
if ok:
self._remember_points_metadata_sync(
telegram_id=int(telegram_id),
points=points,
)
return True
except Exception as exc:
logger.warning(
"supabase points sync error tg={} suid={}: {}",
telegram_id,
supabase_user_id,
exc,
)
return False
return ok
def _sync_supabase_profile_telegram_fields(
self,
@@ -305,35 +275,15 @@ class DBManager:
"telegram_username": str(telegram_username or "").strip() or None,
"updated_at": datetime.now().isoformat(),
}
try:
response = requests.patch(
endpoint,
params={"id": f"eq.{normalized_uid}"},
json=payload,
headers=headers,
timeout=8,
)
if response.status_code not in {200, 204}:
logger.warning(
"supabase profiles telegram sync failed user_id={} status={} body={}",
normalized_uid,
response.status_code,
(response.text or "")[:240],
)
return False
admin = get_supabase_admin_client()
ok = admin.patch_profile(normalized_uid, payload)
if ok:
self._remember_profile_sync(
supabase_user_id=normalized_uid,
telegram_id=telegram_id,
telegram_username=telegram_username,
)
return True
except Exception as exc:
logger.warning(
"supabase profiles telegram sync error user_id={}: {}",
normalized_uid,
exc,
)
return False
return ok
def _sync_bound_supabase_profiles_for_telegram(
self,
+2
View File
@@ -28,6 +28,8 @@ class _DummyMetarSource(MetarSourceMixin):
def __init__(self):
self._metar_cache = {}
self._metar_cache_lock = threading.Lock()
from src.data_collection.weather_cache import WeatherCacheManager
self.cache = WeatherCacheManager()
self.metar_timeout_sec = 0.0
self.metar_latest_timeout_sec = 0.0
self.timeout = 0.0
+2 -1
View File
@@ -2,6 +2,7 @@ import sqlite3
from datetime import datetime, timezone
from src.database.db_manager import DBManager
from src.database.sqlite_connection import connect_sqlite
from web.realtime_event_store import RealtimeEventStore
from web.realtime_patch_schema import normalize_observation_patch
@@ -75,7 +76,7 @@ def test_event_store_cleanup_uses_short_replay_retention(tmp_path):
old_event = store.append_event(_event("taipei", 30.0))
fresh_event = store.append_event(_event("taipei", 31.0))
with sqlite3.connect(db_path) as conn:
with connect_sqlite(db_path) as conn:
conn.execute(
"UPDATE observation_patch_events SET created_at = ? WHERE revision = ?",
("2026-05-26T00:00:00+00:00", old_event["revision"]),
+1 -289
View File
@@ -46,14 +46,7 @@ from web.services.observation_freshness import (
build_observation_freshness as _build_observation_freshness,
observation_age_min as _observation_age_min,
)
from web.services.analysis_utils import (
add_signal as _add_signal,
bucket_label as _bucket_label,
bucket_label_from_value as _bucket_label_from_value,
format_clock_minutes as _format_clock_minutes,
next_observation_clock as _next_observation_clock,
top_probability_bucket as _top_probability_bucket,
)
from web.services.intraday_meteorology import build_intraday_meteorology as _build_intraday_meteorology
from web.services.analysis_signals import (
_build_deviation_monitor,
_build_taf_signal,
@@ -403,287 +396,6 @@ def _set_cached_summary(city: str, payload: Dict[str, Any]) -> None:
def _build_intraday_meteorology(data: Dict[str, Any]) -> Dict[str, Any]:
"""Build a paid-product intraday meteorology read from existing layers."""
current = data.get("current") or {}
probabilities = data.get("probabilities") or {}
distribution = probabilities.get("distribution") or []
top_bucket = _top_probability_bucket(distribution)
unit = str(data.get("temp_symbol") or "°C")
deb = data.get("deb") or {}
peak = data.get("peak") or {}
deviation = data.get("deviation_monitor") or {}
taf_signal = (
((data.get("taf") or {}).get("signal") or {})
if isinstance(data.get("taf"), dict)
else {}
)
vertical = data.get("vertical_profile_signal") or {}
current_temp = _sf(current.get("temp"))
max_so_far = _sf(current.get("max_so_far"))
deb_prediction = _sf(deb.get("prediction"))
base_value = _sf(top_bucket.get("value")) if isinstance(top_bucket, dict) else None
if base_value is None:
base_value = deb_prediction
if base_value is None:
base_value = max_so_far if max_so_far is not None else current_temp
base_case_bucket = _bucket_label(top_bucket, unit) or _bucket_label_from_value(base_value, unit)
upside_bucket = _bucket_label_from_value(base_value + 1.0, unit) if base_value is not None else None
downside_bucket = _bucket_label_from_value(base_value - 1.0, unit) if base_value is not None else None
signals: list = []
support_score = 0
suppress_score = 0
available_layers = 0
direction = str(deviation.get("direction") or "").lower()
severity = str(deviation.get("severity") or "normal").lower()
delta = _sf(deviation.get("current_delta"))
if direction:
available_layers += 1
strength = "strong" if severity == "strong" else ("medium" if severity == "light" else "weak")
if direction == "hot":
support_score += 2 if strength == "strong" else 1
_add_signal(
signals,
label="日内节奏",
label_en="Intraday pace",
direction="support",
strength=strength,
summary=f"实测较预期路径偏高 {abs(delta or 0):.1f}{unit},峰值仍有上修空间。",
summary_en=f"Observed temperature is running {abs(delta or 0):.1f}{unit} above the expected path; the peak still has upside room.",
)
elif direction == "cold":
suppress_score += 2 if strength == "strong" else 1
_add_signal(
signals,
label="日内节奏",
label_en="Intraday pace",
direction="suppress",
strength=strength,
summary=f"实测较预期路径偏低 {abs(delta or 0):.1f}{unit},追更高温档需要等待后续观测确认。",
summary_en=f"Observed temperature is running {abs(delta or 0):.1f}{unit} below the expected path; higher buckets need confirmation from later observations.",
)
else:
_add_signal(
signals,
label="日内节奏",
label_en="Intraday pace",
direction="neutral",
strength="weak",
summary="实测大体贴近当前预期路径,下一步主要看峰值窗口内是否继续抬升。",
summary_en="Observed temperature is broadly tracking the expected path; the next question is whether it keeps lifting through the peak window.",
)
heating_setup = str(vertical.get("heating_setup") or "").lower()
suppression_risk = str(vertical.get("suppression_risk") or "").lower()
if heating_setup or suppression_risk:
available_layers += 1
if heating_setup == "supportive":
support_score += 2
_add_signal(
signals,
label="边界层结构",
label_en="Boundary-layer setup",
direction="support",
strength="strong",
summary=str(vertical.get("summary_zh") or "边界层结构支持白天继续混合升温。"),
summary_en=str(vertical.get("summary_en") or "The boundary-layer setup supports continued daytime mixing and warming."),
)
elif heating_setup == "suppressed" or suppression_risk == "high":
suppress_score += 2
_add_signal(
signals,
label="边界层结构",
label_en="Boundary-layer setup",
direction="suppress",
strength="strong",
summary=str(vertical.get("summary_zh") or "边界层或云雨结构对午后峰值形成压制。"),
summary_en=str(vertical.get("summary_en") or "Boundary-layer or cloud/rain structure is capping the afternoon peak."),
)
else:
_add_signal(
signals,
label="边界层结构",
label_en="Boundary-layer setup",
direction="neutral",
strength="medium",
summary=str(vertical.get("summary_zh") or "边界层结构暂未给出单边信号。"),
summary_en=str(vertical.get("summary_en") or "The boundary-layer setup does not yet provide a one-sided signal."),
)
taf_suppression = str(taf_signal.get("suppression_level") or "").lower()
taf_disruption = str(taf_signal.get("disruption_level") or "").lower()
taf_has_cloud_rain_cap = taf_suppression in {"medium", "high"} or taf_disruption in {
"medium",
"high",
}
structural_cap = False
if taf_signal.get("available") or taf_suppression:
available_layers += 1
if taf_suppression == "high" or taf_disruption == "high":
suppress_score += 2
direction_value = "suppress"
strength = "strong"
elif taf_suppression == "medium" or taf_disruption == "medium":
suppress_score += 1
direction_value = "suppress"
strength = "medium"
else:
support_score += 1
direction_value = "support"
strength = "weak"
_add_signal(
signals,
label="TAF 云雨扰动",
label_en="TAF cloud/rain disruption",
direction=direction_value,
strength=strength,
summary=str(taf_signal.get("summary_zh") or "TAF 暂未提示强云雨压温信号。"),
summary_en=str(taf_signal.get("summary_en") or "TAF does not yet flag a strong cloud/rain temperature cap."),
)
airport_delta = _sf(data.get("airport_vs_network_delta"))
lead_signal = data.get("network_lead_signal") or {}
if airport_delta is not None:
available_layers += 1
leader = str(lead_signal.get("leader_station_label") or lead_signal.get("leader_station_code") or "").strip()
sync_status = str(lead_signal.get("leader_sync_status") or "").strip().lower()
sync_delta = _sf(lead_signal.get("leader_time_delta_vs_anchor_minutes"))
sync_suffix_zh = ""
sync_suffix_en = ""
if sync_status in {"near_realtime", "lagged"} and sync_delta is not None:
sync_suffix_zh = f";但与机场锚点约差 {sync_delta:.0f} 分钟,作为降权信号处理"
sync_suffix_en = f"; timing differs from the airport anchor by about {sync_delta:.0f} minutes, so this signal is down-weighted"
elif sync_status == "unknown":
sync_suffix_zh = ";周边站观测时间不可完全校验,作为弱参考"
sync_suffix_en = "; station timing is not fully verified, so this is treated as a weak reference"
if airport_delta <= -0.4:
support_score += 1
_add_signal(
signals,
label="站网对比",
label_en="Station-network comparison",
direction="support",
strength="weak" if sync_suffix_zh else "medium",
summary=f"周边站网较机场锚点偏热 {abs(airport_delta):.1f}{unit}{f',领先点位 {leader}' if leader else ''}{sync_suffix_zh}",
summary_en=f"Nearby stations are {abs(airport_delta):.1f}{unit} warmer than the airport anchor{f'; leading site: {leader}' if leader else ''}{sync_suffix_en}.",
)
elif airport_delta >= 0.4:
suppress_score += 1
_add_signal(
signals,
label="站网对比",
label_en="Station-network comparison",
direction="suppress",
strength="weak" if sync_suffix_zh else "medium",
summary=f"机场锚点较周边站网偏热 {abs(airport_delta):.1f}{unit},继续上修需要机场自身后续报文确认{sync_suffix_zh}",
summary_en=f"The airport anchor is {abs(airport_delta):.1f}{unit} warmer than nearby stations; further upside needs confirmation from later airport reports{sync_suffix_en}.",
)
else:
_add_signal(
signals,
label="站网对比",
label_en="Station-network comparison",
direction="neutral",
strength="weak",
summary="机场锚点与周边站网基本同步,暂不构成单独上修或下修理由。",
summary_en="The airport anchor and nearby station network are broadly aligned, so this layer does not independently argue for upside or downside.",
)
peak_status = str(peak.get("status") or "").lower()
first_h = _sf(peak.get("first_h"))
last_h = _sf(peak.get("last_h"))
peak_window = (
f"{int(first_h):02d}:00-{int(last_h):02d}:59"
if first_h is not None and last_h is not None
else "--"
)
if peak_status == "past":
headline = "峰值窗口已过,后续更偏向确认最终高点而非继续上修。"
headline_en = "The peak window has passed; the read now shifts toward confirming the final high rather than chasing further upside."
confidence = "high" if available_layers >= 2 else "medium"
elif suppress_score >= support_score + 2:
structural_cap = any(
signal.get("direction") == "suppress"
and signal.get("label") in {"边界层结构", "站网对比", "日内节奏"}
for signal in signals
)
if taf_has_cloud_rain_cap and structural_cap:
headline = "峰值同时存在 TAF 云雨扰动和结构压制,当前更偏防守高温上修。"
headline_en = "Both TAF cloud/rain disruption and structural signals are capping the peak; defend against aggressive high-temperature upside for now."
elif taf_has_cloud_rain_cap:
headline = "TAF 提示峰值窗口有云雨扰动,当前更偏防守高温上修。"
headline_en = "TAF flags cloud/rain disruption near the peak window; defend against aggressive high-temperature upside for now."
else:
headline = "峰值主要受结构信号压制,TAF 云雨层暂未构成主压温理由。"
headline_en = "The peak is mainly capped by structural signals; TAF cloud/rain is not the primary suppression reason for now."
confidence = "high" if available_layers >= 3 else "medium"
elif support_score >= suppress_score + 2:
headline = "峰值仍有上修空间,后续重点看峰值窗口内报文能否继续抬升。"
headline_en = "The peak still has upside room; the next check is whether reports keep lifting through the peak window."
confidence = "high" if available_layers >= 3 else "medium"
elif available_layers == 0:
headline = "关键日内层仍在补齐,先以观测锚点和下一次报文为主。"
headline_en = "Key intraday layers are still filling in; anchor the read on observations and the next report."
confidence = "low"
else:
headline = "当前处于分歧判断区,峰值窗口内的下一组观测将决定方向。"
headline_en = "The setup is in a split-decision zone; the next observations inside the peak window should decide direction."
confidence = "medium" if available_layers >= 2 else "low"
next_observation = _next_observation_clock(data.get("local_time") or current.get("obs_time"))
threshold = base_value
invalidation_rules = []
invalidation_rules_en = []
confirmation_rules = []
confirmation_rules_en = []
if peak_status == "past":
invalidation_rules.append("若后续官方结算源补录更高值,以结算源最终高点为准。")
invalidation_rules_en.append("If the official settlement source later backfills a higher reading, defer to the final settlement-source high.")
confirmation_rules.append("若峰值窗口后连续两次观测不再创新高,当前高点基本确认。")
confirmation_rules_en.append("If two consecutive post-peak observations fail to make a new high, the current high is broadly confirmed.")
else:
watch_clock = _format_clock_minutes(int(first_h or 13) * 60 + 30)
if threshold is not None:
invalidation_rules.append(f"{watch_clock} 前若仍未接近 {threshold:.0f}{unit},上修路径降级。")
invalidation_rules_en.append(f"If observations are still not near {threshold:.0f}{unit} before {watch_clock}, downgrade the upside path.")
confirmation_rules.append(f"峰值窗口内任一结算源观测触达或超过 {threshold:.0f}{unit},基准路径确认度上升。")
confirmation_rules_en.append(f"If any settlement-source observation reaches or exceeds {threshold:.0f}{unit} inside the peak window, confidence in the base path rises.")
invalidation_rules.append("若 TAF 或实况报文出现阵雨、雷暴或低云/云雨压制,高温上沿需要下调。")
invalidation_rules_en.append("If TAF or live reports show showers, thunderstorms, or low-cloud/cloud-rain suppression, lower the upper temperature bound.")
confirmation_rules.append("若实测继续贴近 DEB 曲线且云雨信号不增强,维持当前主路径。")
confirmation_rules_en.append("If observations keep tracking the DEB curve and cloud/rain signals do not strengthen, maintain the current main path.")
if not signals:
_add_signal(
signals,
label="数据完整性",
label_en="Data completeness",
direction="neutral",
strength="weak",
summary="当前缺少足够的日内结构层,等待下一次观测刷新后再提高判断权重。",
summary_en="There are not enough intraday structure layers yet; wait for the next observation refresh before raising confidence.",
)
return {
"headline": headline,
"headline_en": headline_en,
"confidence": confidence,
"base_case_bucket": base_case_bucket,
"upside_bucket": upside_bucket,
"downside_bucket": downside_bucket,
"next_observation_time": next_observation,
"peak_window": peak_window,
"invalidation_rules": invalidation_rules[:4],
"invalidation_rules_en": invalidation_rules_en[:4],
"confirmation_rules": confirmation_rules[:3],
"confirmation_rules_en": confirmation_rules_en[:3],
"signal_contributions": signals[:5],
}
def _archive_intraday_path_snapshot(city: str, result: Dict[str, Any]) -> None:
+269
View File
@@ -0,0 +1,269 @@
"""Authentication guards and identity resolution for PolyWeather."""
import os
from typing import Any, Dict
from fastapi import HTTPException, Request
from loguru import logger
from src.auth.supabase_entitlement import SUPABASE_ENTITLEMENT, extract_bearer_token
def _env_bool(name: str, default: bool = False) -> bool:
raw = os.getenv(name)
if raw is None:
return default
return raw.strip().lower() in {"1", "true", "yes", "on"}
_ENTITLEMENT_HEADER = "x-polyweather-entitlement"
_FORWARDED_SUPABASE_USER_ID_HEADER = "x-polyweather-auth-user-id"
_FORWARDED_SUPABASE_EMAIL_HEADER = "x-polyweather-auth-email"
_OPS_ADMIN_EMAILS = {
item.strip().lower()
for item in str(os.getenv("POLYWEATHER_OPS_ADMIN_EMAILS") or "").split(",")
if item.strip()
}
# Config values imported lazily from web.core to avoid circular imports
# and allow monkeypatching to target web.core directly.
def _get_entitlement_token():
import web.core as _core
return _core._ENTITLEMENT_TOKEN
def _get_supabase_auth_required():
import web.core as _core
return _core._SUPABASE_AUTH_REQUIRED
def _get_entitlement_guard_enabled():
import web.core as _core
return _core._ENTITLEMENT_GUARD_ENABLED
def _legacy_service_token_valid(request: Request) -> bool:
token = request.headers.get(_ENTITLEMENT_HEADER)
if not token:
token = extract_bearer_token(request.headers.get("authorization"))
token_hint = _get_entitlement_token()
return bool(token_hint and token == token_hint)
def _bind_forwarded_supabase_identity(request: Request) -> bool:
if not _legacy_service_token_valid(request):
return False
forwarded_user_id = str(
request.headers.get(_FORWARDED_SUPABASE_USER_ID_HEADER) or ""
).strip()
if not forwarded_user_id:
return False
request.state.auth_user_id = forwarded_user_id
request.state.auth_email = str(
request.headers.get(_FORWARDED_SUPABASE_EMAIL_HEADER) or ""
).strip()
return True
def _bind_optional_supabase_identity(request: Request) -> None:
if _bind_forwarded_supabase_identity(request):
return
if not SUPABASE_ENTITLEMENT.configured:
return
access_token = extract_bearer_token(request.headers.get("authorization"))
if not access_token:
return
identity = SUPABASE_ENTITLEMENT.get_identity(access_token)
if not identity:
return
request.state.auth_user_id = identity.user_id
request.state.auth_email = identity.email
request.state.auth_points = identity.points
request.state.auth_created_at = identity.created_at
from src.utils.online_tracker import record_activity
record_activity(identity.user_id)
def _resolve_auth_points(request: Request, account_db=None) -> int:
from src.database.db_manager import DBManager
if account_db is None:
# imported lazily to avoid circular dependency at module level
from web.core import _account_db as _db
account_db = _db
raw_points = getattr(request.state, "auth_points", 0)
try:
points = max(0, int(raw_points or 0))
except Exception:
points = 0
user_id = str(getattr(request.state, "auth_user_id", "") or "").strip()
if user_id:
try:
db_points = account_db.get_points_by_supabase_user_id(user_id)
if db_points > points:
request.state.auth_points = db_points
points = db_points
except Exception as exc:
logger.warning(f"auth points fallback failed user_id={user_id}: {exc}")
if points <= 0:
email = str(getattr(request.state, "auth_email", "") or "").strip().lower()
if email:
try:
email_points = account_db.get_points_by_supabase_email(email)
if email_points > points:
request.state.auth_points = email_points
points = email_points
except Exception as exc:
logger.warning(
f"auth points email fallback failed email={email}: {exc}"
)
return points
def _resolve_weekly_profile(request: Request, account_db=None) -> Dict[str, Any]:
from src.database.db_manager import DBManager
if account_db is None:
from web.core import _account_db as _db
account_db = _db
user_id = str(getattr(request.state, "auth_user_id", "") or "").strip()
if not user_id:
return {"weekly_points": 0, "weekly_rank": None}
try:
profile = account_db.get_weekly_profile_by_supabase_user_id(user_id)
return {
"weekly_points": int(profile.get("weekly_points") or 0),
"weekly_rank": profile.get("weekly_rank"),
}
except Exception as exc:
logger.warning(f"auth weekly profile fallback failed user_id={user_id}: {exc}")
return {"weekly_points": 0, "weekly_rank": None}
def _assert_entitlement(request: Request) -> None:
if SUPABASE_ENTITLEMENT.enabled:
if _legacy_service_token_valid(request):
if _bind_forwarded_supabase_identity(request):
return
bearer_token = extract_bearer_token(request.headers.get("authorization"))
if not bearer_token or bearer_token == _get_entitlement_token():
return
if not _get_supabase_auth_required():
_bind_optional_supabase_identity(request)
return
if not SUPABASE_ENTITLEMENT.configured:
raise HTTPException(
status_code=503,
detail="Supabase auth is enabled but SUPABASE_URL / SUPABASE_ANON_KEY is not configured",
)
access_token = extract_bearer_token(request.headers.get("authorization"))
if not access_token:
raise HTTPException(status_code=401, detail="Unauthorized")
identity = SUPABASE_ENTITLEMENT.get_identity(access_token)
if not identity:
raise HTTPException(status_code=401, detail="Unauthorized")
skip_subscription_gate = bool(
getattr(request.state, "skip_subscription_gate", False)
)
if (
not skip_subscription_gate
and not SUPABASE_ENTITLEMENT.has_active_subscription(identity.user_id)
):
raise HTTPException(status_code=403, detail="Subscription required")
request.state.auth_user_id = identity.user_id
request.state.auth_email = identity.email
request.state.auth_points = identity.points
request.state.auth_created_at = identity.created_at
from src.utils.online_tracker import record_activity
record_activity(identity.user_id)
return
if not _get_entitlement_guard_enabled():
return
if not _get_entitlement_token():
raise HTTPException(
status_code=503,
detail="Entitlement guard is enabled but backend token is not configured",
)
if not _legacy_service_token_valid(request):
raise HTTPException(status_code=401, detail="Unauthorized")
def _require_supabase_identity(request: Request) -> Dict[str, str]:
if not SUPABASE_ENTITLEMENT.enabled:
raise HTTPException(
status_code=503, detail="payment requires POLYWEATHER_AUTH_ENABLED=true"
)
if not SUPABASE_ENTITLEMENT.configured:
raise HTTPException(
status_code=503,
detail="payment requires SUPABASE_URL and SUPABASE_ANON_KEY",
)
state_user_id = str(getattr(request.state, "auth_user_id", "") or "").strip()
if state_user_id:
state_email = str(getattr(request.state, "auth_email", "") or "").strip()
return {"user_id": state_user_id, "email": state_email}
token = extract_bearer_token(request.headers.get("authorization"))
if token:
identity = SUPABASE_ENTITLEMENT.get_identity(token)
if identity:
return {"user_id": identity.user_id, "email": identity.email}
legacy_ok = _legacy_service_token_valid(request)
if legacy_ok:
forwarded_user_id = str(
request.headers.get(_FORWARDED_SUPABASE_USER_ID_HEADER) or ""
).strip()
if forwarded_user_id:
forwarded_email = str(
request.headers.get(_FORWARDED_SUPABASE_EMAIL_HEADER) or ""
).strip()
return {"user_id": forwarded_user_id, "email": forwarded_email}
return {"user_id": "entitlement", "email": ""}
logger.warning(
"payment auth identity missing state_user={} auth_bearer={} legacy_ok={} forwarded_user={}".format(
bool(state_user_id),
bool(token),
bool(legacy_ok),
bool(
str(
request.headers.get(_FORWARDED_SUPABASE_USER_ID_HEADER) or ""
).strip()
),
)
)
raise HTTPException(status_code=401, detail="Unauthorized")
def _require_ops_admin(request: Request) -> Dict[str, str]:
identity = _require_supabase_identity(request)
email = str(identity.get("email") or "").strip().lower()
if email and email in _OPS_ADMIN_EMAILS:
return identity
user_id = identity.get("user_id")
if user_id and user_id.lower() == "entitlement":
raise HTTPException(
status_code=401,
detail="entitlement bearer is not suitable for ops admin authorization",
)
raise HTTPException(
status_code=403,
detail="ops access restricted to configured admin emails",
)
+97 -718
View File
@@ -1,38 +1,28 @@
"""
PolyWeather Web Core Context
Holds module-level singletons (app, config, weather collector, DB manager, cache)
and re-exports symbols from domain-specific modules for backward compatibility.
"""
import json
import os
import sqlite3
import threading
import time
from collections import OrderedDict
from datetime import datetime, timezone
from typing import Any, Dict, Optional
from fastapi import FastAPI, HTTPException, Request
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from loguru import logger
from src.utils.config_loader import load_config
from src.utils.config_validation import validate_runtime_env
from src.data_collection.weather_sources import WeatherDataCollector
from src.data_collection.country_networks import provider_coverage_summary
from src.data_collection.city_risk_profiles import CITY_RISK_PROFILES # noqa: F401
from src.auth.supabase_entitlement import SUPABASE_ENTITLEMENT, extract_bearer_token
from src.utils.metrics import (
build_metrics_summary,
counter_inc,
gauge_set,
histogram_observe,
)
from src.database.db_manager import DBManager
from src.database.runtime_state import get_state_storage_mode
from src.payments import PAYMENT_CHECKOUT, PaymentCheckoutError # noqa: F401
from src.data_collection.city_registry import CITY_REGISTRY
# ---------------------------------------------------------------------------
# FastAPI app singleton
# ---------------------------------------------------------------------------
app = FastAPI(title="PolyWeather Map", version="1.0")
_cors_origins = os.getenv(
@@ -47,6 +37,9 @@ app.add_middleware(
allow_headers=["*"],
)
# ---------------------------------------------------------------------------
# Core singletons (must remain module-level for backward compat)
# ---------------------------------------------------------------------------
_config = load_config()
_config_validation = validate_runtime_env("web")
for _warning in _config_validation.warnings:
@@ -79,6 +72,9 @@ SETTLEMENT_SOURCE_LABELS: Dict[str, str] = {
"wunderground": "Wunderground",
}
# ---------------------------------------------------------------------------
# LRUDict — simple size-bounded cache
# ---------------------------------------------------------------------------
class LRUDict:
"""Size-bounded ordered dict that evicts oldest entries on overflow."""
@@ -111,405 +107,112 @@ CACHE_TTL_ANKARA = 60
CACHE_TTL_KOREAN_AMOS = 60
_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# ---------------------------------------------------------------------------
# Middleware — imported from web.middleware.http
# ---------------------------------------------------------------------------
from web.middleware.http import metrics_middleware as _metrics_middleware
from web.middleware.http import etag_middleware as _etag_middleware
@app.middleware("http")
async def _metrics_middleware(request: Request, call_next):
started = time.perf_counter()
try:
response = await call_next(request)
except Exception:
duration_ms = (time.perf_counter() - started) * 1000.0
counter_inc(
"polyweather_http_requests_total",
method=request.method,
path=request.url.path,
status="500",
)
histogram_observe(
"polyweather_http_request_duration_ms",
duration_ms,
method=request.method,
path=request.url.path,
status="500",
)
raise
app.middleware("http")(_metrics_middleware)
app.middleware("http")(_etag_middleware)
duration_ms = (time.perf_counter() - started) * 1000.0
status_code = str(response.status_code)
counter_inc(
"polyweather_http_requests_total",
method=request.method,
path=request.url.path,
status=status_code,
)
histogram_observe(
"polyweather_http_request_duration_ms",
duration_ms,
method=request.method,
path=request.url.path,
status=status_code,
)
return response
# Re-export middleware for external consumers
metrics_middleware = _metrics_middleware
etag_middleware = _etag_middleware
# ---------------------------------------------------------------------------
# Schemas — re-exported from web.schemas
# ---------------------------------------------------------------------------
from web.schemas.auth import ( # noqa: E402, F401
AnalyticsEventRequest,
FeedbackRewardRequest,
GrantPointsRequest,
ReferralApplyRequest,
TelegramBindTokenRequest,
TelegramLoginRequest,
UserFeedbackRequest,
)
from web.schemas.payments import ( # noqa: E402, F401
ConfirmPaymentTxRequest,
CreatePaymentIntentRequest,
SubmitPaymentTxRequest,
ValidatePaymentTxRequest,
WalletBindRequest,
WalletChallengeRequest,
WalletUnbindRequest,
WalletVerifyRequest,
)
@app.middleware("http")
async def _etag_middleware(request: Request, call_next):
"""Add ETag to GET /api/* responses; return 304 on If-None-Match hit."""
response = await call_next(request)
# ---------------------------------------------------------------------------
# Auth guards — imported from web.auth.guards, wrappers for account_db
# ---------------------------------------------------------------------------
from src.auth.supabase_entitlement import SUPABASE_ENTITLEMENT # noqa: E402
if request.method != "GET" or response.status_code != 200:
return response
import web.auth.guards as _auth_guards # noqa: E402
path = request.url.path
if not path.startswith("/api/") or path.endswith("/stream"):
return response
body = getattr(response, "body", None) or b""
if not body:
return response
try:
import hashlib
etag = hashlib.md5(body).hexdigest()
except Exception:
return response
etag_value = f'"{etag}"'
if_none_match = request.headers.get("If-None-Match", "")
if if_none_match == etag_value:
from fastapi.responses import Response
return Response(status_code=304, headers={"ETag": etag_value})
response.headers["ETag"] = etag_value
response.headers["Cache-Control"] = "private, max-age=30"
return response
def _env_bool(name: str, default: bool = False) -> bool:
raw = os.getenv(name)
if raw is None:
return default
return raw.strip().lower() in {"1", "true", "yes", "on"}
_ENTITLEMENT_GUARD_ENABLED = _env_bool("POLYWEATHER_REQUIRE_ENTITLEMENT", False)
_ENTITLEMENT_HEADER = "x-polyweather-entitlement"
_ENTITLEMENT_TOKEN = (os.getenv("POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN") or "").strip()
_FORWARDED_SUPABASE_USER_ID_HEADER = "x-polyweather-auth-user-id"
_FORWARDED_SUPABASE_EMAIL_HEADER = "x-polyweather-auth-email"
_SUPABASE_AUTH_REQUIRED = _env_bool(
# Auth config flags — defined directly here so tests can monkeypatch web.core
_SUPABASE_AUTH_REQUIRED = _auth_guards._env_bool(
"POLYWEATHER_AUTH_REQUIRED",
SUPABASE_ENTITLEMENT.enabled,
)
_OPS_ADMIN_EMAILS = {
item.strip().lower()
for item in str(os.getenv("POLYWEATHER_OPS_ADMIN_EMAILS") or "").split(",")
if item.strip()
}
_ENTITLEMENT_GUARD_ENABLED = _auth_guards._env_bool("POLYWEATHER_REQUIRE_ENTITLEMENT", False)
_ENTITLEMENT_TOKEN = (os.getenv("POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN") or "").strip()
def _legacy_service_token_valid(request: Request) -> bool:
token = request.headers.get(_ENTITLEMENT_HEADER)
if not token:
token = extract_bearer_token(request.headers.get("authorization"))
return bool(_ENTITLEMENT_TOKEN and token == _ENTITLEMENT_TOKEN)
def _legacy_service_token_valid(request):
return _auth_guards._legacy_service_token_valid(request)
def _bind_forwarded_supabase_identity(request: Request) -> bool:
if not _legacy_service_token_valid(request):
return False
forwarded_user_id = str(
request.headers.get(_FORWARDED_SUPABASE_USER_ID_HEADER) or ""
).strip()
if not forwarded_user_id:
return False
request.state.auth_user_id = forwarded_user_id
request.state.auth_email = str(
request.headers.get(_FORWARDED_SUPABASE_EMAIL_HEADER) or ""
).strip()
return True
def _bind_forwarded_supabase_identity(request):
return _auth_guards._bind_forwarded_supabase_identity(request)
def _bind_optional_supabase_identity(request: Request) -> None:
if _bind_forwarded_supabase_identity(request):
return
if not SUPABASE_ENTITLEMENT.configured:
return
access_token = extract_bearer_token(request.headers.get("authorization"))
if not access_token:
return
identity = SUPABASE_ENTITLEMENT.get_identity(access_token)
if not identity:
return
request.state.auth_user_id = identity.user_id
request.state.auth_email = identity.email
request.state.auth_points = identity.points
request.state.auth_created_at = identity.created_at
from src.utils.online_tracker import record_activity
record_activity(identity.user_id)
def _bind_optional_supabase_identity(request):
return _auth_guards._bind_optional_supabase_identity(request)
def _resolve_auth_points(request: Request) -> int:
raw_points = getattr(request.state, "auth_points", 0)
try:
points = max(0, int(raw_points or 0))
except Exception:
points = 0
user_id = str(getattr(request.state, "auth_user_id", "") or "").strip()
if user_id:
try:
db_points = _account_db.get_points_by_supabase_user_id(user_id)
if db_points > points:
request.state.auth_points = db_points
points = db_points
except Exception as exc:
logger.warning(f"auth points fallback failed user_id={user_id}: {exc}")
if points <= 0:
email = str(getattr(request.state, "auth_email", "") or "").strip().lower()
if email:
try:
email_points = _account_db.get_points_by_supabase_email(email)
if email_points > points:
request.state.auth_points = email_points
points = email_points
except Exception as exc:
logger.warning(
f"auth points email fallback failed email={email}: {exc}"
)
return points
def _resolve_auth_points(request):
return _auth_guards._resolve_auth_points(request, account_db=_account_db)
def _resolve_weekly_profile(request: Request) -> Dict[str, Any]:
user_id = str(getattr(request.state, "auth_user_id", "") or "").strip()
if not user_id:
return {"weekly_points": 0, "weekly_rank": None}
try:
profile = _account_db.get_weekly_profile_by_supabase_user_id(user_id)
return {
"weekly_points": int(profile.get("weekly_points") or 0),
"weekly_rank": profile.get("weekly_rank"),
}
except Exception as exc:
logger.warning(f"auth weekly profile fallback failed user_id={user_id}: {exc}")
return {"weekly_points": 0, "weekly_rank": None}
def _resolve_weekly_profile(request):
return _auth_guards._resolve_weekly_profile(request, account_db=_account_db)
def _assert_entitlement(request: Request) -> None:
if SUPABASE_ENTITLEMENT.enabled:
if _legacy_service_token_valid(request):
if _bind_forwarded_supabase_identity(request):
return
bearer_token = extract_bearer_token(request.headers.get("authorization"))
if not bearer_token or bearer_token == _ENTITLEMENT_TOKEN:
return
if not _SUPABASE_AUTH_REQUIRED:
_bind_optional_supabase_identity(request)
return
if not SUPABASE_ENTITLEMENT.configured:
raise HTTPException(
status_code=503,
detail="Supabase auth is enabled but SUPABASE_URL / SUPABASE_ANON_KEY is not configured",
)
access_token = extract_bearer_token(request.headers.get("authorization"))
if not access_token:
raise HTTPException(status_code=401, detail="Unauthorized")
identity = SUPABASE_ENTITLEMENT.get_identity(access_token)
if not identity:
raise HTTPException(status_code=401, detail="Unauthorized")
skip_subscription_gate = bool(
getattr(request.state, "skip_subscription_gate", False)
)
if (
not skip_subscription_gate
and not SUPABASE_ENTITLEMENT.has_active_subscription(identity.user_id)
):
raise HTTPException(status_code=403, detail="Subscription required")
request.state.auth_user_id = identity.user_id
request.state.auth_email = identity.email
request.state.auth_points = identity.points
request.state.auth_created_at = identity.created_at
from src.utils.online_tracker import record_activity
record_activity(identity.user_id)
return
if not _ENTITLEMENT_GUARD_ENABLED:
return
if not _ENTITLEMENT_TOKEN:
raise HTTPException(
status_code=503,
detail="Entitlement guard is enabled but backend token is not configured",
)
if not _legacy_service_token_valid(request):
raise HTTPException(status_code=401, detail="Unauthorized")
def _assert_entitlement(request):
return _auth_guards._assert_entitlement(request)
def _require_supabase_identity(request: Request) -> Dict[str, str]:
if not SUPABASE_ENTITLEMENT.enabled:
raise HTTPException(
status_code=503, detail="payment requires POLYWEATHER_AUTH_ENABLED=true"
)
if not SUPABASE_ENTITLEMENT.configured:
raise HTTPException(
status_code=503,
detail="payment requires SUPABASE_URL and SUPABASE_ANON_KEY",
)
def _require_supabase_identity(request):
return _auth_guards._require_supabase_identity(request)
state_user_id = str(getattr(request.state, "auth_user_id", "") or "").strip()
if state_user_id:
state_email = str(getattr(request.state, "auth_email", "") or "").strip()
return {"user_id": state_user_id, "email": state_email}
token = extract_bearer_token(request.headers.get("authorization"))
if token:
identity = SUPABASE_ENTITLEMENT.get_identity(token)
if identity:
return {"user_id": identity.user_id, "email": identity.email}
def _require_ops_admin(request):
return _auth_guards._require_ops_admin(request)
legacy_ok = _legacy_service_token_valid(request)
if legacy_ok:
forwarded_user_id = str(
request.headers.get(_FORWARDED_SUPABASE_USER_ID_HEADER) or ""
).strip()
if forwarded_user_id:
forwarded_email = str(
request.headers.get(_FORWARDED_SUPABASE_EMAIL_HEADER) or ""
).strip()
return {"user_id": forwarded_user_id, "email": forwarded_email}
# Entitlement token is valid but forwarded headers are missing.
# Return a placeholder identity — callers (e.g. _require_ops_admin)
# can decide whether to accept it.
return {"user_id": "entitlement", "email": ""}
logger.warning(
"payment auth identity missing state_user={} auth_bearer={} legacy_ok={} forwarded_user={}".format(
bool(state_user_id),
bool(token),
bool(legacy_ok),
bool(
str(
request.headers.get(_FORWARDED_SUPABASE_USER_ID_HEADER) or ""
).strip()
),
)
# ---------------------------------------------------------------------------
# Diagnostics — re-exported from web.diagnostics.health
# ---------------------------------------------------------------------------
from web.diagnostics.health import ( # noqa: E402
build_health_payload as _build_health_payload_raw,
build_system_status_payload as _build_system_status_payload_raw,
)
def build_health_payload():
return _build_health_payload_raw(_account_db, len(CITIES))
def build_system_status_payload():
return _build_system_status_payload_raw(
_account_db, _config, _weather, _cache, len(_cache), len(CITIES), CITY_REGISTRY
)
raise HTTPException(status_code=401, detail="Unauthorized")
def _require_ops_admin(request: Request) -> Dict[str, str]:
is_entitlement = _legacy_service_token_valid(request)
identity = _require_supabase_identity(request)
if not _OPS_ADMIN_EMAILS:
raise HTTPException(
status_code=503,
detail="ops admin is not configured; set POLYWEATHER_OPS_ADMIN_EMAILS",
)
email = str(identity.get("email") or "").strip().lower()
if email and email in _OPS_ADMIN_EMAILS:
return identity
# If identity lacks an email (e.g. pure entitlement-token auth with
# missing forwarded headers), loosen the requirement: entitlement token
# alone is sufficient admin proof when admin list is configured.
if is_entitlement:
granted = {
"user_id": "admin:entitlement",
"email": next(iter(_OPS_ADMIN_EMAILS)),
}
return granted
raise HTTPException(status_code=403, detail="ops admin required")
class WalletChallengeRequest(BaseModel):
address: str = Field(..., min_length=8)
class WalletVerifyRequest(BaseModel):
address: str = Field(..., min_length=8)
nonce: str = Field(..., min_length=6)
signature: str = Field(..., min_length=20)
class WalletUnbindRequest(BaseModel):
address: str = Field(..., min_length=8)
class CreatePaymentIntentRequest(BaseModel):
plan_code: str = Field(default="pro_monthly", min_length=2)
payment_mode: str = Field(default="strict")
allowed_wallet: Optional[str] = None
token_address: Optional[str] = None
chain_id: Optional[int] = None
use_points: bool = False
points_to_consume: Optional[int] = None
metadata: Dict[str, Any] = Field(default_factory=dict)
class SubmitPaymentTxRequest(BaseModel):
tx_hash: str = Field(..., min_length=10)
from_address: Optional[str] = None
class ValidatePaymentTxRequest(BaseModel):
tx_hash: str = Field(..., min_length=10)
class ConfirmPaymentTxRequest(BaseModel):
tx_hash: Optional[str] = None
class ReferralApplyRequest(BaseModel):
code: str = Field(..., min_length=3, max_length=32)
class TelegramLoginRequest(BaseModel):
id: int
first_name: Optional[str] = None
last_name: Optional[str] = None
username: Optional[str] = None
photo_url: Optional[str] = None
auth_date: int
hash: str = Field(..., min_length=10)
class TelegramBindTokenRequest(BaseModel):
token: str = Field(..., min_length=8)
class AnalyticsEventRequest(BaseModel):
event_type: str = Field(..., min_length=3, max_length=64)
client_id: Optional[str] = Field(default=None, max_length=128)
session_id: Optional[str] = Field(default=None, max_length=128)
payload: Dict[str, Any] = Field(default_factory=dict)
class UserFeedbackRequest(BaseModel):
category: str = Field(default="bug", min_length=2, max_length=40)
message: str = Field(..., min_length=3, max_length=5000)
source: str = Field(default="terminal", max_length=40)
contact: Optional[str] = Field(default=None, max_length=180)
context: Dict[str, Any] = Field(default_factory=dict)
class GrantPointsRequest(BaseModel):
email: str = Field(..., min_length=3)
points: int = Field(..., gt=0, le=100000)
class FeedbackRewardRequest(BaseModel):
points: int = Field(..., gt=0, le=100000)
reason: str = Field(default="", max_length=500)
# ---------------------------------------------------------------------------
# Utility helpers
# ---------------------------------------------------------------------------
def _sf(v) -> Optional[float]:
if v is None:
return None
@@ -523,335 +226,11 @@ def _is_excluded_model_name(model_name: str) -> bool:
return False
def _sqlite_health() -> Dict[str, Any]:
try:
with sqlite3.connect(_account_db.db_path, timeout=0.05) as conn:
conn.execute("SELECT 1").fetchone()
return {"ok": True, "db_path": _account_db.db_path}
except Exception as exc:
return {"ok": False, "db_path": _account_db.db_path, "error": str(exc)}
def _cache_summary() -> Dict[str, Any]:
from web.analysis_service import get_analysis_cache_stats
open_meteo_forecast_entries = len(getattr(_weather, "_open_meteo_cache", {}) or {})
open_meteo_ensemble_entries = len(getattr(_weather, "_ensemble_cache", {}) or {})
open_meteo_multi_model_entries = len(
getattr(_weather, "_multi_model_cache", {}) or {}
)
metar_entries = len(getattr(_weather, "_metar_cache", {}) or {})
taf_entries = len(getattr(_weather, "_taf_cache", {}) or {})
settlement_entries = len(getattr(_weather, "_settlement_cache", {}) or {})
gauge_set("polyweather_api_cache_entries", len(_cache))
gauge_set(
"polyweather_open_meteo_forecast_cache_entries", open_meteo_forecast_entries
)
gauge_set(
"polyweather_open_meteo_ensemble_cache_entries", open_meteo_ensemble_entries
)
gauge_set(
"polyweather_open_meteo_multi_model_cache_entries",
open_meteo_multi_model_entries,
)
gauge_set("polyweather_metar_cache_entries", metar_entries)
gauge_set("polyweather_taf_cache_entries", taf_entries)
gauge_set("polyweather_settlement_cache_entries", settlement_entries)
return {
"api_cache_entries": len(_cache),
"open_meteo_forecast_entries": open_meteo_forecast_entries,
"open_meteo_ensemble_entries": open_meteo_ensemble_entries,
"open_meteo_multi_model_entries": open_meteo_multi_model_entries,
"metar_entries": metar_entries,
"taf_entries": taf_entries,
"settlement_entries": settlement_entries,
"analysis": get_analysis_cache_stats(),
}
def _feature_flags_summary() -> Dict[str, Any]:
return {
"auth_enabled": bool(SUPABASE_ENTITLEMENT.enabled),
"auth_required": bool(_SUPABASE_AUTH_REQUIRED),
"entitlement_guard_enabled": bool(_ENTITLEMENT_GUARD_ENABLED),
"payment_enabled": bool(getattr(PAYMENT_CHECKOUT, "enabled", False)),
"state_storage_mode": get_state_storage_mode(),
}
def _integration_summary() -> Dict[str, Any]:
weather_cfg = _config.get("weather", {}) if isinstance(_config, dict) else {}
return {
"supabase_configured": bool(SUPABASE_ENTITLEMENT.configured),
"telegram_bot_configured": bool(
(_config.get("telegram", {}) or {}).get("bot_token")
),
"walletconnect_configured": bool(
os.getenv("NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID")
),
"weather_sources": {
"openweather": bool(weather_cfg.get("openweather_api_key")),
"wunderground": bool(weather_cfg.get("wunderground_api_key")),
"visualcrossing": bool(weather_cfg.get("visualcrossing_api_key")),
},
}
def _probability_summary() -> Dict[str, Any]:
return {
"engine_mode": "legacy",
}
def _read_json_file(path: str) -> Optional[Dict[str, Any]]:
if not os.path.exists(path):
return None
try:
with open(path, "r", encoding="utf-8") as fh:
payload = json.load(fh)
if isinstance(payload, dict):
return payload
except Exception:
return None
return None
def _table_date_summary(conn: sqlite3.Connection, table_name: str) -> Dict[str, Any]:
try:
row = conn.execute(
f"""
SELECT COUNT(*) AS row_count,
COUNT(DISTINCT city) AS cities_count,
MIN(target_date) AS min_date,
MAX(target_date) AS max_date
FROM {table_name}
"""
).fetchone()
except Exception as exc:
return {"ok": False, "error": str(exc), "row_count": 0, "cities_count": 0}
return {
"ok": True,
"row_count": int(row["row_count"] or 0),
"cities_count": int(row["cities_count"] or 0),
"min_date": row["min_date"],
"max_date": row["max_date"],
}
def _truth_source_counts(conn: sqlite3.Connection) -> Dict[str, int]:
try:
rows = conn.execute(
"""
SELECT COALESCE(NULLIF(TRIM(settlement_source), ''), 'unknown') AS settlement_source,
COUNT(*) AS row_count
FROM truth_records_store
GROUP BY COALESCE(NULLIF(TRIM(settlement_source), ''), 'unknown')
ORDER BY row_count DESC, settlement_source ASC
"""
).fetchall()
except Exception:
return {}
return {str(row["settlement_source"]): int(row["row_count"] or 0) for row in rows}
def _truth_revisions_summary(conn: sqlite3.Connection) -> Dict[str, Any]:
try:
row = conn.execute(
"""
SELECT COUNT(*) AS row_count,
MAX(updated_at) AS last_updated_at
FROM truth_revisions_store
"""
).fetchone()
except Exception as exc:
return {"ok": False, "error": str(exc), "row_count": 0}
return {
"ok": True,
"row_count": int(row["row_count"] or 0),
"last_updated_at": row["last_updated_at"],
}
def _city_coverage_summary(conn: sqlite3.Connection) -> Dict[str, Any]:
truth_rows = conn.execute(
"""
SELECT city, COUNT(*) AS row_count, MIN(target_date) AS min_date, MAX(target_date) AS max_date
FROM truth_records_store
GROUP BY city
"""
).fetchall()
feature_rows = conn.execute(
"""
SELECT city, COUNT(*) AS row_count, MIN(target_date) AS min_date, MAX(target_date) AS max_date
FROM training_feature_records_store
GROUP BY city
"""
).fetchall()
truth_index = {
str(row["city"]): {
"truth_rows": int(row["row_count"] or 0),
"truth_min_date": row["min_date"],
"truth_max_date": row["max_date"],
}
for row in truth_rows
}
feature_index = {
str(row["city"]): {
"feature_rows": int(row["row_count"] or 0),
"feature_min_date": row["min_date"],
"feature_max_date": row["max_date"],
}
for row in feature_rows
}
entries = []
for city, meta in CITY_REGISTRY.items():
truth_payload = truth_index.get(city, {})
feature_payload = feature_index.get(city, {})
entries.append(
{
"city": city,
"name": str(meta.get("name") or city),
"settlement_source": str(meta.get("settlement_source") or "metar"),
"settlement_station_code": str(
meta.get("settlement_station_code") or meta.get("icao") or ""
),
"truth_rows": int(truth_payload.get("truth_rows") or 0),
"feature_rows": int(feature_payload.get("feature_rows") or 0),
"truth_min_date": truth_payload.get("truth_min_date"),
"truth_max_date": truth_payload.get("truth_max_date"),
"feature_min_date": feature_payload.get("feature_min_date"),
"feature_max_date": feature_payload.get("feature_max_date"),
}
)
highlighted = [
entry for entry in entries if entry["city"] in {"taipei", "shenzhen"}
]
gaps = sorted(
entries,
key=lambda entry: (
entry["feature_rows"] > 0,
entry["truth_rows"] > 0,
entry["truth_rows"],
entry["feature_rows"],
entry["city"],
),
)[:10]
return {
"total_cities": len(entries),
"with_truth_rows": sum(1 for entry in entries if entry["truth_rows"] > 0),
"with_feature_rows": sum(1 for entry in entries if entry["feature_rows"] > 0),
"entries": entries,
"highlighted": highlighted,
"top_gaps": gaps,
}
def _model_city_coverage_summary(
city_entries: Any,
) -> Dict[str, Any]:
rows = []
for entry in city_entries or []:
city = str(entry.get("city") or "").strip().lower()
rows.append(
{
"city": city,
"name": entry.get("name") or city,
"settlement_source": entry.get("settlement_source"),
"truth_rows": int(entry.get("truth_rows") or 0),
"feature_rows": int(entry.get("feature_rows") or 0),
}
)
weakest = sorted(
rows,
key=lambda row: (
row["truth_rows"] > 0,
row["truth_rows"],
row["city"],
),
)[:12]
strongest = sorted(
rows,
key=lambda row: (
-row["truth_rows"],
row["city"],
),
)[:8]
return {
"weakest": weakest,
"strongest": strongest,
}
def _training_data_summary() -> Dict[str, Any]:
db_path = _account_db.db_path
truth_records = {"ok": False, "row_count": 0, "cities_count": 0}
truth_revisions = {"ok": False, "row_count": 0}
training_features = {"ok": False, "row_count": 0, "cities_count": 0}
try:
with sqlite3.connect(db_path) as conn:
conn.row_factory = sqlite3.Row
truth_records = _table_date_summary(conn, "truth_records_store")
if truth_records.get("ok"):
truth_records["source_counts"] = _truth_source_counts(conn)
truth_revisions = _truth_revisions_summary(conn)
training_features = _table_date_summary(
conn, "training_feature_records_store"
)
city_coverage = _city_coverage_summary(conn)
except Exception as exc:
return {
"db_path": db_path,
"db_ok": False,
"error": str(exc),
"truth_records": truth_records,
"truth_revisions": truth_revisions,
"training_features": training_features,
"city_coverage": {},
"model_city_coverage": {},
}
return {
"db_path": db_path,
"db_ok": True,
"truth_records": truth_records,
"truth_revisions": truth_revisions,
"training_features": training_features,
"city_coverage": city_coverage,
"model_city_coverage": _model_city_coverage_summary(
city_coverage.get("entries") or [],
),
}
def build_health_payload() -> Dict[str, Any]:
db = _sqlite_health()
return {
"status": "ok" if db.get("ok") else "degraded",
"time_utc": datetime.now(timezone.utc).isoformat(),
"db": db,
"state_storage_mode": get_state_storage_mode(),
"cities_count": len(CITIES),
}
def build_system_status_payload() -> Dict[str, Any]:
return {
"status": build_health_payload()["status"],
"time_utc": datetime.now(timezone.utc).isoformat(),
"state_storage_mode": get_state_storage_mode(),
"db": _sqlite_health(),
"features": _feature_flags_summary(),
"integrations": _integration_summary(),
"cache": _cache_summary(),
"metrics": build_metrics_summary(),
"probability": _probability_summary(),
"training_data": _training_data_summary(),
"station_networks": provider_coverage_summary(),
"cities_count": len(CITIES),
}
# ---------------------------------------------------------------------------
# Compatibility re-exports — symbols that were accessible via web.core
# in the old monolithic module, re-exported for backward compatibility.
# ---------------------------------------------------------------------------
from src.data_collection.city_registry import CITY_REGISTRY # noqa: E402, F401
from src.data_collection.city_risk_profiles import CITY_RISK_PROFILES # noqa: E402, F401
from src.payments import PAYMENT_CHECKOUT, PaymentCheckoutError # noqa: E402, F401
from src.auth.supabase_entitlement import SUPABASE_ENTITLEMENT # noqa: E402, F401
+371
View File
@@ -0,0 +1,371 @@
"""Health check and system status diagnostics for PolyWeather."""
import json
import os
from datetime import datetime, timezone
from typing import Any, Dict, Optional
from src.data_collection.country_networks import provider_coverage_summary
from src.utils.metrics import build_metrics_summary, gauge_set
def _env_bool(name: str, default: bool = False) -> bool:
raw = os.getenv(name)
if raw is None:
return default
return raw.strip().lower() in {"1", "true", "yes", "on"}
def _sqlite_health(account_db) -> Dict[str, Any]:
try:
from src.database.sqlite_connection import connect_sqlite
with connect_sqlite(account_db.db_path, timeout=0.05) as conn:
conn.execute("SELECT 1").fetchone()
return {"ok": True, "db_path": account_db.db_path}
except Exception as exc:
return {"ok": False, "db_path": account_db.db_path, "error": str(exc)}
def _cache_summary(weather_collector, analysis_cache, cache_entries: int) -> Dict[str, Any]:
from web.analysis_service import get_analysis_cache_stats
open_meteo_forecast_entries = len(getattr(weather_collector, "_open_meteo_cache", {}) or {})
open_meteo_ensemble_entries = len(getattr(weather_collector, "_ensemble_cache", {}) or {})
open_meteo_multi_model_entries = len(
getattr(weather_collector, "_multi_model_cache", {}) or {}
)
metar_entries = weather_collector.cache.store_size("metar") if hasattr(weather_collector, "cache") else 0
taf_entries = len(getattr(weather_collector, "_taf_cache", {}) or {})
settlement_entries = len(getattr(weather_collector, "_settlement_cache", {}) or {})
gauge_set("polyweather_api_cache_entries", cache_entries)
gauge_set("polyweather_open_meteo_forecast_cache_entries", open_meteo_forecast_entries)
gauge_set("polyweather_open_meteo_ensemble_cache_entries", open_meteo_ensemble_entries)
gauge_set(
"polyweather_open_meteo_multi_model_cache_entries",
open_meteo_multi_model_entries,
)
gauge_set("polyweather_metar_cache_entries", metar_entries)
gauge_set("polyweather_taf_cache_entries", taf_entries)
gauge_set("polyweather_settlement_cache_entries", settlement_entries)
return {
"api_cache_entries": cache_entries,
"open_meteo_forecast_entries": open_meteo_forecast_entries,
"open_meteo_ensemble_entries": open_meteo_ensemble_entries,
"open_meteo_multi_model_entries": open_meteo_multi_model_entries,
"metar_entries": metar_entries,
"taf_entries": taf_entries,
"settlement_entries": settlement_entries,
"analysis": get_analysis_cache_stats(),
}
def _feature_flags_summary(config: dict) -> Dict[str, Any]:
from src.auth.supabase_entitlement import SUPABASE_ENTITLEMENT
from src.payments import PAYMENT_CHECKOUT
from src.database.runtime_state import get_state_storage_mode
_SUPABASE_AUTH_REQUIRED = _env_bool(
"POLYWEATHER_AUTH_REQUIRED",
SUPABASE_ENTITLEMENT.enabled,
)
_ENTITLEMENT_GUARD_ENABLED = _env_bool("POLYWEATHER_REQUIRE_ENTITLEMENT", False)
return {
"auth_enabled": bool(SUPABASE_ENTITLEMENT.enabled),
"auth_required": bool(_SUPABASE_AUTH_REQUIRED),
"entitlement_guard_enabled": bool(_ENTITLEMENT_GUARD_ENABLED),
"payment_enabled": bool(getattr(PAYMENT_CHECKOUT, "enabled", False)),
"state_storage_mode": get_state_storage_mode(),
}
def _integration_summary(config: dict) -> Dict[str, Any]:
from src.auth.supabase_entitlement import SUPABASE_ENTITLEMENT
weather_cfg = config.get("weather", {}) if isinstance(config, dict) else {}
return {
"supabase_configured": bool(SUPABASE_ENTITLEMENT.configured),
"telegram_bot_configured": bool(
(config.get("telegram", {}) or {}).get("bot_token")
),
"walletconnect_configured": bool(
os.getenv("NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID")
),
"weather_sources": {
"openweather": bool(weather_cfg.get("openweather_api_key")),
"wunderground": bool(weather_cfg.get("wunderground_api_key")),
"visualcrossing": bool(weather_cfg.get("visualcrossing_api_key")),
},
}
def _probability_summary() -> Dict[str, Any]:
return {
"engine_mode": "legacy",
}
def _read_json_file(path: str) -> Optional[Dict[str, Any]]:
if not os.path.exists(path):
return None
try:
with open(path, "r", encoding="utf-8") as fh:
payload = json.load(fh)
if isinstance(payload, dict):
return payload
except Exception:
return None
return None
def _table_date_summary(conn, table_name: str) -> Dict[str, Any]:
try:
row = conn.execute(
f"""
SELECT COUNT(*) AS row_count,
COUNT(DISTINCT city) AS cities_count,
MIN(target_date) AS min_date,
MAX(target_date) AS max_date
FROM {table_name}
"""
).fetchone()
except Exception as exc:
return {"ok": False, "error": str(exc), "row_count": 0, "cities_count": 0}
return {
"ok": True,
"row_count": int(row["row_count"] or 0),
"cities_count": int(row["cities_count"] or 0),
"min_date": row["min_date"],
"max_date": row["max_date"],
}
def _truth_source_counts(conn) -> Dict[str, int]:
try:
rows = conn.execute(
"""
SELECT COALESCE(NULLIF(TRIM(settlement_source), ''), 'unknown') AS settlement_source,
COUNT(*) AS row_count
FROM truth_records_store
GROUP BY COALESCE(NULLIF(TRIM(settlement_source), ''), 'unknown')
ORDER BY row_count DESC, settlement_source ASC
"""
).fetchall()
except Exception:
return {}
return {str(row["settlement_source"]): int(row["row_count"] or 0) for row in rows}
def _truth_revisions_summary(conn) -> Dict[str, Any]:
try:
row = conn.execute(
"""
SELECT COUNT(*) AS row_count,
MAX(updated_at) AS last_updated_at
FROM truth_revisions_store
"""
).fetchone()
except Exception as exc:
return {"ok": False, "error": str(exc), "row_count": 0}
return {
"ok": True,
"row_count": int(row["row_count"] or 0),
"last_updated_at": row["last_updated_at"],
}
def _city_coverage_summary(conn, city_registry) -> Dict[str, Any]:
truth_rows = conn.execute(
"""
SELECT city, COUNT(*) AS row_count, MIN(target_date) AS min_date, MAX(target_date) AS max_date
FROM truth_records_store
GROUP BY city
"""
).fetchall()
feature_rows = conn.execute(
"""
SELECT city, COUNT(*) AS row_count, MIN(target_date) AS min_date, MAX(target_date) AS max_date
FROM training_feature_records_store
GROUP BY city
"""
).fetchall()
truth_index = {
str(row["city"]): {
"truth_rows": int(row["row_count"] or 0),
"truth_min_date": row["min_date"],
"truth_max_date": row["max_date"],
}
for row in truth_rows
}
feature_index = {
str(row["city"]): {
"feature_rows": int(row["row_count"] or 0),
"feature_min_date": row["min_date"],
"feature_max_date": row["max_date"],
}
for row in feature_rows
}
entries = []
for city, meta in city_registry.items():
truth_payload = truth_index.get(city, {})
feature_payload = feature_index.get(city, {})
entries.append(
{
"city": city,
"name": str(meta.get("name") or city),
"settlement_source": str(meta.get("settlement_source") or "metar"),
"settlement_station_code": str(
meta.get("settlement_station_code") or meta.get("icao") or ""
),
"truth_rows": int(truth_payload.get("truth_rows") or 0),
"feature_rows": int(feature_payload.get("feature_rows") or 0),
"truth_min_date": truth_payload.get("truth_min_date"),
"truth_max_date": truth_payload.get("truth_max_date"),
"feature_min_date": feature_payload.get("feature_min_date"),
"feature_max_date": feature_payload.get("feature_max_date"),
}
)
highlighted = [
entry for entry in entries if entry["city"] in {"taipei", "shenzhen"}
]
gaps = sorted(
entries,
key=lambda entry: (
entry["feature_rows"] > 0,
entry["truth_rows"] > 0,
entry["truth_rows"],
entry["feature_rows"],
entry["city"],
),
)[:10]
return {
"total_cities": len(entries),
"with_truth_rows": sum(1 for entry in entries if entry["truth_rows"] > 0),
"with_feature_rows": sum(1 for entry in entries if entry["feature_rows"] > 0),
"entries": entries,
"highlighted": highlighted,
"top_gaps": gaps,
}
def _model_city_coverage_summary(city_entries) -> Dict[str, Any]:
rows = []
for entry in city_entries or []:
city = str(entry.get("city") or "").strip().lower()
rows.append(
{
"city": city,
"name": entry.get("name") or city,
"settlement_source": entry.get("settlement_source"),
"truth_rows": int(entry.get("truth_rows") or 0),
"feature_rows": int(entry.get("feature_rows") or 0),
}
)
weakest = sorted(
rows,
key=lambda row: (
row["truth_rows"] > 0,
row["truth_rows"],
row["city"],
),
)[:12]
strongest = sorted(
rows,
key=lambda row: (
-row["truth_rows"],
row["city"],
),
)[:8]
return {
"weakest": weakest,
"strongest": strongest,
}
def _training_data_summary(account_db, city_registry) -> Dict[str, Any]:
from src.database.sqlite_connection import connect_sqlite
import sqlite3
db_path = account_db.db_path
truth_records = {"ok": False, "row_count": 0, "cities_count": 0}
truth_revisions = {"ok": False, "row_count": 0}
training_features = {"ok": False, "row_count": 0, "cities_count": 0}
try:
with connect_sqlite(db_path, row_factory=sqlite3.Row) as conn:
truth_records = _table_date_summary(conn, "truth_records_store")
if truth_records.get("ok"):
truth_records["source_counts"] = _truth_source_counts(conn)
truth_revisions = _truth_revisions_summary(conn)
training_features = _table_date_summary(
conn, "training_feature_records_store"
)
city_coverage = _city_coverage_summary(conn, city_registry)
except Exception as exc:
return {
"db_path": db_path,
"db_ok": False,
"error": str(exc),
"truth_records": truth_records,
"truth_revisions": truth_revisions,
"training_features": training_features,
"city_coverage": {},
"model_city_coverage": {},
}
return {
"db_path": db_path,
"db_ok": True,
"truth_records": truth_records,
"truth_revisions": truth_revisions,
"training_features": training_features,
"city_coverage": city_coverage,
"model_city_coverage": _model_city_coverage_summary(
city_coverage.get("entries") or [],
),
}
def build_health_payload(account_db, cities_count: int) -> Dict[str, Any]:
from src.database.runtime_state import get_state_storage_mode
db = _sqlite_health(account_db)
return {
"status": "ok" if db.get("ok") else "degraded",
"time_utc": datetime.now(timezone.utc).isoformat(),
"db": db,
"state_storage_mode": get_state_storage_mode(),
"cities_count": cities_count,
}
def build_system_status_payload(
account_db,
config: dict,
weather_collector,
analysis_cache,
cache_entries: int,
cities_count: int,
city_registry,
) -> Dict[str, Any]:
from src.database.runtime_state import get_state_storage_mode
return {
"status": build_health_payload(account_db, cities_count)["status"],
"time_utc": datetime.now(timezone.utc).isoformat(),
"state_storage_mode": get_state_storage_mode(),
"db": _sqlite_health(account_db),
"features": _feature_flags_summary(config),
"integrations": _integration_summary(config),
"cache": _cache_summary(weather_collector, analysis_cache, cache_entries),
"metrics": build_metrics_summary(),
"probability": _probability_summary(),
"training_data": _training_data_summary(account_db, city_registry),
"station_networks": provider_coverage_summary(),
"cities_count": cities_count,
}
+78
View File
@@ -0,0 +1,78 @@
"""HTTP middlewares for the PolyWeather FastAPI application."""
import hashlib
import time
from fastapi import Request
from fastapi.responses import Response
from src.utils.metrics import counter_inc, histogram_observe
async def metrics_middleware(request: Request, call_next):
started = time.perf_counter()
try:
response = await call_next(request)
except Exception:
duration_ms = (time.perf_counter() - started) * 1000.0
counter_inc(
"polyweather_http_requests_total",
method=request.method,
path=request.url.path,
status="500",
)
histogram_observe(
"polyweather_http_request_duration_ms",
duration_ms,
method=request.method,
path=request.url.path,
status="500",
)
raise
duration_ms = (time.perf_counter() - started) * 1000.0
status_code = str(response.status_code)
counter_inc(
"polyweather_http_requests_total",
method=request.method,
path=request.url.path,
status=status_code,
)
histogram_observe(
"polyweather_http_request_duration_ms",
duration_ms,
method=request.method,
path=request.url.path,
status=status_code,
)
return response
async def etag_middleware(request: Request, call_next):
"""Add ETag to GET /api/* responses; return 304 on If-None-Match hit."""
response = await call_next(request)
if request.method != "GET" or response.status_code != 200:
return response
path = request.url.path
if not path.startswith("/api/") or path.endswith("/stream"):
return response
body = getattr(response, "body", None) or b""
if not body:
return response
try:
etag = hashlib.md5(body).hexdigest()
except Exception:
return response
etag_value = f'"{etag}"'
if_none_match = request.headers.get("If-None-Match", "")
if if_none_match == etag_value:
return Response(status_code=304, headers={"ETag": etag_value})
response.headers["ETag"] = etag_value
response.headers["Cache-Control"] = "private, max-age=30"
return response
+2 -4
View File
@@ -9,6 +9,7 @@ from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Optional, Set
from src.database.db_manager import DBManager
from src.database.sqlite_connection import connect_sqlite
from web.realtime_patch_schema import EVENT_TYPE
@@ -72,10 +73,7 @@ class RealtimeEventStore:
self._ensure_table()
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path, timeout=10)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=5000")
return conn
return connect_sqlite(self.db_path, timeout=10)
def _ensure_table(self) -> None:
db_dir = os.path.dirname(self.db_path)
+48
View File
@@ -0,0 +1,48 @@
"""Pydantic request models for PolyWeather auth-related endpoints."""
from typing import Optional
from pydantic import BaseModel, Field
class TelegramLoginRequest(BaseModel):
id: int
first_name: Optional[str] = None
last_name: Optional[str] = None
username: Optional[str] = None
photo_url: Optional[str] = None
auth_date: int
hash: str = Field(..., min_length=10)
class TelegramBindTokenRequest(BaseModel):
token: str = Field(..., min_length=8)
class ReferralApplyRequest(BaseModel):
code: str = Field(..., min_length=3, max_length=32)
class AnalyticsEventRequest(BaseModel):
event_type: str = Field(..., min_length=3, max_length=64)
client_id: Optional[str] = Field(default=None, max_length=128)
session_id: Optional[str] = Field(default=None, max_length=128)
payload: dict = Field(default_factory=dict)
class UserFeedbackRequest(BaseModel):
category: str = Field(default="bug", min_length=2, max_length=40)
message: str = Field(..., min_length=3, max_length=5000)
source: str = Field(default="terminal", max_length=40)
contact: Optional[str] = Field(default=None, max_length=180)
context: dict = Field(default_factory=dict)
class GrantPointsRequest(BaseModel):
email: str = Field(..., min_length=3)
points: int = Field(..., gt=0, le=100000)
class FeedbackRewardRequest(BaseModel):
points: int = Field(..., gt=0, le=100000)
reason: str = Field(default="", max_length=500)
+47
View File
@@ -0,0 +1,47 @@
"""Pydantic request models for PolyWeather payment endpoints."""
from typing import Any, Dict, Optional
from pydantic import BaseModel, Field
class WalletBindRequest(BaseModel):
address: str = Field(..., min_length=8)
class WalletChallengeRequest(BaseModel):
address: str = Field(..., min_length=8)
class WalletVerifyRequest(BaseModel):
address: str = Field(..., min_length=8)
nonce: str = Field(..., min_length=6)
signature: str = Field(..., min_length=20)
class WalletUnbindRequest(BaseModel):
address: str = Field(..., min_length=8)
class CreatePaymentIntentRequest(BaseModel):
plan_code: str = Field(default="pro_monthly", min_length=2)
payment_mode: str = Field(default="strict")
allowed_wallet: Optional[str] = None
token_address: Optional[str] = None
chain_id: Optional[int] = None
use_points: bool = False
points_to_consume: Optional[int] = None
metadata: Dict[str, Any] = Field(default_factory=dict)
class SubmitPaymentTxRequest(BaseModel):
tx_hash: str = Field(..., min_length=10)
from_address: Optional[str] = None
class ValidatePaymentTxRequest(BaseModel):
tx_hash: str = Field(..., min_length=10)
class ConfirmPaymentTxRequest(BaseModel):
tx_hash: Optional[str] = None
+311
View File
@@ -0,0 +1,311 @@
"""Intraday meteorology — paid-product signal layer.
Reads existing analysis layers (current, probabilities, DEB, peak, deviation,
TAF, vertical profile, station network) and produces a structured meteorology
read with headline, confidence, signals, and invalidation/confirmation rules.
"""
from __future__ import annotations
from typing import Any, Dict
from web.services.analysis_utils import (
add_signal as _add_signal,
bucket_label as _bucket_label,
bucket_label_from_value as _bucket_label_from_value,
format_clock_minutes as _format_clock_minutes,
next_observation_clock as _next_observation_clock,
top_probability_bucket as _top_probability_bucket,
)
def _sf(v: Any) -> Any:
if v is None:
return None
try:
return float(v)
except Exception:
return None
def build_intraday_meteorology(data: Dict[str, Any]) -> Dict[str, Any]:
"""Build a paid-product intraday meteorology read from existing layers."""
current = data.get("current") or {}
probabilities = data.get("probabilities") or {}
distribution = probabilities.get("distribution") or []
top_bucket = _top_probability_bucket(distribution)
unit = str(data.get("temp_symbol") or "°C")
deb = data.get("deb") or {}
peak = data.get("peak") or {}
deviation = data.get("deviation_monitor") or {}
taf_signal = (
((data.get("taf") or {}).get("signal") or {})
if isinstance(data.get("taf"), dict)
else {}
)
vertical = data.get("vertical_profile_signal") or {}
current_temp = _sf(current.get("temp"))
max_so_far = _sf(current.get("max_so_far"))
deb_prediction = _sf(deb.get("prediction"))
base_value = _sf(top_bucket.get("value")) if isinstance(top_bucket, dict) else None
if base_value is None:
base_value = deb_prediction
if base_value is None:
base_value = max_so_far if max_so_far is not None else current_temp
base_case_bucket = _bucket_label(top_bucket, unit) or _bucket_label_from_value(base_value, unit)
upside_bucket = _bucket_label_from_value(base_value + 1.0, unit) if base_value is not None else None
downside_bucket = _bucket_label_from_value(base_value - 1.0, unit) if base_value is not None else None
signals: list = []
support_score = 0
suppress_score = 0
available_layers = 0
direction = str(deviation.get("direction") or "").lower()
severity = str(deviation.get("severity") or "normal").lower()
delta = _sf(deviation.get("current_delta"))
if direction:
available_layers += 1
strength = "strong" if severity == "strong" else ("medium" if severity == "light" else "weak")
if direction == "hot":
support_score += 2 if strength == "strong" else 1
_add_signal(
signals,
label="日内节奏",
label_en="Intraday pace",
direction="support",
strength=strength,
summary=f"实测较预期路径偏高 {abs(delta or 0):.1f}{unit},峰值仍有上修空间。",
summary_en=f"Observed temperature is running {abs(delta or 0):.1f}{unit} above the expected path; the peak still has upside room.",
)
elif direction == "cold":
suppress_score += 2 if strength == "strong" else 1
_add_signal(
signals,
label="日内节奏",
label_en="Intraday pace",
direction="suppress",
strength=strength,
summary=f"实测较预期路径偏低 {abs(delta or 0):.1f}{unit},追更高温档需要等待后续观测确认。",
summary_en=f"Observed temperature is running {abs(delta or 0):.1f}{unit} below the expected path; higher buckets need confirmation from later observations.",
)
else:
_add_signal(
signals,
label="日内节奏",
label_en="Intraday pace",
direction="neutral",
strength="weak",
summary="实测大体贴近当前预期路径,下一步主要看峰值窗口内是否继续抬升。",
summary_en="Observed temperature is broadly tracking the expected path; the next question is whether it keeps lifting through the peak window.",
)
heating_setup = str(vertical.get("heating_setup") or "").lower()
suppression_risk = str(vertical.get("suppression_risk") or "").lower()
if heating_setup or suppression_risk:
available_layers += 1
if heating_setup == "supportive":
support_score += 2
_add_signal(
signals,
label="边界层结构",
label_en="Boundary-layer setup",
direction="support",
strength="strong",
summary=str(vertical.get("summary_zh") or "边界层结构支持白天继续混合升温。"),
summary_en=str(vertical.get("summary_en") or "The boundary-layer setup supports continued daytime mixing and warming."),
)
elif heating_setup == "suppressed" or suppression_risk == "high":
suppress_score += 2
_add_signal(
signals,
label="边界层结构",
label_en="Boundary-layer setup",
direction="suppress",
strength="strong",
summary=str(vertical.get("summary_zh") or "边界层或云雨结构对午后峰值形成压制。"),
summary_en=str(vertical.get("summary_en") or "Boundary-layer or cloud/rain structure is capping the afternoon peak."),
)
else:
_add_signal(
signals,
label="边界层结构",
label_en="Boundary-layer setup",
direction="neutral",
strength="medium",
summary=str(vertical.get("summary_zh") or "边界层结构暂未给出单边信号。"),
summary_en=str(vertical.get("summary_en") or "The boundary-layer setup does not yet provide a one-sided signal."),
)
taf_suppression = str(taf_signal.get("suppression_level") or "").lower()
taf_disruption = str(taf_signal.get("disruption_level") or "").lower()
taf_has_cloud_rain_cap = taf_suppression in {"medium", "high"} or taf_disruption in {
"medium",
"high",
}
structural_cap = False
if taf_signal.get("available") or taf_suppression:
available_layers += 1
if taf_suppression == "high" or taf_disruption == "high":
suppress_score += 2
direction_value = "suppress"
strength = "strong"
elif taf_suppression == "medium" or taf_disruption == "medium":
suppress_score += 1
direction_value = "suppress"
strength = "medium"
else:
support_score += 1
direction_value = "support"
strength = "weak"
_add_signal(
signals,
label="TAF 云雨扰动",
label_en="TAF cloud/rain disruption",
direction=direction_value,
strength=strength,
summary=str(taf_signal.get("summary_zh") or "TAF 暂未提示强云雨压温信号。"),
summary_en=str(taf_signal.get("summary_en") or "TAF does not yet flag a strong cloud/rain temperature cap."),
)
airport_delta = _sf(data.get("airport_vs_network_delta"))
lead_signal = data.get("network_lead_signal") or {}
if airport_delta is not None:
available_layers += 1
leader = str(lead_signal.get("leader_station_label") or lead_signal.get("leader_station_code") or "").strip()
sync_status = str(lead_signal.get("leader_sync_status") or "").strip().lower()
sync_delta = _sf(lead_signal.get("leader_time_delta_vs_anchor_minutes"))
sync_suffix_zh = ""
sync_suffix_en = ""
if sync_status in {"near_realtime", "lagged"} and sync_delta is not None:
sync_suffix_zh = f";但与机场锚点约差 {sync_delta:.0f} 分钟,作为降权信号处理"
sync_suffix_en = f"; timing differs from the airport anchor by about {sync_delta:.0f} minutes, so this signal is down-weighted"
elif sync_status == "unknown":
sync_suffix_zh = ";周边站观测时间不可完全校验,作为弱参考"
sync_suffix_en = "; station timing is not fully verified, so this is treated as a weak reference"
if airport_delta <= -0.4:
support_score += 1
_add_signal(
signals,
label="站网对比",
label_en="Station-network comparison",
direction="support",
strength="weak" if sync_suffix_zh else "medium",
summary=f"周边站网较机场锚点偏热 {abs(airport_delta):.1f}{unit}{f',领先点位 {leader}' if leader else ''}{sync_suffix_zh}",
summary_en=f"Nearby stations are {abs(airport_delta):.1f}{unit} warmer than the airport anchor{f'; leading site: {leader}' if leader else ''}{sync_suffix_en}.",
)
elif airport_delta >= 0.4:
suppress_score += 1
_add_signal(
signals,
label="站网对比",
label_en="Station-network comparison",
direction="suppress",
strength="weak" if sync_suffix_zh else "medium",
summary=f"机场锚点较周边站网偏热 {abs(airport_delta):.1f}{unit},继续上修需要机场自身后续报文确认{sync_suffix_zh}",
summary_en=f"The airport anchor is {abs(airport_delta):.1f}{unit} warmer than nearby stations; further upside needs confirmation from later airport reports{sync_suffix_en}.",
)
else:
_add_signal(
signals,
label="站网对比",
label_en="Station-network comparison",
direction="neutral",
strength="weak",
summary="机场锚点与周边站网基本同步,暂不构成单独上修或下修理由。",
summary_en="The airport anchor and nearby station network are broadly aligned, so this layer does not independently argue for upside or downside.",
)
peak_status = str(peak.get("status") or "").lower()
first_h = _sf(peak.get("first_h"))
last_h = _sf(peak.get("last_h"))
peak_window = (
f"{int(first_h):02d}:00-{int(last_h):02d}:59"
if first_h is not None and last_h is not None
else "--"
)
if peak_status == "past":
headline = "峰值窗口已过,后续更偏向确认最终高点而非继续上修。"
headline_en = "The peak window has passed; the read now shifts toward confirming the final high rather than chasing further upside."
confidence = "high" if available_layers >= 2 else "medium"
elif suppress_score >= support_score + 2:
structural_cap = any(
signal.get("direction") == "suppress"
and signal.get("label") in {"边界层结构", "站网对比", "日内节奏"}
for signal in signals
)
if taf_has_cloud_rain_cap and structural_cap:
headline = "峰值同时存在 TAF 云雨扰动和结构压制,当前更偏防守高温上修。"
headline_en = "Both TAF cloud/rain disruption and structural signals are capping the peak; defend against aggressive high-temperature upside for now."
elif taf_has_cloud_rain_cap:
headline = "TAF 提示峰值窗口有云雨扰动,当前更偏防守高温上修。"
headline_en = "TAF flags cloud/rain disruption near the peak window; defend against aggressive high-temperature upside for now."
else:
headline = "峰值主要受结构信号压制,TAF 云雨层暂未构成主压温理由。"
headline_en = "The peak is mainly capped by structural signals; TAF cloud/rain is not the primary suppression reason for now."
confidence = "high" if available_layers >= 3 else "medium"
elif support_score >= suppress_score + 2:
headline = "峰值仍有上修空间,后续重点看峰值窗口内报文能否继续抬升。"
headline_en = "The peak still has upside room; the next check is whether reports keep lifting through the peak window."
confidence = "high" if available_layers >= 3 else "medium"
elif available_layers == 0:
headline = "关键日内层仍在补齐,先以观测锚点和下一次报文为主。"
headline_en = "Key intraday layers are still filling in; anchor the read on observations and the next report."
confidence = "low"
else:
headline = "当前处于分歧判断区,峰值窗口内的下一组观测将决定方向。"
headline_en = "The setup is in a split-decision zone; the next observations inside the peak window should decide direction."
confidence = "medium" if available_layers >= 2 else "low"
next_observation = _next_observation_clock(data.get("local_time") or current.get("obs_time"))
threshold = base_value
invalidation_rules = []
invalidation_rules_en = []
confirmation_rules = []
confirmation_rules_en = []
if peak_status == "past":
invalidation_rules.append("若后续官方结算源补录更高值,以结算源最终高点为准。")
invalidation_rules_en.append("If the official settlement source later backfills a higher reading, defer to the final settlement-source high.")
confirmation_rules.append("若峰值窗口后连续两次观测不再创新高,当前高点基本确认。")
confirmation_rules_en.append("If two consecutive post-peak observations fail to make a new high, the current high is broadly confirmed.")
else:
watch_clock = _format_clock_minutes(int(first_h or 13) * 60 + 30)
if threshold is not None:
invalidation_rules.append(f"{watch_clock} 前若仍未接近 {threshold:.0f}{unit},上修路径降级。")
invalidation_rules_en.append(f"If observations are still not near {threshold:.0f}{unit} before {watch_clock}, downgrade the upside path.")
confirmation_rules.append(f"峰值窗口内任一结算源观测触达或超过 {threshold:.0f}{unit},基准路径确认度上升。")
confirmation_rules_en.append(f"If any settlement-source observation reaches or exceeds {threshold:.0f}{unit} inside the peak window, confidence in the base path rises.")
invalidation_rules.append("若 TAF 或实况报文出现阵雨、雷暴或低云/云雨压制,高温上沿需要下调。")
invalidation_rules_en.append("If TAF or live reports show showers, thunderstorms, or low-cloud/cloud-rain suppression, lower the upper temperature bound.")
confirmation_rules.append("若实测继续贴近 DEB 曲线且云雨信号不增强,维持当前主路径。")
confirmation_rules_en.append("If observations keep tracking the DEB curve and cloud/rain signals do not strengthen, maintain the current main path.")
if not signals:
_add_signal(
signals,
label="数据完整性",
label_en="Data completeness",
direction="neutral",
strength="weak",
summary="当前缺少足够的日内结构层,等待下一次观测刷新后再提高判断权重。",
summary_en="There are not enough intraday structure layers yet; wait for the next observation refresh before raising confidence.",
)
return {
"headline": headline,
"headline_en": headline_en,
"confidence": confidence,
"base_case_bucket": base_case_bucket,
"upside_bucket": upside_bucket,
"downside_bucket": downside_bucket,
"next_observation_time": next_observation,
"peak_window": peak_window,
"invalidation_rules": invalidation_rules[:4],
"invalidation_rules_en": invalidation_rules_en[:4],
"confirmation_rules": confirmation_rules[:3],
"confirmation_rules_en": confirmation_rules_en[:3],
"signal_contributions": signals[:5],
}
+1
View File
@@ -0,0 +1 @@
"""Ops service sub-package."""
+724
View File
@@ -0,0 +1,724 @@
"""Ops config / subscriptions / logs / telegram service functions."""
from __future__ import annotations
import concurrent.futures
import os
import sqlite3
import subprocess
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional
import requests as _requests
from fastapi import HTTPException, Request
from src.database.db_manager import DBManager # type hints
from src.utils.runtime_secrets import get_runtime_secret_status
def _get_db():
from src.database.db_manager import DBManager as _DBManager
return _DBManager()
# ── Config key definitions ──────────────────────────────────────────
_EDITABLE_CONFIG_KEYS: dict[str, str] = {
"POLYWEATHER_AUTH_REQUIRED": "是否强制要求 Supabase 登录访问 API",
"POLYWEATHER_PAYMENT_ENABLED": "是否启用支付功能",
"POLYWEATHER_PAYMENT_POINTS_ENABLED": "是否启用积分抵扣",
"POLYWEATHER_TELEGRAM_ALERT_PUSH_ENABLED": "是否启用 Telegram 告警推送",
"POLYWEATHER_GROUP_MEMBER_PRICE_USDC": "群成员月费 (USDC)",
"POLYWEATHER_PUBLIC_PRICE_USDC": "公开月费 (USDC)",
"POLYWEATHER_PAYMENT_POINTS_PER_USDC": "积分兑换汇率 (积分/USDC)",
"POLYWEATHER_PAYMENT_POINTS_MAX_DISCOUNT_USDC": "积分最高抵扣金额 (USDC)",
"POLYWEATHER_PAYMENT_DIRECT_RECEIVER_ADDRESS": "手动转账收款钱包地址",
}
_SENSITIVE_CONFIG_KEYS: dict[str, dict[str, str]] = {
"POLYWEATHER_AMSC_SESSION_ID": {
"label": "AMSC AWOS sessionId",
"description": "中国跑道观测接口 sessionId,用于上海/北京/广州等 AMSC AWOS 数据源。",
},
}
# ── Helpers ─────────────────────────────────────────────────────────
def _require_ops(request: Request) -> Dict[str, Any] | None:
from web.services.ops_api import _require_ops as _real
return _real(request)
def _parse_iso_datetime(value: Any) -> Optional[datetime]:
text = str(value or "").strip()
if not text:
return None
try:
parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
except Exception:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
def _to_utc_iso(value: datetime) -> str:
if value.tzinfo is None:
value = value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
def _supabase_rest_rows(
table: str,
params: Dict[str, Any],
*,
timeout: int = 10,
) -> List[Dict[str, Any]]:
supabase_url = str(os.getenv("SUPABASE_URL") or "").strip().rstrip("/")
service_role_key = str(os.getenv("SUPABASE_SERVICE_ROLE_KEY") or "").strip()
if not supabase_url or not service_role_key:
raise HTTPException(status_code=503, detail="Supabase not configured")
headers = {
"apikey": service_role_key,
"Authorization": f"Bearer {service_role_key}",
}
resp = _requests.get(
f"{supabase_url}/rest/v1/{table}",
headers=headers,
params=params,
timeout=timeout,
)
if not resp.ok:
raise HTTPException(
status_code=502,
detail=f"Supabase query failed for {table}: {resp.status_code}",
)
rows = resp.json() if resp.content else []
if not isinstance(rows, list):
return []
return [row for row in rows if isinstance(row, dict)]
def _supabase_service_headers(
service_role_key: str,
*,
prefer: str | None = None,
) -> dict[str, str]:
headers = {
"apikey": service_role_key,
"Authorization": f"Bearer {service_role_key}",
"Content-Type": "application/json",
}
if prefer:
headers["Prefer"] = prefer
return headers
def _lookup_supabase_user_id_by_email(
supabase_url: str,
service_role_key: str,
email: str,
) -> str:
normalized_email = str(email or "").strip().lower()
if not normalized_email:
return ""
base = str(supabase_url or "").strip().rstrip("/")
headers = _supabase_service_headers(service_role_key)
profile_resp = _requests.get(
f"{base}/rest/v1/profiles",
headers=headers,
params={
"select": "id",
"email": f"eq.{normalized_email}",
"limit": "1",
},
timeout=10,
)
if profile_resp.ok:
profiles = profile_resp.json() if profile_resp.content else []
if isinstance(profiles, list) and profiles:
user_id = str((profiles[0] or {}).get("id") or "").strip()
if user_id:
return user_id
user_resp = _requests.get(
f"{base}/auth/v1/admin/users",
headers=headers,
params={"filter": f"email.eq.{normalized_email}"},
timeout=10,
)
users = user_resp.json().get("users", []) if user_resp.ok else []
return str(users[0].get("id") or "").strip() if users else ""
# ── Config ──────────────────────────────────────────────────────────
def get_ops_config(request: Request) -> dict[str, Any]:
_require_ops(request)
configs: list[dict[str, Any]] = []
for key, desc in _EDITABLE_CONFIG_KEYS.items():
configs.append(
{
"key": key,
"value": os.getenv(key) or "",
"description": desc,
}
)
return {"configs": configs}
def update_ops_config(request: Request, key: str, value: str) -> dict[str, Any]:
_require_ops(request)
normalized_key = str(key or "").strip()
if normalized_key not in _EDITABLE_CONFIG_KEYS:
raise HTTPException(
status_code=400, detail=f"config key '{normalized_key}' is not editable"
)
os.environ[normalized_key] = str(value)
return {
"key": normalized_key,
"value": value,
"ok": True,
}
def _sensitive_config_payload(key: str) -> dict[str, Any]:
definition = _SENSITIVE_CONFIG_KEYS.get(key) or {}
metadata = get_runtime_secret_status(key)
return {
"key": key,
"label": definition.get("label") or key,
"description": definition.get("description") or "",
"configured": bool(metadata.get("configured")),
"masked": str(metadata.get("masked") or ""),
"length": int(metadata.get("length") or 0),
"updated_at": str(metadata.get("updated_at") or ""),
"updated_by": str(metadata.get("updated_by") or ""),
"source": str(metadata.get("source") or "runtime_store"),
}
def get_ops_sensitive_config(request: Request) -> dict[str, Any]:
_require_ops(request)
return {
"configs": [
_sensitive_config_payload(key)
for key in _SENSITIVE_CONFIG_KEYS
]
}
def update_ops_sensitive_config(
request: Request,
key: str,
value: str,
) -> dict[str, Any]:
admin = _require_ops(request) or {}
normalized_key = str(key or "").strip()
if normalized_key not in _SENSITIVE_CONFIG_KEYS:
raise HTTPException(
status_code=400,
detail=f"sensitive config key '{normalized_key}' is not editable",
)
secret_value = str(value or "").strip()
if not 12 <= len(secret_value) <= 256 or any(ch.isspace() for ch in secret_value):
raise HTTPException(
status_code=400,
detail="sessionId must be 12-256 non-whitespace characters",
)
db = _get_db()
try:
config = db.set_runtime_secret(
normalized_key,
secret_value,
updated_by=str(admin.get("email") or ""),
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
os.environ[normalized_key] = secret_value
response_config = _sensitive_config_payload(normalized_key)
response_config.update(
{
"configured": bool(config.get("configured")),
"masked": str(config.get("masked") or ""),
"length": int(config.get("length") or 0),
"updated_at": str(config.get("updated_at") or ""),
"updated_by": str(config.get("updated_by") or ""),
"source": str(config.get("source") or "runtime_store"),
}
)
# Lazy import to avoid circular dependency with ops_api
from web.services.ops_api import _check_amsc_awos_health
health = (
_check_amsc_awos_health(timeout=8)
if normalized_key == "POLYWEATHER_AMSC_SESSION_ID"
else None
)
return {"ok": True, "config": response_config, "health": health}
# ── Subscriptions ───────────────────────────────────────────────────
def grant_ops_subscription(
request: Request,
email: str,
plan_code: str = "pro_monthly",
days: int = 30,
deduct_points: int = 0,
) -> dict[str, Any]:
_require_ops(request)
from datetime import datetime, timedelta
import web.routes as legacy_routes # lazy avoid circular import
supabase_url = str(os.getenv("SUPABASE_URL") or "").strip().rstrip("/")
service_role_key = str(os.getenv("SUPABASE_SERVICE_ROLE_KEY") or "").strip()
if not supabase_url or not service_role_key:
raise HTTPException(status_code=503, detail="Supabase not configured")
allowed_plans = {"pro_monthly"}
if plan_code not in allowed_plans:
raise HTTPException(
status_code=400, detail=f"invalid plan_code, allowed: {allowed_plans}"
)
safe_days = max(1, min(365, int(days or 30)))
safe_deduct = max(0, int(deduct_points or 0))
normalized_email = str(email or "").strip().lower()
if not normalized_email:
raise HTTPException(status_code=400, detail="email is required")
user_id = _lookup_supabase_user_id_by_email(
supabase_url,
service_role_key,
normalized_email,
)
if not user_id:
raise HTTPException(
status_code=404, detail=f"user not found: {normalized_email}"
)
now = datetime.utcnow()
starts_at = now.isoformat() + "Z"
expires_at = (now + timedelta(days=safe_days)).isoformat() + "Z"
payload = {
"user_id": user_id,
"email": normalized_email,
"plan_code": plan_code,
"starts_at": starts_at,
"expires_at": expires_at,
"source": "ops_manual_grant",
"created_at": now.isoformat() + "Z",
}
resp = _requests.post(
f"{supabase_url}/rest/v1/subscriptions",
headers=_supabase_service_headers(service_role_key, prefer="return=minimal"),
json=payload,
timeout=10,
)
if not resp.ok:
raise HTTPException(
status_code=500, detail=f"Supabase insert failed: {resp.text[:200]}"
)
legacy_routes.SUPABASE_ENTITLEMENT.invalidate_subscription_cache(user_id)
result: dict[str, Any] = {
"ok": True,
"user_id": user_id,
"plan_code": plan_code,
"days": safe_days,
"expires_at": expires_at,
}
# Optionally deduct points from the user (manual Pro grant with points payment)
if safe_deduct > 0:
db = _get_db()
deduct_result = db.deduct_points_by_supabase_email(
normalized_email, safe_deduct
)
result["points_deducted"] = safe_deduct
result["points_result"] = deduct_result
return result
def extend_ops_subscription(
request: Request,
email: str,
additional_days: int = 30,
) -> dict[str, Any]:
_require_ops(request)
from datetime import datetime, timedelta
import web.routes as legacy_routes # lazy avoid circular import
supabase_url = str(os.getenv("SUPABASE_URL") or "").strip().rstrip("/")
service_role_key = str(os.getenv("SUPABASE_SERVICE_ROLE_KEY") or "").strip()
if not supabase_url or not service_role_key:
raise HTTPException(status_code=503, detail="Supabase not configured")
safe_days = max(1, min(365, int(additional_days or 30)))
normalized_email = str(email or "").strip().lower()
if not normalized_email:
raise HTTPException(status_code=400, detail="email is required")
headers = _supabase_service_headers(service_role_key)
user_id = _lookup_supabase_user_id_by_email(
supabase_url,
service_role_key,
normalized_email,
)
if not user_id:
raise HTTPException(
status_code=404, detail=f"user not found: {normalized_email}"
)
# Find latest active subscription
subs_resp = _requests.get(
f"{supabase_url}/rest/v1/subscriptions",
headers=headers,
params={
"select": "id,expires_at",
"user_id": f"eq.{user_id}",
"status": "eq.active",
"order": "expires_at.desc",
"limit": "1",
},
timeout=10,
)
subs = subs_resp.json() if subs_resp.ok else []
if not subs:
raise HTTPException(
status_code=404, detail=f"no subscription found for {normalized_email}"
)
sub = subs[0]
current_expiry = sub.get("expires_at", "")
try:
dt = datetime.fromisoformat(current_expiry.replace("Z", "+00:00"))
new_expiry = (dt + timedelta(days=safe_days)).isoformat()
except Exception:
new_expiry = (datetime.utcnow() + timedelta(days=safe_days)).isoformat() + "Z"
patch_resp = _requests.patch(
f"{supabase_url}/rest/v1/subscriptions?id=eq.{sub['id']}",
headers=_supabase_service_headers(service_role_key, prefer="return=minimal"),
json={"expires_at": new_expiry},
timeout=10,
)
if patch_resp.ok:
legacy_routes.SUPABASE_ENTITLEMENT.invalidate_subscription_cache(user_id)
return {
"ok": True,
"email": normalized_email,
"additional_days": safe_days,
"new_expires_at": new_expiry,
}
raise HTTPException(
status_code=500, detail=f"Supabase update failed: {patch_resp.text[:200]}"
)
def get_ops_user_subscriptions(
request: Request,
email: str,
) -> dict[str, Any]:
"""Return ALL subscription rows for a user (by email), regardless of status."""
_require_ops(request)
supabase_url = str(os.getenv("SUPABASE_URL") or "").strip().rstrip("/")
service_role_key = str(os.getenv("SUPABASE_SERVICE_ROLE_KEY") or "").strip()
if not supabase_url or not service_role_key:
raise HTTPException(status_code=503, detail="Supabase not configured")
normalized_email = str(email or "").strip().lower()
if not normalized_email:
raise HTTPException(status_code=400, detail="email is required")
headers = _supabase_service_headers(service_role_key)
user_id = _lookup_supabase_user_id_by_email(
supabase_url,
service_role_key,
normalized_email,
)
if not user_id:
raise HTTPException(
status_code=404, detail=f"user not found: {normalized_email}"
)
# Fetch all subscription rows for this user (no status filter)
subs_resp = _requests.get(
f"{supabase_url}/rest/v1/subscriptions",
headers=headers,
params={
"select": "id,user_id,status,plan_code,source,starts_at,expires_at,created_at,updated_at",
"user_id": f"eq.{user_id}",
"order": "created_at.desc",
"limit": "50",
},
timeout=10,
)
rows = subs_resp.json() if subs_resp.ok and subs_resp.content else []
if not isinstance(rows, list):
rows = []
return {
"email": normalized_email,
"user_id": user_id,
"subscriptions": rows,
"count": len(rows),
}
# ── Logs ────────────────────────────────────────────────────────────
def get_ops_logs(
request: Request,
level: str = "",
lines: int = 100,
) -> dict[str, Any]:
_require_ops(request)
safe_lines = max(10, min(1000, int(lines or 100)))
log_text = ""
try:
# Read from Docker logs
result = subprocess.run(
["docker", "logs", "--tail", str(safe_lines), "polyweather_bot"],
capture_output=True,
text=True,
timeout=10,
)
log_text = result.stdout or result.stderr or ""
except Exception:
pass
# Fallback to local log file if docker logs returns empty
if not log_text.strip():
log_file = "data/logs/polyweather.log"
if os.path.exists(log_file):
try:
with open(log_file, "r", encoding="utf-8", errors="ignore") as f:
all_lines = f.readlines()
log_text = "".join(all_lines[-safe_lines:])
except Exception:
pass
log_lines = log_text.strip().split("\n") if log_text.strip() else []
if level:
level_upper = level.upper()
log_lines = [line for line in log_lines if level_upper in line.upper()]
return {
"lines": log_lines[-safe_lines:],
"total": len(log_lines),
}
# ── Telegram audit ──────────────────────────────────────────────────
def get_ops_telegram_audit(request: Request) -> Dict[str, Any]:
_require_ops(request)
from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env
# Lazy imports to avoid circular dependency with ops_api
import web.routes as legacy_routes
from web.services.ops.payments import _list_active_subscriptions_with_windows
db = _get_db()
# 1. Fetch all distinct telegram users from database
with db._get_connection() as conn:
conn.row_factory = sqlite3.Row
users_rows = conn.execute("SELECT telegram_id, username FROM users").fetchall()
bindings_rows = conn.execute(
"SELECT telegram_id, supabase_user_id, supabase_email FROM supabase_bindings"
).fetchall()
user_info = {}
for r in users_rows:
tid = int(r["telegram_id"])
user_info[tid] = {
"telegram_id": tid,
"username": r["username"] or f"ID: {tid}",
"supabase_user_id": None,
"supabase_email": None,
"is_bound": False,
}
for r in bindings_rows:
tid = int(r["telegram_id"])
if tid not in user_info:
user_info[tid] = {
"telegram_id": tid,
"username": f"ID: {tid}",
"supabase_user_id": r["supabase_user_id"],
"supabase_email": r["supabase_email"],
"is_bound": True,
}
else:
user_info[tid]["supabase_user_id"] = r["supabase_user_id"]
user_info[tid]["supabase_email"] = r["supabase_email"]
user_info[tid]["is_bound"] = True
# 2. Get Telegram Bot settings
bot_token = str(os.getenv("TELEGRAM_BOT_TOKEN") or "").strip()
chat_ids = get_telegram_chat_ids_from_env()
# Add other group IDs if configured
for env_name in [
"POLYWEATHER_TELEGRAM_GROUP_ID",
"POLYWEATHER_TELEGRAM_TOPICS_GROUP_ID",
]:
val = str(os.getenv(env_name) or "").strip()
if val and val not in chat_ids:
chat_ids.append(val)
if not bot_token or not chat_ids:
return {
"error": "Telegram Bot Token or Chat IDs not configured",
"anomalies": [],
}
# 3. Check membership status for all users in parallel
results = []
def check_user_chat(tg_id, chat_id):
try:
resp = _requests.get(
f"https://api.telegram.org/bot{bot_token}/getChatMember",
params={"chat_id": chat_id, "user_id": tg_id},
timeout=5,
)
if resp.status_code == 200:
data = resp.json()
if data.get("ok"):
res_status = data["result"].get("status")
return chat_id, res_status
return chat_id, None
except Exception:
return chat_id, None
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
futures = {}
for tg_id in user_info.keys():
for c_id in chat_ids:
f = executor.submit(check_user_chat, tg_id, c_id)
futures[f] = (tg_id, c_id)
for f in concurrent.futures.as_completed(futures):
tg_id, c_id = futures[f]
try:
_, status = f.result()
if status in {"creator", "administrator", "member"}:
results.append((tg_id, c_id, status))
except Exception:
pass
# 4. Filter and categorize members
anomalies = []
valid_members = []
active_subs, _, used_active_window_query = _list_active_subscriptions_with_windows(
limit=5000
)
if not used_active_window_query:
active_subs = legacy_routes.SUPABASE_ENTITLEMENT.list_active_subscriptions(
limit=5000
)
active_subs_map = {}
for sub in active_subs:
uid = str(sub.get("user_id") or "").strip().lower()
if uid:
active_subs_map[uid] = sub
for tg_id, chat_id, status in results:
info = user_info[tg_id]
if not info["is_bound"]:
anomalies.append(
{
"telegram_id": tg_id,
"username": info["username"],
"chat_id": chat_id,
"status": status,
"anomaly_type": "unbound",
"reason": "未绑定网页账号",
"email": None,
"expires_at": None,
}
)
else:
uid = str(info["supabase_user_id"]).strip().lower()
sub = active_subs_map.get(uid)
is_paid = False
plan_code = ""
expires_at = None
if sub:
plan_code = str(sub.get("plan_code") or "").strip().lower()
source = str(sub.get("source") or "").strip().lower()
is_paid = "trial" not in plan_code and "trial" not in source
expires_at = sub.get("expires_at")
if not sub:
anomalies.append(
{
"telegram_id": tg_id,
"username": info["username"],
"chat_id": chat_id,
"status": status,
"anomaly_type": "expired",
"reason": "没有有效的会员订阅",
"email": info["supabase_email"],
"expires_at": None,
}
)
elif not is_paid:
anomalies.append(
{
"telegram_id": tg_id,
"username": info["username"],
"chat_id": chat_id,
"status": status,
"anomaly_type": "trial_only",
"reason": f"仅拥有试用会员 ({plan_code})",
"email": info["supabase_email"],
"expires_at": expires_at,
}
)
else:
valid_members.append(
{
"telegram_id": tg_id,
"username": info["username"],
"chat_id": chat_id,
"status": status,
"email": info["supabase_email"],
"plan_code": plan_code,
"expires_at": expires_at,
}
)
return {
"anomalies": anomalies,
"valid_count": len(valid_members),
"anomaly_count": len(anomalies),
}
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+213
View File
@@ -0,0 +1,213 @@
"""Ops users / feedback / points / analytics service functions."""
from __future__ import annotations
from typing import Any, Dict, Optional
from fastapi import HTTPException, Request
from web.core import GrantPointsRequest
import web.routes as legacy_routes
def _get_db():
from web.services.ops_api import DBManager
return DBManager()
# _require_ops is a lightweight auth guard duplicated in each ops submodule
# to avoid circular imports with the ops_api re-export hub.
def _require_ops(request: Request):
from web.services.ops_api import _require_ops as _real
return _real(request)
# ═══════════════════════════════════════════════════════════════════════
# Internal helpers
# ═══════════════════════════════════════════════════════════════════════
def _sf(value: Any) -> Optional[float]:
try:
if value is None or value == "":
return None
return float(value)
except (TypeError, ValueError):
return None
def _round_metric(value: Optional[float], digits: int = 1) -> Optional[float]:
return None if value is None else round(float(value), digits)
def _app_analytics_actor_key(row: Dict[str, Any]) -> str:
payload = row.get("payload")
payload = payload if isinstance(payload, dict) else {}
user_id = str(row.get("user_id") or payload.get("user_id") or "").strip().lower()
client_id = str(row.get("client_id") or "").strip()
session_id = str(row.get("session_id") or "").strip()
if user_id:
return f"user:{user_id}"
if client_id:
return f"client:{client_id}"
if session_id:
return f"session:{session_id}"
return f"event:{row.get('id')}"
# ═══════════════════════════════════════════════════════════════════════
# Users
# ═══════════════════════════════════════════════════════════════════════
def search_ops_users(request: Request, q: str = "", limit: int = 20) -> Dict[str, Any]:
_require_ops(request)
db = _get_db()
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 = _get_db()
return {"leaderboard": db.get_weekly_leaderboard(limit=limit)}
# ═══════════════════════════════════════════════════════════════════════
# Points
# ═══════════════════════════════════════════════════════════════════════
def grant_ops_points(request: Request, body: GrantPointsRequest) -> Dict[str, Any]:
admin = _require_ops(request) or {}
db = _get_db()
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 transfer_ops_points(
request: Request,
from_email: str = "",
to_email: str = "",
amount: int = 0,
) -> Dict[str, Any]:
"""Transfer points from one user to another."""
admin = _require_ops(request) or {}
from_email = str(from_email or "").strip()
to_email = str(to_email or "").strip()
amount = int(amount or 0)
if not from_email or not to_email:
raise HTTPException(
status_code=400, detail="from_email and to_email are required"
)
if amount <= 0:
raise HTTPException(status_code=400, detail="amount must be positive")
db = _get_db()
result = db.transfer_points_by_email(from_email, to_email, amount)
result["operator_email"] = admin.get("email")
if not result.get("ok"):
raise HTTPException(status_code=400, detail=result)
return result
# ═══════════════════════════════════════════════════════════════════════
# Analytics
# ═══════════════════════════════════════════════════════════════════════
def get_ops_analytics_funnel(request: Request, days: int = 30) -> Dict[str, Any]:
_require_ops(request)
db = _get_db()
return db.get_app_analytics_funnel_summary(days=days)
# ═══════════════════════════════════════════════════════════════════════
# Feedback
# ═══════════════════════════════════════════════════════════════════════
def list_ops_feedback(
request: Request,
*,
limit: int = 100,
status: str = "",
) -> Dict[str, Any]:
_require_ops(request)
db = _get_db()
rows = db.list_user_feedback(limit=limit, status=status or None)
status_counts: Dict[str, int] = {}
recent_rows = db.list_user_feedback(limit=500)
for row in recent_rows:
key = str(row.get("status") or "unknown")
status_counts[key] = status_counts.get(key, 0) + 1
return {
"feedback": rows,
"total": len(rows),
"status_counts": status_counts,
}
def update_ops_feedback_status(
request: Request,
*,
feedback_id: int,
status: str,
) -> Dict[str, Any]:
_require_ops(request)
normalized = str(status or "").strip().lower()
allowed = {"open", "triaged", "investigating", "resolved", "closed"}
if normalized not in allowed:
raise HTTPException(status_code=400, detail="unsupported feedback status")
updated = _get_db().update_user_feedback_status(feedback_id, status=normalized)
if not updated:
raise HTTPException(status_code=404, detail="feedback not found")
return {"ok": True, "feedback": updated}
def grant_ops_feedback_reward(
request: Request,
*,
feedback_id: int,
points: int,
reason: str = "",
) -> Dict[str, Any]:
admin = _require_ops(request) or {}
db = _get_db()
result = db.grant_feedback_reward(
feedback_id,
points=points,
reason=reason,
)
if not result.get("ok") and str(result.get("reason") or "") == "user_not_found":
feedback = result.get("feedback") if isinstance(result.get("feedback"), dict) else {}
reward_status = str(feedback.get("reward_status") or "").strip().lower()
reward_points = int(feedback.get("reward_points") or 0)
supabase_user_id = str(feedback.get("user_id") or "").strip().lower()
if supabase_user_id and not (reward_status == "granted" and reward_points > 0):
try:
fallback = legacy_routes.SUPABASE_ENTITLEMENT.grant_points_to_user(
supabase_user_id,
points,
)
except Exception as exc:
fallback = {"ok": False, "reason": f"supabase_points_grant_failed:{exc}"}
if fallback.get("ok"):
updated_feedback = db.update_user_feedback_reward(
feedback_id,
points=points,
reason=reason,
status="granted",
)
result = {
**fallback,
"ok": True,
"feedback_id": int(feedback_id),
"supabase_user_id": supabase_user_id,
"feedback": updated_feedback,
}
result["operator_email"] = admin.get("email")
if not result.get("ok"):
reason_code = str(result.get("reason") or "feedback_reward_failed")
status_code = 404 if reason_code in {"feedback_not_found", "user_not_found"} else 400
if reason_code == "already_rewarded":
status_code = 409
raise HTTPException(status_code=status_code, detail=result)
return result
+76 -2868
View File
File diff suppressed because it is too large Load Diff