feat: 增加全国气象预警对账补偿

CHANGELOG.md

@@ -5,6 +5,21 @@
 <!-- markdownlint-disable MD041 -->
 # ChangeLog
 
+# Unreleased
+
+## ✨ New Features (新功能)
+
+- 新增默认关闭的中国天气网全国活动预警对账补偿，可在不替换 FAN Studio 实时流的前提下补齐可能遗漏的中间气象预警。
+- 气象预警事件 ID 去重缓存扩展为 24 小时、最多 4096 条，避免 FAN 与 HTTP 对账因到达顺序不同而重复推送。
+
+## 🐛 Bug Fixes (修复)
+
+- 按预警语义规范化气象预警类型编码，避免上游紧凑类型码被误当作国家标准 `11B` 图标码而展示错误灾种图片。
+
+## 🧪 Tests (测试)
+
+- 新增配置边界、首次基线、详情失败隔离与重试、跨传输去重、URL 安全校验、运行时取消、公开端点请求头和预警编码规范化测试。
+
 # 2026/07/12 v1.6.0-beta.1
 
 ## 🚀 What's Changed

README.md

@@ -117,7 +117,7 @@
 - **日本气象厅地震情报** (P2P / Wolfx) - 详细地震情报。
 - **USGS地震测定** (FAN Studio) - 美国地质调查局地震信息。
 - **Global Quake服务器** - 全球地震测站实时计算推送，精度有限。
-- **中国气象局气象预警** (FAN Studio) - 气象灾害预警。
+- **中国气象局气象预警** (FAN Studio + 可选 China Weather 对账) - 实时推送配合全国活动预警低频补偿。
 - **自然资源部海啸预警中心** (FAN Studio) - 海啸预警信息。
 - **日本气象厅海啸预报** (P2P) - 日本海啸预报信息。
 
@@ -419,6 +419,7 @@ https://obs.nmefc.cn/Warning/TsunamiAdvice/202607111826_2_file/Earthquake_Pos.jp
 | 日本气象厅地震情报 | Wolfx | Info | ✅ |
 | 美国地质调查局 | FAN Studio | Info | ✅ |
 | 中国气象局 | FAN Studio | Weather | ✅ |
+| 中国天气网活动预警索引 | China Weather（可选对账） | Weather | 🧪 |
 | 中国海啸预警中心 | FAN Studio | Tsunami | ✅ |
 | 日本气象厅海啸预报 | P2P | Tsunami | 🧪 |
 
@@ -440,7 +441,7 @@ https://obs.nmefc.cn/Warning/TsunamiAdvice/202607111826_2_file/Earthquake_Pos.jp
 - **台湾中央气象署地震报告 (CWA)**：约 0.5-2s
 - **日本气象厅地震情报 (JMA)**：约 0.5-2s
 - **美国地质调查局地震测定 (USGS)**：约 0.5-15s
-- **中国气象局气象预警 (CMA)**：约 3-15 分钟
+- **中国气象局气象预警 (CMA)**：约 3-15 分钟；可选对账源按配置轮询，仅用于补偿覆盖率，不保证实时性
   - 开启地图瓦片渲染 / GQ 卡片还将额外增加约 1-2.5s 的延迟。(地图瓦片仅影响地震情报延迟，不阻塞地震预警类事件的推送)
   - 不同数据源间的**推送**延迟一般为 Fan = P2P ＜ Wolfx ＜ Global Quake
 
@@ -541,6 +542,29 @@ https://obs.nmefc.cn/Warning/TsunamiAdvice/202607111826_2_file/Earthquake_Pos.jp
 - **原理**: 连接到 Global Quake 服务器。这些数据是由全球数千个测站通过算法实时计算得出的。
 - **特点**: 在偏远地区或国际海域，由于官方机构反应时间较长，GQ 往往能最先提供初步数据，但震级和位置可能随报数更新而有较大波动。
 
