From 1ae9b55509775688a7e8b5d5ada4cb950c8055f4 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Tue, 17 Mar 2026 23:15:13 +0800 Subject: [PATCH] Add RP5 forecast scraping support --- .env.example | 7 +- frontend/app/subscription-help/page.tsx | 2 +- frontend/components/dashboard/DetailPanel.tsx | 4 +- .../dashboard/FutureForecastModal.tsx | 4 +- .../components/dashboard/HistoryModal.tsx | 47 ++- .../components/dashboard/PanelSections.tsx | 4 +- .../subscription/UnlockProOverlay.tsx | 4 +- frontend/lib/dashboard-types.ts | 6 + frontend/lib/dashboard-utils.ts | 127 ++++++- frontend/lib/i18n.ts | 26 +- frontend/lib/types.ts | 1 + frontend/public/legacy/index.html | 2 +- scripts/rp5_scrape_forecast.py | 51 +++ src/analysis/ai_analyzer.py | 196 ---------- src/analysis/city_query_service.py | 43 +-- src/analysis/metar_narrator.py | 307 ++++++++++++++++ src/data_collection/rp5_scraper.py | 341 ++++++++++++++++++ src/data_collection/weather_sources.py | 132 ++++++- web/app.py | 87 ++++- 19 files changed, 1113 insertions(+), 278 deletions(-) create mode 100644 scripts/rp5_scrape_forecast.py delete mode 100644 src/analysis/ai_analyzer.py create mode 100644 src/analysis/metar_narrator.py create mode 100644 src/data_collection/rp5_scraper.py diff --git a/.env.example b/.env.example index 9112a01f..7baaf75f 100644 --- a/.env.example +++ b/.env.example @@ -21,13 +21,14 @@ TELEGRAM_ALERT_MIN_SEVERITY=medium TELEGRAM_ALERT_MISPRICING_MAX_YES_BUY=0.10 TELEGRAM_ALERT_CITIES=ankara,london,paris,seoul,hong kong,shanghai,singapore,tokyo,tel aviv,toronto,buenos aires,wellington,new york,chicago,dallas,miami,atlanta,seattle,lucknow,sao paulo,munich -# AI -GROQ_API_KEY=your_groq_api_key_here - # Open-Meteo (forecast data changes ~hourly, no need to refresh more often) OPEN_METEO_CACHE_TTL_SEC=7200 OPEN_METEO_ENSEMBLE_CACHE_TTL_SEC=7200 OPEN_METEO_MULTI_MODEL_CACHE_TTL_SEC=7200 +OPEN_METEO_MULTI_MODEL_CACHE_VERSION=v2 +# RP5 public page scrape (for extra model in multi-model forecasts) +RP5_MULTI_MODEL_ENABLED=true +RP5_HTTP_TIMEOUT_SEC=20 # Proxy Setting (optional) HTTPS_PROXY=http://127.0.0.1:7890 diff --git a/frontend/app/subscription-help/page.tsx b/frontend/app/subscription-help/page.tsx index 5b0fab8f..a9c77c56 100644 --- a/frontend/app/subscription-help/page.tsx +++ b/frontend/app/subscription-help/page.tsx @@ -22,7 +22,7 @@ const TELEGRAM_GROUP_URL = String( const FAQ_ITEMS = [ { q: "Pro 包含哪些功能？", - a: "开通后可解锁：今日日内深度分析（含高温时段）、历史对账 + 未来日期分析、全平台智能气象推送。", + a: "开通后可解锁：今日日内机场报文规则分析（含高温时段）、历史对账 + 未来日期分析、全平台智能气象推送。", }, { q: "当前订阅价格是多少？", diff --git a/frontend/components/dashboard/DetailPanel.tsx b/frontend/components/dashboard/DetailPanel.tsx index 91f0b979..5544c171 100644 --- a/frontend/components/dashboard/DetailPanel.tsx +++ b/frontend/components/dashboard/DetailPanel.tsx @@ -60,7 +60,9 @@ function DetailMiniTemperatureChart({ detail }: { detail: CityDetail }) { borderWidth: 0, data: chartData.datasets.metarPoints, fill: false, - label: locale === "en-US" ? "METAR Observation" : "METAR 实测", + label: + chartData.observationLabel || + (locale === "en-US" ? "METAR Observation" : "METAR 实况"), pointHoverRadius: 6, pointRadius: 3.8, showLine: false, diff --git a/frontend/components/dashboard/FutureForecastModal.tsx b/frontend/components/dashboard/FutureForecastModal.tsx index 82d856b6..c7c651ae 100644 --- a/frontend/components/dashboard/FutureForecastModal.tsx +++ b/frontend/components/dashboard/FutureForecastModal.tsx @@ -248,7 +248,9 @@ function DailyTemperatureChart({ dateStr }: { dateStr: string }) { borderWidth: 0, data: todayChartData.datasets.metarPoints, fill: false, - label: locale === "en-US" ? "METAR Observation" : "METAR 实测", + label: + todayChartData.observationLabel || + (locale === "en-US" ? "METAR Observation" : "METAR 实况"), order: 0, pointHoverRadius: 7, pointRadius: 5, diff --git a/frontend/components/dashboard/HistoryModal.tsx b/frontend/components/dashboard/HistoryModal.tsx index df1f8ae6..c6288730 100644 --- a/frontend/components/dashboard/HistoryModal.tsx +++ b/frontend/components/dashboard/HistoryModal.tsx @@ -19,6 +19,10 @@ function HistoryChart() { const hasMgm = store.selectedCity === "ankara" && summary.mgms.some((value) => value != null); + const hasBestBaseline = + Boolean(summary.bestModelName) && + summary.bestModelName !== "MGM" && + summary.bestModelSeries.some((value) => value != null); const canvasRef = useChart(() => { const datasets: NonNullable< @@ -62,6 +66,23 @@ function HistoryChart() { }); } + if (hasBestBaseline) { + datasets.push({ + backgroundColor: "transparent", + borderColor: "#60a5fa", + borderDash: [4, 3], + borderWidth: 2, + data: summary.bestModelSeries, + label: + locale === "en-US" + ? `Best Baseline (${summary.bestModelName})` + : `最佳单模型 (${summary.bestModelName})`, + pointHoverRadius: 6, + pointRadius: 4, + tension: 0.2, + }); + } + return { data: { datasets, @@ -114,7 +135,7 @@ function HistoryChart() { }, type: "line", } satisfies ChartConfiguration<"line">; - }, [hasMgm, summary, locale]); + }, [hasBestBaseline, hasMgm, summary, locale]); if (!summary.recentData.length) return null; @@ -195,17 +216,37 @@ export function HistoryModal() { ) : ( <>
- {t("history.hitRate")} + {t("history.debHitRate")} {summary.hitRate != null ? `${summary.hitRate}%` : "--"}
- {t("history.mae")} + {t("history.debMae")} {summary.debMae != null ? `${summary.debMae}°` : "--"}
+
+ {t("history.bestModelMae")} + + {summary.bestModelMae != null + ? `${summary.bestModelMae}°${ + summary.bestModelName + ? ` (${summary.bestModelName})` + : "" + }` + : "--"} + +
+
+ {t("history.debVsBest")} + + {summary.debWinRateVsBest != null + ? `${summary.debWinRateVsBest}% (${summary.debWinDaysVsBest}/${summary.debVsBestComparableDays})` + : "--"} + +
{t("history.sample")} diff --git a/frontend/components/dashboard/PanelSections.tsx b/frontend/components/dashboard/PanelSections.tsx index 30dd94d0..2cbecc47 100644 --- a/frontend/components/dashboard/PanelSections.tsx +++ b/frontend/components/dashboard/PanelSections.tsx @@ -287,7 +287,9 @@ export function TemperatureChart() { borderWidth: 0, data: chartData.datasets.metarPoints, fill: false, - label: locale === "en-US" ? "METAR Observation" : "METAR 实测", + label: + chartData.observationLabel || + (locale === "en-US" ? "METAR Observation" : "METAR 实况"), order: 0, pointHoverRadius: 7, pointRadius: 5, diff --git a/frontend/components/subscription/UnlockProOverlay.tsx b/frontend/components/subscription/UnlockProOverlay.tsx index 6cc53198..bbd303fb 100644 --- a/frontend/components/subscription/UnlockProOverlay.tsx +++ b/frontend/components/subscription/UnlockProOverlay.tsx @@ -56,12 +56,12 @@ type UnlockProOverlayProps = { const FEATURES = { "zh-CN": [ - "今日日内深度分析（含高温时段）", + "今日日内机场报文规则分析（含高温时段）", "历史对账 + 未来日期分析", "全平台智能气象推送", ], "en-US": [ - "Intraday deep analysis with peak-time window", + "Intraday METAR rule-based analysis with peak-time window", "Historical reconciliation + future-date analysis", "Cross-platform alerts", ], diff --git a/frontend/lib/dashboard-types.ts b/frontend/lib/dashboard-types.ts index d6ed5b38..218d17a5 100644 --- a/frontend/lib/dashboard-types.ts +++ b/frontend/lib/dashboard-types.ts @@ -287,6 +287,10 @@ export interface CityDetail { time?: string; temp?: number | null; }>; + settlement_today_obs?: Array<{ + time?: string; + temp?: number | null; + }>; trend?: TrendInfo; peak?: PeakInfo; ai_analysis?: string | AiAnalysisStructured | null; @@ -300,7 +304,9 @@ export interface HistoryPoint { date: string; actual: number | null; deb: number | null; + mu?: number | null; mgm?: number | null; + forecasts?: Record; } export interface LoadingState { diff --git a/frontend/lib/dashboard-utils.ts b/frontend/lib/dashboard-utils.ts index 761292b5..fb9d0d79 100644 --- a/frontend/lib/dashboard-utils.ts +++ b/frontend/lib/dashboard-utils.ts @@ -30,6 +30,24 @@ function isEnglish(locale: Locale) { return locale === "en-US"; } +function getObservationSourceCode(detail: CityDetail): string { + return String(detail.current?.settlement_source || "metar") + .trim() + .toLowerCase(); +} + +function getObservationSourceTag(detail: CityDetail): string { + const label = String(detail.current?.settlement_source_label || "") + .trim() + .toUpperCase(); + if (label) return label; + const code = getObservationSourceCode(detail); + if (code === "hko") return "HKO"; + if (code === "cwa") return "CWA"; + if (code === "mgm") return "MGM"; + return "METAR"; +} + function normalizeCloudSummary( cloudDesc: string | null | undefined, locale: Locale, @@ -119,6 +137,7 @@ export function getWeatherSummary(detail: CityDetail, locale: Locale = "zh-CN") export function getHeroMetaItems(detail: CityDetail, locale: Locale = "zh-CN") { const current = detail.current || {}; const parts: string[] = []; + const sourceTag = getObservationSourceTag(detail); if (current.obs_time) { const ageText = @@ -127,7 +146,7 @@ export function getHeroMetaItems(detail: CityDetail, locale: Locale = "zh-CN") { ? ` (${current.obs_age_min} min ago)` : `（${current.obs_age_min} 分钟前）` : ""; - parts.push(`✈️ METAR ${current.obs_time}${ageText}`); + parts.push(`✈️ ${sourceTag} ${current.obs_time}${ageText}`); } if (current.wx_desc) { @@ -209,12 +228,18 @@ export function getTemperatureChartData( currentIndex < 0 || index >= currentIndex ? temp : null, ); - const metarPoints = new Array(times.length).fill(null); - const metarSource = detail.metar_today_obs?.length - ? detail.metar_today_obs - : detail.trend?.recent || []; + const observationTag = getObservationSourceTag(detail); + const observationCode = getObservationSourceCode(detail); + const settlementSource = + observationCode === "hko" || observationCode === "cwa"; + const observationSource = settlementSource + ? detail.settlement_today_obs || [] + : detail.metar_today_obs?.length + ? detail.metar_today_obs + : detail.trend?.recent || []; - metarSource.forEach((item) => { + const metarPoints = new Array(times.length).fill(null); + observationSource.forEach((item) => { const parts = String(item.time || "").split(":"); let hour = Number.parseInt(parts[0], 10); const minute = Number.parseInt(parts[1] || "0", 10); @@ -286,13 +311,17 @@ export function getTemperatureChartData( : "已使用 MGM 小时预报替代 DEB 曲线", ); } - if (detail.trend?.recent?.length) { - const recentText = [...detail.trend.recent] + if ((detail.trend?.recent?.length || 0) > 0 || observationSource.length > 0) { + const recentData = + observationSource.length > 0 + ? [...observationSource] + : [...(detail.trend?.recent || [])]; + const recentText = recentData .slice(0, 4) .reverse() .map((item) => `${item.temp}${detail.temp_symbol}@${item.time}`) .join(" -> "); - legendParts.push(`METAR: ${recentText}`); + legendParts.push(`${observationTag}: ${recentText}`); } return { @@ -306,6 +335,9 @@ export function getTemperatureChartData( offset, temps, }, + observationLabel: isEnglish(locale) + ? `${observationTag} Observation` + : `${observationTag} 实况`, legendText: legendParts.join(" | "), max, min, @@ -973,6 +1005,13 @@ export function getHistorySummary( history: HistoryPoint[], cityLocalDate?: string | null, ) { + const toFinite = (value: unknown): number | null => { + const numeric = Number(value); + return Number.isFinite(numeric) ? numeric : null; + }; + const isExcludedModel = (name: string) => + String(name || "").toLowerCase().includes("meteoblue"); + const cutoff = new Date(); cutoff.setHours(0, 0, 0, 0); cutoff.setDate(cutoff.getDate() - 14); @@ -992,15 +1031,65 @@ export function getHistorySummary( let hits = 0; const debErrors: number[] = []; + const modelErrors: Record = {}; + settledData.forEach((row) => { - if (row.actual != null && row.deb != null) { - debErrors.push(Math.abs(row.actual - row.deb)); - if (wuRound(row.actual) === wuRound(row.deb)) { + const actual = toFinite(row.actual); + const deb = toFinite(row.deb); + if (actual != null && deb != null) { + debErrors.push(Math.abs(actual - deb)); + if (wuRound(actual) === wuRound(deb)) { hits += 1; } } + + const forecasts = row.forecasts || {}; + Object.entries(forecasts).forEach(([modelName, modelValue]) => { + if (isExcludedModel(modelName)) return; + const mv = toFinite(modelValue); + if (actual == null || mv == null) return; + if (!modelErrors[modelName]) { + modelErrors[modelName] = []; + } + modelErrors[modelName].push(Math.abs(actual - mv)); + }); }); + const modelMaeList = Object.entries(modelErrors) + .map(([name, errors]) => ({ + mae: + errors.length > 0 + ? errors.reduce((sum, value) => sum + value, 0) / errors.length + : Number.POSITIVE_INFINITY, + model: name, + sampleCount: errors.length, + })) + .filter((row) => Number.isFinite(row.mae) && row.sampleCount > 0) + .sort((a, b) => a.mae - b.mae); + + const primaryModelMaeList = modelMaeList.filter((row) => row.sampleCount >= 2); + const bestModel = (primaryModelMaeList[0] || modelMaeList[0]) ?? null; + const bestModelName = bestModel?.model || null; + const bestModelMae = bestModel ? Number(bestModel.mae.toFixed(1)) : null; + const bestModelSeries = recentData.map((row) => + bestModelName ? toFinite(row.forecasts?.[bestModelName]) : null, + ); + + let debWinDaysVsBest = 0; + let debVsBestComparableDays = 0; + if (bestModelName) { + settledData.forEach((row) => { + const actual = toFinite(row.actual); + const deb = toFinite(row.deb); + const bestModelVal = toFinite(row.forecasts?.[bestModelName]); + if (actual == null || deb == null || bestModelVal == null) return; + debVsBestComparableDays += 1; + if (Math.abs(deb - actual) <= Math.abs(bestModelVal - actual)) { + debWinDaysVsBest += 1; + } + }); + } + return { dates: recentData.map((row) => row.date), debMae: debErrors.length @@ -1011,6 +1100,20 @@ export function getHistorySummary( ) : null, debs: recentData.map((row) => row.deb), + bestModelName, + bestModelMae, + bestModelSeries, + modelMaeRanks: modelMaeList.map((row) => ({ + model: row.model, + mae: Number(row.mae.toFixed(1)), + sampleCount: row.sampleCount, + })), + debWinDaysVsBest, + debVsBestComparableDays, + debWinRateVsBest: + debVsBestComparableDays > 0 + ? Number(((debWinDaysVsBest / debVsBestComparableDays) * 100).toFixed(0)) + : null, hitRate: debErrors.length ? Number(((hits / debErrors.length) * 100).toFixed(0)) : null, diff --git a/frontend/lib/i18n.ts b/frontend/lib/i18n.ts index bb83bdc3..0dc97b06 100644 --- a/frontend/lib/i18n.ts +++ b/frontend/lib/i18n.ts @@ -59,6 +59,11 @@ const MESSAGES: Record> = { "history.empty": "近 15 天暂无该城市历史数据", "history.hitRate": "DEB 结算胜率 (WU)", "history.mae": "DEB MAE", + "history.debHitRate": "DEB 结算胜率 (WU)", + "history.debMae": "DEB MAE", + "history.muMae": "μ MAE", + "history.bestModelMae": "最佳单模型 MAE", + "history.debVsBest": "DEB 优于最佳模型", "history.sample": "近 15 天已结算样本", "history.sampleDays": "{count} 天", @@ -86,8 +91,8 @@ const MESSAGES: Record> = { "future.judgement": "判断", "future.confidence": "置信度", "future.maxPrecip": "最大降水概率", - "future.ai": "AI 深度分析", - "future.noAi": "暂无 AI 分析，当前以结构化气象与模型数据为主。", + "future.ai": "机场报文解读", + "future.noAi": "暂无机场报文解读，当前以结构化气象与模型数据为主。", "future.weatherGov": "weather.gov 文本", "future.risk": "结算与偏差风险", "future.climate": "当地气候主要受什么影响", @@ -104,8 +109,8 @@ const MESSAGES: Record> = { "section.noProb": "暂无概率数据", "section.models": "多模型预报", "section.noModels": "暂无多模型预报", - "section.ai": "AI 深度分析", - "section.aiEmpty": "暂无 AI 分析，当前以结构化气象与模型数据为主。", + "section.ai": "机场报文解读", + "section.aiEmpty": "暂无机场报文解读，当前以结构化气象与模型数据为主。", "section.risk": "数据偏差风险", "section.noRiskProfile": "暂无风险档案", "section.airport": "机场", @@ -214,6 +219,11 @@ const MESSAGES: Record> = { "history.empty": "No historical records for this city in the last 15 days", "history.hitRate": "DEB Settlement Hit Rate (WU)", "history.mae": "DEB MAE", + "history.debHitRate": "DEB Settlement Hit Rate (WU)", + "history.debMae": "DEB MAE", + "history.muMae": "μ MAE", + "history.bestModelMae": "Best Single-model MAE", + "history.debVsBest": "DEB vs Best Model", "history.sample": "Settled Samples (Last 15 Days)", "history.sampleDays": "{count} days", @@ -241,9 +251,9 @@ const MESSAGES: Record> = { "future.judgement": "Judgement", "future.confidence": "Confidence", "future.maxPrecip": "Max Precip Probability", - "future.ai": "AI Deep Analysis", + "future.ai": "Airport METAR Narrative", "future.noAi": - "No AI analysis available. Structured meteorological and model data are used as baseline.", + "No airport bulletin narrative is available. Structured meteorological and model data are used as baseline.", "future.weatherGov": "weather.gov text", "future.risk": "Settlement & Deviation Risk", "future.climate": "What Mainly Drives Local Climate", @@ -261,9 +271,9 @@ const MESSAGES: Record> = { "section.noProb": "No probability data available", "section.models": "Multi-model Forecast", "section.noModels": "No multi-model forecast available", - "section.ai": "AI Deep Analysis", + "section.ai": "Airport METAR Narrative", "section.aiEmpty": - "No AI analysis available. Structured meteorological and model data are currently used.", + "No airport bulletin narrative is available. Structured meteorological data are currently used.", "section.risk": "Data Deviation Risk", "section.noRiskProfile": "No risk profile available", "section.airport": "Airport", diff --git a/frontend/lib/types.ts b/frontend/lib/types.ts index 1cfefa13..1025d916 100644 --- a/frontend/lib/types.ts +++ b/frontend/lib/types.ts @@ -343,6 +343,7 @@ export interface CityDetail { timeseries: { metar_recent_obs: any[]; metar_today_obs: any[]; + settlement_today_obs?: any[]; hourly: any; mgm_hourly: any[]; forecast_daily: any[]; diff --git a/frontend/public/legacy/index.html b/frontend/public/legacy/index.html index 683dee52..ea42f978 100644 --- a/frontend/public/legacy/index.html +++ b/frontend/public/legacy/index.html @@ -106,7 +106,7 @@
-

AI 深度分析

+

机场报文解读

点击城市后加载...
diff --git a/scripts/rp5_scrape_forecast.py b/scripts/rp5_scrape_forecast.py new file mode 100644 index 00000000..d8a58e54 --- /dev/null +++ b/scripts/rp5_scrape_forecast.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +import argparse +import json +import os +import sys + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) + +from src.data_collection.rp5_scraper import scrape_rp5_forecast + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Scrape public RP5 forecast page and output JSON.", + ) + parser.add_argument( + "--url", + required=True, + help="RP5 city weather URL, e.g. https://rp5.am/Weather_in_Ankara%2C_Esenboga_%28airport%29", + ) + parser.add_argument( + "--timeout", + type=int, + default=20, + help="HTTP timeout seconds (default: 20)", + ) + parser.add_argument( + "--out", + default="", + help="Optional output file path. If empty, print to stdout.", + ) + args = parser.parse_args() + + data = scrape_rp5_forecast(args.url, timeout_sec=args.timeout) + payload = json.dumps(data, ensure_ascii=False, indent=2) + + if args.out: + with open(args.out, "w", encoding="utf-8") as f: + f.write(payload + "\n") + print(f"saved: {args.out}") + else: + print(payload) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) + diff --git a/src/analysis/ai_analyzer.py b/src/analysis/ai_analyzer.py deleted file mode 100644 index 4a2d4a22..00000000 --- a/src/analysis/ai_analyzer.py +++ /dev/null @@ -1,196 +0,0 @@ -import hashlib -import os -import threading -import time -import requests -from loguru import logger - -# 主力模型 + 备用模型（当主力 500 时自动降级） -MODELS = [ - "llama-3.3-70b-versatile", - "llama-3.1-8b-instant", -] - -# ── 本地缓存 ────────────────────────────────────────────── -# key: sha1(city_name + weather_insights[:200]) -# value: {"result": str, "t": float} -_ai_cache: dict = {} -_ai_cache_lock = threading.Lock() - -# 全局 429 冷却期：触发限流后暂停一段时间内的所有 Groq 请求 -_rate_limit_until: float = 0.0 -_rate_limit_lock = threading.Lock() - - -def get_ai_analysis(weather_insights: str, city_name: str, temp_symbol: str) -> str: - """ - 通过 Groq API (LLaMA 3.3 70B) 对天气态势进行极速交易分析 - 内置自动重试 + 模型降级机制 + 本地 TTL 缓存 + 全局 429 冷却期 - """ - api_key = os.getenv("GROQ_API_KEY") - if not api_key: - logger.warning("GROQ_API_KEY 未配置，跳过 AI 分析") - return "" - - global _rate_limit_until # 必须在函数顶部声明，不能放在 with 块内 - - # ── 缓存配置 ───────────────────────────────────────── - cache_ttl = int(os.getenv("GROQ_CACHE_TTL_SEC", "1200")) # 默认 20 分钟 - rl_cooldown = int(os.getenv("GROQ_RATE_LIMIT_COOLDOWN_SEC", "600")) # 默认 10 分钟 - - # 缓存 key：城市名 + 天气摘要前 200 字符（同城市、同数据不重复打 API） - cache_raw = f"{city_name}:{weather_insights[:200]}" - cache_key = hashlib.sha1(cache_raw.encode("utf-8")).hexdigest()[:16] - - now = time.time() - - # ── 命中缓存则直接返回 ──────────────────────────────── - with _ai_cache_lock: - cached = _ai_cache.get(cache_key) - if cached and now - cached["t"] < cache_ttl: - logger.debug(f"Groq AI cache hit city={city_name} age={int(now - cached['t'])}s") - return cached["result"] - - # ── 全局 429 冷却期检查 ─────────────────────────────── - with _rate_limit_lock: - if now < _rate_limit_until: - remaining = int(_rate_limit_until - now) - logger.warning(f"Groq 冷却期中，还需等待 {remaining}s，跳过本次请求") - # 如果有旧缓存，返回旧结果（过期但总比没有好） - with _ai_cache_lock: - stale = _ai_cache.get(cache_key) - if stale: - return stale["result"] + "\n（AI 分析来自缓存，数据可能略旧）" - return "\n⚠️ Groq AI 限流中，请稍后再试" - - url = "https://api.groq.com/openai/v1/chat/completions" - headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} - - prompt = f""" -你是一个专业的天气衍生品交易员。你的任务是分析当前气象实况，判断今日实测最高温的结算落点。 -结算基准为合约指定源（通常为 METAR 口径整数四舍五入），以下用"结算值"代替具体源名称。 - -请综合以下提供的【{city_name}】气象特征进行深度推理。 - -【气象特征与事实】 -{weather_insights} - -【分析框架】（按此顺序逐项检查，前置项可约束后置项的结论） - -P0 **预报失准检测**： - - 若数据含"🚨 预报崩盘"或"⚠️ 预报差距"标记，判定预报失准。 - - 失准等级：轻(偏差2-3°) / 中(3-5°) / 重(>5°)。 - - 但"失准"≠"已定局"：还需检查近2报斜率是否≤0 且 风向/云量不支持二次抬升，才能判定结算锁定。 - - 若斜率仍>0或有暖平流迹象，应注明"预报偏高但仍有上行空间"。 - -P1 **实况节奏**： - - 近2-4条METAR的温度走势：连涨/持平/回落？ - - 连续2报创新高 → 升温未止；连续2报未创新高且斜率≤0 → 偏死盘。 - - 升温出现在低辐射时段 → 可能有多因子叠加（平流/混合层/热岛），不可单因子归因。 - -P1.5 **高温时段约束（强制）**： - - 先读取输入里的峰值窗口状态（before / in_window / past）。 - - 若状态是 before（尚未进入峰值窗口）： - - 禁止给出“已确认底线/已锁定/大概率到顶”。 - - 盘口结论必须是“上沿待确认”或“仍有上行变数”。 - - 若状态是 in_window： - - 允许谨慎偏空，但仍不能直接“锁定”；需强调继续观察后续报文是否再创新高。 - - 只有状态是 past 且同时满足“连续未创新高+回落/抑制因子”，才可使用“已确认底线/锁定”。 - -P2 **阻碍因子**（需结合城市特性判断）： - - 降水已出现(非trace) → 强压温。 - - 高湿度+厚云层持续2报以上 → 压温可能有效，但阈值因城市（海洋型 vs 大陆型）而异，不可套用固定数值。 - - 若仅单因子（如仅多云），不足以断定"升温受限"。 - -P3 **概率与一致性校验**： - - 参考结算概率分布，与 P1 实况做一致性检查。 - - 若概率分布与实况趋势矛盾，以实况为准并说明偏离原因。 - - 概率可辅助判断边界进位风险（如 X.5 线附近）。 - -P4 **预报背景**（最低优先级）： - - 可参考 DEB/预报做上沿空间评估。 - - 当实测已显著偏离预报时，禁止继续引用预报值作为目标。 - -【输出要求】 -1. 正常场景控制在 300 字左右；异常场景（预报失准/极端走势）可扩展到 450 字。 -2. 严格按照以下 HTML 格式输出: - -🤖 Groq AI 决策 -- 🎲 盘口: [给出结算判断。用"已确认底线 X{temp_symbol}"表示下限确定；用"上沿待确认，关注 Y{temp_symbol}"表示仍有变数。若预报严重失准，注明失准等级和原因。禁止在升温未止时用"锁定"。] -- 💡 逻辑: [3-5 句深度分析。含具体数值。预报失准时重点分析偏差成因。正常时分析实测与预报的动态博弈。] -- 🎯 置信度: [1-10]/10 - -3. **禁止输出分析框架本身**。不要输出 P0/P1/P2/P3/P4 的分析过程或标题。只输出上方三行格式，不要多余内容。 -4. 若输入出现“尚未进入峰值窗口 / 距最热时段开始还有 / 状态=before”，盘口行必须包含“上沿待确认”或同义表达，且逻辑行必须明确“时间窗未到，不能锁定”。 -""" - - # Use proxy if configured - proxies = {} - proxy_url = os.getenv("HTTPS_PROXY") or os.getenv("HTTP_PROXY") - if proxy_url: - proxies = {"http": proxy_url, "https": proxy_url} - - for model in MODELS: - for attempt in range(2): # 每个模型最多重试 2 次 - try: - payload = { - "model": model, - "messages": [ - { - "role": "system", - "content": "你是不讲废话、只看数据的专业气象分析师。", - }, - {"role": "user", "content": prompt}, - ], - "temperature": 0.5, - "max_tokens": 400, - } - - response = requests.post( - url, json=payload, headers=headers, timeout=15, proxies=proxies - ) - response.raise_for_status() - - result = response.json() - content = result["choices"][0]["message"]["content"].strip() - - if model != MODELS[0]: - logger.info(f"Groq 降级到备用模型 {model} 成功") - # ── 写入缓存 ───────────────────────────────── - with _ai_cache_lock: - _ai_cache[cache_key] = {"result": content, "t": time.time()} - return content - - except requests.exceptions.HTTPError as e: - status = e.response.status_code if e.response is not None else 0 - error_body = "" - try: - error_body = e.response.text - except: - pass - logger.warning( - f"Groq {model} 失败 (HTTP {status}): {error_body}. 尝试下一个..." - ) - if status == 429: - # 触发限流：设置全局冷却期，后续请求不再尝试 - with _rate_limit_lock: - _rate_limit_until = time.time() + rl_cooldown - logger.warning(f"Groq 触发限流，设置 {rl_cooldown}s 全局冷却期") - break # 不再尝试其他模型，直接走 stale cache 逻辑 - if status in (500, 502, 503) and attempt == 0: - time.sleep(1.5) - continue - else: - break # 换下一个模型 - except Exception as e: - logger.warning(f"Groq {model} 异常: {str(e)}，尝试下一个模型...") - break - - logger.error("所有 Groq 模型均不可用") - # ── 有旧缓存则返回旧结果 ────────────────────────────── - with _ai_cache_lock: - stale = _ai_cache.get(cache_key) - if stale: - logger.info(f"Groq 不可用，返回旧缓存结果 city={city_name} age={int(time.time()-stale['t'])}s") - return stale["result"] + "\n（⚠️ AI 分析来自上次缓存）" - return "" diff --git a/src/analysis/city_query_service.py b/src/analysis/city_query_service.py index 4b5e21b8..20f630c2 100644 --- a/src/analysis/city_query_service.py +++ b/src/analysis/city_query_service.py @@ -3,8 +3,7 @@ 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.metar_narrator import describe_metar_report 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 @@ -543,33 +542,29 @@ def build_city_query_report( 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) + 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}") + metar_narrative = describe_metar_report( + raw_metar=str(primary_current.get("raw_metar") or metar_current.get("raw_metar") or ""), + temp_symbol=temp_symbol, + fallback={ + "icao": metar.get("icao"), + "station_name": metar.get("station_name"), + "temp": cur_temp, + "wind_speed_kt": _sf(primary_current.get("wind_speed_kt")), + "wind_dir": _sf(primary_current.get("wind_dir")), + "altimeter": _sf(primary_current.get("altimeter")), + "wx_desc": primary_current.get("wx_desc"), + "clouds": primary_current.get("clouds", []), + }, + ) + if metar_narrative: + msg_lines.append("\n🛰️ 机场报文解读:") + msg_lines.append(metar_narrative) msg_lines.append(f"\n💸 本次消耗 {city_query_cost} 积分。") return "\n".join(msg_lines) diff --git a/src/analysis/metar_narrator.py b/src/analysis/metar_narrator.py new file mode 100644 index 00000000..3228f80d --- /dev/null +++ b/src/analysis/metar_narrator.py @@ -0,0 +1,307 @@ +from __future__ import annotations + +import re +from typing import Any, Dict, Iterable, Optional, Tuple + +_WIND_TOKEN_RE = re.compile(r"^(VRB|\d{3})(\d{2,3})(G(\d{2,3}))?KT$") +_WIND_VAR_RE = re.compile(r"^(\d{3})V(\d{3})$") +_TEMP_DEW_RE = re.compile(r"^(M?\d{2}|//)/(M?\d{2}|//)$") +_PRESSURE_Q_RE = re.compile(r"^Q(\d{4})$") +_PRESSURE_A_RE = re.compile(r"^A(\d{4})$") +_CLOUD_RE = re.compile(r"^(FEW|SCT|BKN|OVC|VV|SKC|CLR|NSC)(\d{3})?$") +_WX_CODE_RE = re.compile(r"^[-+]?([A-Z]{2,})$") + +_WIND_DIR_16 = [ + "北方", + "北偏东北方向", + "东北方向", + "东偏东北方向", + "东方", + "东偏东南方向", + "东南方向", + "南偏东南方向", + "南方", + "南偏西南方向", + "西南方向", + "西偏西南方向", + "西方", + "西偏西北方向", + "西北方向", + "北偏西北方向", +] + +_CLOUD_DESC = { + "CLR": "晴空", + "SKC": "晴空", + "NSC": "晴空", + "FEW": "少云", + "SCT": "多变云天", + "BKN": "多云", + "OVC": "阴天", + "VV": "低云压顶", +} + +_WEATHER_DESC = { + "RA": "有降雨", + "DZ": "有毛毛雨", + "SN": "有降雪", + "TS": "有雷暴", + "TSRA": "有雷阵雨", + "FG": "有雾", + "BR": "有轻雾", + "HZ": "有霾", + "SHRA": "有阵雨", + "FZRA": "有冻雨", +} + + +def _safe_float(value: Any) -> Optional[float]: + if value is None: + return None + try: + return float(value) + except Exception: + return None + + +def _parse_metar_signed_temp(raw: str) -> Optional[float]: + if raw in {"", "//"}: + return None + sign = -1.0 if raw.startswith("M") else 1.0 + value = raw[1:] if raw.startswith("M") else raw + try: + return sign * float(int(value)) + except Exception: + return None + + +def _pick_station(tokens: Iterable[str]) -> str: + token_list = list(tokens) + if not token_list: + return "" + first = token_list[0] + if first in {"METAR", "SPECI"} and len(token_list) >= 2: + first = token_list[1] + if re.fullmatch(r"[A-Z]{4}", first): + return first + return "" + + +def _direction_desc(direction_deg: float) -> str: + idx = int(((direction_deg % 360) + 11.25) // 22.5) % 16 + return _WIND_DIR_16[idx] + + +def _wind_level_desc(ms: float) -> str: + if ms < 0.3: + return "静风" + if ms < 1.6: + return "软风" + if ms < 3.4: + return "轻风" + if ms < 5.5: + return "微风" + if ms < 8.0: + return "和风" + if ms < 10.8: + return "清劲风" + if ms < 13.9: + return "强风" + if ms < 17.2: + return "疾风" + return "大风" + + +def _format_temp(temp: float, symbol: str) -> str: + rounded = round(temp, 1) + if abs(rounded - round(rounded)) < 0.05: + body = str(int(round(rounded))) + else: + body = f"{rounded:.1f}" + if rounded > 0: + body = f"+{body}" + return f"{body}{symbol}" + + +def _format_ms(ms: float) -> str: + rounded = round(ms, 1) + if abs(rounded - round(rounded)) < 0.05: + return str(int(round(rounded))) + return f"{rounded:.1f}" + + +def _pressure_desc(hpa: float) -> str: + hp = round(hpa) + if hp < 1000: + return f"偏低气压({hp} hPa)" + if hp > 1030: + return f"偏高气压({hp} hPa)" + return f"在正常范围内的大气压({hp} hPa)" + + +def _best_cloud_code(tokens: Iterable[str], fallback_clouds: Any) -> str: + best = "" + rank = {"CLR": 0, "SKC": 0, "NSC": 0, "FEW": 1, "SCT": 2, "BKN": 3, "OVC": 4, "VV": 5} + best_rank = -1 + + for token in tokens: + m = _CLOUD_RE.match(token) + if not m: + continue + code = m.group(1) + score = rank.get(code, -1) + if score > best_rank: + best_rank = score + best = code + + if best: + return best + + if isinstance(fallback_clouds, list): + for row in fallback_clouds: + if not isinstance(row, dict): + continue + code = str(row.get("cover") or "").upper().strip() + if not code: + continue + score = rank.get(code, -1) + if score > best_rank: + best_rank = score + best = code + return best + + +def describe_metar_report( + raw_metar: str, + temp_symbol: str = "°C", + fallback: Optional[Dict[str, Any]] = None, +) -> str: + """ + Convert METAR bulletin into deterministic human-language description. + Style is inspired by rp5 bulletin narration: temperature, cloud, pressure, wind. + """ + fallback = fallback or {} + raw = str(raw_metar or "").strip().upper() + tokens = [token for token in raw.split() if token] + if not tokens and not fallback: + return "" + + station = _pick_station(tokens) or str(fallback.get("icao") or "").upper().strip() + station_name = str(fallback.get("station_name") or "").strip() + + wind_dir = _safe_float(fallback.get("wind_dir")) + wind_kt = _safe_float(fallback.get("wind_speed_kt")) + wind_var: Optional[Tuple[float, float]] = None + for token in tokens: + m = _WIND_TOKEN_RE.match(token) + if not m: + continue + dir_token = m.group(1) + spd_token = m.group(2) + if dir_token != "VRB": + wind_dir = _safe_float(dir_token) + wind_kt = _safe_float(spd_token) + break + for token in tokens: + mv = _WIND_VAR_RE.match(token) + if mv: + left = _safe_float(mv.group(1)) + right = _safe_float(mv.group(2)) + if left is not None and right is not None: + wind_var = (left, right) + break + + temp_c = None + for token in tokens: + tm = _TEMP_DEW_RE.match(token) + if tm: + temp_c = _parse_metar_signed_temp(tm.group(1)) + break + fallback_temp = _safe_float(fallback.get("temp")) + if temp_c is None and fallback_temp is not None: + temp_c = fallback_temp if temp_symbol == "°C" else (fallback_temp - 32.0) * 5.0 / 9.0 + + pressure_hpa = None + for token in tokens: + qm = _PRESSURE_Q_RE.match(token) + if qm: + pressure_hpa = _safe_float(qm.group(1)) + break + am = _PRESSURE_A_RE.match(token) + if am: + inhg = _safe_float(am.group(1)) + if inhg is not None: + pressure_hpa = (inhg / 100.0) * 33.8639 + break + if pressure_hpa is None: + altim = _safe_float(fallback.get("altimeter")) + if altim is not None: + pressure_hpa = altim * 33.8639 if altim < 200 else altim + + cloud_code = _best_cloud_code(tokens, fallback.get("clouds")) + cloud_desc = _CLOUD_DESC.get(cloud_code, "") + + wx_desc = "" + wx_raw = str(fallback.get("wx_desc") or "").upper().strip() + if wx_raw: + for key, value in _WEATHER_DESC.items(): + if key in wx_raw: + wx_desc = value + break + if not wx_desc: + for token in tokens: + if not _WX_CODE_RE.match(token): + continue + for key, value in _WEATHER_DESC.items(): + if key in token: + wx_desc = value + break + if wx_desc: + break + + station_label = "" + if station: + station_label = f"{station} 机场" + elif station_name: + station_label = station_name + else: + station_label = "机场" + + parts = [] + if temp_c is not None: + display_temp = temp_c if temp_symbol == "°C" else temp_c * 9.0 / 5.0 + 32.0 + parts.append(f"{station_label} {_format_temp(display_temp, temp_symbol)}") + else: + parts.append(station_label) + + if cloud_desc: + parts.append(cloud_desc) + + if pressure_hpa is not None: + parts.append(_pressure_desc(pressure_hpa)) + + if wind_kt is not None: + wind_ms = float(wind_kt) * 0.514444 + wind_level = _wind_level_desc(wind_ms) + if wind_dir is not None: + wind_sentence = ( + f"从{_direction_desc(wind_dir)}吹来的{wind_level}" + f"({_format_ms(wind_ms)}米/秒)" + ) + else: + wind_sentence = f"{wind_level}({_format_ms(wind_ms)}米/秒)" + if wind_var is not None: + left, right = wind_var + wind_sentence += ( + f"，风向在{_direction_desc(left)}与{_direction_desc(right)}之间摆动" + ) + parts.append(wind_sentence) + + if wx_desc: + parts.append(wx_desc) + + if "NOSIG" in tokens: + parts.append("短时无显著变化") + + text = "，".join([p for p in parts if str(p or "").strip()]) + return f"{text}。" if text else "" diff --git a/src/data_collection/rp5_scraper.py b/src/data_collection/rp5_scraper.py new file mode 100644 index 00000000..c9d38ca1 --- /dev/null +++ b/src/data_collection/rp5_scraper.py @@ -0,0 +1,341 @@ +from __future__ import annotations + +import re +import unicodedata +from datetime import datetime, timezone +from html import unescape +from typing import Any, Dict, List, Optional +from urllib.parse import quote, unquote, urljoin + +import requests + +DEFAULT_TIMEOUT_SEC = 20 +DEFAULT_UA = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0.0.0 Safari/537.36" +) + +_TABLE_BY_ID_RE = r'(?is)]*\bid=["\']{table_id}["\'][^>]*>.*?' +_ROW_RE = re.compile(r"(?is)]*>.*?") +_CELL_RE = re.compile(r"(?is)<(td|th)\b([^>]*)>(.*?)") +_TITLE_RE = re.compile(r"(?is)(.*?)") +_COLSPAN_RE = re.compile(r'(?is)\bcolspan\s*=\s*["\']?(\d+)') +_SPACE_RE = re.compile(r"\s+") +_TAG_RE = re.compile(r"(?is)<[^>]+>") +_NUM_RE = re.compile(r"[+-]?\d+(?:\.\d+)?") + +RP5_BASE_URL = "https://rp5.am" +RP5_CITY_URL_OVERRIDES: Dict[str, str] = { + # City pages that do not resolve correctly via simple /Weather_in_ + "london": "https://rp5.am/Weather_in_London%2C_St._James%27s_Park", + "paris": "https://rp5.am/Weather_in_Paris,_France", + "toronto": "https://rp5.am/Weather_in_Toronto,_Canada", + "new york": "https://rp5.am/Weather_in_New_York,_USA", + "warsaw": "https://rp5.am/Weather_in_Warsaw%2C_Okecie_%28airport%29", + "dallas": "https://rp5.am/Weather_in_Dallas%2C_Love_Field_%28airport%29", + "miami": "https://rp5.am/Weather_in_Miami_%28airport%29%2C_Florida", + "atlanta": "https://rp5.am/Weather_in_Atlanta%2C_Georgia", + "sao paulo": "https://rp5.am/Weather_in_Sao_Paulo", + "hong kong": "https://rp5.am/Weather_in_Hong_Kong_%28airport%29", + "singapore": "https://rp5.am/Weather_in_Singapore_%28airport%29", + "madrid": "https://rp5.am/Weather_in_Madrid,_Barajas_(airport)", +} + + +def _strip_html(raw: str) -> str: + text = re.sub(r"(?is)<(script|style)\b[^>]*>.*?", " ", raw) + text = _TAG_RE.sub(" ", text) + text = unescape(text).replace("\xa0", " ") + return _SPACE_RE.sub(" ", text).strip() + + +def _extract_title(html: str) -> str: + match = _TITLE_RE.search(html) + if not match: + return "" + return _strip_html(match.group(1)) + + +def _extract_table_html(html: str, table_id: str) -> str: + pattern = re.compile(_TABLE_BY_ID_RE.format(table_id=re.escape(table_id))) + match = pattern.search(html) + return match.group(0) if match else "" + + +def _parse_cells(row_html: str) -> List[Dict[str, Any]]: + cells: List[Dict[str, Any]] = [] + for match in _CELL_RE.finditer(row_html): + attrs = match.group(2) or "" + inner = match.group(3) or "" + colspan_match = _COLSPAN_RE.search(attrs) + colspan = int(colspan_match.group(1)) if colspan_match else 1 + text = _strip_html(inner) + cells.append({"text": text, "colspan": max(1, colspan)}) + return cells + + +def _parse_table(table_html: str) -> List[List[Dict[str, Any]]]: + rows: List[List[Dict[str, Any]]] = [] + for row_match in _ROW_RE.finditer(table_html): + row_cells = _parse_cells(row_match.group(0)) + if row_cells: + rows.append(row_cells) + return rows + + +def _expand_row_values(cells: List[Dict[str, Any]]) -> List[str]: + expanded: List[str] = [] + for cell in cells: + expanded.extend([cell.get("text", "")] * int(cell.get("colspan") or 1)) + return expanded + + +def _to_float_first(text: str) -> Optional[float]: + if not text: + return None + values = _NUM_RE.findall(text) + if not values: + return None + try: + return float(values[0]) + except Exception: + return None + + +def _to_float_last(text: str) -> Optional[float]: + if not text: + return None + values = _NUM_RE.findall(text) + if not values: + return None + try: + return float(values[-1]) + except Exception: + return None + + +def _find_row(rows: List[List[Dict[str, Any]]], prefix: str) -> List[Dict[str, Any]]: + low = prefix.lower() + for row in rows: + if not row: + continue + label = str(row[0].get("text") or "").strip().lower() + if label.startswith(low): + return row + return [] + + +def _build_timeseries(rows: List[List[Dict[str, Any]]]) -> List[Dict[str, Any]]: + if len(rows) < 2: + return [] + + header_row = rows[0] + local_row = _find_row(rows, "Local time") + if not header_row or not local_row: + return [] + + day_slots = _expand_row_values(header_row) + time_slots = _expand_row_values(local_row[1:]) + slot_count = min(len(day_slots), len(time_slots)) + if slot_count <= 0: + return [] + + day_slots = day_slots[:slot_count] + time_slots = time_slots[:slot_count] + + temp_row = _find_row(rows, "Temperature") + pressure_row = _find_row(rows, "Pressure") + wind_speed_row = _find_row(rows, "Wind: speed") + wind_dir_row = _find_row(rows, "direction") + humidity_row = _find_row(rows, "Humidity") + precip_row = _find_row(rows, "Precipitation") + + temp_values = _expand_row_values(temp_row[1:])[:slot_count] if temp_row else [""] * slot_count + pressure_values = _expand_row_values(pressure_row[1:])[:slot_count] if pressure_row else [""] * slot_count + wind_speed_values = _expand_row_values(wind_speed_row[1:])[:slot_count] if wind_speed_row else [""] * slot_count + wind_dir_values = _expand_row_values(wind_dir_row[1:])[:slot_count] if wind_dir_row else [""] * slot_count + humidity_values = _expand_row_values(humidity_row[1:])[:slot_count] if humidity_row else [""] * slot_count + precip_values = _expand_row_values(precip_row[1:])[:slot_count] if precip_row else [""] * slot_count + + out: List[Dict[str, Any]] = [] + for idx in range(slot_count): + entry = { + "day_label": day_slots[idx], + "local_time": time_slots[idx], + "temp_c": _to_float_first(temp_values[idx]), + "pressure_hpa": _to_float_last(pressure_values[idx]), + "wind_mps": _to_float_first(wind_speed_values[idx]), + "wind_dir": (wind_dir_values[idx] or "").strip() or None, + "humidity_pct": _to_float_first(humidity_values[idx]), + "precip_mm": _to_float_first(precip_values[idx]), + } + out.append(entry) + return out + + +def _extract_summary(html: str) -> str: + block = re.search( + r"(?is)]*>\s*Today we expect.*?", + html, + ) + if block: + return _strip_html(block.group(0)) + + idx = html.find("Today we expect") + if idx < 0: + return "" + chunk = html[idx : idx + 1400] + text = _strip_html(chunk) + match = re.search(r"Today we expect.*?(?:Tomorrow:.*?$)", text, re.I) + if match: + return match.group(0).strip() + return "" + + +def _extract_weather_links(html: str) -> List[str]: + links = re.findall(r'(?is)href=["\'](/Weather_in_[^"\']+)["\']', html) + seen = set() + out = [] + for link in links: + if link in seen: + continue + seen.add(link) + out.append(link) + return out + + +def _normalize_for_match(value: str) -> str: + text = unquote(str(value or "")).lower() + text = unicodedata.normalize("NFKD", text) + text = "".join(ch for ch in text if not unicodedata.combining(ch)) + text = re.sub(r"[^a-z0-9]+", " ", text) + return re.sub(r"\s+", " ", text).strip() + + +def _score_weather_link(link: str, city_hint: str) -> int: + low_raw = unquote(str(link or "")).lower() + low = _normalize_for_match(link) + city_norm = _normalize_for_match(city_hint) + words = [w for w in city_norm.split() if len(w) >= 3] + matched = sum(1 for w in words if w in low) + exact_phrase = bool(city_norm and city_norm in low) + starts_with_city = bool(city_norm and low.startswith(f"weather in {city_norm}")) + tokens = set(low.split()) + has_airport = "airport" in tokens + + if words and matched == 0: + return -300 + + score = 0 + score += matched * 30 + if exact_phrase: + score += 20 + if starts_with_city: + score += 20 + if has_airport: + score += 35 if matched > 0 else -60 + if "_region" in low_raw: + score -= 20 + if "_district" in low_raw: + score -= 25 + if "_county" in low_raw or "_province" in low_raw: + score -= 15 + for bad in [ + "weather_in_the_world", + "weather_in_russia", + "weather_in_ukraine", + "weather_in_belarus", + "weather_in_lithuania", + ]: + if bad in low_raw: + score -= 120 + return score + + +def build_rp5_city_url_candidates(city_name: str, city_key: str = "") -> List[str]: + key = str(city_key or city_name).strip().lower() + name = str(city_name or city_key).strip() + out: List[str] = [] + + if key in RP5_CITY_URL_OVERRIDES: + out.append(RP5_CITY_URL_OVERRIDES[key]) + + if name: + tokens = [name.replace(" ", "_"), name] + for token in tokens: + out.append(f"{RP5_BASE_URL}/Weather_in_{quote(token)}") + + # Keep order, deduplicate. + seen = set() + unique: List[str] = [] + for item in out: + if item in seen: + continue + seen.add(item) + unique.append(item) + return unique + + +def scrape_rp5_forecast( + url: str, + timeout_sec: int = DEFAULT_TIMEOUT_SEC, + city_hint: str = "", + max_hops: int = 2, +) -> Dict[str, Any]: + sess = requests.Session() + headers = { + "User-Agent": DEFAULT_UA, + "Accept-Language": "en-US,en;q=0.9", + } + visited = set() + current_url = url + last_payload: Dict[str, Any] = {} + + for _ in range(max(1, int(max_hops))): + if current_url in visited: + break + visited.add(current_url) + + resp = sess.get(current_url, timeout=timeout_sec, headers=headers) + resp.raise_for_status() + html = resp.text + + table_html = _extract_table_html(html, "forecastTable") + rows = _parse_table(table_html) if table_html else [] + points = _build_timeseries(rows) + + last_payload = { + "source": "rp5_html", + "url": url, + "resolved_url": resp.url, + "fetched_at_utc": datetime.now(timezone.utc).isoformat(), + "title": _extract_title(html), + "summary": _extract_summary(html), + "points": points, + } + if points: + return last_payload + + links = _extract_weather_links(html) + if not links: + break + ranked = sorted( + links, + key=lambda item: _score_weather_link(item, city_hint), + reverse=True, + ) + best = ranked[0] + if _score_weather_link(best, city_hint) < 1: + break + current_url = urljoin(resp.url, best) + + return last_payload or { + "source": "rp5_html", + "url": url, + "resolved_url": url, + "fetched_at_utc": datetime.now(timezone.utc).isoformat(), + "title": "", + "summary": "", + "points": [], + } diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index edb1fb69..d77fef78 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -7,6 +7,10 @@ import threading from typing import Optional, Dict, List, Any from datetime import datetime, timedelta, timezone from loguru import logger +from src.data_collection.rp5_scraper import ( + build_rp5_city_url_candidates, + scrape_rp5_forecast, +) class WeatherDataCollector: @@ -67,6 +71,13 @@ class WeatherDataCollector: self.open_meteo_multi_model_cache_ttl_sec = int( os.getenv("OPEN_METEO_MULTI_MODEL_CACHE_TTL_SEC", "900") ) + self.multi_model_cache_version = str( + os.getenv("OPEN_METEO_MULTI_MODEL_CACHE_VERSION", "v2") + ).strip() or "v2" + self.rp5_multi_model_enabled = str( + os.getenv("RP5_MULTI_MODEL_ENABLED", "true") + ).strip().lower() in {"1", "true", "yes", "on"} + self.rp5_timeout_sec = max(5, int(os.getenv("RP5_HTTP_TIMEOUT_SEC", "20"))) self._open_meteo_cache: Dict[str, Dict] = {} self._ensemble_cache: Dict[str, Dict] = {} self._multi_model_cache: Dict[str, Dict] = {} @@ -1685,10 +1696,109 @@ class WeatherDataCollector: return fallback return None + def _merge_rp5_into_daily_forecasts( + self, + city: str, + dates: List[str], + daily_forecasts: Dict[str, Dict[str, float]], + use_fahrenheit: bool, + ) -> Optional[Dict[str, Any]]: + """ + Scrape RP5 public forecast page and merge as an extra model line ("RP5"). + """ + if not self.rp5_multi_model_enabled: + return None + city_key = str(city or "").strip().lower() + if not city_key: + return None + + city_meta = self.CITY_REGISTRY.get(city_key, {}) + city_name = str(city_meta.get("name") or city).strip() + candidates = build_rp5_city_url_candidates(city_name=city_name, city_key=city_key) + if not candidates: + return None + + payload: Dict[str, Any] = {} + points: List[Dict[str, Any]] = [] + for url in candidates: + try: + probe = scrape_rp5_forecast( + url=url, + timeout_sec=self.rp5_timeout_sec, + city_hint=city_name, + max_hops=3, + ) + except Exception as exc: + logger.debug(f"RP5 scrape failed city={city_name} url={url}: {exc}") + continue + p = probe.get("points") + if isinstance(p, list) and p: + payload = probe + points = p + break + + if not points: + return None + + by_label: Dict[str, float] = {} + ordered_labels: List[str] = [] + for row in points: + if not isinstance(row, dict): + continue + label = str(row.get("day_label") or "").strip() + if not label: + continue + temp_c = row.get("temp_c") + if temp_c is None: + continue + try: + temp_c_val = float(temp_c) + except Exception: + continue + if label not in by_label: + by_label[label] = temp_c_val + ordered_labels.append(label) + elif temp_c_val > by_label[label]: + by_label[label] = temp_c_val + + if not ordered_labels: + return None + + mapped_values: List[float] = [by_label[k] for k in ordered_labels] + mapped_count = min(len(dates), len(mapped_values)) + for idx in range(mapped_count): + date_key = dates[idx] + value_c = mapped_values[idx] + value = value_c * 9 / 5 + 32 if use_fahrenheit else value_c + day_bucket = daily_forecasts.setdefault(date_key, {}) + day_bucket["RP5"] = round(value, 1) + + if not dates and mapped_values: + # Fallback: if Open-Meteo dates missing unexpectedly, create today bucket. + today_key = datetime.now(timezone.utc).strftime("%Y-%m-%d") + value_c = mapped_values[0] + value = value_c * 9 / 5 + 32 if use_fahrenheit else value_c + daily_forecasts.setdefault(today_key, {})["RP5"] = round(value, 1) + + logger.info( + "RP5 merged city={} mapped_days={} url={}", + city_name, + mapped_count if dates else 1, + payload.get("resolved_url") or payload.get("url") or "", + ) + return { + "source": "rp5_html", + "url": payload.get("url"), + "resolved_url": payload.get("resolved_url"), + "summary": payload.get("summary"), + "mapped_labels": ordered_labels[:mapped_count] if dates else ordered_labels[:1], + } + def fetch_multi_model( self, lat: float, lon: float, + city: str = "", use_fahrenheit: bool = False, ) -> Optional[Dict]: """ @@ -1704,9 +1814,10 @@ class WeatherDataCollector: 返回 3 天的预报数据，支持今日+明日共识分析 """ + cache_city = str(city or "").strip().lower() cache_key = ( - f"{round(float(lat), 4)}:{round(float(lon), 4)}:" - f"{'f' if use_fahrenheit else 'c'}" + f"{round(float(lat), 4)}:{round(float(lon), 4)}:{cache_city}:" + f"{'f' if use_fahrenheit else 'c'}:{self.multi_model_cache_version}" ) self._maybe_reload_open_meteo_disk_cache() now_ts = time.time() @@ -1777,6 +1888,13 @@ class WeatherDataCollector: if day_data: daily_forecasts[date_str] = day_data + rp5_meta = self._merge_rp5_into_daily_forecasts( + city=city, + dates=dates, + daily_forecasts=daily_forecasts, + use_fahrenheit=use_fahrenheit, + ) + if not daily_forecasts: logger.warning("Multi-model: 无有效模型数据") return None @@ -1795,6 +1913,7 @@ class WeatherDataCollector: "forecasts": forecasts, # 今天 {"ECMWF": 12.3, "GFS": 11.8, ...} (向后兼容) "daily_forecasts": daily_forecasts, # 按天 {"2026-02-23": {...}, "2026-02-24": {...}} "dates": dates, + "rp5": rp5_meta or {}, "unit": "fahrenheit" if use_fahrenheit else "celsius", } with self._multi_model_cache_lock: @@ -2034,7 +2153,10 @@ class WeatherDataCollector: unit = "f" if use_fahrenheit else "c" open_meteo_key = f"{base}:14:{unit}" ensemble_key = f"{base}:{unit}" - multi_model_key = ensemble_key + cache_city = str(city or "").strip().lower() + multi_model_key = ( + f"{base}:{cache_city}:{unit}:{self.multi_model_cache_version}" + ) with self._open_meteo_cache_lock: self._open_meteo_cache.pop(open_meteo_key, None) @@ -2180,7 +2302,7 @@ class WeatherDataCollector: # 多模型预报 (所有城市通用，用于共识评分) mm_data = self.fetch_multi_model( - lat, lon, use_fahrenheit=use_fahrenheit + lat, lon, city=city, use_fahrenheit=use_fahrenheit ) if mm_data: results["multi_model"] = mm_data @@ -2228,7 +2350,7 @@ class WeatherDataCollector: if ens_data: results["ensemble"] = ens_data mm_data = self.fetch_multi_model( - lat, lon, use_fahrenheit=use_fahrenheit + lat, lon, city=city, use_fahrenheit=use_fahrenheit ) if mm_data: results["multi_model"] = mm_data diff --git a/web/app.py b/web/app.py index 6fc7cec4..67cc2d57 100644 --- a/web/app.py +++ b/web/app.py @@ -36,6 +36,7 @@ from src.auth.supabase_entitlement import ( SUPABASE_ENTITLEMENT, extract_bearer_token, ) +from src.analysis.metar_narrator import describe_metar_report from src.database.db_manager import DBManager from src.payments import PAYMENT_CHECKOUT, PaymentCheckoutError @@ -431,6 +432,29 @@ def _analyze(city: str, force_refresh: bool = False) -> Dict[str, Any]: except Exception: obs_time_str = str(obs_t)[:16] + settlement_today_obs = [] + if use_settlement_current: + if obs_time_str and cur_temp is not None: + settlement_today_obs.append({"time": obs_time_str, "temp": cur_temp}) + if ( + max_temp_time + and max_so_far is not None + and str(max_temp_time) != str(obs_time_str) + ): + settlement_today_obs.append({"time": str(max_temp_time), "temp": max_so_far}) + + metar_today_obs_payload = ( + [] + if use_settlement_current + else [ + {"time": t, "temp": v} + for t, v in (metar.get("today_obs", []) if metar else []) + ] + ) + metar_recent_obs_payload = ( + [] if use_settlement_current else (metar.get("recent_obs", []) if metar else []) + ) + # ── 3. Local time parsing ── local_time_full = om.get("current", {}).get("local_time", "") local_hour, local_minute = 12, 0 @@ -610,13 +634,12 @@ def _analyze(city: str, force_refresh: bool = False) -> Dict[str, Any]: # This single call replaces the duplicate probability engine, dead market # detection, forecast bust grading, and AI context building. from src.analysis.trend_engine import analyze_weather_trend as _trend_analyze, calculate_prob_distribution - from src.analysis.ai_analyzer import get_ai_analysis probabilities = [] mu = None ai_text = "" try: - _, ai_context, sd = _trend_analyze(raw, sym, city) + _, _ai_context, sd = _trend_analyze(raw, sym, city) # Use structured data from shared engine mu = sd.get("mu") @@ -631,17 +654,23 @@ def _analyze(city: str, force_refresh: bool = False) -> Dict[str, Any]: deb_val = sd["deb_prediction"] deb_weights = sd.get("deb_weights", "") - # Append multi-model divergence for AI - if current_forecasts and ai_context: - mm_str = " | ".join( - [f"{k}:{v}{sym}" for k, v in current_forecasts.items() if v] - ) - ai_context += f"\n模型分歧: {mm_str}" - - if ai_context: - ai_text = get_ai_analysis(ai_context, city, sym) except Exception as e: - logger.warning(f"Analysis/AI skipped for {city}: {e}") + logger.warning(f"Structured analysis skipped for {city}: {e}") + + ai_text = describe_metar_report( + raw_metar=str(primary_current.get("raw_metar") or mc.get("raw_metar") or ""), + temp_symbol=sym, + fallback={ + "icao": metar.get("icao"), + "station_name": metar.get("station_name"), + "temp": cur_temp, + "wind_speed_kt": _sf(primary_current.get("wind_speed_kt")), + "wind_dir": _sf(primary_current.get("wind_dir")), + "altimeter": _sf(primary_current.get("altimeter")), + "wx_desc": primary_current.get("wx_desc"), + "clouds": primary_current.get("clouds", []) or mc.get("clouds", []), + }, + ) # ── 12. Hourly data (today only, for chart) ── today_hourly: Dict[str, list] = {"times": [], "temps": [], "radiation": []} @@ -911,11 +940,9 @@ def _analyze(city: str, force_refresh: bool = False) -> Dict[str, Any]: }, "hourly": today_hourly, "hourly_next_48h": next_48h_hourly, - "metar_today_obs": [ - {"time": t, "temp": v} - for t, v in (metar.get("today_obs", []) if metar else []) - ], - "metar_recent_obs": metar.get("recent_obs", []) if metar else [], + "metar_today_obs": metar_today_obs_payload, + "metar_recent_obs": metar_recent_obs_payload, + "settlement_today_obs": settlement_today_obs, "ai_analysis": ai_text, "updated_at": datetime.now(timezone.utc).isoformat(), } @@ -1138,6 +1165,7 @@ def _build_city_detail_payload( "timeseries": { "metar_recent_obs": data.get("metar_recent_obs") or [], "metar_today_obs": data.get("metar_today_obs") or [], + "settlement_today_obs": data.get("settlement_today_obs") or [], "hourly": data.get("hourly") or {}, "mgm_hourly": (data.get("mgm") or {}).get("hourly", []), "forecast_daily": (data.get("forecast") or {}).get("daily", []), @@ -1170,7 +1198,12 @@ async def city_history(request: Request, name: str): data = load_history(history_file) if name not in data: - return {"history": []} + source = str(CITIES.get(name, {}).get("settlement_source") or "metar").strip().lower() + return { + "history": [], + "settlement_source": source, + "settlement_source_label": SETTLEMENT_SOURCE_LABELS.get(source, source.upper()), + } city_data = data[name] out = [] @@ -1178,7 +1211,15 @@ async def city_history(request: Request, name: str): act = rec.get("actual_high") deb = rec.get("deb_prediction") mu = rec.get("mu") - mgm = rec.get("forecasts", {}).get("MGM") + forecasts_raw = rec.get("forecasts", {}) or {} + forecasts = {} + if isinstance(forecasts_raw, dict): + for model_name, model_value in forecasts_raw.items(): + if _is_excluded_model_name(str(model_name)): + continue + fv = _sf(model_value) + forecasts[str(model_name)] = fv if fv is not None else None + mgm = forecasts.get("MGM") # Only return items where we have at least an actual or a prediction out.append({ @@ -1187,8 +1228,14 @@ async def city_history(request: Request, name: str): "deb": float(deb) if deb is not None else None, "mu": float(mu) if mu is not None else None, "mgm": float(mgm) if mgm is not None else None, + "forecasts": forecasts, }) - return {"history": out} + source = str(CITIES.get(name, {}).get("settlement_source") or "metar").strip().lower() + return { + "history": out, + "settlement_source": source, + "settlement_source_label": SETTLEMENT_SOURCE_LABELS.get(source, source.upper()), + } @app.get("/api/auth/me")