From af4bee12f5c0f560895f49d5a7c3353d327fb258 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Wed, 11 Mar 2026 08:46:32 +0800 Subject: [PATCH] feat: Introduce a web frontend with new city API routes and refactor bot city query logic into a dedicated service. --- .env.example | 4 +- bot_listener.py | 631 +----------------- frontend/.env.example | 5 + frontend/app/api/cities/route.ts | 3 +- frontend/app/api/city/[name]/detail/route.ts | 3 +- frontend/app/api/city/[name]/route.ts | 3 +- frontend/app/api/city/[name]/summary/route.ts | 3 +- frontend/app/api/history/[name]/route.ts | 3 +- frontend/app/entitlement-required/page.tsx | 46 ++ frontend/lib/backend-auth.ts | 14 + frontend/middleware.ts | 69 ++ src/analysis/city_query_service.py | 468 +++++++++++++ src/analysis/market_alert_engine.py | 157 +++++ src/utils/telegram_push.py | 62 +- web/app.py | 55 +- 15 files changed, 893 insertions(+), 633 deletions(-) create mode 100644 frontend/app/entitlement-required/page.tsx create mode 100644 frontend/lib/backend-auth.ts create mode 100644 frontend/middleware.ts create mode 100644 src/analysis/city_query_service.py diff --git a/.env.example b/.env.example index 36c68bc1..d47084a9 100644 --- a/.env.example +++ b/.env.example @@ -30,6 +30,9 @@ HTTP_PROXY=http://127.0.0.1:7890 LOG_LEVEL=INFO ENV=production POLYWEATHER_MAP_URL=https://polyweather-pro.vercel.app/ +# Backend entitlement guard (for /api/cities, /api/city/*, /api/history/*) +POLYWEATHER_REQUIRE_ENTITLEMENT=false +POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN= # Polymarket P0 Read-Only Market Layer POLYMARKET_MARKET_SCAN_ENABLED=true @@ -74,4 +77,3 @@ POLYMARKET_WALLET_ACTIVITY_BOOTSTRAP_ALERT=false POLYMARKET_WALLET_ACTIVITY_AVG_PRICE_SHOW_MIN=0.01 POLYMARKET_WALLET_ACTIVITY_AVG_PRICE_SHOW_MAX=0.99 - diff --git a/bot_listener.py b/bot_listener.py index 929aff21..5c490d66 100644 --- a/bot_listener.py +++ b/bot_listener.py @@ -1,6 +1,5 @@ import sys import os -from typing import List import telebot # type: ignore from loguru import logger # type: ignore @@ -14,10 +13,11 @@ from src.utils.telegram_push import start_trade_alert_push_loop # type: ignore from src.onchain.polygon_wallet_watcher import start_polygon_wallet_watch_loop # type: ignore # noqa: E402 from src.onchain.polymarket_wallet_activity_watcher import start_polymarket_wallet_activity_loop # type: ignore # noqa: E402 from src.data_collection.weather_sources import WeatherDataCollector # type: ignore # noqa: E402 -from src.data_collection.city_registry import CITY_REGISTRY # noqa: E402 -from src.data_collection.city_risk_profiles import get_city_risk_profile # type: ignore # noqa: E402 -from src.analysis.deb_algorithm import calculate_dynamic_weights, update_daily_record # noqa: E402 from src.database.db_manager import DBManager +from src.analysis.city_query_service import ( + resolve_city_name, + build_city_query_report, +) MESSAGE_POINTS = 4 MESSAGE_DAILY_CAP = 50 @@ -27,13 +27,6 @@ CITY_QUERY_COST = 1 DEB_QUERY_COST = 1 -def analyze_weather_trend(weather_data, temp_symbol, city_name=None): - """Thin wrapper 鈥?delegates to shared trend_engine module.""" - from src.analysis.trend_engine import analyze_weather_trend as _analyze - display_str, ai_context, _structured = _analyze(weather_data, temp_symbol, city_name) - return display_str, ai_context - - def start_bot(): config = load_config() token = os.getenv("TELEGRAM_BOT_TOKEN") @@ -310,7 +303,6 @@ def start_bot(): bot.reply_to(message, f"❌ 查询失败: {e}") @bot.message_handler(commands=["city"]) - def get_city_info(message): """查询指定城市的天气详情""" try: @@ -323,38 +315,13 @@ def start_bot(): ) return - from src.data_collection.city_registry import ALIASES, CITY_REGISTRY city_input = parts[1].strip().lower() - - # --- 使用统一注册表解析城市 --- - SUPPORTED_CITIES = list(CITY_REGISTRY.keys()) - - # 1. 第一优先级:全称或别名完全匹配 - city_name = ALIASES.get(city_input) - if not city_name and city_input in SUPPORTED_CITIES: - city_name = city_input - - # 2. 第二优先级:前缀模糊匹配 - if not city_name and len(city_input) >= 2: - # 搜别名 - for k, v in ALIASES.items(): - if k.startswith(city_input): - city_name = v - break - # 搜城市全名 - if not city_name: - for full_name in SUPPORTED_CITIES: - if full_name.startswith(city_input): - city_name = full_name - break - - # 3. 未找到 ➔ 报错 + city_name, supported_cities = resolve_city_name(city_input) if not city_name: - city_list = ", ".join(sorted(SUPPORTED_CITIES)) + city_list = ", ".join(supported_cities) bot.reply_to( message, - f"❌ 未找到城市: {city_input}\n\n" - f"支持的城市: {city_list}", + f"❌ 未找到城市: {city_input}\n\n支持的城市: {city_list}", parse_mode="HTML", ) return @@ -374,586 +341,12 @@ def start_bot(): weather_data = weather.fetch_all_sources( city_name, lat=coords["lat"], lon=coords["lon"] ) - open_meteo = weather_data.get("open-meteo", {}) - metar = weather_data.get("metar", {}) - mgm = weather_data.get("mgm") or {} - city_meta = CITY_REGISTRY.get(city_name.lower(), {}) - fallback_utc_offset = int(city_meta.get("tz_offset", 0)) - nws_periods = (weather_data.get("nws", {}) or {}).get("forecast_periods", []) or [] - if nws_periods: - try: - from datetime import datetime as _dt - - first_start = nws_periods[0].get("start_time") - if first_start: - maybe_dt = _dt.fromisoformat(str(first_start)) - if maybe_dt.utcoffset() is not None: - fallback_utc_offset = int(maybe_dt.utcoffset().total_seconds()) - except Exception: - pass - - # 鏁板€煎綊涓€鍖? - def _sf(v): - if v is None: - return None - try: - return float(v) - except Exception: - return None - - city_is_fahrenheit = city_name.strip().lower() in ["dallas", "new york", "chicago", "miami", "atlanta", "seattle"] - temp_symbol = "°F" if city_is_fahrenheit else "°C" - - # --- 1. 紧凑 Header (城市 + 时间 + 风险状态) --- - local_time = open_meteo.get("current", {}).get("local_time", "") - time_str = local_time.split(" ")[1][:5] if " " in local_time else "N/A" - if time_str == "N/A": - metar_obs = metar.get("observation_time", "") if metar else "" - if "T" in metar_obs: - try: - from datetime import datetime, timezone, timedelta - - dt = datetime.fromisoformat(metar_obs.replace("Z", "+00:00")) - utc_offset_for_view = open_meteo.get("utc_offset") - if utc_offset_for_view is None: - utc_offset_for_view = fallback_utc_offset - local_dt = dt.astimezone( - timezone(timedelta(seconds=int(utc_offset_for_view))) - ) - time_str = local_dt.strftime("%H:%M") - except Exception: - time_str = metar_obs.split("T")[1][:5] - elif " " in metar_obs: - time_str = metar_obs.split(" ")[1][:5] - elif metar_obs: - time_str = str(metar_obs)[:5] - else: - try: - from datetime import datetime, timezone, timedelta - - local_now = datetime.now(timezone.utc).astimezone( - timezone(timedelta(seconds=int(fallback_utc_offset))) - ) - time_str = local_now.strftime("%H:%M") - except Exception: - pass - - risk_profile = get_city_risk_profile(city_name) - risk_emoji = risk_profile.get("risk_level", "⚠️") if risk_profile else "⚠️" - - msg_header = f"📍 {city_name.title()} ({time_str}) {risk_emoji}" - msg_lines = [msg_header] - - # --- 2. 紧凑 风险提示 --- - if risk_profile: - bias = risk_profile.get("bias", "±0.0") - msg_lines.append( - f"⚠️ {risk_profile.get('airport_name', '')}: {bias}{temp_symbol} | {risk_profile.get('warning', '')}" - ) - - # --- 3. 紧凑 预测区 --- - daily = open_meteo.get("daily", {}) - dates = daily.get("time", [])[:3] - max_temps = daily.get("temperature_2m_max", [])[:3] - - nws_high = _sf(weather_data.get("nws", {}).get("today_high")) - mgm_high = _sf(mgm.get("today_high")) - mb_high = _sf(weather_data.get("meteoblue", {}).get("today_high")) - metar_max_so_far = _sf(metar.get("current", {}).get("max_temp_so_far")) if metar else None - - # 今天对比 - today_t = _sf(max_temps[0]) if max_temps else None - fallback_source = None - metar_only_fallback = False - if today_t is None: - for source_name, candidate in ( - ("MB", mb_high), - ("NWS", nws_high), - ("MGM", mgm_high), - ): - if candidate is not None: - today_t = candidate - fallback_source = source_name - break - if today_t is None and metar_max_so_far is not None: - # Last-resort display only: do not treat METAR as a forecast source - today_t = metar_max_so_far - metar_only_fallback = True - today_t_display = ( - f"{today_t:.1f}" if isinstance(today_t, (int, float)) else "N/A" + city_report = build_city_query_report( + city_name=city_name, + weather_data=weather_data, + city_query_cost=CITY_QUERY_COST, ) - comp_parts = [] - sources = ["Open-Meteo"] if max_temps else [] - - if mb_high is not None: - if "MB" not in sources: - sources.append("MB") - # 只在非 fallback(即 Open-Meteo 存在)时显示为对比,否则作为 today_t 已显示 - if fallback_source != "MB": - comp_parts.append( - f"MB: {mb_high:.1f}{temp_symbol}" - if isinstance(mb_high, (int, float)) - else f"MB: {mb_high}" - ) - if nws_high is not None: - if "NWS" not in sources: - sources.append("NWS") - if fallback_source != "NWS": - comp_parts.append( - f"NWS: {nws_high:.1f}{temp_symbol}" - if isinstance(nws_high, (int, float)) - else f"NWS: {nws_high}" - ) - if mgm_high is not None: - if "MGM" not in sources: - sources.append("MGM") - if fallback_source != "MGM": - comp_parts.append( - f"MGM: {mgm_high:.1f}{temp_symbol}" - if isinstance(mgm_high, (int, float)) - else f"MGM: {mgm_high}" - ) - if fallback_source and fallback_source not in sources: - sources.append(fallback_source) - if metar_only_fallback: - if not sources: - sources = ["Model unavailable"] - comp_parts.append(f"METAR实测回退: {metar_max_so_far:.1f}{temp_symbol}") - if not sources: - sources = ["N/A"] - - # 检查是否有显著分歧 (超过 5°F 或 2.5°C) - divergence_warning = "" - base_for_divergence = _sf(max_temps[0]) if max_temps else today_t - if mb_high is not None and base_for_divergence is not None: - diff = abs(mb_high - base_for_divergence) - threshold = 5.0 if temp_unit == "fahrenheit" else 2.5 - if diff > threshold: - divergence_warning = ( - f" ⚠️ 模型显著分歧 ({diff:.1f}{temp_symbol})" - ) - - comp_str = f" ({' | '.join(comp_parts)})" if comp_parts else "" - sources_str = " | ".join(sources) - - msg_lines.append(f"\n📊 预报 ({sources_str})") - msg_lines.append( - f"👉 今天: {today_t_display}{temp_symbol}{comp_str}{divergence_warning}" - ) - - # 明后天 - mgm_daily = mgm.get("daily_forecasts", {}) or {} - mm_raw = weather_data.get("multi_model", {}) or {} - mm_daily = mm_raw.get("daily_forecasts", {}) if isinstance(mm_raw, dict) else {} - mb_daily = weather_data.get("meteoblue", {}).get("daily_highs", []) or [] - nws_periods = weather_data.get("nws", {}).get("forecast_periods", []) or [] - if len(dates) > 1: - future_forecasts = [] - for d, t in zip(dates[1:], max_temps[1:]): - # 检查 MGM 是否有该日期的预报 - mgm_f = mgm_daily.get(d) - if mgm_f is not None: - future_forecasts.append( - f"{d[5:]}: {t}{temp_symbol} | 🇺🇸 MGM: {mgm_f}{temp_symbol}" - ) - else: - future_forecasts.append(f"{d[5:]}: {t}{temp_symbol}") - msg_lines.append("📅 " + " | ".join(future_forecasts)) - elif mgm_daily: - # Open-Meteo missing: still show next 2 days from MGM daily forecast - from datetime import datetime, timezone, timedelta - - local_now = datetime.now(timezone.utc).astimezone( - timezone(timedelta(seconds=int(fallback_utc_offset))) - ) - today_local = local_now.strftime("%Y-%m-%d") - future_forecasts = [] - for d in sorted(mgm_daily.keys()): - if d <= today_local: - continue - t = mgm_daily.get(d) - if t is None: - continue - future_forecasts.append(f"{d[5:]}: {t}{temp_symbol}") - if len(future_forecasts) >= 2: - break - if future_forecasts: - msg_lines.append("📅 " + " | ".join(future_forecasts)) - elif isinstance(mm_daily, dict) and mm_daily: - # Open-Meteo missing: fallback to multi-model daily medians - from datetime import datetime, timezone, timedelta - - local_now = datetime.now(timezone.utc).astimezone( - timezone(timedelta(seconds=int(fallback_utc_offset))) - ) - today_local = local_now.strftime("%Y-%m-%d") - future_forecasts = [] - for d in sorted(mm_daily.keys()): - if d <= today_local: - continue - day_models = mm_daily.get(d, {}) or {} - vals = [] - for v in day_models.values(): - vv = _sf(v) - if vv is not None: - vals.append(vv) - if not vals: - continue - vals.sort() - median_v = vals[len(vals) // 2] - future_forecasts.append(f"{d[5:]}: MM中位 {median_v:.1f}{temp_symbol}") - if len(future_forecasts) >= 2: - break - if future_forecasts: - msg_lines.append("📅 " + " | ".join(future_forecasts)) - elif isinstance(mb_daily, list) and len(mb_daily) > 1: - # Open-Meteo missing: fallback to Meteoblue daily highs - from datetime import datetime, timezone, timedelta - - local_now = datetime.now(timezone.utc).astimezone( - timezone(timedelta(seconds=int(fallback_utc_offset))) - ) - future_forecasts = [] - for idx in range(1, min(3, len(mb_daily))): - t = _sf(mb_daily[idx]) - if t is None: - continue - d = (local_now + timedelta(days=idx)).strftime("%m-%d") - future_forecasts.append(f"{d}: MB {t:.1f}{temp_symbol}") - if future_forecasts: - msg_lines.append("📅 " + " | ".join(future_forecasts)) - elif isinstance(nws_periods, list) and nws_periods: - # US fallback: use next daytime NWS periods - from datetime import datetime, timezone, timedelta - - local_now = datetime.now(timezone.utc).astimezone( - timezone(timedelta(seconds=int(fallback_utc_offset))) - ) - today_local = local_now.strftime("%Y-%m-%d") - future_forecasts = [] - seen_days = set() - for p in nws_periods: - if not p.get("is_daytime"): - continue - temp_v = _sf(p.get("temperature")) - start_time = str(p.get("start_time") or "") - if temp_v is None or "T" not in start_time: - continue - day_str = start_time[:10] - if day_str <= today_local or day_str in seen_days: - continue - seen_days.add(day_str) - future_forecasts.append(f"{day_str[5:]}: NWS {temp_v:.0f}{temp_symbol}") - if len(future_forecasts) >= 2: - break - if future_forecasts: - msg_lines.append("📅 " + " | ".join(future_forecasts)) - - # --- 3.5 日出日落 + 日照时长 --- - sunrises = daily.get("sunrise", []) - sunsets = daily.get("sunset", []) - sunshine_durations = daily.get("sunshine_duration", []) - if sunrises and sunsets: - sunrise_t = ( - sunrises[0].split("T")[1][:5] - if "T" in str(sunrises[0]) - else sunrises[0] - ) - sunset_t = ( - sunsets[0].split("T")[1][:5] - if "T" in str(sunsets[0]) - else sunsets[0] - ) - sun_line = f"🌅 日出 {sunrise_t} | 🌇 日落 {sunset_t}" - if sunshine_durations: - sunshine_hours = sunshine_durations[0] / 3600 # 秒 -> 小时 - sun_line += f" | ☀️ 日照 {sunshine_hours:.1f}h" - msg_lines.append(sun_line) - - # --- 4. 核心 实测区 (合并 METAR 和 MGM) --- - # 基础数据优先用 METAR - cur_temp = _sf( - metar.get("current", {}).get("temp") - if metar - else mgm.get("current", {}).get("temp") - ) - max_p = _sf( - metar.get("current", {}).get("max_temp_so_far") if metar else None - ) - max_p_time = ( - metar.get("current", {}).get("max_temp_time") if metar else None - ) - obs_t_str = "N/A" - metar_age_min = None # METAR 数据年龄(分钟) - main_source = "METAR" if metar else "MGM" - - if metar: - obs_t = metar.get("observation_time", "") - try: - if "T" in obs_t: - from datetime import datetime, timezone, timedelta - - dt = datetime.fromisoformat(obs_t.replace("Z", "+00:00")) - utc_offset = open_meteo.get("utc_offset") - if utc_offset is None: - utc_offset = fallback_utc_offset - local_dt = dt.astimezone( - timezone(timedelta(seconds=int(utc_offset))) - ) - obs_t_str = local_dt.strftime("%H:%M") - # 计算数据年龄 - now_utc = datetime.now(timezone.utc) - metar_age_min = int((now_utc - dt).total_seconds() / 60) - elif " " in obs_t: - obs_t_str = obs_t.split(" ")[1][:5] - else: - obs_t_str = obs_t - except Exception: - obs_t_str = obs_t[:16] - elif mgm: - m_time = mgm.get("current", {}).get("time", "") - if "T" in m_time: - from datetime import datetime, timezone, timedelta - - dt = datetime.fromisoformat(m_time.replace("Z", "+00:00")) - m_time = dt.astimezone(timezone(timedelta(hours=3))).strftime( - "%H:%M" - ) - elif " " in m_time: - m_time = m_time.split(" ")[1][:5] - obs_t_str = m_time - - # 数据年龄标注 - age_tag = "" - if metar_age_min is not None: - if metar_age_min >= 60: - age_tag = f" ⚠️{metar_age_min}分钟前" - elif metar_age_min >= 30: - age_tag = f" 🔔{metar_age_min}分钟前" - - max_str = "" - if max_p is not None: - import math - - settled_val = math.floor(max_p + 0.5) - max_str = f" (最高: {max_p}{temp_symbol}" - if max_p_time: - max_str += f" @{max_p_time}" - max_str += f" → WU {settled_val}{temp_symbol})" - - # --- 天气状况总结 --- - wx_summary = "" - # 优先使用 METAR 天气现象 - metar_wx = metar.get("current", {}).get("wx_desc", "") if metar else "" - metar_clouds = metar.get("current", {}).get("clouds", []) if metar else [] - mgm_cloud = mgm.get("current", {}).get("cloud_cover") if mgm else None - - if metar_wx: - wx_upper = metar_wx.upper().strip() - wx_tokens = set(wx_upper.split()) - rain_codes = { - "RA", - "DZ", - "-RA", - "+RA", - "-DZ", - "+DZ", - "TSRA", - "SHRA", - "FZRA", - } - snow_codes = {"SN", "GR", "GS", "-SN", "+SN", "BLSN"} - fog_codes = {"FG", "BR", "HZ", "FZFG"} - ts_codes = {"TS", "TSRA"} - if ts_codes & wx_tokens: - wx_summary = "⛈️ 雷暴" - elif {"+RA", "+SN"} & wx_tokens: - wx_summary = "🌧️ 大雨" if "+RA" in wx_tokens else "❄️ 大雪" - elif rain_codes & wx_tokens: - wx_summary = ( - "🌧️ 小雨" if {"-RA", "-DZ", "DZ"} & wx_tokens else "🌧️ 下雨" - ) - elif snow_codes & wx_tokens: - wx_summary = "❄️ 下雪" - elif fog_codes & wx_tokens: - wx_summary = "🌫️ 雾 / 霾" - - # 如果 METAR 没有特殊现象,用云量推断 - if not wx_summary: - # 优先 METAR 云层,回退 MGM - cover_code = "" - if metar_clouds: - cover_code = metar_clouds[-1].get("cover", "") - - if cover_code in ("SKC", "CLR") or ( - cover_code == "" and mgm_cloud is not None and mgm_cloud <= 1 - ): - wx_summary = "☀️ 晴" - elif cover_code == "FEW" or ( - cover_code == "" and mgm_cloud is not None and mgm_cloud <= 2 - ): - wx_summary = "🌤️ 晴间少云" - elif cover_code == "SCT" or ( - cover_code == "" and mgm_cloud is not None and mgm_cloud <= 4 - ): - wx_summary = "⛅ 晴间多云" - elif cover_code == "BKN" or ( - cover_code == "" and mgm_cloud is not None and mgm_cloud <= 6 - ): - wx_summary = "🌥️ 多云" - elif cover_code == "OVC" or ( - cover_code == "" and mgm_cloud is not None and mgm_cloud <= 8 - ): - wx_summary = "☁️ 阴天" - elif mgm_cloud is not None: - cloud_names = { - 0: "☀️ 晴", - 1: "☀️ 晴", - 2: "🌤️ 少云", - 3: "⛅ 散云", - 4: "⛅ 散云", - 5: "🌥️ 多云", - 6: "🌥️ 多云", - 7: "☁️ 阴", - 8: "☁️ 阴天", - } - wx_summary = cloud_names.get(mgm_cloud, "") - - wx_display = f" {wx_summary}" if wx_summary else "" - msg_lines.append( - f"\n✈️ 实测 ({main_source}): {cur_temp}{temp_symbol}{max_str} |{wx_display} | {obs_t_str}{age_tag}" - ) - - if mgm: - m_c = mgm.get("current", {}) - # 翻译风向 - wind_dir = m_c.get("wind_dir") - wind_speed_ms = m_c.get("wind_speed_ms") - dir_str = "" - if wind_dir is not None: - dirs = ["北", "东北", "东", "东南", "南", "西南", "西", "西北"] - dir_str = dirs[int((float(wind_dir) + 22.5) % 360 / 45)] + "风" - - # 体感和湿度(跳过缺失数据) - feels_like = m_c.get("feels_like") - humidity = m_c.get("humidity") - if feels_like is not None or humidity is not None: - parts = [] - if feels_like is not None: - parts.append(f"🌡️ 体感: {feels_like}°C") - - # 针对安卡拉,补充市区(Center)实测值 - ankara_center = next((s for s in weather_data.get("mgm_nearby", []) if "Bölge/Center" in s.get("name", "")), None) - if ankara_center: - parts.append(f"Ankara (Bölge/Center): {ankara_center['temp']}°C") - - if humidity is not None: - parts.append(f"💧 {humidity}%") - msg_lines.append(f" [MGM] {' | '.join(parts)}") - - # 风况(跳过缺失数据) - if wind_dir is not None and wind_speed_ms is not None: - msg_lines.append( - f" [MGM] 🌬️ {dir_str}{wind_dir}° ({wind_speed_ms} m/s) | 💧 降水: {m_c.get('rain_24h') or 0}mm" - ) - - # 新增:气压和云量 - extra_parts = [] - pressure = m_c.get("pressure") - if pressure is not None: - extra_parts.append(f"🌡 气压: {pressure}hPa") - cloud_cover = m_c.get("cloud_cover") - if cloud_cover is not None: - cloud_desc_map = { - 0: "晴朗", - 1: "少云", - 2: "少云", - 3: "散云", - 4: "散云", - 5: "多云", - 6: "多云", - 7: "很多云", - 8: "阴天", - } - cloud_text = cloud_desc_map.get(cloud_cover, f"{cloud_cover}/8") - extra_parts.append(f"☁️ 云量: {cloud_text}({cloud_cover}/8)") - mgm_max = m_c.get("mgm_max_temp") - if mgm_max is not None: - extra_parts.append(f"🌡️ MGM最高: {mgm_max}°C") - if extra_parts: - msg_lines.append(f" [MGM] {' | '.join(extra_parts)}") - - if metar: - m_c = metar.get("current", {}) - wind = m_c.get("wind_speed_kt") - wind_dir = m_c.get("wind_dir") - vis = m_c.get("visibility_mi") - clouds = m_c.get("clouds", []) - - cloud_desc = "" - if clouds: - c_map = { - "BKN": "多云", - "OVC": "阴天", - "FEW": "少云", - "SCT": "散云", - "SKC": "晴", - "CLR": "晴", - } - main = clouds[-1] - cloud_desc = f"☁️ {c_map.get(main.get('cover'), main.get('cover'))}" - - prefix = "[METAR]" if mgm else " " - if not mgm: - msg_lines.append( - f" {prefix} 🌪 {wind or 0}kt ({wind_dir or 0}°) | 👁️ {vis or 10}mi" - ) - - if cloud_desc: - msg_lines.append( - f" {prefix} {cloud_desc} | 👁️ {vis or 10}mi | 🌪 {wind or 0}kt" - ) - - # --- 5. 态势特征提取 --- - feature_str, ai_context = analyze_weather_trend( - weather_data, temp_symbol, city_name - ) - if feature_str: - # 仅将最核心的信息展示给用户作为"态势分析" - # 但后面会把更全的数据传给 AI - msg_lines.append("\n💡 分析:") - for line in feature_str.split("\n"): - if line.strip(): - msg_lines.append(f"- {line.strip()}") - - # --- 6. Groq AI 深度分析 --- - try: - from src.analysis.ai_analyzer import get_ai_analysis - # 构建更全的背景数据给 AI - - # 补充多模型分歧 - mm = weather_data.get("multi_model", {}) or {} - if not isinstance(mm, dict): - mm = {} - if mm.get("forecasts"): - mm_str = " | ".join( - [ - f"{k}:{v}{temp_symbol}" - for k, v in mm["forecasts"].items() - if v - ] - ) - ai_context += f"\n模型分歧: {mm_str}" - - ai_result = get_ai_analysis(ai_context, city_name, temp_symbol) - if ai_result: - msg_lines.append(f"\n{ai_result}") - except Exception as e: - logger.error(f"调用 Groq AI 分析失败: {e}") - - msg_lines.append(f"\n💸 本次消耗 {CITY_QUERY_COST} 积分。") - bot.send_message(message.chat.id, "\n".join(msg_lines), parse_mode="HTML") - + bot.send_message(message.chat.id, city_report, parse_mode="HTML") except Exception as e: import traceback diff --git a/frontend/.env.example b/frontend/.env.example index c049a0d8..2aa4f4e7 100644 --- a/frontend/.env.example +++ b/frontend/.env.example @@ -1 +1,6 @@ POLYWEATHER_API_BASE_URL=http://127.0.0.1:8000 +# Optional dashboard guard (Next.js middleware) +# If set, open dashboard with: /?access_token= +POLYWEATHER_DASHBOARD_ACCESS_TOKEN= +# Shared secret forwarded by Next API routes to backend +POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN= diff --git a/frontend/app/api/cities/route.ts b/frontend/app/api/cities/route.ts index d8787c79..0150acd4 100644 --- a/frontend/app/api/cities/route.ts +++ b/frontend/app/api/cities/route.ts @@ -1,4 +1,5 @@ import { NextResponse } from "next/server"; +import { buildBackendRequestHeaders } from "@/lib/backend-auth"; const API_BASE = process.env.POLYWEATHER_API_BASE_URL; @@ -12,7 +13,7 @@ export async function GET() { try { const res = await fetch(`${API_BASE}/api/cities`, { - headers: { Accept: "application/json" }, + headers: buildBackendRequestHeaders(), next: { revalidate: 120 }, }); if (!res.ok) { diff --git a/frontend/app/api/city/[name]/detail/route.ts b/frontend/app/api/city/[name]/detail/route.ts index 2b30015f..e237653c 100644 --- a/frontend/app/api/city/[name]/detail/route.ts +++ b/frontend/app/api/city/[name]/detail/route.ts @@ -1,4 +1,5 @@ import { NextRequest, NextResponse } from "next/server"; +import { buildBackendRequestHeaders } from "@/lib/backend-auth"; const API_BASE = process.env.POLYWEATHER_API_BASE_URL; @@ -26,7 +27,7 @@ export async function GET( try { const res = await fetch(url, { - headers: { Accept: "application/json" }, + headers: buildBackendRequestHeaders(), cache: "no-store", }); if (!res.ok) { diff --git a/frontend/app/api/city/[name]/route.ts b/frontend/app/api/city/[name]/route.ts index f463496e..74de954d 100644 --- a/frontend/app/api/city/[name]/route.ts +++ b/frontend/app/api/city/[name]/route.ts @@ -1,4 +1,5 @@ import { NextRequest, NextResponse } from "next/server"; +import { buildBackendRequestHeaders } from "@/lib/backend-auth"; const API_BASE = process.env.POLYWEATHER_API_BASE_URL; @@ -19,7 +20,7 @@ export async function GET( try { const res = await fetch(url, { - headers: { Accept: "application/json" }, + headers: buildBackendRequestHeaders(), cache: "no-store", }); if (!res.ok) { diff --git a/frontend/app/api/city/[name]/summary/route.ts b/frontend/app/api/city/[name]/summary/route.ts index 77781abf..20beb876 100644 --- a/frontend/app/api/city/[name]/summary/route.ts +++ b/frontend/app/api/city/[name]/summary/route.ts @@ -1,4 +1,5 @@ import { NextRequest, NextResponse } from "next/server"; +import { buildBackendRequestHeaders } from "@/lib/backend-auth"; const API_BASE = process.env.POLYWEATHER_API_BASE_URL; @@ -19,7 +20,7 @@ export async function GET( try { const res = await fetch(url, { - headers: { Accept: "application/json" }, + headers: buildBackendRequestHeaders(), cache: "no-store", }); if (!res.ok) { diff --git a/frontend/app/api/history/[name]/route.ts b/frontend/app/api/history/[name]/route.ts index 76969c1d..1e2e55bb 100644 --- a/frontend/app/api/history/[name]/route.ts +++ b/frontend/app/api/history/[name]/route.ts @@ -1,4 +1,5 @@ import { NextResponse } from "next/server"; +import { buildBackendRequestHeaders } from "@/lib/backend-auth"; const API_BASE = process.env.POLYWEATHER_API_BASE_URL; @@ -18,7 +19,7 @@ export async function GET( try { const res = await fetch(url, { - headers: { Accept: "application/json" }, + headers: buildBackendRequestHeaders(), cache: "no-store", }); if (!res.ok) { diff --git a/frontend/app/entitlement-required/page.tsx b/frontend/app/entitlement-required/page.tsx new file mode 100644 index 00000000..7ef054d3 --- /dev/null +++ b/frontend/app/entitlement-required/page.tsx @@ -0,0 +1,46 @@ +type Props = { + searchParams?: Promise<{ next?: string }>; +}; + +export default async function EntitlementRequiredPage({ searchParams }: Props) { + const params = (await searchParams) || {}; + const nextPath = params.next || "/"; + + return ( +
+
+