+#### 🔹 中国天气网全国活动预警对账 (`weather_alarm_fallback`)
+
+这是一个默认关闭的低频补偿源，用于对账 FAN Studio 气象实时流可能遗漏的中间事件，不替代 FAN Studio WebSocket。
+
+- **启用 (`enabled`)**: 仅在 FAN Studio 的中国气象预警子源同时启用时生效。
+- **首次快照**: 首次成功读取只建立基线，不会补发当前全部活动预警。
+- **后续处理**: 只请求新出现的官方预警详情，并复用现有关键词、等级、会话、展示、统计与发送规则。
+- **轮询间隔 (`poll_interval_seconds`)**: 默认 `180` 秒，范围 `60 - 3600` 秒。
+- **请求超时 (`request_timeout_seconds`)**: 默认 `15` 秒，范围 `3 - 30` 秒。
+- **详情并发 (`detail_concurrency`)**: 默认 `4`，范围 `1 - 8`。
+- **去重**: FAN 与 HTTP 对账使用相同逻辑事件 ID，并通过 24 小时、最多 4096 条的有界缓存抑制重复推送。
+
+```json
+{
+  "weather_alarm_fallback": {
+    "enabled": false,
+    "poll_interval_seconds": 180,
+    "request_timeout_seconds": 15,
+    "detail_concurrency": 4
+  }
+}
+```
+
 ---
 
 ### 📍 3. 本地预估烈度 (`local_monitoring`)

_conf_schema.json

@@ -187,6 +187,52 @@
       }
     }
   },
+  "weather_alarm_fallback": {
+    "description": "可选的全国 China Weather 气象预警校准源",
+    "type": "object",
+    "hint": "可选的全国 China Weather 校准源；首次成功快照仅建立基线，不发送通知；后续事件复用现有气象预警规则。默认关闭。",
+    "items": {
+      "enabled": {
+        "description": "启用 China Weather 全国气象预警校准",
+        "type": "bool",
+        "hint": "仅在 FAN Studio 中国气象局气象预警源已启用时运行；首次成功快照仅建立基线。",
+        "default": false
+      },
+      "poll_interval_seconds": {
+        "description": "校准轮询间隔",
+        "type": "int",
+        "hint": "单位：秒；范围 60 至 3600。",
+        "default": 180,
+        "slider": {
+          "min": 60,
+          "max": 3600,
+          "step": 10
+        }
+      },
+      "request_timeout_seconds": {
+        "description": "单次请求超时",
+        "type": "int",
+        "hint": "单位：秒；范围 3 至 30。",
+        "default": 15,
+        "slider": {
+          "min": 3,
+          "max": 30,
+          "step": 1
+        }
+      },
+      "detail_concurrency": {
+        "description": "详情请求并发数",
+        "type": "int",
+        "hint": "每轮新增预警详情的最大并发请求数；范围 1 至 8。后续事件复用现有气象预警规则。",
+        "default": 4,
+        "slider": {
+          "min": 1,
+          "max": 8,
+          "step": 1
+        }
+      }
+    }
+  },
   "local_monitoring": {
     "description": "📍 本地烈度监控配置",
     "type": "object",
@@ -919,4 +965,4 @@
       }
     }
   }
-}
+}

core/app/runtime/disaster_service_runtime.py

@@ -9,9 +9,26 @@
 import asyncio
 import json
 
+import aiohttp
 from astrbot.api import logger
 
 from ...services.query.source_runtime_query_service import SourceRuntimeQueryService
+from ...services.weather import (
+    ChinaWeatherReconciler,
+    WeatherFallbackConfig,
+    resolve_fallback_config,
+)
+
+
+_CHINA_WEATHER_INDEX_URL = "https://product.weather.com.cn/alarm/grepalarm_cn.php"
+_CHINA_WEATHER_DETAIL_BASE = "https://product.weather.com.cn/alarm/webdata/"
+_CHINA_WEATHER_HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+        "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36"
+    ),
+    "Referer": "http://www.weather.com.cn/alarm/",
+}
 
 
 class DisasterServiceRuntimeService:
@@ -86,6 +103,8 @@ async def _connect_with_timeout(name, uri, info):
     async def start_scheduled_http_fetch(self) -> None:
         """启动定时 HTTP 数据获取。"""
 
+        fallback_config = resolve_fallback_config(self.service.config)
+
         async def fetch_wolfx_data():
             # Wolfx 列表属于低频补偿型数据：
             # 一方面用于补齐地震列表缓存，另一方面在对应数据源启用时也可转成事件，
@@ -163,6 +182,98 @@ async def fetch_wolfx_data():
         task = asyncio.create_task(fetch_wolfx_data(), name="dw_http_fetch_wolfx")
         self.service.scheduled_tasks.append(task)
 
