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

移除 Polymarket 价格拉取：删除 _market_layer、market_scan 返回空、清理健康检查和配置验证

Browse Source
This commit is contained in:
2569718930@qq.com
2026-05-23 12:04:40 +08:00
parent 442a9c8560
commit eeb5f563db
5 changed files with 407 additions and 324 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/utils/config_validation.py
+12 -9
View File
@@ -71,7 +71,6 @@ def validate_runtime_env(
entitlement_guard = _env_bool("POLYWEATHER_REQUIRE_ENTITLEMENT", False)
payment_enabled = _env_bool("POLYWEATHER_PAYMENT_ENABLED", False)
weekly_reward_enabled = _env_bool("POLYWEATHER_WEEKLY_REWARD_ENABLED", False)
wallet_activity_enabled = _env_bool("POLYMARKET_WALLET_ACTIVITY_ENABLED", False)
polygon_watch_enabled = _env_bool("POLYGON_WALLET_WATCH_ENABLED", False)
if component_key == "bot":
@@ -79,7 +78,9 @@ def validate_runtime_env(
if missing:
report.errors.append(f"Bot 启动缺少必填变量: {', '.join(missing)}")
if not (_has("TELEGRAM_CHAT_ID") or _has("TELEGRAM_CHAT_IDS")):
report.warnings.append("未配置 TELEGRAM_CHAT_ID / TELEGRAM_CHAT_IDS，机器人推送目标为空")
report.warnings.append(
"未配置 TELEGRAM_CHAT_ID / TELEGRAM_CHAT_IDS，机器人推送目标为空"
)
if auth_enabled:
missing = _missing(["SUPABASE_URL", "SUPABASE_ANON_KEY"])
@@ -93,12 +94,16 @@ def validate_runtime_env(
if entitlement_guard:
missing = _missing(["POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN"])
if missing:
report.errors.append(f"已启用 backend entitlement guard，但缺少变量: {', '.join(missing)}")
report.errors.append(
f"已启用 backend entitlement guard，但缺少变量: {', '.join(missing)}"
)
if payment_enabled:
payment_missing = _missing(["POLYWEATHER_PAYMENT_RPC_URL"])
if payment_missing:
report.errors.append(f"已启用支付，但缺少变量: {', '.join(payment_missing)}")
report.errors.append(
f"已启用支付，但缺少变量: {', '.join(payment_missing)}"
)
has_receiver = _has("POLYWEATHER_PAYMENT_RECEIVER_CONTRACT")
has_tokens_json = _has("POLYWEATHER_PAYMENT_ACCEPTED_TOKENS_JSON")
if not (has_receiver or has_tokens_json):
@@ -106,13 +111,11 @@ def validate_runtime_env(
"已启用支付，但未配置 POLYWEATHER_PAYMENT_RECEIVER_CONTRACT 或 POLYWEATHER_PAYMENT_ACCEPTED_TOKENS_JSON"
)
if wallet_activity_enabled:
if not _has("POLYMARKET_WALLET_ACTIVITY_USERS"):
report.warnings.append("已启用 wallet activity watcher，但未配置 POLYMARKET_WALLET_ACTIVITY_USERS")
if polygon_watch_enabled:
if not _has("POLYGON_WALLET_WATCH_ADDRESSES"):
report.warnings.append("已启用 polygon watcher，但未配置 POLYGON_WALLET_WATCH_ADDRESSES")
report.warnings.append(
"已启用 polygon watcher，但未配置 POLYGON_WALLET_WATCH_ADDRESSES"
)
if component_key == "web":
if auth_enabled and not auth_required:
src/utils/telegram_chat_ids.py
-21
View File
@@ -40,27 +40,6 @@ def get_telegram_chat_ids_from_env() -> List[str]:
)
def get_polymarket_wallet_activity_chat_ids_from_env() -> List[str]:
"""
Preferred envs for wallet activity push:
- POLYMARKET_WALLET_ACTIVITY_CHAT_IDS (comma-separated)
- POLYMARKET_WALLET_ACTIVITY_CHAT_ID (single)
Falls back to global TELEGRAM_CHAT_IDS / TELEGRAM_CHAT_ID for compatibility.
"""
dedicated = parse_telegram_chat_ids(
os.getenv("POLYMARKET_WALLET_ACTIVITY_CHAT_IDS"),
os.getenv("POLYMARKET_WALLET_ACTIVITY_CHAT_ID"),
)
if dedicated:
return dedicated
return parse_telegram_chat_ids(
os.getenv("TELEGRAM_CHAT_IDS"),
os.getenv("TELEGRAM_CHAT_ID"),
)
def get_primary_telegram_chat_id_from_env() -> str:
chat_ids = get_telegram_chat_ids_from_env()
return chat_ids[0] if chat_ids else ""
web/core.py
+59 -23
View File
@@ -19,7 +19,6 @@ 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.data_collection.polymarket_readonly import PolymarketReadOnlyLayer
from src.auth.supabase_entitlement import SUPABASE_ENTITLEMENT, extract_bearer_token
from src.utils.metrics import (
build_metrics_summary,
@@ -53,7 +52,6 @@ for _warning in _config_validation.warnings:
if _config_validation.errors:
raise RuntimeError(" | ".join(_config_validation.errors))
_weather = WeatherDataCollector(_config)
_market_layer = PolymarketReadOnlyLayer()
_account_db = DBManager()
CITIES: Dict[str, Dict[str, Any]] = {
@@ -62,7 +60,10 @@ CITIES: Dict[str, Dict[str, Any]] = {
"lon": info["lon"],
"f": info["use_fahrenheit"],
"tz": info["tz_offset"],
"settlement_source": str(info.get("settlement_source") or "metar").strip().lower() or "metar",
"settlement_source": str(info.get("settlement_source") or "metar")
.strip()
.lower()
or "metar",
}
for cid, info in CITY_REGISTRY.items()
}
@@ -141,6 +142,7 @@ async def _etag_middleware(request: Request, call_next):
try:
import hashlib
etag = hashlib.md5(body).hexdigest()
except Exception:
return response
@@ -149,12 +151,14 @@ async def _etag_middleware(request: Request, call_next):
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:
@@ -227,7 +231,9 @@ def _resolve_auth_points(request: Request) -> int:
request.state.auth_points = email_points
points = email_points
except Exception as exc:
logger.warning(f"auth points email fallback failed email={email}: {exc}")
logger.warning(
f"auth points email fallback failed email={email}: {exc}"
)
return points
@@ -289,9 +295,14 @@ def _assert_entitlement(request: Request) -> None:
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")
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")
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:
@@ -306,9 +317,13 @@ def _require_supabase_identity(request: Request) -> Dict[str, str]:
legacy_ok = _legacy_service_token_valid(request)
if legacy_ok:
forwarded_user_id = str(request.headers.get(_FORWARDED_SUPABASE_USER_ID_HEADER) or "").strip()
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()
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)
@@ -316,12 +331,15 @@ def _require_supabase_identity(request: Request) -> Dict[str, str]:
return {"user_id": "entitlement", "email": ""}
logger.warning(
"payment auth identity missing state_user={} auth_bearer={} legacy_ok={} forwarded_user={}"
.format(
"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()),
bool(
str(
request.headers.get(_FORWARDED_SUPABASE_USER_ID_HEADER) or ""
).strip()
),
)
)
raise HTTPException(status_code=401, detail="Unauthorized")
@@ -342,7 +360,10 @@ def _require_ops_admin(request: Request) -> Dict[str, str]:
# 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))}
granted = {
"user_id": "admin:entitlement",
"email": next(iter(_OPS_ADMIN_EMAILS)),
}
return granted
raise HTTPException(status_code=403, detail="ops admin required")
@@ -437,15 +458,24 @@ def _cache_summary() -> Dict[str, Any]:
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 {})
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_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)
@@ -475,8 +505,12 @@ 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")),
"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")),
@@ -603,7 +637,9 @@ def _city_coverage_summary(conn: sqlite3.Connection) -> Dict[str, Any]:
"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 ""),
"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"),
@@ -614,9 +650,7 @@ def _city_coverage_summary(conn: sqlite3.Connection) -> Dict[str, Any]:
)
highlighted = [
entry
for entry in entries
if entry["city"] in {"taipei", "shenzhen"}
entry for entry in entries if entry["city"] in {"taipei", "shenzhen"}
]
gaps = sorted(
entries,
@@ -687,7 +721,9 @@ def _training_data_summary() -> Dict[str, Any]:
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")
training_features = _table_date_summary(
conn, "training_feature_records_store"
)
city_coverage = _city_coverage_summary(conn)
except Exception as exc:
return {
web/services/city_payloads.py
+23 -131
View File
@@ -4,8 +4,7 @@ from __future__ import annotations
from typing import Any, Dict, Optional
from src.analysis.settlement_rounding import apply_city_settlement
from web.core import _is_excluded_model_name, _market_layer, _sf
from web.core import _is_excluded_model_name
TURKISH_MGM_CITIES = {"ankara", "istanbul"}
@@ -22,7 +21,9 @@ def build_city_summary_payload(data: Dict[str, Any]) -> Dict[str, Any]:
"temp": data.get("current", {}).get("temp"),
"obs_time": data.get("current", {}).get("obs_time"),
"settlement_source": data.get("current", {}).get("settlement_source"),
"settlement_source_label": data.get("current", {}).get("settlement_source_label"),
"settlement_source_label": data.get("current", {}).get(
"settlement_source_label"
),
},
"deb": {"prediction": data.get("deb", {}).get("prediction")},
"deviation_monitor": data.get("deviation_monitor") or {},
@@ -41,135 +42,13 @@ def build_city_market_scan_payload(
lite: bool = False,
scan_filters: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
city = str(data.get("name") or "").strip().lower()
local_date = str(data.get("local_date") or "").strip()
requested_date = str(target_date or "").strip()
selected_date = requested_date or local_date
multi_model_daily = data.get("multi_model_daily") or {}
selected_daily = (
multi_model_daily.get(selected_date)
if isinstance(multi_model_daily, dict)
else None
)
if not isinstance(selected_daily, dict):
selected_daily = {}
selected_date = local_date
distribution = selected_daily.get("probabilities")
if not isinstance(distribution, list) or not distribution:
distribution = data.get("probabilities", {}).get("distribution", []) or []
distribution_all = selected_daily.get("probabilities_all")
if not isinstance(distribution_all, list) or not distribution_all:
distribution_all = data.get("probabilities", {}).get("distribution_all", []) or []
if not distribution_all:
distribution_all = distribution
model_map = selected_daily.get("models") or data.get("multi_model") or {}
if not isinstance(model_map, dict):
model_map = {}
anchor_temp = None
anchor_model = None
for model_name, raw_value in model_map.items():
value = _sf(raw_value)
if value is None:
continue
if anchor_temp is None or value > anchor_temp:
anchor_temp = value
anchor_model = str(model_name or "").strip() or None
anchor_temp_c = anchor_temp
temp_symbol = str(data.get("temp_symbol") or "")
if anchor_temp_c is not None and "F" in temp_symbol.upper():
anchor_temp_c = (anchor_temp_c - 32.0) * 5.0 / 9.0
anchor_settlement = apply_city_settlement(city, anchor_temp_c) if anchor_temp_c is not None else None
primary_bucket = None
if isinstance(distribution, list) and distribution:
ranked_buckets = []
temp_symbol_upper = str(temp_symbol or "").upper()
max_primary_bucket_delta = 16.0 if "F" in temp_symbol_upper else 8.0
for idx, row in enumerate(distribution_all):
if not isinstance(row, dict):
continue
bucket_value = _sf(
row.get("temp")
if row.get("temp") is not None
else row.get("value")
if row.get("value") is not None
else row.get("lower")
)
if (
anchor_temp is not None
and bucket_value is not None
and abs(float(bucket_value) - float(anchor_temp)) > max_primary_bucket_delta
):
continue
bucket_prob = _sf(row.get("probability"))
prob_rank = bucket_prob if bucket_prob is not None else -1.0
ranked_buckets.append((-prob_rank, idx, row))
if ranked_buckets:
ranked_buckets.sort(key=lambda x: (x[0], x[1]))
primary_bucket = ranked_buckets[0][2]
elif anchor_temp is None:
primary_bucket = distribution[0]
model_probability = None
if isinstance(primary_bucket, dict) and primary_bucket.get("probability") is not None:
try:
raw_probability = float(primary_bucket.get("probability"))
model_probability = raw_probability / 100.0 if raw_probability > 1.0 else raw_probability
except Exception:
model_probability = None
fallback_sparkline = [
p.get("probability", 0)
for p in distribution_all[:8]
if isinstance(p, dict)
]
current = data.get("current") or {}
selected_deb = selected_daily.get("deb") if isinstance(selected_daily.get("deb"), dict) else {}
current_deb = data.get("deb") if isinstance(data.get("deb"), dict) else {}
scan_context = {
"local_date": data.get("local_date"),
"local_time": data.get("local_time"),
"peak": data.get("peak") or {},
"current_max_so_far": current.get("max_so_far"),
"current_temp": current.get("temp"),
"trend": data.get("trend") or {},
"network_lead_signal": data.get("network_lead_signal") or {},
"models": model_map,
"deb_prediction": selected_deb.get("prediction") or current_deb.get("prediction"),
}
market_scan = _market_layer.build_market_scan(
city=data.get("name"),
target_date=selected_date or data.get("local_date"),
temperature_bucket=primary_bucket if isinstance(primary_bucket, dict) else None,
model_probability=model_probability,
probability_distribution=distribution_all,
temp_symbol=temp_symbol,
fallback_sparkline=fallback_sparkline,
forced_market_slug=market_slug,
include_related_buckets=not lite,
scan_filters=scan_filters,
scan_context=scan_context,
)
if isinstance(market_scan, dict):
market_scan["anchor_model"] = anchor_model
market_scan["anchor_high"] = anchor_temp
market_scan["anchor_settlement"] = anchor_settlement
market_scan["open_meteo_settlement"] = anchor_settlement
probabilities = data.get("probabilities") or {}
market_scan["probability_engine"] = str(
probabilities.get("engine") or "legacy"
).strip() or "legacy"
market_scan["probability_calibration_mode"] = str(
probabilities.get("calibration_mode") or "legacy"
).strip() or "legacy"
return {
"market_scan": market_scan,
"selected_date": selected_date or data.get("local_date"),
"market_scan": {"available": False},
"selected_date": selected_date,
"fetched_at": data.get("updated_at"),
}
@@ -200,7 +79,9 @@ def build_city_detail_payload(
"temp_symbol": data.get("temp_symbol"),
"current_temp": data.get("current", {}).get("temp"),
"settlement_source": data.get("current", {}).get("settlement_source"),
"settlement_source_label": data.get("current", {}).get("settlement_source_label"),
"settlement_source_label": data.get("current", {}).get(
"settlement_source_label"
),
"settlement_station": data.get("settlement_station") or {},
"deb_prediction": data.get("deb", {}).get("prediction"),
"risk_level": data.get("risk", {}).get("level"),
@@ -221,7 +102,12 @@ def build_city_detail_payload(
"weather_gov": {},
"mgm": data.get("mgm") or {},
"mgm_nearby": data.get("mgm_nearby") or [],
"nearby_source": data.get("nearby_source") or ("mgm" if str(data.get("name") or "").lower() in TURKISH_MGM_CITIES else "metar_cluster"),
"nearby_source": data.get("nearby_source")
or (
"mgm"
if str(data.get("name") or "").lower() in TURKISH_MGM_CITIES
else "metar_cluster"
),
"airport_primary": data.get("airport_primary") or {},
"airport_primary_today_obs": data.get("airport_primary_today_obs") or [],
"official_nearby": data.get("official_nearby") or [],
@@ -248,7 +134,8 @@ def build_city_detail_payload(
"deb": data.get("deb") or {},
"multi_model_daily": data.get("multi_model_daily") or {},
"probabilities": data.get("probabilities") or {"mu": None, "distribution": []},
"dynamic_commentary": data.get("dynamic_commentary") or {"summary": "", "notes": []},
"dynamic_commentary": data.get("dynamic_commentary")
or {"summary": "", "notes": []},
"intraday_meteorology": data.get("intraday_meteorology")
or _build_intraday_meteorology(data),
"vertical_profile_signal": data.get("vertical_profile_signal") or {},
@@ -266,7 +153,12 @@ def build_city_detail_payload(
"airport_vs_network_delta": data.get("airport_vs_network_delta"),
"airport_current": data.get("airport_current") or {},
"amos": data.get("amos") or {},
"nearby_source": data.get("nearby_source") or ("mgm" if str(data.get("name") or "").lower() in TURKISH_MGM_CITIES else "metar_cluster"),
"nearby_source": data.get("nearby_source")
or (
"mgm"
if str(data.get("name") or "").lower() in TURKISH_MGM_CITIES
else "metar_cluster"
),
"ai_analysis": data.get("ai_analysis") or "",
"errors": {},
}
web/services/ops_api.py
+313 -140
View File
@@ -39,7 +39,9 @@ def list_ops_memberships(request: Request, limit: int = 200) -> Dict[str, Any]:
)
except Exception:
pass
subscriptions = legacy_routes.SUPABASE_ENTITLEMENT.list_active_subscriptions(limit=limit)
subscriptions = legacy_routes.SUPABASE_ENTITLEMENT.list_active_subscriptions(
limit=limit
)
subscription_user_ids = [str(item.get("user_id") or "") for item in subscriptions]
user_map = db.get_users_by_supabase_user_ids(subscription_user_ids)
unresolved_user_ids = [
@@ -47,19 +49,24 @@ def list_ops_memberships(request: Request, limit: int = 200) -> Dict[str, Any]:
for user_id in subscription_user_ids
if str(user_id or "").strip().lower()
and not str(
(user_map.get(str(user_id).strip().lower(), {}) or {}).get("supabase_email") or ""
(user_map.get(str(user_id).strip().lower(), {}) or {}).get("supabase_email")
or ""
).strip()
]
auth_user_map = legacy_routes.SUPABASE_ENTITLEMENT.get_auth_users(unresolved_user_ids)
auth_user_map = legacy_routes.SUPABASE_ENTITLEMENT.get_auth_users(
unresolved_user_ids
)
deduped: dict[str, dict] = {}
for item in subscriptions:
user_id = str(item.get("user_id") or "").strip().lower()
local_user = user_map.get(user_id, {})
auth_user = auth_user_map.get(user_id, {})
subscription_window = legacy_routes.SUPABASE_ENTITLEMENT.get_subscription_window(
user_id,
respect_requirement=False,
bypass_cache=True,
subscription_window = (
legacy_routes.SUPABASE_ENTITLEMENT.get_subscription_window(
user_id,
respect_requirement=False,
bypass_cache=True,
)
)
current_expires_at = item.get("expires_at")
total_expires_at = (
@@ -78,13 +85,18 @@ def list_ops_memberships(request: Request, limit: int = 200) -> Dict[str, Any]:
else 0
)
source = str(item.get("source") or "")
is_trial = source == "signup_trial" or str(item.get("plan_code") or "").startswith("signup_trial")
is_trial = source == "signup_trial" or str(
item.get("plan_code") or ""
).startswith("signup_trial")
row = {
"user_id": user_id,
"email": str(auth_user.get("email") or local_user.get("supabase_email") or ""),
"email": str(
auth_user.get("email") or local_user.get("supabase_email") or ""
),
"telegram_id": local_user.get("telegram_id"),
"username": local_user.get("username"),
"registered_at": local_user.get("created_at") or auth_user.get("created_at"),
"registered_at": local_user.get("created_at")
or auth_user.get("created_at"),
"plan_code": item.get("plan_code"),
"source": source,
"is_trial": is_trial,
@@ -113,7 +125,9 @@ def get_ops_memberships_growth(request: Request, days: int = 90) -> dict[str, An
from datetime import datetime, timedelta
safe_days = max(7, min(365, int(days or 90)))
subscriptions = legacy_routes.SUPABASE_ENTITLEMENT.list_active_subscriptions(limit=5000)
subscriptions = legacy_routes.SUPABASE_ENTITLEMENT.list_active_subscriptions(
limit=5000
)
now = datetime.utcnow()
cutoff = now - timedelta(days=safe_days)
@@ -150,7 +164,15 @@ def get_ops_memberships_growth(request: Request, days: int = 90) -> dict[str, An
pc = paid_by_day.get(key, 0)
total = tc + pc
running += total
daily.append({"date": key, "trial": tc, "paid": pc, "total": total, "cumulative": running})
daily.append(
{
"date": key,
"trial": tc,
"paid": pc,
"total": total,
"cumulative": running,
}
)
cursor += timedelta(days=1)
return {"days": safe_days, "daily": daily}
@@ -186,7 +208,9 @@ def list_ops_payment_incidents(
def resolve_ops_payment_incident(request: Request, event_id: int) -> Dict[str, Any]:
admin = _require_ops(request) or {}
db = DBManager()
resolved = db.mark_payment_audit_event_resolved(event_id, str(admin.get("email") or ""))
resolved = db.mark_payment_audit_event_resolved(
event_id, str(admin.get("email") or "")
)
if not resolved:
raise HTTPException(status_code=404, detail="payment_incident_not_found")
return {"ok": True, "incident": resolved}
@@ -251,7 +275,9 @@ def transfer_ops_points(
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")
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 = DBManager()
@@ -300,7 +326,8 @@ def get_ops_truth_history(
{
"city": row_city,
"display_name": str(
(legacy_routes.CITY_REGISTRY.get(row_city) or {}).get("name") or row_city
(legacy_routes.CITY_REGISTRY.get(row_city) or {}).get("name")
or row_city
),
"target_date": str(target_date),
"actual_high": payload.get("actual_high"),
@@ -314,7 +341,9 @@ def get_ops_truth_history(
}
)
rows.sort(key=lambda item: (str(item["target_date"]), str(item["city"])), reverse=True)
rows.sort(
key=lambda item: (str(item["target_date"]), str(item["city"])), reverse=True
)
filtered_count = len(rows)
rows = rows[:max_limit]
available_cities = [
@@ -358,27 +387,34 @@ _EDITABLE_CONFIG_KEYS: dict[str, str] = {
def get_ops_config(request: Request) -> dict[str, Any]:
_require_ops(request)
import os
configs: list[dict[str, str]] = []
for key, desc in _EDITABLE_CONFIG_KEYS.items():
configs.append({
"key": key,
"value": os.getenv(key) or "",
"description": desc,
})
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)
import os
if key not in _EDITABLE_CONFIG_KEYS:
raise HTTPException(status_code=400, detail=f"config key '{key}' is not editable")
raise HTTPException(
status_code=400, detail=f"config key '{key}' is not editable"
)
os.environ[key] = str(value)
return {"key": key, "value": value, "ok": True}
# ── Subscriptions ───────────────────────────────────────────────────
def grant_ops_subscription(
request: Request,
email: str,
@@ -398,7 +434,9 @@ def grant_ops_subscription(
allowed_plans = {"pro_monthly"}
if plan_code not in allowed_plans:
raise HTTPException(status_code=400, detail=f"invalid plan_code, allowed: {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))
@@ -423,7 +461,9 @@ def grant_ops_subscription(
users = user_resp.json().get("users", []) if user_resp.ok else []
user_id = users[0].get("id") if users else None
if not user_id:
raise HTTPException(status_code=404, detail=f"user not found: {normalized_email}")
raise HTTPException(
status_code=404, detail=f"user not found: {normalized_email}"
)
now = datetime.utcnow()
starts_at = now.isoformat() + "Z"
@@ -446,17 +486,24 @@ def grant_ops_subscription(
timeout=10,
)
if not resp.ok:
raise HTTPException(status_code=500, detail=f"Supabase insert failed: {resp.text[:200]}")
raise HTTPException(
status_code=500, detail=f"Supabase insert failed: {resp.text[:200]}"
)
result: dict[str, Any] = {
"ok": True, "user_id": user_id, "plan_code": plan_code,
"days": safe_days, "expires_at": expires_at,
"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 = DBManager()
deduct_result = db.deduct_points_by_supabase_email(normalized_email, safe_deduct)
deduct_result = db.deduct_points_by_supabase_email(
normalized_email, safe_deduct
)
result["points_deducted"] = safe_deduct
result["points_result"] = deduct_result
@@ -504,7 +551,9 @@ def extend_ops_subscription(
)
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}")
raise HTTPException(
status_code=404, detail=f"no subscription found for {normalized_email}"
)
sub = subs[0]
current_expiry = sub.get("expires_at", "")
@@ -521,8 +570,15 @@ def extend_ops_subscription(
timeout=10,
)
if patch_resp.ok:
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]}")
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(
@@ -559,7 +615,9 @@ def get_ops_user_subscriptions(
users = user_resp.json().get("users", []) if user_resp.ok else []
user_id = users[0].get("id") if users else None
if not user_id:
raise HTTPException(status_code=404, detail=f"user not found: {normalized_email}")
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(
@@ -587,6 +645,7 @@ def get_ops_user_subscriptions(
# ── Logs ────────────────────────────────────────────────────────────
def get_ops_logs(
request: Request,
level: str = "",
@@ -595,6 +654,7 @@ def get_ops_logs(
_require_ops(request)
import os
import subprocess
safe_lines = max(10, min(1000, int(lines or 100)))
log_text = ""
try:
@@ -649,8 +709,19 @@ def get_ops_health_check(request: Request) -> dict[str, Any]:
supabase_key = str(os.getenv("SUPABASE_SERVICE_ROLE_KEY") or "").strip()
if supabase_url and supabase_key:
try:
r = _r.get(f"{supabase_url}/rest/v1/", headers={"apikey": supabase_key, "Authorization": f"Bearer {supabase_key}"}, timeout=timeout)
results["supabase"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round(r.elapsed.total_seconds() * 1000)}
r = _r.get(
f"{supabase_url}/rest/v1/",
headers={
"apikey": supabase_key,
"Authorization": f"Bearer {supabase_key}",
},
timeout=timeout,
)
results["supabase"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round(r.elapsed.total_seconds() * 1000),
}
except Exception as e:
results["supabase"] = {"ok": False, "error": str(e)[:100]}
else:
@@ -659,16 +730,30 @@ def get_ops_health_check(request: Request) -> dict[str, Any]:
# Open-Meteo
try:
t0 = _time.perf_counter()
r = _r.get("https://api.open-meteo.com/v1/forecast?latitude=52.52&longitude=13.41&daily=temperature_2m_max&timezone=auto&forecast_days=1", timeout=timeout)
results["open_meteo"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
r = _r.get(
"https://api.open-meteo.com/v1/forecast?latitude=52.52&longitude=13.41&daily=temperature_2m_max&timezone=auto&forecast_days=1",
timeout=timeout,
)
results["open_meteo"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["open_meteo"] = {"ok": False, "error": str(e)[:100]}
# METAR (aviationweather)
try:
t0 = _time.perf_counter()
r = _r.get("https://aviationweather.gov/api/data/metar?ids=KJFK&format=json", timeout=timeout)
results["metar"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
r = _r.get(
"https://aviationweather.gov/api/data/metar?ids=KJFK&format=json",
timeout=timeout,
)
results["metar"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["metar"] = {"ok": False, "error": str(e)[:100]}
@@ -677,8 +762,16 @@ def get_ops_health_check(request: Request) -> dict[str, Any]:
if knmi_key:
try:
t0 = _time.perf_counter()
r = _r.get("https://api.dataplatform.knmi.nl/open-data/v1/datasets/10-minute-in-situ-meteorological-observations/versions/1.0/files?maxKeys=1", headers={"Authorization": knmi_key}, timeout=timeout)
results["knmi"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
r = _r.get(
"https://api.dataplatform.knmi.nl/open-data/v1/datasets/10-minute-in-situ-meteorological-observations/versions/1.0/files?maxKeys=1",
headers={"Authorization": knmi_key},
timeout=timeout,
)
results["knmi"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["knmi"] = {"ok": False, "error": str(e)[:100]}
else:
@@ -687,8 +780,15 @@ def get_ops_health_check(request: Request) -> dict[str, Any]:
# MADIS (NOAA)
try:
t0 = _time.perf_counter()
r = _r.get("https://madis-data.ncep.noaa.gov/madisPublic1/data/LDAD/hfmetar/netCDF/", timeout=timeout)
results["madis"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
r = _r.get(
"https://madis-data.ncep.noaa.gov/madisPublic1/data/LDAD/hfmetar/netCDF/",
timeout=timeout,
)
results["madis"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["madis"] = {"ok": False, "error": str(e)[:100]}
@@ -697,8 +797,14 @@ def get_ops_health_check(request: Request) -> dict[str, Any]:
if bot_token:
try:
t0 = _time.perf_counter()
r = _r.get(f"https://api.telegram.org/bot{bot_token}/getMe", timeout=timeout)
results["telegram"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
r = _r.get(
f"https://api.telegram.org/bot{bot_token}/getMe", timeout=timeout
)
results["telegram"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["telegram"] = {"ok": False, "error": str(e)[:100]}
else:
@@ -708,86 +814,125 @@ def get_ops_health_check(request: Request) -> dict[str, Any]:
try:
t0 = _time.perf_counter()
r = _r.get("https://www.jma.go.jp/bosai/forecast/", timeout=timeout)
results["jma"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
results["jma"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["jma"] = {"ok": False, "error": str(e)[:100]}
# MGM (Turkish State Meteorological Service)
try:
t0 = _time.perf_counter()
r = _r.get("https://servis.mgm.gov.tr/web/sondurumlar?istno=17130&_=1", timeout=timeout, headers={"Origin": "https://www.mgm.gov.tr"})
results["mgm"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
r = _r.get(
"https://servis.mgm.gov.tr/web/sondurumlar?istno=17130&_=1",
timeout=timeout,
headers={"Origin": "https://www.mgm.gov.tr"},
)
results["mgm"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["mgm"] = {"ok": False, "error": str(e)[:100]}
# FMI (Finnish Meteorological Institute)
try:
t0 = _time.perf_counter()
r = _r.get("https://opendata.fmi.fi/wfs?service=WFS&version=2.0.0&request=GetCapabilities", timeout=timeout)
results["fmi"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
r = _r.get(
"https://opendata.fmi.fi/wfs?service=WFS&version=2.0.0&request=GetCapabilities",
timeout=timeout,
)
results["fmi"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["fmi"] = {"ok": False, "error": str(e)[:100]}
# KMA (Korea Meteorological Administration)
try:
t0 = _time.perf_counter()
r = _r.get("https://www.weather.go.kr/wgis-nuri/js/info/sfc.geojson", timeout=timeout)
results["kma"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
r = _r.get(
"https://www.weather.go.kr/wgis-nuri/js/info/sfc.geojson", timeout=timeout
)
results["kma"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["kma"] = {"ok": False, "error": str(e)[:100]}
# HKO (Hong Kong Observatory)
try:
t0 = _time.perf_counter()
r = _r.get("https://data.weather.gov.hk/weatherAPI/hko_data/regional-weather/latest_1min_temperature.csv", timeout=timeout)
results["hko"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
r = _r.get(
"https://data.weather.gov.hk/weatherAPI/hko_data/regional-weather/latest_1min_temperature.csv",
timeout=timeout,
)
results["hko"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["hko"] = {"ok": False, "error": str(e)[:100]}
# Singapore MSS (data.gov.sg)
try:
t0 = _time.perf_counter()
r = _r.get("https://api.data.gov.sg/v1/environment/air-temperature", timeout=timeout)
results["singapore_mss"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
r = _r.get(
"https://api.data.gov.sg/v1/environment/air-temperature", timeout=timeout
)
results["singapore_mss"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["singapore_mss"] = {"ok": False, "error": str(e)[:100]}
# CWA (Taiwan Central Weather Administration)
cwa_key = str(os.getenv("CWA_API_KEY") or os.getenv("CWA_OPEN_DATA_AUTH") or os.getenv("CWA_OPEN_DATA_API_KEY") or "").strip()
cwa_key = str(
os.getenv("CWA_API_KEY")
or os.getenv("CWA_OPEN_DATA_AUTH")
or os.getenv("CWA_OPEN_DATA_API_KEY")
or ""
).strip()
if cwa_key:
try:
t0 = _time.perf_counter()
r = _r.get(f"https://opendata.cwa.gov.tw/api/v1/rest/datastore/O-A0003-001?Authorization={cwa_key}&limit=1", timeout=timeout)
results["cwa"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
r = _r.get(
f"https://opendata.cwa.gov.tw/api/v1/rest/datastore/O-A0003-001?Authorization={cwa_key}&limit=1",
timeout=timeout,
)
results["cwa"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["cwa"] = {"ok": False, "error": str(e)[:100]}
else:
results["cwa"] = {"ok": False, "error": "not configured"}
# Polymarket Gamma API
try:
t0 = _time.perf_counter()
r = _r.get("https://gamma-api.polymarket.com/events?limit=1", timeout=timeout)
results["polymarket_gamma"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
except Exception as e:
results["polymarket_gamma"] = {"ok": False, "error": str(e)[:100]}
# Polymarket CLOB API — GET / returns 200 on healthy CLOB
try:
t0 = _time.perf_counter()
r = _r.get("https://clob.polymarket.com/", timeout=timeout)
results["polymarket_clob"] = {"ok": r.status_code < 500, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
except Exception as e:
results["polymarket_clob"] = {"ok": False, "error": str(e)[:100]}
# Polymarket health checks have been removed from the project
# AMOS (Korea runway sensors)
try:
t0 = _time.perf_counter()
r = _r.get("https://global.amo.go.kr/amosobsnew/AmosRealTimeImage.do", timeout=timeout)
results["amos"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
r = _r.get(
"https://global.amo.go.kr/amosobsnew/AmosRealTimeImage.do", timeout=timeout
)
results["amos"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["amos"] = {"ok": False, "error": str(e)[:100]}
@@ -802,11 +947,17 @@ def get_ops_health_check(request: Request) -> dict[str, Any]:
verify=False,
headers={
"Accept": "application/json, text/plain, */*",
"Referer": os.getenv("AMSC_AWOS_REFERER", "https://www.amsc.net.cn/"),
"Referer": os.getenv(
"AMSC_AWOS_REFERER", "https://www.amsc.net.cn/"
),
"User-Agent": "PolyWeather/1.0",
},
)
results["amsc_awos"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
results["amsc_awos"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["amsc_awos"] = {"ok": False, "error": str(e)[:100]}
else:
@@ -816,12 +967,20 @@ def get_ops_health_check(request: Request) -> dict[str, Any]:
try:
t0 = _time.perf_counter()
r = _r.get("https://www.weather.gov/wrh/timeseries?site=KJFK", timeout=timeout)
results["noaa_wrh"] = {"ok": r.ok, "status": r.status_code, "latency_ms": round((_time.perf_counter() - t0) * 1000)}
results["noaa_wrh"] = {
"ok": r.ok,
"status": r.status_code,
"latency_ms": round((_time.perf_counter() - t0) * 1000),
}
except Exception as e:
results["noaa_wrh"] = {"ok": False, "error": str(e)[:100]}
all_ok = all(v.get("ok") for v in results.values())
return {"ok": all_ok, "checked_at": __import__("datetime").datetime.utcnow().isoformat() + "Z", "services": results}
return {
"ok": all_ok,
"checked_at": __import__("datetime").datetime.utcnow().isoformat() + "Z",
"services": results,
}
def get_ops_training_accuracy(request: Request) -> Dict[str, Any]:
@@ -841,7 +1000,7 @@ def get_ops_training_accuracy(request: Request) -> Dict[str, Any]:
"hit_rate": deb_acc[0],
"mae": deb_acc[1],
"total_days": deb_acc[2],
"details_str": deb_acc[3]
"details_str": deb_acc[3],
}
# Calculate Mu accuracy
@@ -853,24 +1012,21 @@ def get_ops_training_accuracy(request: Request) -> Dict[str, Any]:
"hit_rate": mu_acc[1],
"brier_score": mu_acc[2],
"total_days": mu_acc[3],
"details_str": mu_acc[4]
"details_str": mu_acc[4],
}
if deb_payload or mu_payload:
accuracy_data.append({
"city_id": city_id,
"name": name,
"deb": deb_payload,
"mu": mu_payload
})
accuracy_data.append(
{"city_id": city_id, "name": name, "deb": deb_payload, "mu": mu_payload}
)
# Sort by total days of DEB or Mu
accuracy_data.sort(
key=lambda x: max(
x["deb"]["total_days"] if x["deb"] else 0,
x["mu"]["total_days"] if x["mu"] else 0
x["mu"]["total_days"] if x["mu"] else 0,
),
reverse=True
reverse=True,
)
return {"accuracy": accuracy_data}
@@ -891,7 +1047,9 @@ def get_ops_telegram_audit(request: Request) -> Dict[str, Any]:
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()
bindings_rows = conn.execute(
"SELECT telegram_id, supabase_user_id, supabase_email FROM supabase_bindings"
).fetchall()
user_info = {}
for r in users_rows:
@@ -901,7 +1059,7 @@ def get_ops_telegram_audit(request: Request) -> Dict[str, Any]:
"username": r["username"] or f"ID: {tid}",
"supabase_user_id": None,
"supabase_email": None,
"is_bound": False
"is_bound": False,
}
for r in bindings_rows:
@@ -912,7 +1070,7 @@ def get_ops_telegram_audit(request: Request) -> Dict[str, Any]:
"username": f"ID: {tid}",
"supabase_user_id": r["supabase_user_id"],
"supabase_email": r["supabase_email"],
"is_bound": True
"is_bound": True,
}
else:
user_info[tid]["supabase_user_id"] = r["supabase_user_id"]
@@ -924,13 +1082,19 @@ def get_ops_telegram_audit(request: Request) -> Dict[str, Any]:
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"]:
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": []}
return {
"error": "Telegram Bot Token or Chat IDs not configured",
"anomalies": [],
}
# 3. Check membership status for all users in parallel
results = []
@@ -940,7 +1104,7 @@ def get_ops_telegram_audit(request: Request) -> Dict[str, Any]:
resp = requests.get(
f"https://api.telegram.org/bot{bot_token}/getChatMember",
params={"chat_id": chat_id, "user_id": tg_id},
timeout=5
timeout=5,
)
if resp.status_code == 200:
data = resp.json()
@@ -972,7 +1136,10 @@ def get_ops_telegram_audit(request: Request) -> Dict[str, Any]:
valid_members = []
from web.services.ops_api import legacy_routes
active_subs = legacy_routes.SUPABASE_ENTITLEMENT.list_active_subscriptions(limit=5000)
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()
@@ -983,16 +1150,18 @@ def get_ops_telegram_audit(request: Request) -> Dict[str, Any]:
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,
})
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)
@@ -1007,42 +1176,46 @@ def get_ops_telegram_audit(request: Request) -> Dict[str, Any]:
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,
})
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,
})
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,
})
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),
}