+ Entitlement Required +

+

+ This dashboard is protected. Append{" "} + ?access_token=<your-token> to the URL once, and + the session cookie will be set automatically. +

+

+ Requested path: {nextPath} +

+
+
+ ); +} diff --git a/frontend/lib/backend-auth.ts b/frontend/lib/backend-auth.ts new file mode 100644 index 00000000..165eaad7 --- /dev/null +++ b/frontend/lib/backend-auth.ts @@ -0,0 +1,14 @@ +export const BACKEND_ENTITLEMENT_HEADER = "x-polyweather-entitlement"; + +export function buildBackendRequestHeaders(): HeadersInit { + const headers: HeadersInit = { + Accept: "application/json", + }; + + const token = process.env.POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN?.trim(); + if (token) { + headers[BACKEND_ENTITLEMENT_HEADER] = token; + } + + return headers; +} diff --git a/frontend/middleware.ts b/frontend/middleware.ts new file mode 100644 index 00000000..743a4645 --- /dev/null +++ b/frontend/middleware.ts @@ -0,0 +1,69 @@ +import { NextRequest, NextResponse } from "next/server"; + +const SESSION_COOKIE = "polyweather_entitlement"; + +function isStaticAsset(pathname: string) { + return ( + pathname.startsWith("/_next/") || + pathname.startsWith("/favicon") || + pathname.startsWith("/robots.txt") || + pathname.startsWith("/sitemap.xml") || + pathname.startsWith("/icons/") || + pathname.startsWith("/images/") || + pathname.startsWith("/static/") + ); +} + +function isPublicPage(pathname: string) { + return pathname === "/entitlement-required"; +} + +export function middleware(request: NextRequest) { + const requiredToken = process.env.POLYWEATHER_DASHBOARD_ACCESS_TOKEN?.trim(); + if (!requiredToken) { + return NextResponse.next(); + } + + const { pathname, searchParams } = request.nextUrl; + if (isStaticAsset(pathname) || isPublicPage(pathname)) { + return NextResponse.next(); + } + + const cookieToken = request.cookies.get(SESSION_COOKIE)?.value; + if (cookieToken && cookieToken === requiredToken) { + return NextResponse.next(); + } + + const queryToken = searchParams.get("access_token"); + if (queryToken && queryToken === requiredToken) { + const cleanUrl = request.nextUrl.clone(); + cleanUrl.searchParams.delete("access_token"); + + const response = NextResponse.redirect(cleanUrl); + response.cookies.set(SESSION_COOKIE, requiredToken, { + httpOnly: true, + sameSite: "lax", + secure: cleanUrl.protocol === "https:", + path: "/", + maxAge: 60 * 60 * 12, + }); + return response; + } + + if (pathname.startsWith("/api/")) { + return NextResponse.json( + { error: "Unauthorized", detail: "Entitlement token required" }, + { status: 401 }, + ); + } + + const deniedUrl = request.nextUrl.clone(); + deniedUrl.pathname = "/entitlement-required"; + deniedUrl.search = ""; + deniedUrl.searchParams.set("next", pathname); + return NextResponse.redirect(deniedUrl); +} + +export const config = { + matcher: ["/((?!_next/static|_next/image).*)"], +}; diff --git a/src/analysis/city_query_service.py b/src/analysis/city_query_service.py new file mode 100644 index 00000000..d7f215c3 --- /dev/null +++ b/src/analysis/city_query_service.py @@ -0,0 +1,468 @@ +from __future__ import annotations + +from datetime import datetime, timezone, timedelta +from typing import Any, Dict, List, Optional, Tuple + +from loguru import logger + +from src.analysis.trend_engine import analyze_weather_trend +from src.data_collection.city_registry import ALIASES, CITY_REGISTRY +from src.data_collection.city_risk_profiles import get_city_risk_profile + + +FAHRENHEIT_CITIES = { + "dallas", + "new york", + "chicago", + "miami", + "atlanta", + "seattle", +} + + +def _sf(value: Any) -> Optional[float]: + if value is None: + return None + try: + return float(value) + except Exception: + return None + + +def resolve_city_name(city_input: str) -> Tuple[Optional[str], List[str]]: + city_input_norm = city_input.strip().lower() + supported = list(CITY_REGISTRY.keys()) + + # 1) Exact alias/name + city_name = ALIASES.get(city_input_norm) + if not city_name and city_input_norm in supported: + city_name = city_input_norm + + # 2) Prefix match + if not city_name and len(city_input_norm) >= 2: + for alias, canonical in ALIASES.items(): + if alias.startswith(city_input_norm): + city_name = canonical + break + if not city_name: + for canonical in supported: + if canonical.startswith(city_input_norm): + city_name = canonical + break + + return city_name, sorted(supported) + + +def _render_local_time( + open_meteo: Dict[str, Any], + metar: Dict[str, Any], + fallback_utc_offset: int, +) -> str: + local_time = (open_meteo.get("current") or {}).get("local_time", "") + if " " in str(local_time): + return str(local_time).split(" ")[1][:5] + + metar_obs = metar.get("observation_time", "") if metar else "" + if "T" in str(metar_obs): + try: + dt = datetime.fromisoformat(str(metar_obs).replace("Z", "+00:00")) + utc_offset = open_meteo.get("utc_offset") + if utc_offset is None: + utc_offset = fallback_utc_offset + local_dt = dt.astimezone(timezone(timedelta(seconds=int(utc_offset)))) + return local_dt.strftime("%H:%M") + except Exception: + return str(metar_obs).split("T")[1][:5] + + if " " in str(metar_obs): + return str(metar_obs).split(" ")[1][:5] + if metar_obs: + return str(metar_obs)[:5] + + try: + local_now = datetime.now(timezone.utc).astimezone( + timezone(timedelta(seconds=int(fallback_utc_offset))) + ) + return local_now.strftime("%H:%M") + except Exception: + return "N/A" + + +def _append_future_forecast_lines( + lines: List[str], + weather_data: Dict[str, Any], + dates: List[str], + max_temps: List[Any], + temp_symbol: str, + fallback_utc_offset: int, +) -> None: + mgm = weather_data.get("mgm") or {} + mgm_daily = (mgm.get("daily_forecasts") or {}) if isinstance(mgm, dict) else {} + mm_raw = weather_data.get("multi_model") or {} + mm_daily = mm_raw.get("daily_forecasts", {}) if isinstance(mm_raw, dict) else {} + mb_daily = (weather_data.get("meteoblue") or {}).get("daily_highs", []) or [] + nws_periods = (weather_data.get("nws") or {}).get("forecast_periods", []) or [] + + if len(dates) > 1: + future_forecasts = [] + for d, t in zip(dates[1:], max_temps[1:]): + mgm_value = mgm_daily.get(d) if isinstance(mgm_daily, dict) else None + if mgm_value is not None: + future_forecasts.append( + f"{d[5:]}: {t}{temp_symbol} | 🇺🇸 MGM: {mgm_value}{temp_symbol}" + ) + else: + future_forecasts.append(f"{d[5:]}: {t}{temp_symbol}") + lines.append("📅 " + " | ".join(future_forecasts)) + return + + local_now = datetime.now(timezone.utc).astimezone( + timezone(timedelta(seconds=int(fallback_utc_offset))) + ) + today_local = local_now.strftime("%Y-%m-%d") + + if isinstance(mgm_daily, dict) and mgm_daily: + future = [] + for day in sorted(mgm_daily.keys()): + if day <= today_local: + continue + day_temp = mgm_daily.get(day) + if day_temp is None: + continue + future.append(f"{day[5:]}: {day_temp}{temp_symbol}") + if len(future) >= 2: + break + if future: + lines.append("📅 " + " | ".join(future)) + return + + if isinstance(mm_daily, dict) and mm_daily: + future = [] + for day in sorted(mm_daily.keys()): + if day <= today_local: + continue + models = mm_daily.get(day, {}) or {} + vals = [_sf(v) for v in models.values()] + vals = [v for v in vals if v is not None] + if not vals: + continue + vals.sort() + median = vals[len(vals) // 2] + future.append(f"{day[5:]}: MM中位 {median:.1f}{temp_symbol}") + if len(future) >= 2: + break + if future: + lines.append("📅 " + " | ".join(future)) + return + + if isinstance(mb_daily, list) and len(mb_daily) > 1: + future = [] + for idx in range(1, min(3, len(mb_daily))): + day_temp = _sf(mb_daily[idx]) + if day_temp is None: + continue + day = (local_now + timedelta(days=idx)).strftime("%m-%d") + future.append(f"{day}: MB {day_temp:.1f}{temp_symbol}") + if future: + lines.append("📅 " + " | ".join(future)) + return + + if isinstance(nws_periods, list) and nws_periods: + future = [] + seen_days = set() + for period in nws_periods: + if not period.get("is_daytime"): + continue + day_temp = _sf(period.get("temperature")) + start_time = str(period.get("start_time") or "") + if day_temp is None or "T" not in start_time: + continue + day = start_time[:10] + if day <= today_local or day in seen_days: + continue + seen_days.add(day) + future.append(f"{day[5:]}: NWS {day_temp:.0f}{temp_symbol}") + if len(future) >= 2: + break + if future: + lines.append("📅 " + " | ".join(future)) + + +def _build_wx_summary( + metar_current: Dict[str, Any], + metar_clouds: List[Dict[str, Any]], + mgm_cloud: Optional[Any], +) -> str: + wx_desc = str(metar_current.get("wx_desc") or "").upper().strip() + if wx_desc: + tokens = set(wx_desc.split()) + rain_codes = {"RA", "DZ", "-RA", "+RA", "-DZ", "+DZ", "TSRA", "SHRA", "FZRA"} + snow_codes = {"SN", "GR", "GS", "-SN", "+SN", "BLSN"} + fog_codes = {"FG", "BR", "HZ", "FZFG"} + ts_codes = {"TS", "TSRA"} + if ts_codes & tokens: + return "⛈️ 雷暴" + if {"+RA", "+SN"} & tokens: + return "🌧️ 大雨" if "+RA" in tokens else "❄️ 大雪" + if rain_codes & tokens: + return "🌧️ 小雨" if {"-RA", "-DZ", "DZ"} & tokens else "🌧️ 下雨" + if snow_codes & tokens: + return "❄️ 下雪" + if fog_codes & tokens: + return "🌫️ 雾 / 霾" + + cover_code = "" + if metar_clouds: + cover_code = str((metar_clouds[-1] or {}).get("cover") or "") + + if cover_code in ("SKC", "CLR") or (cover_code == "" and mgm_cloud is not None and mgm_cloud <= 1): + return "☀️ 晴" + if cover_code == "FEW" or (cover_code == "" and mgm_cloud is not None and mgm_cloud <= 2): + return "🌤️ 晴间少云" + if cover_code == "SCT" or (cover_code == "" and mgm_cloud is not None and mgm_cloud <= 4): + return "⛅ 晴间多云" + if cover_code == "BKN" or (cover_code == "" and mgm_cloud is not None and mgm_cloud <= 6): + return "🌥️ 多云" + if cover_code == "OVC" or (cover_code == "" and mgm_cloud is not None and mgm_cloud <= 8): + return "☁️ 阴天" + if mgm_cloud is not None: + cloud_names = { + 0: "☀️ 晴", + 1: "☀️ 晴", + 2: "🌤️ 少云", + 3: "⛅ 散云", + 4: "⛅ 散云", + 5: "🌥️ 多云", + 6: "🌥️ 多云", + 7: "☁️ 阴", + 8: "☁️ 阴天", + } + return cloud_names.get(int(mgm_cloud), "") + return "" + + +def build_city_query_report( + city_name: str, + weather_data: Dict[str, Any], + city_query_cost: int, +) -> str: + open_meteo = weather_data.get("open-meteo", {}) or {} + metar = weather_data.get("metar", {}) or {} + mgm = weather_data.get("mgm") or {} + city_meta = CITY_REGISTRY.get(city_name.lower(), {}) + fallback_utc_offset = int(city_meta.get("tz_offset", 0)) + nws_periods = ((weather_data.get("nws") or {}).get("forecast_periods") or []) + if nws_periods: + try: + first_start = nws_periods[0].get("start_time") + if first_start: + maybe_dt = datetime.fromisoformat(str(first_start)) + if maybe_dt.utcoffset() is not None: + fallback_utc_offset = int(maybe_dt.utcoffset().total_seconds()) + except Exception: + pass + + city_is_fahrenheit = city_name.strip().lower() in FAHRENHEIT_CITIES + temp_symbol = "°F" if city_is_fahrenheit else "°C" + + time_str = _render_local_time(open_meteo, metar, fallback_utc_offset) + risk_profile = get_city_risk_profile(city_name) + risk_emoji = risk_profile.get("risk_level", "⚠️") if risk_profile else "⚠️" + + msg_lines = [f"📍 {city_name.title()} ({time_str}) {risk_emoji}"] + if risk_profile: + bias = risk_profile.get("bias", "±0.0") + msg_lines.append( + f"⚠️ {risk_profile.get('airport_name', '')}: {bias}{temp_symbol} | {risk_profile.get('warning', '')}" + ) + + daily = open_meteo.get("daily", {}) or {} + dates = (daily.get("time") or [])[:3] + max_temps = (daily.get("temperature_2m_max") or [])[:3] + + nws_high = _sf((weather_data.get("nws") or {}).get("today_high")) + mgm_high = _sf((mgm.get("today_high") if isinstance(mgm, dict) else None)) + mb_high = _sf((weather_data.get("meteoblue") or {}).get("today_high")) + metar_max_so_far = _sf((metar.get("current") or {}).get("max_temp_so_far")) + + today_t = _sf(max_temps[0]) if max_temps else None + fallback_source = None + metar_only_fallback = False + if today_t is None: + for source_name, candidate in (("MB", mb_high), ("NWS", nws_high), ("MGM", mgm_high)): + if candidate is not None: + today_t = candidate + fallback_source = source_name + break + if today_t is None and metar_max_so_far is not None: + today_t = metar_max_so_far + metar_only_fallback = True + + today_t_display = f"{today_t:.1f}" if isinstance(today_t, (int, float)) else "N/A" + sources = ["Open-Meteo"] if max_temps else [] + comp_parts: List[str] = [] + + if mb_high is not None: + if "MB" not in sources: + sources.append("MB") + if fallback_source != "MB": + comp_parts.append(f"MB: {mb_high:.1f}{temp_symbol}") + if nws_high is not None: + if "NWS" not in sources: + sources.append("NWS") + if fallback_source != "NWS": + comp_parts.append(f"NWS: {nws_high:.1f}{temp_symbol}") + if mgm_high is not None: + if "MGM" not in sources: + sources.append("MGM") + if fallback_source != "MGM": + comp_parts.append(f"MGM: {mgm_high:.1f}{temp_symbol}") + if fallback_source and fallback_source not in sources: + sources.append(fallback_source) + if metar_only_fallback: + if not sources: + sources = ["Model unavailable"] + comp_parts.append(f"METAR实测回退: {metar_max_so_far:.1f}{temp_symbol}") + if not sources: + sources = ["N/A"] + + divergence_warning = "" + base_for_divergence = _sf(max_temps[0]) if max_temps else today_t + if mb_high is not None and base_for_divergence is not None: + diff = abs(mb_high - base_for_divergence) + threshold = 5.0 if city_is_fahrenheit else 2.5 + if diff > threshold: + divergence_warning = f" ⚠️ 模型显著分歧 ({diff:.1f}{temp_symbol})" + + comp_str = f" ({' | '.join(comp_parts)})" if comp_parts else "" + msg_lines.append(f"\n📊 预报 ({' | '.join(sources)})") + msg_lines.append( + f"👉 今天: {today_t_display}{temp_symbol}{comp_str}{divergence_warning}" + ) + + _append_future_forecast_lines( + lines=msg_lines, + weather_data=weather_data, + dates=dates, + max_temps=max_temps, + temp_symbol=temp_symbol, + fallback_utc_offset=fallback_utc_offset, + ) + + sunrises = daily.get("sunrise", []) or [] + sunsets = daily.get("sunset", []) or [] + sunshine_durations = daily.get("sunshine_duration", []) or [] + if sunrises and sunsets: + sunrise_t = str(sunrises[0]).split("T")[1][:5] if "T" in str(sunrises[0]) else str(sunrises[0]) + sunset_t = str(sunsets[0]).split("T")[1][:5] if "T" in str(sunsets[0]) else str(sunsets[0]) + sun_line = f"🌅 日出 {sunrise_t} | 🌇 日落 {sunset_t}" + if sunshine_durations: + sun_line += f" | ☀️ 日照 {float(sunshine_durations[0]) / 3600:.1f}h" + msg_lines.append(sun_line) + + metar_current = metar.get("current", {}) if isinstance(metar, dict) else {} + mgm_current = mgm.get("current", {}) if isinstance(mgm, dict) else {} + cur_temp = _sf(metar_current.get("temp")) + if cur_temp is None: + cur_temp = _sf(mgm_current.get("temp")) + max_p = _sf(metar_current.get("max_temp_so_far")) + max_p_time = metar_current.get("max_temp_time") + obs_t_str = "N/A" + metar_age_min = None + main_source = "METAR" if metar else "MGM" + + if metar and metar.get("observation_time"): + obs_t = str(metar.get("observation_time")) + try: + if "T" in obs_t: + dt = datetime.fromisoformat(obs_t.replace("Z", "+00:00")) + utc_offset = open_meteo.get("utc_offset") + if utc_offset is None: + utc_offset = fallback_utc_offset + local_dt = dt.astimezone(timezone(timedelta(seconds=int(utc_offset)))) + obs_t_str = local_dt.strftime("%H:%M") + metar_age_min = int((datetime.now(timezone.utc) - dt).total_seconds() / 60) + elif " " in obs_t: + obs_t_str = obs_t.split(" ")[1][:5] + else: + obs_t_str = obs_t + except Exception: + obs_t_str = obs_t[:16] + elif mgm: + mgm_time = str(mgm_current.get("time") or "") + if "T" in mgm_time: + dt = datetime.fromisoformat(mgm_time.replace("Z", "+00:00")) + mgm_time = dt.astimezone(timezone(timedelta(hours=3))).strftime("%H:%M") + elif " " in mgm_time: + mgm_time = mgm_time.split(" ")[1][:5] + obs_t_str = mgm_time or "N/A" + + age_tag = "" + if metar_age_min is not None: + if metar_age_min >= 60: + age_tag = f" ⚠️{metar_age_min}分钟前" + elif metar_age_min >= 30: + age_tag = f" 🔔{metar_age_min}分钟前" + + max_str = "" + if max_p is not None: + settled_val = int(max_p + 0.5) + max_str = f" (最高: {max_p}{temp_symbol}" + if max_p_time: + max_str += f" @{max_p_time}" + max_str += f" → WU {settled_val}{temp_symbol})" + + metar_clouds = metar_current.get("clouds", []) if isinstance(metar_current, dict) else [] + mgm_cloud = mgm_current.get("cloud_cover") if isinstance(mgm_current, dict) else None + wx_summary = _build_wx_summary(metar_current, metar_clouds, mgm_cloud) + wx_display = f" {wx_summary}" if wx_summary else "" + msg_lines.append( + f"\n✈️ 实测 ({main_source}): {cur_temp}{temp_symbol}{max_str} |{wx_display} | {obs_t_str}{age_tag}" + ) + + if metar: + wind = metar_current.get("wind_speed_kt") + wind_dir = metar_current.get("wind_dir") + vis = metar_current.get("visibility_mi") + if not mgm: + msg_lines.append(f" [METAR] 🌪 {wind or 0}kt ({wind_dir or 0}°) | 👁️ {vis or 10}mi") + if mgm: + wind_dir = mgm_current.get("wind_dir") + wind_speed_ms = mgm_current.get("wind_speed_ms") + if wind_dir is not None and wind_speed_ms is not None: + dirs = ["北", "东北", "东", "东南", "南", "西南", "西", "西北"] + dir_str = dirs[int((float(wind_dir) + 22.5) % 360 / 45)] + "风" + msg_lines.append( + f" [MGM] 🌬️ {dir_str}{wind_dir}° ({wind_speed_ms} m/s) | 💧 降水: {mgm_current.get('rain_24h') or 0}mm" + ) + + feature_str, ai_context, _structured = analyze_weather_trend(weather_data, temp_symbol, city_name) + if feature_str: + msg_lines.append("\n💡 分析:") + for line in feature_str.split("\n"): + if line.strip(): + msg_lines.append(f"- {line.strip()}") + + try: + from src.analysis.ai_analyzer import get_ai_analysis + + mm = weather_data.get("multi_model", {}) or {} + if not isinstance(mm, dict): + mm = {} + if mm.get("forecasts"): + mm_parts = [ + f"{k}:{v}{temp_symbol}" + for k, v in (mm.get("forecasts") or {}).items() + if v is not None + ] + if mm_parts: + ai_context += f"\n模型分歧: {' | '.join(mm_parts)}" + + ai_result = get_ai_analysis(ai_context, city_name, temp_symbol) + if ai_result: + msg_lines.append(f"\n{ai_result}") + except Exception as exc: + logger.error(f"调用 Groq AI 分析失败: {exc}") + + msg_lines.append(f"\n💸 本次消耗 {city_query_cost} 积分。") + return "\n".join(msg_lines) diff --git a/src/analysis/market_alert_engine.py b/src/analysis/market_alert_engine.py index de68f123..83ec024a 100644 --- a/src/analysis/market_alert_engine.py +++ b/src/analysis/market_alert_engine.py @@ -1013,6 +1013,154 @@ def _build_telegram_messages_mispricing( return {"zh": "\n".join(lines_zh), "en": "\n".join(lines_en)} +def _select_rule_evidence(rule: Dict[str, Any], keys: List[str]) -> Dict[str, Any]: + out: Dict[str, Any] = {} + for key in keys: + if key in rule: + out[key] = rule.get(key) + return out + + +def _build_alert_evidence( + city_weather: Dict[str, Any], + rules: Dict[str, Dict[str, Any]], + triggered: List[Dict[str, Any]], + suppression: Dict[str, Any], + market_snapshot: Dict[str, Any], + temp_symbol: str, +) -> Dict[str, Any]: + current = city_weather.get("current") or {} + deb = city_weather.get("deb") or {} + + momentum = rules.get("momentum_spike") or {} + breakthrough = rules.get("forecast_breakthrough") or {} + advection = rules.get("advection") or {} + ankara_center = rules.get("ankara_center_deb_hit") or {} + + top_rows = [] + for row in (market_snapshot.get("top_bucket_rows") or [])[:4]: + if not isinstance(row, dict): + continue + top_rows.append( + { + "label": row.get("label"), + "probability": row.get("probability"), + "yes_buy": row.get("yes_buy"), + "yes_sell": row.get("yes_sell"), + "market_url": row.get("market_url"), + } + ) + + trigger_types = [row.get("type") for row in triggered if row.get("type")] + forecast_bucket = market_snapshot.get("forecast_bucket") or {} + + return { + "version": 1, + "city": city_weather.get("name"), + "generated_local_time": city_weather.get("local_time"), + "observed_at": current.get("obs_time"), + "temp_symbol": temp_symbol, + "inputs": { + "current_temp": _sf(current.get("temp")), + "deb_prediction": _sf(deb.get("prediction")), + "wu_settle": current.get("wu_settle"), + "obs_age_min": current.get("obs_age_min"), + }, + "trigger_summary": { + "trigger_count": len(trigger_types), + "trigger_types": trigger_types, + "suppressed": bool(suppression.get("suppressed")), + "suppression_reason": suppression.get("reason"), + "suppression_snapshot": _select_rule_evidence( + suppression, + [ + "minutes_since_peak", + "rollback", + "rollback_threshold", + "max_temp_time", + "max_so_far", + "current_temp", + ], + ), + }, + "rules": { + "momentum_spike": _select_rule_evidence( + momentum, + [ + "triggered", + "direction", + "delta_temp", + "delta_minutes", + "slope_30m", + "threshold_30m", + ], + ), + "forecast_breakthrough": _select_rule_evidence( + breakthrough, + [ + "triggered", + "baseline_model", + "baseline_high", + "current_temp", + "margin", + "threshold", + "model_coverage", + ], + ), + "advection": _select_rule_evidence( + advection, + [ + "triggered", + "lead_delta", + "threshold_delta", + "wind_now", + "wind_prev", + "turned_southerly", + "wind_alignment_deg", + "lead_window_minutes", + ], + ), + "ankara_center_deb_hit": _select_rule_evidence( + ankara_center, + [ + "triggered", + "deb_prediction", + "airport_temp", + "margin_vs_deb", + "center_lead_vs_airport", + ], + ), + }, + "market": { + "available": bool(market_snapshot.get("available")), + "market_prob": market_snapshot.get("market_prob"), + "model_prob": market_snapshot.get("model_prob"), + "edge_percent": market_snapshot.get("edge_percent"), + "yes_buy": market_snapshot.get("yes_buy"), + "yes_sell": market_snapshot.get("yes_sell"), + "spread": market_snapshot.get("spread"), + "signal_label": market_snapshot.get("signal_label"), + "confidence": market_snapshot.get("confidence"), + "top_bucket": market_snapshot.get("top_bucket"), + "top_bucket_prob": market_snapshot.get("top_bucket_prob"), + "open_meteo_today_high_c": market_snapshot.get("open_meteo_today_high_c"), + "open_meteo_settlement": market_snapshot.get("open_meteo_settlement"), + "forecast_bucket": { + "label": forecast_bucket.get("label"), + "probability": forecast_bucket.get("probability"), + "yes_buy": forecast_bucket.get("yes_buy"), + "yes_sell": forecast_bucket.get("yes_sell"), + "market_url": forecast_bucket.get("market_url"), + } + if isinstance(forecast_bucket, dict) + else None, + "top4": top_rows, + "market_url": market_snapshot.get("market_url"), + "primary_market_url": market_snapshot.get("primary_market_url"), + }, + } + + def build_trading_alerts( city_weather: Dict[str, Any], map_url: Optional[str] = None, @@ -1065,6 +1213,14 @@ def build_trading_alerts( rules=rules, market_snapshot=market_snapshot, ) + evidence = _build_alert_evidence( + city_weather=city_weather, + rules=rules, + triggered=triggered, + suppression=suppression, + market_snapshot=market_snapshot, + temp_symbol=temp_symbol, + ) return { "city": city, @@ -1076,6 +1232,7 @@ def build_trading_alerts( "market_snapshot": market_snapshot, "suppression": suppression, "triggered_alerts": triggered, + "evidence": evidence, "telegram": telegram, } diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index b6a451d3..237b036a 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -187,6 +187,53 @@ def _trigger_type_key(alert_payload: Dict[str, Any]) -> str: return "|".join(trigger_types) +def _evidence_brief(alert_payload: Dict[str, Any]) -> str: + evidence = alert_payload.get("evidence") or {} + if not isinstance(evidence, dict): + return "--" + + trigger_summary = evidence.get("trigger_summary") or {} + rules = evidence.get("rules") or {} + market = evidence.get("market") or {} + momentum = rules.get("momentum_spike") or {} + advection = rules.get("advection") or {} + breakthrough = rules.get("forecast_breakthrough") or {} + + parts: List[str] = [] + trigger_types = trigger_summary.get("trigger_types") + if isinstance(trigger_types, list) and trigger_types: + parts.append(f"triggers={','.join(str(t) for t in trigger_types)}") + + slope = momentum.get("slope_30m") + if slope is not None: + parts.append(f"slope_30m={slope}") + + lead_delta = advection.get("lead_delta") + if lead_delta is not None: + parts.append(f"lead_delta={lead_delta}") + + margin = breakthrough.get("margin") + if margin is not None: + parts.append(f"break_margin={margin}") + + edge = market.get("edge_percent") + if edge is not None: + parts.append(f"edge_pct={edge}") + + forecast_bucket = market.get("forecast_bucket") or {} + if isinstance(forecast_bucket, dict): + label = str(forecast_bucket.get("label") or "").strip() + yes_buy = forecast_bucket.get("yes_buy") + if label: + parts.append(f"bucket={label}") + if yes_buy is not None: + parts.append(f"yes_buy={yes_buy}") + + if not parts: + return "--" + return "; ".join(parts) + + def _alert_signature(alert_payload: Dict[str, Any]) -> str: rules = alert_payload.get("rules") or {} center_deb = rules.get("ankara_center_deb_hit") or {} @@ -321,11 +368,13 @@ def _maybe_send_alert( "severity": alert_payload.get("severity"), "ts": now_ts, "active": True, + "evidence": alert_payload.get("evidence"), } state.setdefault("by_signature", {})[signature] = now_ts logger.info( f"trade alert pushed city={city} severity={alert_payload.get('severity')} " - f"trigger_count={alert_payload.get('trigger_count')} trigger_key={trigger_key}" + f"trigger_count={alert_payload.get('trigger_count')} trigger_key={trigger_key} " + f"evidence={_evidence_brief(alert_payload)}" ) return True @@ -340,7 +389,13 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th logger.warning("telegram alert push loop skipped: TELEGRAM_CHAT_ID is not set") return None - interval_sec = max(60, _env_int("TELEGRAM_ALERT_PUSH_INTERVAL_SEC", 300)) + mispricing_only = _env_bool("TELEGRAM_ALERT_MISPRICING_ONLY", True) + if mispricing_only: + interval_sec = max( + 300, _env_int("TELEGRAM_ALERT_MISPRICING_INTERVAL_SEC", 7200) + ) + else: + interval_sec = max(60, _env_int("TELEGRAM_ALERT_PUSH_INTERVAL_SEC", 300)) cooldown_sec = max(interval_sec, _env_int("TELEGRAM_ALERT_PUSH_COOLDOWN_SEC", 1800)) min_trigger_count = max(1, _env_int("TELEGRAM_ALERT_MIN_TRIGGER_COUNT", 2)) min_severity = os.getenv("TELEGRAM_ALERT_MIN_SEVERITY", "medium").strip().lower() @@ -353,7 +408,8 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th except Exception: logger.exception(f"failed to initialize telegram push state path={state_path}") logger.info( - f"telegram alert push loop started cities={len(cities)} interval={interval_sec}s " + f"telegram alert push loop started mode={'mispricing-only' if mispricing_only else 'full'} " + f"cities={len(cities)} interval={interval_sec}s " f"cooldown={cooldown_sec}s min_triggers={min_trigger_count} min_severity={min_severity} " f"state_path={state_path}" ) diff --git a/web/app.py b/web/app.py index ca1c437f..adc51de6 100644 --- a/web/app.py +++ b/web/app.py @@ -21,7 +21,7 @@ if _root not in sys.path: if _file_dir not in sys.path: sys.path.insert(0, _file_dir) -from fastapi import FastAPI, HTTPException +from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from loguru import logger @@ -77,6 +77,45 @@ CACHE_TTL = 300 CACHE_TTL_ANKARA = 60 # Ankara measurement updates frequent, narrower cache +def _env_bool(name: str, default: bool = False) -> bool: + raw = os.getenv(name) + if raw is None: + return default + return raw.strip().lower() in {"1", "true", "yes", "on"} + + +_ENTITLEMENT_GUARD_ENABLED = _env_bool("POLYWEATHER_REQUIRE_ENTITLEMENT", False) +_ENTITLEMENT_HEADER = "x-polyweather-entitlement" +_ENTITLEMENT_TOKEN = (os.getenv("POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN") or "").strip() + + +def _extract_bearer_token(auth_header: Optional[str]) -> Optional[str]: + if not auth_header: + return None + parts = auth_header.strip().split() + if len(parts) == 2 and parts[0].lower() == "bearer": + return parts[1].strip() + return None + + +def _assert_entitlement(request: Request) -> None: + if not _ENTITLEMENT_GUARD_ENABLED: + return + + if not _ENTITLEMENT_TOKEN: + raise HTTPException( + status_code=503, + detail="Entitlement guard is enabled but backend token is not configured", + ) + + token = request.headers.get(_ENTITLEMENT_HEADER) + if not token: + token = _extract_bearer_token(request.headers.get("authorization")) + + if token != _ENTITLEMENT_TOKEN: + raise HTTPException(status_code=401, detail="Unauthorized") + + def _sf(v) -> Optional[float]: """Safe float conversion.""" if v is None: @@ -672,8 +711,9 @@ def _analyze(city: str, force_refresh: bool = False) -> Dict[str, Any]: # Routes # ────────────────────────────────────────────────────────── @app.get("/api/cities") -async def list_cities(): +async def list_cities(request: Request): """Return all supported cities with coordinates and risk level.""" + _assert_entitlement(request) try: out = [] for name, info in CITIES.items(): @@ -701,8 +741,9 @@ async def list_cities(): @app.get("/api/city/{name}") -async def city_detail(name: str, force_refresh: bool = False): +async def city_detail(request: Request, name: str, force_refresh: bool = False): """Return full weather analysis for a single city.""" + _assert_entitlement(request) name = name.lower().strip().replace("-", " ") name = ALIASES.get(name, name) if name not in CITIES: @@ -819,8 +860,9 @@ def _build_city_detail_payload( @app.get("/api/history/{name}") -async def city_history(name: str): +async def city_history(request: Request, name: str): """Return historical accuracy data (DEB, mu, actuals) for a city.""" + _assert_entitlement(request) name = name.lower().strip().replace("-", " ") name = ALIASES.get(name, name) @@ -854,7 +896,8 @@ async def city_history(name: str): @app.get("/api/city/{name}/summary") -async def city_summary(name: str, force_refresh: bool = False): +async def city_summary(request: Request, name: str, force_refresh: bool = False): + _assert_entitlement(request) city = _normalize_city_or_404(name) data = _analyze(city, force_refresh=force_refresh) return _build_city_summary_payload(data) @@ -862,10 +905,12 @@ async def city_summary(name: str, force_refresh: bool = False): @app.get("/api/city/{name}/detail") async def city_detail_aggregate( + request: Request, name: str, force_refresh: bool = False, market_slug: Optional[str] = None, ): + _assert_entitlement(request) city = _normalize_city_or_404(name) data = _analyze(city, force_refresh=force_refresh) return _build_city_detail_payload(data, market_slug=market_slug)