+        if fallback_config.enabled and self._source_runtime_query.is_source_enabled(
+            "china_weather_fanstudio"
+        ):
+            china_weather_task = asyncio.create_task(
+                self._run_china_weather_loop(fallback_config),
+                name="dw_http_fetch_china_weather",
+            )
+            self.service.scheduled_tasks.append(china_weather_task)
+
+    async def _run_china_weather_loop(
+        self,
+        fallback_config: WeatherFallbackConfig,
+        *,
+        session_factory=None,
+        sleep=None,
+    ) -> None:
+        """Run the independent China Weather reconciliation polling loop."""
+        session_factory = session_factory or aiohttp.ClientSession
+        sleep = sleep or asyncio.sleep
+        reconciler = ChinaWeatherReconciler(
+            detail_concurrency=fallback_config.detail_concurrency
+        )
+        timeout = aiohttp.ClientTimeout(total=fallback_config.request_timeout_seconds)
+        async with session_factory(
+            timeout=timeout,
+            headers=_CHINA_WEATHER_HEADERS,
+        ) as session:
+
+            async def fetch_text(url: str) -> str:
+                async with session.get(url) as response:
+                    response.raise_for_status()
+                    return await response.text()
+
+            async def fetch_detail(detail_path: str) -> str:
+                return await fetch_text(f"{_CHINA_WEATHER_DETAIL_BASE}{detail_path}")
+
+            while self.service.running:
+                try:
+                    await self._reconcile_china_weather_cycle(
+                        reconciler, fetch_text, fetch_detail
+                    )
+                except asyncio.CancelledError:
+                    raise
+                except Exception as e:
+                    logger.warning(
+                        "[灾害预警] China Weather 索引请求失败，保留上次有效快照: "
+                        f"{type(e).__name__}: {e}"
+                    )
+
+                await sleep(fallback_config.poll_interval_seconds)
+
+    async def _reconcile_china_weather_cycle(
+        self, reconciler, fetch_text, fetch_detail
+    ) -> None:
+        """Fetch and process one China Weather reconciliation snapshot."""
+        index_script = await fetch_text(_CHINA_WEATHER_INDEX_URL)
+        result = await reconciler.reconcile(
+            index_script,
+            fetch_detail,
+            self._dispatch_china_weather_payload,
+        )
+        if not result.index_valid:
+            logger.warning("[灾害预警] China Weather 校准索引无效，保留上次有效快照")
+        elif (
+            result.new_count
+            or result.failed_identifiers
+            or result.consumed_error_identifiers
+        ):
+            logger.debug(
+                "[灾害预警] China Weather 校准完成: "
+                f"索引 {result.reference_count}, 新增 {result.new_count}, "
+                f"已处理 {result.consumed_count}, 失败 {len(result.failed_identifiers)}"
+            )
+        for identifier in result.failed_identifiers:
+            logger.warning(
+                f"[灾害预警] China Weather 详情处理失败，保留重试资格: {identifier}"
+            )
+        for identifier in result.consumed_error_identifiers:
+            logger.warning(
+                "[灾害预警] China Weather 已进入发送链但调用异常，"
+                f"按最多一次策略不自动重试: {identifier}"
+            )
+
+    async def _dispatch_china_weather_payload(self, payload: dict[str, object]) -> None:
+        """Hand one payload to the authoritative existing event pipeline once."""
+        event = self.service.parse_event(
+            "china_weather_fanstudio",
+            json.dumps(payload, ensure_ascii=False),
+        )
+        if event is not None:
+            await self.service._handle_disaster_event(event)
+
     async def start_cleanup_task(self) -> None:
         """启动清理任务。"""
 

core/parsers/weather_parser.py

@@ -5,26 +5,53 @@
 
 from __future__ import annotations
 
-from collections import deque
 from datetime import datetime
+import re
 from typing import Any
 
 from ...utils.plugin_logger import plugin_logger
 from ..domain.event_identity import EventIdentity
 from ..domain.event_models import EventEnvelope, WeatherEvent
 from ..domain.event_payload import SourcePayload
+from ..services.weather import BoundedTTLSet, normalize_weather_warning_code
 from ..sources.source_catalog import get_source_entry
 from .base_parser import BaseParser
 
 
