From eeb5f563dbfb2e8f60c1cd4c396d4c313518c3df Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Sat, 23 May 2026 12:04:40 +0800 Subject: [PATCH] =?UTF-8?q?=E7=A7=BB=E9=99=A4=20Polymarket=20=E4=BB=B7?= =?UTF-8?q?=E6=A0=BC=E6=8B=89=E5=8F=96=EF=BC=9A=E5=88=A0=E9=99=A4=20=5Fmar?= =?UTF-8?q?ket=5Flayer=E3=80=81market=5Fscan=20=E8=BF=94=E5=9B=9E=E7=A9=BA?= =?UTF-8?q?=E3=80=81=E6=B8=85=E7=90=86=E5=81=A5=E5=BA=B7=E6=A3=80=E6=9F=A5?= =?UTF-8?q?=E5=92=8C=E9=85=8D=E7=BD=AE=E9=AA=8C=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/utils/config_validation.py | 21 +- src/utils/telegram_chat_ids.py | 21 -- web/core.py | 82 ++++-- web/services/city_payloads.py | 154 ++--------- web/services/ops_api.py | 453 +++++++++++++++++++++++---------- 5 files changed, 407 insertions(+), 324 deletions(-) diff --git a/src/utils/config_validation.py b/src/utils/config_validation.py index b68eb372..85c73b09 100644 --- a/src/utils/config_validation.py +++ b/src/utils/config_validation.py @@ -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: diff --git a/src/utils/telegram_chat_ids.py b/src/utils/telegram_chat_ids.py index e3cad3c9..77261d54 100644 --- a/src/utils/telegram_chat_ids.py +++ b/src/utils/telegram_chat_ids.py @@ -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 "" diff --git a/web/core.py b/web/core.py index a8592cd1..3e886f36 100644 --- a/web/core.py +++ b/web/core.py @@ -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 { diff --git a/web/services/city_payloads.py b/web/services/city_payloads.py index 6af6e7bc..1748311a 100644 --- a/web/services/city_payloads.py +++ b/web/services/city_payloads.py @@ -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": {}, } diff --git a/web/services/ops_api.py b/web/services/ops_api.py index 5fa7935f..ec60e102 100644 --- a/web/services/ops_api.py +++ b/web/services/ops_api.py @@ -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), } - -