From 160dd65a545b3d494ea65153332bc51a2407992e Mon Sep 17 00:00:00 2001
From: "2569718930@qq.com" <2569718930@qq.com>
Date: Sat, 23 May 2026 09:59:37 +0800
Subject: [PATCH] =?UTF-8?q?=E6=8E=A5=E5=85=A5=E4=B8=AD=E5=9B=BD=E6=B0=94?=
=?UTF-8?q?=E8=B1=A1=E5=B1=80=20weather.com.cn=20=E9=A2=84=E6=8A=A5?=
=?UTF-8?q?=E6=95=B0=E6=8D=AE=E4=BD=9C=E4=B8=BA=E5=A4=A9=E6=B0=94=E6=97=A5?=
=?UTF-8?q?=E6=8A=A5=E4=B8=BB=E6=95=B0=E6=8D=AE=E6=BA=90?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- 新增 CMA 7日预报页 HTML 爬虫，提取白天天气描述和最高/最低温
- 7 城优先使用 CMA 数据，Open-Meteo 仅做 fallback
- 数据来源在 prompt 中标注 weather.com.cn
---
src/utils/daily_weather_report.py | 229 +++++++++++++++++++++++-------
1 file changed, 174 insertions(+), 55 deletions(-)
diff --git a/src/utils/daily_weather_report.py b/src/utils/daily_weather_report.py
index 19c66a97..4ffe7d76 100644
--- a/src/utils/daily_weather_report.py
+++ b/src/utils/daily_weather_report.py
@@ -4,6 +4,7 @@ from __future__ import annotations
import json
import os
+import re
import threading
import time
from datetime import datetime
@@ -45,9 +46,178 @@ CITY_NAME_ZH: Dict[str, str] = {
"qingdao": "青岛",
}
+# weather.com.cn city codes
+CMA_CITY_CODES: Dict[str, str] = {
+ "beijing": "101010100",
+ "shanghai": "101020100",
+ "guangzhou": "101280101",
+ "chengdu": "101270101",
+ "chongqing": "101040100",
+ "wuhan": "101200101",
+ "qingdao": "101120201",
+}
-def _weather_desc(code: Any) -> str:
- """Translate WMO weather code to Chinese."""
+_CMA_FORECAST_URL = "http://www.weather.com.cn/weather/{code}.shtml"
+
+
+def _env_bool(name: str, default: bool) -> bool:
+ raw = os.getenv(name)
+ if raw is None:
+ return default
+ return raw.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _env_int(name: str, default: int, min_val: int = 0) -> int:
+ try:
+ return max(min_val, int(os.getenv(name, str(default))))
+ except (TypeError, ValueError):
+ return default
+
+
+def _fetch_cma_forecast(city_key: str) -> Optional[Dict[str, Any]]:
+ """Scrape today's forecast from weather.com.cn (CMA)."""
+ code = CMA_CITY_CODES.get(city_key)
+ if not code:
+ return None
+
+ url = _CMA_FORECAST_URL.format(code=code)
+ try:
+ resp = httpx.get(
+ url,
+ headers={
+ "User-Agent": (
+ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+ "AppleWebKit/537.36 (KHTML, like Gecko) "
+ "Chrome/120.0.0.0 Safari/537.36"
+ ),
+ },
+ timeout=httpx.Timeout(timeout=10.0, connect=5.0, read=10.0),
+ follow_redirects=True,
+ )
+ resp.raise_for_status()
+ html = resp.text
+ except Exception as exc:
+ logger.warning(
+ "daily_weather_report: CMA fetch failed for {}: {}", city_key, exc
+ )
+ return None
+
+ # Parse today's weather block from the 7-day forecast page.
+ # The HTML structure has entries like:
+ # 晴转多云
+ # 25℃ / 19℃
+ # We target the first occurrence (today).
+
+ weather = _extract_first(html, r']*class="wea"[^>]*>([^<]+)')
+ tem_text = _extract_first(html, r']*class="tem"[^>]*>(.+?)')
+
+ high_str: Optional[str] = None
+ low_str: Optional[str] = None
+
+ if tem_text:
+ # Patterns: 25℃ or 25°C
+ high_match = re.search(r"]*>(-?\d+)\s*(?:℃|°C|°c)?", tem_text)
+ if high_match:
+ high_str = high_match.group(1)
+ # Night temp in : 19℃
+ low_match = re.search(r"]*>(-?\d+)\s*(?:℃|°C|°c)?", tem_text)
+ if low_match:
+ low_str = low_match.group(1)
+
+ if not weather and not high_str:
+ return None
+
+ result: Dict[str, Any] = {"source": "cma"}
+ if weather:
+ result["weather"] = weather.strip()
+ if high_str:
+ try:
+ result["forecast_high"] = float(high_str)
+ except (TypeError, ValueError):
+ result["forecast_high"] = None
+ if low_str:
+ try:
+ result["forecast_low"] = float(low_str)
+ except (TypeError, ValueError):
+ result["forecast_low"] = None
+
+ return result
+
+
+def _extract_first(html: str, pattern: str) -> Optional[str]:
+ m = re.search(pattern, html, re.IGNORECASE)
+ return m.group(1) if m else None
+
+
+def _fetch_city_data(
+ collector: WeatherDataCollector, city_key: str
+) -> Optional[Dict[str, Any]]:
+ name = CITY_NAME_ZH.get(city_key, city_key)
+
+ # 1. Try CMA first for weather description + official forecast high
+ cma = _fetch_cma_forecast(city_key)
+ if cma and cma.get("weather") and cma.get("forecast_high") is not None:
+ logger.debug(
+ "daily_weather_report: {} using CMA data weather={} high={}",
+ city_key,
+ cma["weather"],
+ cma["forecast_high"],
+ )
+ return {
+ "city": city_key,
+ "name": name,
+ "weather": cma["weather"],
+ "forecast_high": cma["forecast_high"],
+ }
+
+ # 2. Fall back to Open-Meteo
+ info = CITY_REGISTRY.get(city_key)
+ if not info:
+ return None
+
+ try:
+ results = collector.fetch_all_sources(
+ city_key,
+ lat=info["lat"],
+ lon=info["lon"],
+ include_taf=False,
+ include_ensemble=False,
+ include_multi_model=False,
+ )
+ except Exception as exc:
+ logger.warning(f"daily_weather_report: OM fetch failed for {city_key}: {exc}")
+ return None
+
+ if not isinstance(results, dict):
+ return None
+
+ om = results.get("open-meteo", {}) if isinstance(results, dict) else {}
+ current = om.get("current_weather", {}) if isinstance(om, dict) else {}
+ daily = om.get("daily", {}) if isinstance(om, dict) else {}
+
+ daily_highs = daily.get("temperature_2m_max", []) or []
+ today_high = daily_highs[0] if daily_highs else None
+
+ # Use CMA weather if available, fall back to WMO code translation
+ weather = (
+ cma.get("weather")
+ if (cma and cma.get("weather"))
+ else _wmo_to_weather(current.get("weathercode"))
+ )
+ forecast_high = cma.get("forecast_high") if cma else None
+ if forecast_high is None:
+ forecast_high = today_high
+
+ return {
+ "city": city_key,
+ "name": name,
+ "weather": weather,
+ "forecast_high": forecast_high,
+ }
+
+
+def _wmo_to_weather(code: Any) -> str:
+ """Translate WMO weather code to Chinese (fallback only)."""
try:
c = int(code or 0)
except (TypeError, ValueError):
@@ -67,62 +237,11 @@ def _weather_desc(code: Any) -> str:
return "阴"
-def _env_bool(name: str, default: bool) -> bool:
- raw = os.getenv(name)
- if raw is None:
- return default
- return raw.strip().lower() in {"1", "true", "yes", "on"}
-
-
-def _env_int(name: str, default: int, min_val: int = 0) -> int:
- try:
- return max(min_val, int(os.getenv(name, str(default))))
- except (TypeError, ValueError):
- return default
-
-
-def _fetch_city_data(
- collector: WeatherDataCollector, city_key: str
-) -> Optional[Dict[str, Any]]:
- info = CITY_REGISTRY.get(city_key)
- if not info:
- return None
-
- try:
- results = collector.fetch_all_sources(
- city_key,
- lat=info["lat"],
- lon=info["lon"],
- include_taf=False,
- include_ensemble=False,
- include_multi_model=False,
- )
- except Exception as exc:
- logger.warning(f"daily_weather_report: fetch failed for {city_key}: {exc}")
- return None
-
- if not isinstance(results, dict):
- return None
-
- om = results.get("open-meteo", {}) if isinstance(results, dict) else {}
- current = om.get("current_weather", {}) if isinstance(om, dict) else {}
- daily = om.get("daily", {}) if isinstance(om, dict) else {}
-
- daily_highs = daily.get("temperature_2m_max", []) or []
- today_high = daily_highs[0] if daily_highs else None
-
- return {
- "city": city_key,
- "name": CITY_NAME_ZH.get(city_key, city_key),
- "weather": _weather_desc(current.get("weathercode")),
- "forecast_high": today_high,
- }
-
-
def _build_ai_prompt(cities_data: List[Dict[str, Any]], report_date: str) -> str:
data_json = json.dumps(cities_data, ensure_ascii=False, indent=2, default=str)
return (
- f"今天是 {report_date}。以下是今天中国主要城市的天气预报数据（JSON格式）。\n\n"
+ f"今天是 {report_date}。以下是今天中国主要城市的天气预报数据（JSON格式），"
+ "数据来自中国气象局（weather.com.cn）。\n\n"
f"{data_json}\n\n"
"请用自然亲切的中文写一段天气日报。每个城市逐行播报，格式：\n\n"
"城市名 weather，最高 forecast_high 度。一句话体感或穿衣建议。\n\n"