+_WEATHER_ISSUE_TIME_PATTERNS = (
+    re.compile(r"^\d{14}_(?P<issued_at>\d{12}(?:\d{2})?)$"),
+    re.compile(r"^\d{6}-(?P<issued_at>\d{14})-11B\d{4}$"),
+)
+
+
+def _extract_issue_time(
+    event_id: object, effective_time: datetime | None
+) -> datetime | None:
+    """Extract the encoded issue timestamp, falling back to effective time."""
+    event_id_text = str(event_id or "")
+    for pattern in _WEATHER_ISSUE_TIME_PATTERNS:
+        match = pattern.fullmatch(event_id_text)
+        if match is None:
+            continue
+        issued_at = match.group("issued_at")
+        timestamp_format = "%Y%m%d%H%M%S" if len(issued_at) == 14 else "%Y%m%d%H%M"
+        try:
+            return datetime.strptime(issued_at, timestamp_format)
+        except ValueError:
+            break
+    return effective_time
+
+
 class WeatherAlarmParser(BaseParser):
     """中国气象局气象预警解析器。"""
 
     def __init__(self, message_logger=None):
         """初始化气象预警解析器与短期重复记录缓存。"""
         super().__init__("china_weather_fanstudio", message_logger)
-        # 用双端队列在内存中缓存最近 10 条已处理过气象预警标识，用于快速防重过滤
-        self._processed_weather_ids = deque(maxlen=10)
+        self._processed_weather_ids = BoundedTTLSet(
+            ttl_seconds=86_400,
+            max_entries=4_096,
+        )
 
     def _parse_data(self, data: dict[str, Any]) -> EventEnvelope | None:
         """解析中国气象局气象预警数据。"""
@@ -64,25 +91,7 @@ def _parse_data(self, data: dict[str, Any]) -> EventEnvelope | None:
             effective_time = self._parse_datetime(msg_data.get("effective", ""))
 
             # 预警发布时间优先尝试从标识尾部编码中提取，失败时回退到生效时间。
-            issue_time = None
-            id_str = msg_data.get("id", "")
-            if "_" in id_str:
-                time_part = id_str.split("_")[-1]
-                if len(time_part) >= 12:
-                    try:
-                        year = int(time_part[0:4])
-                        month = int(time_part[4:6])
-                        day = int(time_part[6:8])
-                        hour = int(time_part[8:10])
-                        minute = int(time_part[10:12])
-                        second = int(time_part[12:14]) if len(time_part) >= 14 else 0
-                        issue_time = datetime(year, month, day, hour, minute, second)
-                    except (ValueError, IndexError):
-                        issue_time = effective_time
-                else:
-                    issue_time = effective_time
-            else:
-                issue_time = effective_time
+            issue_time = _extract_issue_time(msg_data.get("id"), effective_time)
 
             headline = msg_data.get("headline", "")
             title = msg_data.get("title", "") or headline
@@ -99,7 +108,7 @@ def _parse_data(self, data: dict[str, Any]) -> EventEnvelope | None:
             source_entry = get_source_entry(self.source_id)
 
             # 多重回退以解析气象编码
-            weather_code = str(
+            source_weather_code = str(
                 msg_data.get("weather_type")
                 or msg_data.get("weatherType")
                 or msg_data.get("alertCode")
@@ -108,6 +117,18 @@ def _parse_data(self, data: dict[str, Any]) -> EventEnvelope | None:
                 or msg_data.get("type")
                 or ""
             ).strip()
+            weather_code = normalize_weather_warning_code(
+                source_weather_code,
+                title,
+                headline,
+                description,
+            )
+            if weather_code != source_weather_code:
+                plugin_logger.warning(
+                    f"[灾害预警] 已纠正冲突的气象预警编码: "
+                    f"{source_weather_code} -> {weather_code}",
+                    is_event_linked=True,
+                )
 
             # 整合元数据
             metadata = {
@@ -172,9 +193,9 @@ def _parse_data(self, data: dict[str, Any]) -> EventEnvelope | None:
                 metadata=metadata,
             )
 
-            # 加入防重去噪队列中
-            if envelope.id:
-                self._processed_weather_ids.append(envelope.id)
+            # 加入防重去噪缓存中
+            if event_id:
+                self._processed_weather_ids.add(event_id)
 
             plugin_logger.info(
                 f"[灾害预警] 气象预警解析成功: {domain_event.title or domain_event.headline}, 生效时间: {issue_time}",