接入中国气象局 weather.com.cn 预报数据作为天气日报主数据源

- 新增 CMA 7日预报页 HTML 爬虫，提取白天天气描述和最高/最低温
- 7 城优先使用 CMA 数据，Open-Meteo 仅做 fallback
- 数据来源在 prompt 中标注 weather.com.cn

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:
+_CMA_FORECAST_URL = "http://www.weather.com.cn/weather/{code}.shtml"
-    """Translate WMO weather code to Chinese."""
+
+
+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:
+    #   <p class="wea">晴转多云</p>
+    #   <p class="tem"><span>25℃</span> / <i>19℃</i></p>
+    # We target the first occurrence (today).
+
+    weather = _extract_first(html, r'<p[^>]*class="wea"[^>]*>([^<]+)</p>')
+    tem_text = _extract_first(html, r'<p[^>]*class="tem"[^>]*>(.+?)</p>')
+
+    high_str: Optional[str] = None
+    low_str: Optional[str] = None
+
+    if tem_text:
+        # Patterns: <span>25℃</span> or <span>25°C</span>
+        high_match = re.search(r"<span[^>]*>(-?\d+)\s*(?:℃|°C|°c)?</span>", tem_text)
+        if high_match:
+            high_str = high_match.group(1)
+        # Night temp in <i>: <i>19℃</i>
+        low_match = re.search(r"<i[^>]*>(-?\d+)\s*(?:℃|°C|°c)?</i>", 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"