Conversation
在 SOURCE_CATALOG 中注册 OpenQuakeAPI CMA 气象预警数据源， priority=3（高于 Fan=1），与 Fan 不做跨源去重，可双推。 payload 直接透传 CMA 预警地图 API 字段。
/ws/all 聚合推送的 source 字段新增 cma 映射到 china_weather_openquake， 更新文档注释说明当前已接入 Global Quake 与 CMA 气象预警。
source_compat 新增 openquake_cma/cma_weather/cma → china_weather_openquake 别名映射；_DISPLAY_MAP 区分 Fan/OQ 气象预警展示名（- Fan / - OQ）。
WeatherAlarmParser 构造时按 source_id 区分数据源，各源维护独立 短窗去重队列。新增 _extract_realtime_payload 方法解包 OpenQuakeAPI 外层结构（source/type/action/payload），仅处理 type=weather 且 action=new 的事件，非 RealtimeEvent 结构回退到 FAN Studio 扁平载荷。 source_family 改从 source_entry 动态读取。
- parser_registry: 新增 weather_alarm_parser 细化分派，按 source_id 注入正确的解析器实例，避免硬编码 china_weather_fanstudio - base_parser: critical_fields 新增 china_weather_openquake 关键字段校验
OpenQuakeAPI 聚合连接的 JSON 消息，从仅放行 earthquake 业务类型 扩展为同时放行 weather（CMA 气象）业务类型，其余仍过滤。
新增 /ws/cma 路径说明、weather/new 事件列表、payload 数据示例 及字段说明表（id/headline/effective/description/longitude/latitude/type/title）。 source 列表表新增 cma 行。
新增 WeatherAggregationService：按会话维度独立缓冲气象预警事件， 时间窗口到期后合并推送。支持合并转发（Comp.Nodes），不支持时降级 为逐条文本发送并启用限流（按颜色级别排序，超出配额丢弃低级别）。 红色预警可配置立即触发推送，缓冲区为空时直接走常规推送流。
- event_pipeline: 按会话级配置判断是否聚合，启用聚合的会话进入 缓冲区，未启用的走常规推送；新增 _flush_weather_buffer 回调， 为每条预警构建含图标的完整消息链后发送，先通过规则链复核 - disaster_service: 装配 WeatherAggregationService 并注入流水线 - lifecycle: 停机前 flush_all 推送缓冲区中尚未发送的事件，避免丢失
- 新增 weather_aggregation 配置组（enabled/time_window_seconds/ max_batch_size/flush_on_red/rate_limit_enabled/rate_limit_max_messages/ rate_limit_window_seconds） - 新增 event_stream_log_level 配置组（all/weather_alarm/global_quake/ earthquake/tsunami/typhoon 独立日志级别覆盖） - OpenQuakeAPI 分组新增 china_weather_alarm 子源开关 - FAN Studio 气象预警默认关闭（由 OQ 高优先级源接管） - 全部 hint 统一补充句号，push_frequency_control hint 措辞修正
- 新增 weather_aggregation 配置项的布尔/数值范围校验 （time_window 60-3600s, max_batch 1-50, rate_limit_max 1-20 等） - OpenQuakeAPI 分组新增 china_weather_alarm 子源开关校验与旧配置回填
PluginLogger 新增 event_stream 参数与 _resolve_stream_action 方法， 支持按事件流类型（weather_alarm/global_quake/earthquake/tsunami/typhoon） 独立控制日志级别，优先级高于 log_mode 总开关。all 全局覆盖优先级最高， 其次按事件流类型匹配，最后回退到总开关。info/warning 方法签名 新增 event_stream 可选参数。
CENC/CEAEEW/烈度速报/FSSN CMT/JMA 地震与预警/S-Net/CWA 报告与预警 等地震类解析器的 plugin_logger.info 调用统一补充 event_stream="earthquake" 标签，支持细粒度日志级别控制。
Global Quake/USGS/ShakeAlert 解析器补充 event_stream="global_quake"/"earthquake"； 海啸解析器补充 "tsunami"；台风解析器补充 "typhoon"；
- source_message_router: 新增 _resolve_stream_by_source_id 静态方法， 按数据源标识解析事件流标签 - event_deduplication_service: 新增 _resolve_event_stream_by_source_id， 确保去重器报数日志与解析器日志流标签一致 - cenc/cwa_eew_fusion_service: 融合策略日志补充 "earthquake" 标签 - push_flow_handler: 新增 _resolve_event_stream 按事件类型解析流标签 - EQSC 轮询服务日志补充对应事件流标签
新增 weather_alarm_code_map 模块，将 CMA 预警地图 API 的 p 编码 （如 p0002003）转换为 Fan Studio 图标接口兼容的 11B 编码 （如 11B03_yellow）。支持 p 编码通用规则解析、特殊短码跳过、 标题文本兜底映射，提供 resolve_weather_icon_code 与 build_weather_icon_url 公共接口。
- 通过 resolve_weather_icon_code 统一解析图标码（p 编码经映射表转换）， 替代直接拼接 URL - 识别气象预警图标接口返回的伪图片（HTML 错误页伪装为 image/png）， 稳定触发的失败降级为 DEBUG 避免控制台刷屏 - 图标下载失败/无可用回退图标的日志从 WARNING 降级为 DEBUG
- weather_constants: 新增 extract_final_weather_color 函数，优先从 "升级为/降级为/变更为"关键词后的文本提取最终颜色 - weather_presenter: 颜色提示图标改用 extract_final_weather_color， 避免显示变更前旧颜色；标题补充"中国气象局"来源标识 - weather_query_service: detect_weather_color 同步改用统一提取逻辑； extract_weather_org 新增气象台/气象局/气象站/气象中心/气象分局 机构名正则，覆盖升级/降级类预警 headline 无"发布"关键词的场景
- startup_silence_coordinator: GateState 新增 fetch_started_at/fetching 字段，超时从实际开始抓取时间起算，避免初始化耗时被误计入； fetching=True 时不触发超时放行，避免首轮抓取耗时较长时被误放行 - 各轮询服务统一调用 note_poll_fetch_started/completed，抽取 _notify_silence_fetch_completed 消除重复代码，确保失败路径通知协调器 - 台风静默期调用 _handle_disaster_event 播种去重指纹，避免重载后 首次推送因去重缓存为空而被误放行（重复推送根因）
_evaluate_dual_threshold_filter 新增 skip_primary/skip_secondary 参数， 被跳过的检查项不参与 any/all 组合判定。震度模式下检测 is_assumption/magnitude_is_placeholder 元数据标记，跳过震级阈值检查， 仅按震度过滤，避免 all 模式误杀 PLUM 报或 any 模式产生误导性失败描述。
connect/sock_connect 从 8s 放宽到 15s（国内到 equake.top 的 TCP+TLS 握手在网络抖动时容易超过 8 秒）；sock_read 从 15s 放宽到 30s （台风列表接口含 20 个台风完整轨迹，响应体较大）；total 从 25s 放宽到 45s（connect+sock_read 上限）。
- 从 connections 视图提取各连接分组的配置启用状态，避免将未启用的 连接（如 FAN Studio 烈度速报）误显示为"异常" - EQSC/S-Net 子源改由 build_sub_source_status 按 config_key 正确归组， 移除硬编码注入逻辑，避免 key 名与 grouped_sources 漂移导致计数错误 - 数据源分组之间增加空行提升可读性 - 补充 FSSN CMT、china_weather_alarm(OpenQuakeAPI) 等缺失的数据源展示
正常区域搜索结果 >1 条时优先尝试合并转发分批发送（max_nodes 从 8 提升到 10），失败时回退文本概要发送。补充 delivery_mode 遥测字段 区分 forward_batches/text_blocks 两种投递模式。
- ConnectionsGrid: OpenQuakeAPI 分组新增 china_weather_alarm/ china_weather_openquake 连接展示 - formatters: normalizeSourceName 新增 openquake_cma/cma_weather/cma 别名归一化；formatSourceName 区分 Fan/OQ 气象预警展示名
地域汇总分支此前完全不读取 detailed_jma_intensity，导致 detailed=false 时仍输出全部震度档地域，与开关描述"仅显示最大震度区域"的语义矛盾。 现于地域汇总输出循环前根据 detailed_jma_intensity 截取 sorted_scales， false 时仅保留最大震度档，使该开关在 region 开启/关闭两种模式下均生效， 实现两个开关的正交双维度控制。
Contributor
Reviewer's Guide集成来自 OpenQuakeAPI 的 CMA 气象预警，新增按会话维度的天气聚合/转发推送服务，引入基于事件流的细粒度日志级别控制，优化启动静默协调和 EQSC/S-Net 轮询行为，并修复若干展示、解析、查询与配置相关问题。 CMA 气象事件聚合与推送的时序图sequenceDiagram
actor OpenQuakeAPI
participant SourceMessageRouter
participant WeatherAlarmParser
participant EventPipeline
participant WeatherAggregationService
participant MessageManager
participant MessageBuildService
participant SessionSender
OpenQuakeAPI->>SourceMessageRouter: WebSocket message (source=cma, type=weather)
SourceMessageRouter->>WeatherAlarmParser: _parse_data(data)
WeatherAlarmParser->>WeatherAlarmParser: _extract_realtime_payload(data)
WeatherAlarmParser-->>SourceMessageRouter: EventEnvelope(weather_alarm)
SourceMessageRouter->>EventPipeline: handle(event)
loop per session
EventPipeline->>WeatherAggregationService: should_aggregate(event, session, runtime_config)
alt should_aggregate == False
EventPipeline->>MessageManager: push_event(event, target_sessions)
else should_aggregate == True
EventPipeline->>WeatherAggregationService: buffer event
end
end
rect rgb(230,230,250)
note over WeatherAggregationService: later, when flush conditions met
WeatherAggregationService->>EventPipeline: _flush_weather_buffer(session, entries, config, mode="forward")
loop each WeatherBufferEntry
EventPipeline->>MessageManager: evaluate_push_decision(entry.event, runtime_config)
alt decision.accepted
EventPipeline->>MessageBuildService: build_message_async(entry.event, runtime_config)
MessageBuildService-->>EventPipeline: MessageChain
else decision.rejected
EventPipeline-->>EventPipeline: skip entry
end
end
alt mode == "forward"
EventPipeline->>SessionSender: send(session, MessageChain(Comp.Nodes))
else mode == "single"
loop each built message
EventPipeline->>SessionSender: send(session, message)
end
end
end
File-Level Changes
Possibly linked issues
Tips and commandsInteracting with Sourcery
Customizing Your Experience访问你的 dashboard 可以：
Getting HelpOriginal review guide in EnglishReviewer's GuideIntegrates CMA weather alerts from OpenQuakeAPI, adds a per-session weather aggregation/forwarding push service, introduces fine-grained log level control by event stream, improves startup silence coordination and EQSC/S-Net polling behavior, and fixes several presentation, parser, query, and configuration issues. Sequence diagram for CMA weather event aggregation and pushsequenceDiagram
actor OpenQuakeAPI
participant SourceMessageRouter
participant WeatherAlarmParser
participant EventPipeline
participant WeatherAggregationService
participant MessageManager
participant MessageBuildService
participant SessionSender
OpenQuakeAPI->>SourceMessageRouter: WebSocket message (source=cma, type=weather)
SourceMessageRouter->>WeatherAlarmParser: _parse_data(data)
WeatherAlarmParser->>WeatherAlarmParser: _extract_realtime_payload(data)
WeatherAlarmParser-->>SourceMessageRouter: EventEnvelope(weather_alarm)
SourceMessageRouter->>EventPipeline: handle(event)
loop per session
EventPipeline->>WeatherAggregationService: should_aggregate(event, session, runtime_config)
alt should_aggregate == False
EventPipeline->>MessageManager: push_event(event, target_sessions)
else should_aggregate == True
EventPipeline->>WeatherAggregationService: buffer event
end
end
rect rgb(230,230,250)
note over WeatherAggregationService: later, when flush conditions met
WeatherAggregationService->>EventPipeline: _flush_weather_buffer(session, entries, config, mode="forward")
loop each WeatherBufferEntry
EventPipeline->>MessageManager: evaluate_push_decision(entry.event, runtime_config)
alt decision.accepted
EventPipeline->>MessageBuildService: build_message_async(entry.event, runtime_config)
MessageBuildService-->>EventPipeline: MessageChain
else decision.rejected
EventPipeline-->>EventPipeline: skip entry
end
end
alt mode == "forward"
EventPipeline->>SessionSender: send(session, MessageChain(Comp.Nodes))
else mode == "single"
loop each built message
EventPipeline->>SessionSender: send(session, message)
end
end
end
File-Level Changes
Possibly linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
😷 屎山代码检测报告
🎯 本次检测概览
🎭 整体印象有点臭味，但还不至于熏死人 🧭 下一步这代码像个叛逆期的青少年，需要适当管教才能成才 📥 详细数据 由 fuck-u-code 自动检测 · 本评论不代表人类观点 |
📝 WalkthroughWalkthrough本次变更接入 CMA OpenQuake 气象预警源，新增天气预警解析、图标与颜色展示、按会话聚合推送和合并转发；同时扩展事件流日志控制、启动静默状态追踪、配置兼容及相关后台状态展示。 ChangesCMA 气象预警接入
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related issues
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
Comment |
|
2 tasks
Not up to standards ⛔🔴 Issues
|
| Category | Results |
|---|---|
| Security | 5 medium |
| Complexity | 4 medium |
🟢 Metrics 124 complexity · 9 duplication
Metric Results Complexity 124 Duplication 9
NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.
2 tasks
coderabbitai
Bot
added
type/design
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - 我发现了 1 个问题，并留下了一些总体反馈：
- 在
WeatherAggregationService._flush_session/_flush_via_forward中，聚合和限流设置始终取自self._config，因此在刷新时会忽略runtime_config.push_frequency_control.weather_aggregation中的任意会话级覆盖配置；建议将“有效的会话配置”贯穿到 flush 逻辑中，以保持与should_aggregate的行为一致。 - 新的事件流日志级别处理逻辑在多个位置被重复实现（例如
EventDeduplicationService._resolve_event_stream_by_source_id和SourceMessageRouter._resolve_stream_by_source_id）；考虑将其抽取到一个共享的辅助方法中，以减少逻辑偏移，并让未来新增事件流时更易维护。
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `WeatherAggregationService._flush_session/_flush_via_forward` the aggregation and rate-limit settings are always taken from `self._config`, so any per-session overrides in `runtime_config.push_frequency_control.weather_aggregation` are ignored during flushing; consider threading the effective session config through to the flush path to keep behavior consistent with `should_aggregate`.
- The new event-stream log level handling logic is duplicated in several places (e.g. `EventDeduplicationService._resolve_event_stream_by_source_id` and `SourceMessageRouter._resolve_stream_by_source_id`); factoring this into a single shared helper would reduce drift and make future additions of new event streams easier to maintain.
## Individual Comments
### Comment 1
<location path="core/message/push/weather_aggregation_service.py" line_range="45-54" />
<code_context>
+_MAX_NODES_PER_FORWARD = 20
</code_context>
<issue_to_address>
**issue (bug_risk):** Aggregation batch size is hard-coded and ignores the validated `max_batch_size` config.
`weather_aggregation.max_batch_size` is validated but `_flush_via_forward` still uses the hard-coded `_MAX_NODES_PER_FORWARD = 20`, so config changes and docs around `max_batch_size` are ineffective. Please either read `max_batch_size` when computing `batches` (e.g. `agg_config.get("max_batch_size", _MAX_NODES_PER_FORWARD)`) or remove `max_batch_size` from validation and documentation if the limit is intended to be fixed.
</issue_to_address>Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Original comment in English
Hey - I've found 1 issue, and left some high level feedback:
- In
WeatherAggregationService._flush_session/_flush_via_forwardthe aggregation and rate-limit settings are always taken fromself._config, so any per-session overrides inruntime_config.push_frequency_control.weather_aggregationare ignored during flushing; consider threading the effective session config through to the flush path to keep behavior consistent withshould_aggregate. - The new event-stream log level handling logic is duplicated in several places (e.g.
EventDeduplicationService._resolve_event_stream_by_source_idandSourceMessageRouter._resolve_stream_by_source_id); factoring this into a single shared helper would reduce drift and make future additions of new event streams easier to maintain.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `WeatherAggregationService._flush_session/_flush_via_forward` the aggregation and rate-limit settings are always taken from `self._config`, so any per-session overrides in `runtime_config.push_frequency_control.weather_aggregation` are ignored during flushing; consider threading the effective session config through to the flush path to keep behavior consistent with `should_aggregate`.
- The new event-stream log level handling logic is duplicated in several places (e.g. `EventDeduplicationService._resolve_event_stream_by_source_id` and `SourceMessageRouter._resolve_stream_by_source_id`); factoring this into a single shared helper would reduce drift and make future additions of new event streams easier to maintain.
## Individual Comments
### Comment 1
<location path="core/message/push/weather_aggregation_service.py" line_range="45-54" />
<code_context>
+_MAX_NODES_PER_FORWARD = 20
</code_context>
<issue_to_address>
**issue (bug_risk):** Aggregation batch size is hard-coded and ignores the validated `max_batch_size` config.
`weather_aggregation.max_batch_size` is validated but `_flush_via_forward` still uses the hard-coded `_MAX_NODES_PER_FORWARD = 20`, so config changes and docs around `max_batch_size` are ineffective. Please either read `max_batch_size` when computing `batches` (e.g. `agg_config.get("max_batch_size", _MAX_NODES_PER_FORWARD)`) or remove `max_batch_size` from validation and documentation if the limit is intended to be fixed.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+45
to
+54
| _MAX_NODES_PER_FORWARD = 20 | ||
|
|
||
|
|
||
| @dataclass | ||
| class WeatherBufferEntry: | ||
| """缓冲区中的单条气象预警条目。""" | ||
|
|
||
| event: EventEnvelope | ||
| color_level: int = 0 | ||
| color_name: str = "白色" |
Contributor
There was a problem hiding this comment.
issue (bug_risk): 聚合批大小被硬编码，忽略了已验证的 max_batch_size 配置。
weather_aggregation.max_batch_size 已经做了验证，但 _flush_via_forward 仍然使用硬编码的 _MAX_NODES_PER_FORWARD = 20，因此针对 max_batch_size 的配置变更和文档说明都不会生效。如果该限制是可配置的，请在计算 batches 时读取 max_batch_size（例如 agg_config.get("max_batch_size", _MAX_NODES_PER_FORWARD)）；如果该限制预期是固定值，则请从配置校验和文档中移除 max_batch_size。
Original comment in English
issue (bug_risk): Aggregation batch size is hard-coded and ignores the validated max_batch_size config.
weather_aggregation.max_batch_size is validated but _flush_via_forward still uses the hard-coded _MAX_NODES_PER_FORWARD = 20, so config changes and docs around max_batch_size are ineffective. Please either read max_batch_size when computing batches (e.g. agg_config.get("max_batch_size", _MAX_NODES_PER_FORWARD)) or remove max_batch_size from validation and documentation if the limit is intended to be fixed.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 14
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (5)
core/app/pipeline/event_pipeline.py (1)
89-105: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win全部会话进入聚合时，统计会记入上一个事件的成功会话。
当
non_aggregated_sessions为空（所有会话都启用聚合）时本次完全不调用push_event，但第 102 行仍用self.service.message_manager.last_success_sessions记录统计。该字段保留的是上一次推送结果，会把无关会话记为本事件的推送目标；真正的聚合推送发生在_flush_weather_buffer，也没有回写统计。建议在聚合分支显式传入空列表（或标记为“已缓冲待推送”），并在 flush 成功后补记统计。🐛 建议修复方向
- await self.service.statistics_manager.record_push( - event, - pushed_sessions=self.service.message_manager.last_success_sessions, - ) + pushed_sessions = self.service.message_manager.last_success_sessions + if aggregated_only: + # 事件仅进入聚合缓冲区，本次未产生实际推送 + pushed_sessions = [] + await self.service.statistics_manager.record_push( + event, + pushed_sessions=pushed_sessions, + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/app/pipeline/event_pipeline.py` around lines 89 - 105, Update the statistics call in the event-processing flow around _push_event_normal and record_push so events whose all sessions enter aggregation do not reuse message_manager.last_success_sessions from a prior event. Track whether the event was actually pushed versus only buffered, pass an empty session list for the fully aggregated case, and ensure _flush_weather_buffer records the sessions after the deferred aggregated push succeeds.core/network/source_message_router.py (1)
267-272: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win按实际子源标记缺失解析器日志。
FAN Studio 映射包含气象、台风和海啸子源；固定为
earthquake会绕过对应事件流的日志策略。改为_resolve_stream_by_source_id(source_id)。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/network/source_message_router.py` around lines 267 - 272, Update the missing-parser warning in the source message routing flow to set event_stream from _resolve_stream_by_source_id(source_id) instead of the hardcoded "earthquake" value. Preserve the existing warning message and is_event_linked behavior.core/services/eqsc/eqsc_typhoon_poll_service.py (1)
362-373: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win在静默播种完成后再报告首轮抓取成功。
当前先调用
note_poll_fetch_completed(success=True)，再检查is_silencing()。若该门闩使协调器进入 READY，首批台风会跳过播种并直接推送。应先按静默状态处理、播种首批指纹，再通知抓取完成。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/services/eqsc/eqsc_typhoon_poll_service.py` around lines 362 - 373, 调整 eqsc 轮询成功处理流程：在设置成功状态并调用 _notify_silence_fetch_completed(success=True) 前，先依据当前静默状态处理首批 active_items，并通过 _process_typhoon_updates 完成指纹播种；确保静默协调器仍处于播种阶段时不会因提前进入 READY 而直接推送首批台风，同时保留 emit_event 为 false 时的返回行为。core/services/identity/event_deduplication_service.py (1)
375-407: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
plugin_logger.debug不能接收事件流参数
core/services/identity/event_deduplication_service.py:375-407里log_level="debug"时，log_fn会变成plugin_logger.debug，但这里仍传了is_event_linked和event_stream。这会把未知关键字透传到底层 logger，触发TypeError并中断去重。仅在info分支传这些参数，debug分支只记录消息即可。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/services/identity/event_deduplication_service.py` around lines 375 - 407, Update the duplicate-filter logging in the deduplication flow around the priority checks so event metadata is passed only when log_fn is plugin_logger.info. Keep debug logging limited to the message argument, preventing is_event_linked and event_stream from reaching plugin_logger.debug, while preserving the existing filtering behavior.core/services/query/weather_query_service.py (1)
282-286: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win气象预警查询结果的
icon_url未复用新的编码映射，OpenQuakeAPI(CMA) 来源的图标会失效。两处
icon_url都是直接用原始weather_type_code拼接 Fan Studio 图标接口 URL。若weather_type_code是 OpenQuakeAPI CMA 的p编码格式（如p0002003），该接口无法识别，会返回message_build_service.py中专门识别处理的"伪图片"错误页——但查询命令这条路径并未接入resolve_weather_icon_code/build_weather_icon_url，导致 CMA 来源预警的查询结果图标必现失效。🐛 建议修复
+from ...message.presenters.weather_alarm_code_map import ( + build_weather_icon_url, + resolve_weather_icon_code, +)detected_type = detect_weather_type(title_text, weather_type_code) detected_color = detect_weather_color(level_text, title_text) color_emoji = COLOR_LEVEL_EMOJI.get(detected_color, "") + icon_code = resolve_weather_icon_code( + weather_type_code, title=title_text, headline=headline_text + ) @@ - "icon_url": ( - f"https://api.fanstudio.tech/we/img/alarm_icon.php?type={weather_type_code}" - if weather_type_code - else None - ), + "icon_url": build_weather_icon_url(icon_code) if icon_code else None,同样的替换需应用到搜索分支（约 400-427 行）。
Also applies to: 308-312, 400-404, 423-427
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/services/query/weather_query_service.py` around lines 282 - 286, 更新天气预警查询结果中两处查询分支的 icon_url 构建逻辑，先通过 resolve_weather_icon_code 将原始 weather_type_code（包括 CMA 的 p 编码）转换为 Fan Studio 可识别的图标编码，再复用 build_weather_icon_url 生成 URL。覆盖当前分支及搜索分支对应的全部 icon_url 拼接位置，保持其他天气类型检测逻辑不变。
🧹 Nitpick comments (3)
core/app/pipeline/event_pipeline.py (1)
139-146: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value回调签名与冗余
raise e。两处小问题：
set_flush_callback的文档约定回调返回bool，此处声明-> None；调用方也未使用返回值，建议统一约定。- 第 241-242 行
except Exception as e: raise e会重置回溯信息，直接去掉try或改为裸raise更合适。♻️ 建议调整
- chain = MessageChain([nodes]) - try: - await message_manager.session_sender.send(session, chain) - plugin_logger.debug( - f"[灾害预警] 气象预警合并转发已发送到 {session}, " - f"含 {len(built_messages)} 条预警", - event_stream="weather_alarm", - ) - except Exception as e: - raise e + chain = MessageChain([nodes]) + # 发送失败向上抛出，由聚合服务降级为逐条发送 + await message_manager.session_sender.send(session, chain) + plugin_logger.debug( + f"[灾害预警] 气象预警合并转发已发送到 {session}, " + f"含 {len(built_messages)} 条预警", + event_stream="weather_alarm", + )Also applies to: 241-242
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/app/pipeline/event_pipeline.py` around lines 139 - 146, Update _flush_weather_buffer to match the set_flush_callback contract by declaring a bool return type, and ensure its implementation returns the appropriate boolean value. In the exception handling around this method, remove the redundant `except Exception as e: raise e` pattern or replace it with a bare raise so the original traceback is preserved.core/rules/intensity_rule.py (1)
217-254: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low value泛化占位震级文案
skip_magnitude也会覆盖isAssumption的假定震源场景，accept_reason直接写“PLUM法占位震级”会把这类报表述错。改成更通用的“假定震源占位震级，已跳过震级过滤”更稳妥。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/rules/intensity_rule.py` around lines 217 - 254, Update the skip_magnitude accept_reason branch in _evaluate_dual_threshold_filter’s caller to use the generic “假定震源占位震级，已跳过震级过滤” wording, covering both is_assumption and magnitude_is_placeholder cases instead of naming only PLUM.core/message/push/message_build_service.py (1)
661-704: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
_resolve_weather_color_key与weather_alarm_code_map.py的 p 编码解析逻辑重复且校验不一致。
_P_FORMAT_COLOR_MAP（54-59 行）与weather_alarm_code_map._P_COLOR_DIGIT_TO_SUFFIX数值完全相同，属于跨文件重复维护；且此处 p 编码判定要求len(code) >= 8（685 行），而weather_alarm_code_map._is_p_code只要求len(code) >= 7——两者对"什么算合法 p 编码"的定义不一致。更关键的是，此处完全不感知weather_alarm_code_map._P_CODE_SKIP_GENERIC特殊短码列表，一旦该列表未来因颜色（而不仅是类型）误判而扩展，本地回退图标的颜色解析会静默地与官方图标解析逻辑产生分歧。建议在
weather_alarm_code_map.py中新增一个统一的resolve_p_code_color(code) -> str | None辅助函数（复用_P_COLOR_DIGIT_TO_SUFFIX并感知_P_CODE_SKIP_GENERIC），让_resolve_weather_color_key直接调用它，而不是维护第二套独立实现。♻️ 重构方向示意
# weather_alarm_code_map.py +def resolve_p_code_color(code: str) -> str | None: + """仅解析 p 编码的颜色部分，供颜色回退场景复用。""" + if not (code and _is_p_code(code)) or code in _P_CODE_SKIP_GENERIC: + return None + digits = code[1:] + if len(digits) < 5: + return None + return _P_COLOR_DIGIT_TO_SUFFIX.get(digits[-1])# message_build_service.py - if code.startswith("p") and len(code) >= 8: - return cls._P_FORMAT_COLOR_MAP.get(code[-1]) + if code.startswith("p"): + return resolve_p_code_color(code)同时建议核实 CMA 真实 p 编码长度是否恒为 8（若是，7 vs 8 的差异目前是休眠状态，但仍建议对齐避免未来隐患）。
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@core/app/runtime/disaster_service_lifecycle.py`:
- Around line 288-299: 将灾害服务停止流程中针对 _weather_aggregation_service.flush_all
的刷新移到后台任务及 ws_manager、http_fetcher 等资源回收之前，确保消息构建和发送依赖仍可用。为 flush_all 增加
asyncio.wait_for 超时保护，超时或其他异常继续记录并忽略，避免阻塞 stop()。
In `@core/message/logging/filters/raw_message_filter.py`:
- Around line 74-80: Update should_filter_message and the string-processing path
around _handle_string_message so OpenQuakeAPI JSON strings are parsed and
subjected to the same global_quake type check as dictionary payloads, allowing
only earthquake and weather. Reuse or extract a shared helper for the type
validation, then preserve the existing filtering result and _handle_dict_message
behavior for accepted messages.
In `@core/message/push/weather_aggregation_service.py`:
- Line 216: Update the background-task creation in the session flush flow around
_flush_session so each asyncio.create_task result is retained in a service-level
task collection or registered through the existing register_background_task
mechanism. Add completion cleanup and ensure exceptions from _flush_session are
observed and recorded, applying the same handling to both affected task-creation
sites.
- Around line 12-21: 实现注释中声明的缓冲条数触发条件：更新 should_aggregate，使单会话缓冲数量达到配置的
max_batch_size 时立即触发推送，并确保读取并使用该配置而非仅依赖
_MAX_NODES_PER_FORWARD。保持时间窗口和红色预警的现有触发逻辑不变，同时避免缓冲区在窗口内无限增长。
- Around line 262-265: Update the session flush flow around _flush_session and
_flush_via_forward so it uses the session’s effective runtime configuration,
matching the configuration used by should_aggregate, instead of
self._get_aggregation_config() and the global self._config. Resolve or preserve
the session-specific configuration through session_config_manager and pass it to
the flush callback so per-session time-window and rate-limit overrides remain
effective.
In `@core/network/source_message_router.py`:
- Around line 150-153: 在未找到解析器的告警分支中，将 `logger.warning` 替换为
`plugin_logger.warning`，并把 `event_stream` 改为调用
`_resolve_stream_by_source_id(source_id)`；保留现有告警消息及 `is_event_linked=True` 参数。
In `@core/parsers/base_parser.py`:
- Around line 172-176: Update BaseParser._parse_datetime() so the
time-parse-failure warning derives event_stream from source_entry.source_type or
the existing source-to-stream mapping instead of hardcoding “earthquake”; ensure
weather, tsunami, typhoon, and other parser sources receive their correct labels
while preserving the existing warning behavior.
In `@core/parsers/global_sources_parser.py`:
- Around line 745-749: 更新 UsgsEarthquakeParser 中地震解析成功日志的 event_stream 配置，将当前的
global_quake 改为 usgs_fanstudio，使 USGS 日志使用其实际来源对应的流配置；保留日志消息及 is_event_linked
行为不变。
In `@core/parsers/weather_parser.py`:
- Around line 34-41: Update _parse_data and _extract_realtime_payload so they
distinguish missing/non-RealtimeEvent wrappers from recognized RealtimeEvent
messages with unsupported type or action. Only fall back to _extract_data when
the input is not a RealtimeEvent; return None immediately for recognized
non-weather or non-new events, preserving the contract described by the
RealtimeEvent parsing logic around lines 203-237.
In `@core/services/config/config_validation_service.py`:
- Around line 1177-1221: Update the weather_aggregation validation in
ConfigValidator so max_batch_size uses the schema contract’s maximum of 20
instead of 50. Also verify the schema bounds for rate_limit_max_messages and
rate_limit_window_seconds, and align their _clamp_number ranges with those
declared limits.
In `@core/services/eqsc/eqsc_cenc_intensity_poll_service.py`:
- Around line 306-314: 修复 eqsc 轮询流程中将请求失败的空列表标记为成功的问题。更新 fetch_list 与调用方，使无
token、非 200 或请求异常可被区分；在 fetch_list 返回结果未知/失败时保持失败状态并调用
_notify_silence_fetch_completed(success=False)，仅在请求确实成功时更新 _last_success_at、重置
_consecutive_failures 并标记 success=True。
In `@core/services/eqsc/eqsc_tsunami_poll_service.py`:
- Around line 219-236: Move silent-gate completion reporting out of reusable
fetch_once methods: in core/services/eqsc/eqsc_tsunami_poll_service.py lines
219-236, remove the success/failure notifications from fetch_once and have
_poll_loop report completion only after its first fetch_once call fully finishes
processing, including _handle_disaster_event; in
core/services/snet/snet_poll_service.py lines 199-204, remove the existing
fetch_once completion notifications as well so fetch_for_query cannot affect
startup silence state.
In `@docs/OpenQuakeAPI.md`:
- Around line 468-472: 更新 docs/OpenQuakeAPI.md 示例中的 timestampMs，使其对应 2026 年 7 月
29 日 12:45，并与同一示例的 id、effective 和说明文本保持一致；不要修改示例中的其他字段。
In `@plugin/commands/plugin_query_command_service.py`:
- Around line 489-529: 修复全国分支在 `_send_forward_batches` 和 `_send_text_blocks`
均失败后继续落入普通搜索分支的问题：在全国分支最终失败路径显式返回，避免重复投递。将全国与普通区域中重复的投递及
`_track_command_feature` 逻辑抽取为内部辅助函数，并通过 `is_nationwide`
等参数区分行为，保持现有成功与回退遥测结果不变。
---
Outside diff comments:
In `@core/app/pipeline/event_pipeline.py`:
- Around line 89-105: Update the statistics call in the event-processing flow
around _push_event_normal and record_push so events whose all sessions enter
aggregation do not reuse message_manager.last_success_sessions from a prior
event. Track whether the event was actually pushed versus only buffered, pass an
empty session list for the fully aggregated case, and ensure
_flush_weather_buffer records the sessions after the deferred aggregated push
succeeds.
In `@core/network/source_message_router.py`:
- Around line 267-272: Update the missing-parser warning in the source message
routing flow to set event_stream from _resolve_stream_by_source_id(source_id)
instead of the hardcoded "earthquake" value. Preserve the existing warning
message and is_event_linked behavior.
In `@core/services/eqsc/eqsc_typhoon_poll_service.py`:
- Around line 362-373: 调整 eqsc 轮询成功处理流程：在设置成功状态并调用
_notify_silence_fetch_completed(success=True) 前，先依据当前静默状态处理首批 active_items，并通过
_process_typhoon_updates 完成指纹播种；确保静默协调器仍处于播种阶段时不会因提前进入 READY 而直接推送首批台风，同时保留
emit_event 为 false 时的返回行为。
In `@core/services/identity/event_deduplication_service.py`:
- Around line 375-407: Update the duplicate-filter logging in the deduplication
flow around the priority checks so event metadata is passed only when log_fn is
plugin_logger.info. Keep debug logging limited to the message argument,
preventing is_event_linked and event_stream from reaching plugin_logger.debug,
while preserving the existing filtering behavior.
In `@core/services/query/weather_query_service.py`:
- Around line 282-286: 更新天气预警查询结果中两处查询分支的 icon_url 构建逻辑，先通过
resolve_weather_icon_code 将原始 weather_type_code（包括 CMA 的 p 编码）转换为 Fan Studio
可识别的图标编码，再复用 build_weather_icon_url 生成 URL。覆盖当前分支及搜索分支对应的全部 icon_url
拼接位置，保持其他天气类型检测逻辑不变。
---
Nitpick comments:
In `@core/app/pipeline/event_pipeline.py`:
- Around line 139-146: Update _flush_weather_buffer to match the
set_flush_callback contract by declaring a bool return type, and ensure its
implementation returns the appropriate boolean value. In the exception handling
around this method, remove the redundant `except Exception as e: raise e`
pattern or replace it with a bare raise so the original traceback is preserved.
In `@core/rules/intensity_rule.py`:
- Around line 217-254: Update the skip_magnitude accept_reason branch in
_evaluate_dual_threshold_filter’s caller to use the generic “假定震源占位震级，已跳过震级过滤”
wording, covering both is_assumption and magnitude_is_placeholder cases instead
of naming only PLUM.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: c0affe2c-7b3c-4b43-b1fb-3aa69e8a703d
📒 Files selected for processing (51)
_conf_schema.jsonadmin/js/components/status/ConnectionsGrid.jsxadmin/js/utils/formatters.jscore/app/disaster_service.pycore/app/pipeline/event_pipeline.pycore/app/runtime/disaster_service_lifecycle.pycore/app/runtime/startup_silence_coordinator.pycore/message/fusion/cenc_fusion_service.pycore/message/fusion/cwa_eew_fusion_service.pycore/message/logging/filters/raw_message_filter.pycore/message/presenters/earthquake_presenter.pycore/message/presenters/weather_alarm_code_map.pycore/message/presenters/weather_constants.pycore/message/presenters/weather_presenter.pycore/message/push/__init__.pycore/message/push/message_build_service.pycore/message/push/push_flow_handler.pycore/message/push/weather_aggregation_service.pycore/network/http/eqsc_http_client.pycore/network/source_message_router.pycore/parsers/base_parser.pycore/parsers/china_earthquake_parser.pycore/parsers/china_eew_parser.pycore/parsers/china_intensity_report_eqsc_parser.pycore/parsers/china_intensity_report_parser.pycore/parsers/fssn_cmt_parser.pycore/parsers/global_sources_parser.pycore/parsers/japan_earthquake_parser.pycore/parsers/japan_eew_parser.pycore/parsers/parser_registry.pycore/parsers/snet_parser.pycore/parsers/taiwan_earthquake_parser.pycore/parsers/taiwan_eew_parser.pycore/parsers/tsunami_parser.pycore/parsers/typhoon_parser.pycore/parsers/weather_parser.pycore/rules/intensity_rule.pycore/services/config/config_validation_service.pycore/services/eqsc/eqsc_cenc_intensity_poll_service.pycore/services/eqsc/eqsc_tsunami_poll_service.pycore/services/eqsc/eqsc_typhoon_poll_service.pycore/services/identity/event_deduplication_service.pycore/services/query/weather_query_service.pycore/services/snet/snet_poll_service.pycore/sources/source_catalog.pycore/sources/source_router.pycore/storage/source_compat.pydocs/OpenQuakeAPI.mdplugin/commands/plugin_admin_command_service.pyplugin/commands/plugin_query_command_service.pyutils/plugin_logger.py
📜 Review details
⏰ Context from checks skipped due to timeout. (2)
- GitHub Check: Sourcery review
- GitHub Check: Codacy Static Code Analysis
⚠️ CI failures not shown inline (2)
GitHub Actions: 🐍 代码格式与质量检查 / 0_Ruff 代码检查.txt: feat: 新增 OpenQuakeAPI 气象预警接入、聚合推送服务、细粒度日志级别控制及多项修复
Conclusion: failure
##[group]Run actions/github-script@v9
with:
script: const fs = require('fs');
const formatLog = fs.readFileSync('format.log', 'utf8');
const checkLog = fs.readFileSync('check.log', 'utf8');
// --- 解析格式化日志 ---
// 示例: "1 file reformatted, 2 files left unchanged."
const formatMatch = formatLog.match(/(\d+) file(s)? reformatted/);
const reformattedCount = formatMatch ? parseInt(formatMatch[1]) : 0;
const unchangedMatch = formatLog.match(/(\d+) file(s)? left unchanged/);
const unchangedCount = unchangedMatch ? parseInt(unchangedMatch[1]) : 0;
// --- 解析检查日志 ---
// 示例: "Found 2 errors (1 fixed, 1 remaining)."
let fixedCount = 0;
let remainingCount = 0;
let totalErrors = 0;
const errorMatch = checkLog.match(/Found (\d+) error/);
if (errorMatch) {
totalErrors = parseInt(errorMatch[1]);
const fixedMatch = checkLog.match(/\((\d+) fixed/);
if (fixedMatch) fixedCount = parseInt(fixedMatch[1]);
const remainingMatch = checkLog.match(/, (\d+) remaining\)/);
if (remainingMatch) {
remainingCount = parseInt(remainingMatch[1]);
} else {
// 如果没有详细的 fixed/remaining 信息，通常意味着没有修复或全部剩余
if (fixedCount === 0) remainingCount = totalErrors;
}
}
// --- 构建报告内容 ---
let summary = '';
let isSuccess = (reformattedCount === 0 && remainingCount === 0);
if (isSuccess) {
summary = '### ✅ All checked pass!';
} else {
summary = '### ⚠️ 代码质量检查报告\n\n';
summary += '> 发现部分文件需要格式化或存在代码问题，建议在本地运行 `ruff format .` 和 `ruff check --fix .`。\n\n';
// 格式化部分
if (reformattedCount > 0) {
summary += `**🖌️ 格式化**: \`${reformattedCount}\` 个文件被重新格式化 (未变更: ${unchangedCount})\n`;
} else {
summary += `**🖌️ 格式化**: ✅ 无需变更\n`;
}
// 代码检查部分
if (totalErrors > 0) {
summary += `**🛡️ 代码检查**: 发现 \`${totalErrors}\` 个问题\n`;
summary += `- 🛠️ 已自动修复: \`${fixedCount}\`\n`;
summary += `- ❌ 剩余未修复: \`${remainingCount}\`\n`;
} else {
summary += `**🛡️ 代码检查**: ✅ 通过\n`;
}
// 错误详情 (如果有剩余错误)
i...
GitHub Actions: 🐍 代码格式与质量检查 / Ruff 代码检查: feat: 新增 OpenQuakeAPI 气象预警接入、聚合推送服务、细粒度日志级别控制及多项修复
Conclusion: failure
##[group]Run actions/github-script@v9
with:
script: const fs = require('fs');
const formatLog = fs.readFileSync('format.log', 'utf8');
const checkLog = fs.readFileSync('check.log', 'utf8');
// --- 解析格式化日志 ---
// 示例: "1 file reformatted, 2 files left unchanged."
const formatMatch = formatLog.match(/(\d+) file(s)? reformatted/);
const reformattedCount = formatMatch ? parseInt(formatMatch[1]) : 0;
const unchangedMatch = formatLog.match(/(\d+) file(s)? left unchanged/);
const unchangedCount = unchangedMatch ? parseInt(unchangedMatch[1]) : 0;
// --- 解析检查日志 ---
// 示例: "Found 2 errors (1 fixed, 1 remaining)."
let fixedCount = 0;
let remainingCount = 0;
let totalErrors = 0;
const errorMatch = checkLog.match(/Found (\d+) error/);
if (errorMatch) {
totalErrors = parseInt(errorMatch[1]);
const fixedMatch = checkLog.match(/\((\d+) fixed/);
if (fixedMatch) fixedCount = parseInt(fixedMatch[1]);
const remainingMatch = checkLog.match(/, (\d+) remaining\)/);
if (remainingMatch) {
remainingCount = parseInt(remainingMatch[1]);
} else {
// 如果没有详细的 fixed/remaining 信息，通常意味着没有修复或全部剩余
if (fixedCount === 0) remainingCount = totalErrors;
}
}
// --- 构建报告内容 ---
let summary = '';
let isSuccess = (reformattedCount === 0 && remainingCount === 0);
if (isSuccess) {
summary = '### ✅ All checked pass!';
} else {
summary = '### ⚠️ 代码质量检查报告\n\n';
summary += '> 发现部分文件需要格式化或存在代码问题，建议在本地运行 `ruff format .` 和 `ruff check --fix .`。\n\n';
// 格式化部分
if (reformattedCount > 0) {
summary += `**🖌️ 格式化**: \`${reformattedCount}\` 个文件被重新格式化 (未变更: ${unchangedCount})\n`;
} else {
summary += `**🖌️ 格式化**: ✅ 无需变更\n`;
}
// 代码检查部分
if (totalErrors > 0) {
summary += `**🛡️ 代码检查**: 发现 \`${totalErrors}\` 个问题\n`;
summary += `- 🛠️ 已自动修复: \`${fixedCount}\`\n`;
summary += `- ❌ 剩余未修复: \`${remainingCount}\`\n`;
} else {
summary += `**🛡️ 代码检查**: ✅ 通过\n`;
}
// 错误详情 (如果有剩余错误)
i...
🧰 Additional context used
🪛 ast-grep (0.45.0)
core/message/presenters/weather_constants.py
[warning] 141-141: XPath query is request-/variable-derived; use parameterized XPath to prevent injection.
Context: text.find(keyword)
Note: [CWE-643] Improper Neutralization of Data within XPath Expressions ('XPath Injection').
(xpath-injection-python)
core/message/push/message_build_service.py
[warning] 804-804: File path is request-/variable-derived; validate and normalize to prevent path traversal.
Context: open(fallback_path, "rb")
Note: [CWE-22] Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal').
(open-filename-from-request)
🪛 GitHub Check: Codacy Static Code Analysis
core/services/snet/snet_poll_service.py
[warning] 164-164: core/services/snet/snet_poll_service.py#L164
Try, Except, Pass detected.
core/services/eqsc/eqsc_tsunami_poll_service.py
[warning] 174-174: core/services/eqsc/eqsc_tsunami_poll_service.py#L174
Try, Except, Pass detected.
core/services/eqsc/eqsc_cenc_intensity_poll_service.py
[warning] 189-189: core/services/eqsc/eqsc_cenc_intensity_poll_service.py#L189
Try, Except, Pass detected.
core/services/eqsc/eqsc_typhoon_poll_service.py
[warning] 195-195: core/services/eqsc/eqsc_typhoon_poll_service.py#L195
Try, Except, Pass detected.
core/app/pipeline/event_pipeline.py
[warning] 139-139: core/app/pipeline/event_pipeline.py#L139
EventPipeline._flush_weather_buffer is too complex (20) (MC0001)
[warning] 139-139: core/app/pipeline/event_pipeline.py#L139
Method _flush_weather_buffer has 88 lines of code (limit is 50)
[warning] 210-210: core/app/pipeline/event_pipeline.py#L210
Try, Except, Pass detected.
core/message/push/weather_aggregation_service.py
[warning] 135-135: core/message/push/weather_aggregation_service.py#L135
Method should_aggregate has 56 lines of code (limit is 50)
[warning] 267-267: core/message/push/weather_aggregation_service.py#L267
Method _flush_via_forward has 71 lines of code (limit is 50)
🔇 Additional comments (53)
core/app/runtime/startup_silence_coordinator.py (1)
42-46: LGTM!Also applies to: 262-283, 296-297, 387-388, 540-552
core/services/eqsc/eqsc_tsunami_poll_service.py (1)
159-175: LGTM!Also applies to: 209-233
core/message/push/__init__.py (1)
14-22: LGTM!core/app/disaster_service.py (1)
204-212: LGTM!plugin/commands/plugin_query_command_service.py (1)
298-298: LGTM!core/parsers/typhoon_parser.py (1)
225-225: LGTM!utils/plugin_logger.py (1)
7-24: LGTM!Also applies to: 33-83, 105-121, 130-140
core/message/fusion/cenc_fusion_service.py (1)
137-137: LGTM!Also applies to: 148-148, 185-185, 196-196, 282-282
core/message/fusion/cwa_eew_fusion_service.py (1)
112-112: LGTM!Also applies to: 123-123, 157-157, 168-168, 268-274
core/message/push/push_flow_handler.py (1)
263-298: LGTM!Also applies to: 337-343
core/network/source_message_router.py (2)
125-138: LGTM!
445-445: LGTM!Also applies to: 519-519
core/parsers/global_sources_parser.py (1)
230-230: LGTM!Also applies to: 257-257, 414-414, 574-574, 924-924
core/parsers/tsunami_parser.py (1)
238-238: LGTM!Also applies to: 409-409, 678-678
core/services/identity/event_deduplication_service.py (1)
112-130: LGTM!Also applies to: 211-211, 238-238, 455-455, 736-744, 862-872
core/services/eqsc/eqsc_cenc_intensity_poll_service.py (1)
174-190: LGTM!Also applies to: 361-361
core/services/eqsc/eqsc_typhoon_poll_service.py (1)
180-196: LGTM!Also applies to: 295-310, 378-378
core/parsers/china_earthquake_parser.py (1)
126-126: LGTM!core/parsers/china_eew_parser.py (1)
144-144: LGTM!Also applies to: 252-252
core/parsers/china_intensity_report_eqsc_parser.py (1)
307-307: LGTM!core/parsers/china_intensity_report_parser.py (1)
351-351: LGTM!core/parsers/fssn_cmt_parser.py (1)
131-131: LGTM!core/parsers/japan_earthquake_parser.py (1)
206-206: LGTM!Also applies to: 357-357
core/parsers/japan_eew_parser.py (1)
126-126: LGTM!Also applies to: 150-150, 273-274, 281-281, 433-433, 671-671
core/parsers/snet_parser.py (1)
631-633: LGTM!core/parsers/taiwan_earthquake_parser.py (1)
238-238: LGTM!core/parsers/taiwan_eew_parser.py (1)
144-147: LGTM!Also applies to: 303-306
core/message/presenters/earthquake_presenter.py (1)
1043-1051: LGTM!Also applies to: 1099-1101
core/rules/intensity_rule.py (1)
86-134: LGTM!core/services/config/config_validation_service.py (1)
1282-1293: LGTM!plugin/commands/plugin_admin_command_service.py (5)
157-157: LGTM!
259-260: LGTM!
351-365: LGTM!
313-335: 🎯 Functional Correctness这处映射不需要改动。
status["connections"]的group_key和这里的connection_label_map使用的是同一套稳定标识，不会把已启用连接误判成“⚪ 未启用”。> Likely an incorrect or invalid review comment.
176-187: 🎯 Functional CorrectnessNIED S-Net 子源键就是
enabled。> Likely an incorrect or invalid review comment.core/network/http/eqsc_http_client.py (1)
79-97: 🩺 Stability & Availability无需修改
_conf_schema.json (1)
228-233: LGTM!Also applies to: 769-837, 1438-1486
admin/js/utils/formatters.js (1)
248-250: LGTM!Also applies to: 369-370
core/message/presenters/weather_alarm_code_map.py (1)
1-203: LGTM!core/message/presenters/weather_constants.py (1)
119-156: LGTM!core/message/presenters/weather_presenter.py (1)
74-81: LGTM!core/message/push/message_build_service.py (2)
269-337: LGTM!Also applies to: 785-822
744-754: 🗄️ Data Integrity & Integration这处
type回退可保留：metadata["type"]在weather_parser.py里写入的是weather_code，不是事件类型。> Likely an incorrect or invalid review comment.core/services/query/weather_query_service.py (1)
97-149: LGTM!admin/js/components/status/ConnectionsGrid.jsx (1)
353-357: 📐 Maintainability & Code Quality这条映射不是死代码
scopedSourceMap按connectionName分组，同一个china_weather_openquake在FAN Studio和OpenQuakeAPI下都可单独保留。> Likely an incorrect or invalid review comment.core/sources/source_catalog.py (1)
793-822: LGTM!core/sources/source_router.py (1)
176-178: LGTM!core/storage/source_compat.py (1)
62-64: LGTM!Also applies to: 118-119
docs/OpenQuakeAPI.md (2)
122-122: LGTM!
445-466: LGTM!Also applies to: 482-493
core/parsers/parser_registry.py (1)
77-81: LGTM!Also applies to: 94-94, 104-104, 114-114
core/parsers/weather_parser.py (1)
21-32: LGTM!Also applies to: 57-57, 134-136, 190-194
core/parsers/base_parser.py (1)
119-126: LGTM!
Comment on lines
+288
to
+299
| # 停机前推送气象预警聚合缓冲区中尚未发送的事件，避免丢失。 | ||
| weather_agg = getattr( | ||
| self.service, "_weather_aggregation_service", None | ||
| ) | ||
| if weather_agg is not None: | ||
| try: | ||
| await weather_agg.flush_all() | ||
| except Exception as flush_err: | ||
| logger.debug( | ||
| f"[灾害预警] 停机时推送气象预警聚合缓冲区失败（已忽略）: {flush_err}" | ||
| ) | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
停机刷新位置偏后且无超时保护。
此处刷新发生在后台任务已取消、ws_manager / http_fetcher 已关闭之后，而 flush_all() 内部会走 message_build_service.build_message_async（图标/地图渲染可能依赖已关闭的 HTTP 资源）并逐条发送消息；同时没有超时，缓冲较多时会明显拖长停机，甚至卡住 stop()。建议把该刷新提前到任务/资源回收之前，并加 asyncio.wait_for 兜底。
🛡️ 建议修复
if weather_agg is not None:
try:
- await weather_agg.flush_all()
+ await asyncio.wait_for(weather_agg.flush_all(), timeout=15)
+ except TimeoutError:
+ logger.warning(
+ "[灾害预警] 停机时推送气象预警聚合缓冲区超时，已放弃"
+ )
except Exception as flush_err:
logger.debug(
f"[灾害预警] 停机时推送气象预警聚合缓冲区失败（已忽略）: {flush_err}"
)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/app/runtime/disaster_service_lifecycle.py` around lines 288 - 299,
将灾害服务停止流程中针对 _weather_aggregation_service.flush_all 的刷新移到后台任务及
ws_manager、http_fetcher 等资源回收之前，确保消息构建和发送依赖仍可用。为 flush_all 增加 asyncio.wait_for
超时保护，超时或其他异常继续记录并忽略，避免阻塞 stop()。
Comment on lines
+74
to
80
| # 针对 JSON 结构的字典数据，如果是 OpenQuakeAPI 聚合连接， | ||
| # 仅保留 earthquake（GQ）和 weather（CMA 气象）业务类型，其余过滤。 | ||
| if "global_quake" in source_id.lower(): | ||
| inner_type = str(payload_data.get("type") or "").lower() | ||
| if inner_type not in ("earthquake", "weather"): | ||
| return f"OpenQuakeAPI 非地震/气象业务JSON消息过滤: {inner_type}" | ||
| return self._handle_dict_message(payload_data, source_id) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
字符串 JSON 会绕过 OpenQuake 类型过滤。
RawMessageLoggingService 传入的仍可能是原始 JSON 字符串，而 should_filter_message() 对字符串会先走 _handle_string_message()；因此 Line 74-79 仅对已经解析成 dict 的消息生效。实际字符串形式的 type=tsunami/status/station 不会被过滤，和“仅保留 earthquake/weather”的规则不一致，并可能增加日志存储和处理压力。
请在字符串解析后复用同一套 OpenQuake 类型检查，或将该检查提取为公共 helper，避免字符串与字典入口产生两套行为。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/message/logging/filters/raw_message_filter.py` around lines 74 - 80,
Update should_filter_message and the string-processing path around
_handle_string_message so OpenQuakeAPI JSON strings are parsed and subjected to
the same global_quake type check as dictionary payloads, allowing only
earthquake and weather. Reuse or extract a shared helper for the type
validation, then preserve the existing filtering result and _handle_dict_message
behavior for accepted messages.
Comment on lines
+12
to
+21
| 设计要点： | ||
| 1. 按会话维度独立缓冲，避免跨群混淆； | ||
| 2. 缓冲期间对同一预警 ID 去重，只保留最新； | ||
| 3. 触发推送条件：时间窗口到期 / 缓冲条数达上限 / 收到红色预警立即推送； | ||
| 红色预警立即推送时仍需通过规则链复核，不绕过已有过滤配置； | ||
| 若红色预警触发推送时缓冲区仅有这一条事件，则直接走常规推送流， | ||
| 避免为单条事件构建合并转发节点的额外开销； | ||
| 4. 合并转发失败时降级为逐条文本发送，此时才启用限流； | ||
| 5. 限流平台按颜色级别排序，超出配额的低级别预警被丢弃并记录 debug 日志。 | ||
| """ |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
文档描述的触发条件与实现不一致：缓冲条数上限未实现。
模块注释（第 15 行）称"缓冲条数达上限"是推送触发条件之一，但 should_aggregate 只依赖时间窗口与红色预警，max_batch_size 配置也从未被读取（分批只使用常量 _MAX_NODES_PER_FORWARD）。结果是单会话缓冲区在窗口内无条数上限，极端情况下（如全国范围批量预警）内存与单次转发体积不可控。建议要么按 max_batch_size 实现提前触发/裁剪，要么修正注释以免误导。
Also applies to: 219-223
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/message/push/weather_aggregation_service.py` around lines 12 - 21,
实现注释中声明的缓冲条数触发条件：更新 should_aggregate，使单会话缓冲数量达到配置的 max_batch_size
时立即触发推送，并确保读取并使用该配置而非仅依赖
_MAX_NODES_PER_FORWARD。保持时间窗口和红色预警的现有触发逻辑不变，同时避免缓冲区在窗口内无限增长。
| is_event_linked=True, | ||
| event_stream="weather_alarm", | ||
| ) | ||
| asyncio.create_task(self._flush_session(session)) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
后台任务未持引用，可能被 GC 回收导致聚合推送丢失。
asyncio.create_task(...) 的返回值未保存，事件循环只持弱引用，任务在完成前可能被垃圾回收；同时 _flush_session 抛出的异常也无人观察。建议持有任务集合（或复用主服务的 register_background_task），并加上完成回调清理与异常记录。
🛡️ 建议修复
def __init__(self, config: dict[str, Any]):
self._config = config
+ # 持有正在执行的 flush 任务引用，避免被 GC 回收
+ self._flush_tasks: set[asyncio.Task] = set()+ def _spawn_flush(self, session: str) -> None:
+ task = asyncio.create_task(self._flush_session(session))
+ self._flush_tasks.add(task)
+ task.add_done_callback(self._flush_tasks.discard)
+- asyncio.create_task(self._flush_session(session))
+ self._spawn_flush(session) timer = loop.call_later(
delay,
- lambda: asyncio.create_task(self._flush_session(session)),
+ lambda: self._spawn_flush(session),
)Also applies to: 241-245
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/message/push/weather_aggregation_service.py` at line 216, Update the
background-task creation in the session flush flow around _flush_session so each
asyncio.create_task result is retained in a service-level task collection or
registered through the existing register_background_task mechanism. Add
completion cleanup and ensure exceptions from _flush_session are observed and
recorded, applying the same handling to both affected task-creation sites.
Comment on lines
+262
to
+265
| agg_config = self._get_aggregation_config() | ||
|
|
||
| # 统一走合并转发，平台是否支持由框架层处理 | ||
| await self._flush_via_forward(session, entries, agg_config) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
flush 阶段回退到全局配置，会话级差异配置失效。
should_aggregate 使用会话级 runtime_config 读取 time_window_seconds 等参数，而 _flush_session/_flush_via_forward 只用 self._get_aggregation_config()（全局），同时 _flush_callback(session, batch, self._config, ...) 也传全局 config。若某会话覆写了限流参数，推送阶段不会生效。建议在缓冲时按会话记录一次生效配置，或在 flush 时通过 session_config_manager 重新解析。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/message/push/weather_aggregation_service.py` around lines 262 - 265,
Update the session flush flow around _flush_session and _flush_via_forward so it
uses the session’s effective runtime configuration, matching the configuration
used by should_aggregate, instead of self._get_aggregation_config() and the
global self._config. Resolve or preserve the session-specific configuration
through session_config_manager and pass it to the flush callback so per-session
time-window and rate-limit overrides remain effective.
Comment on lines
+1177
to
+1221
| # 气象预警聚合推送配置校验 | ||
| agg = cfg.get("weather_aggregation") | ||
| if isinstance(agg, dict): | ||
| ConfigValidator._ensure_bool(agg, "enabled", True) | ||
| ConfigValidator._ensure_bool(agg, "flush_on_red", True) | ||
| ConfigValidator._ensure_bool(agg, "rate_limit_enabled", True) | ||
|
|
||
| agg["time_window_seconds"] = int( | ||
| ConfigValidator._clamp_number( | ||
| agg.get("time_window_seconds", 900), | ||
| minimum=60, | ||
| maximum=3600, | ||
| default=900, | ||
| field_name="气象预警聚合时间窗口", | ||
| ) | ||
| ) | ||
| agg["max_batch_size"] = int( | ||
| ConfigValidator._clamp_number( | ||
| agg.get("max_batch_size", 20), | ||
| minimum=1, | ||
| maximum=50, | ||
| default=20, | ||
| field_name="气象预警聚合单批最大条数", | ||
| ) | ||
| ) | ||
| agg["rate_limit_max_messages"] = int( | ||
| ConfigValidator._clamp_number( | ||
| agg.get("rate_limit_max_messages", 3), | ||
| minimum=1, | ||
| maximum=20, | ||
| default=3, | ||
| field_name="气象预警限流最大消息数", | ||
| ) | ||
| ) | ||
| agg["rate_limit_window_seconds"] = int( | ||
| ConfigValidator._clamp_number( | ||
| agg.get("rate_limit_window_seconds", 900), | ||
| minimum=60, | ||
| maximum=3600, | ||
| default=900, | ||
| field_name="气象预警限流时间窗口", | ||
| ) | ||
| ) | ||
| cfg["weather_aggregation"] = agg | ||
|
|
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
max_batch_size 校验范围与配置契约不一致。
_conf_schema.json 中 weather_aggregation.max_batch_size 的 slider 声明为 min=1, max=20，但此处 _clamp_number 使用 maximum=50。若某种途径写入 21-50 之间的值，校验器会将其当作合法值保留，而不是按契约钳制到 20，导致运行时行为与管理面板/schema 描述不一致。
另外 rate_limit_max_messages、rate_limit_window_seconds 在提供的 schema 片段中被截断，建议一并确认其边界与此处 clamp 范围是否一致。
🐛 建议修复
agg["max_batch_size"] = int(
ConfigValidator._clamp_number(
agg.get("max_batch_size", 20),
minimum=1,
- maximum=50,
+ maximum=20,
default=20,
field_name="气象预警聚合单批最大条数",
)
)🔍 验证脚本：确认 rate_limit 相关字段的 schema 边界
#!/bin/bash
fd _conf_schema.json --exec cat -n {} \; | rg -n -A 30 '"rate_limit_max_messages"|"rate_limit_window_seconds"'📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| # 气象预警聚合推送配置校验 | |
| agg = cfg.get("weather_aggregation") | |
| if isinstance(agg, dict): | |
| ConfigValidator._ensure_bool(agg, "enabled", True) | |
| ConfigValidator._ensure_bool(agg, "flush_on_red", True) | |
| ConfigValidator._ensure_bool(agg, "rate_limit_enabled", True) | |
| agg["time_window_seconds"] = int( | |
| ConfigValidator._clamp_number( | |
| agg.get("time_window_seconds", 900), | |
| minimum=60, | |
| maximum=3600, | |
| default=900, | |
| field_name="气象预警聚合时间窗口", | |
| ) | |
| ) | |
| agg["max_batch_size"] = int( | |
| ConfigValidator._clamp_number( | |
| agg.get("max_batch_size", 20), | |
| minimum=1, | |
| maximum=50, | |
| default=20, | |
| field_name="气象预警聚合单批最大条数", | |
| ) | |
| ) | |
| agg["rate_limit_max_messages"] = int( | |
| ConfigValidator._clamp_number( | |
| agg.get("rate_limit_max_messages", 3), | |
| minimum=1, | |
| maximum=20, | |
| default=3, | |
| field_name="气象预警限流最大消息数", | |
| ) | |
| ) | |
| agg["rate_limit_window_seconds"] = int( | |
| ConfigValidator._clamp_number( | |
| agg.get("rate_limit_window_seconds", 900), | |
| minimum=60, | |
| maximum=3600, | |
| default=900, | |
| field_name="气象预警限流时间窗口", | |
| ) | |
| ) | |
| cfg["weather_aggregation"] = agg | |
| # 气象预警聚合推送配置校验 | |
| agg = cfg.get("weather_aggregation") | |
| if isinstance(agg, dict): | |
| ConfigValidator._ensure_bool(agg, "enabled", True) | |
| ConfigValidator._ensure_bool(agg, "flush_on_red", True) | |
| ConfigValidator._ensure_bool(agg, "rate_limit_enabled", True) | |
| agg["time_window_seconds"] = int( | |
| ConfigValidator._clamp_number( | |
| agg.get("time_window_seconds", 900), | |
| minimum=60, | |
| maximum=3600, | |
| default=900, | |
| field_name="气象预警聚合时间窗口", | |
| ) | |
| ) | |
| agg["max_batch_size"] = int( | |
| ConfigValidator._clamp_number( | |
| agg.get("max_batch_size", 20), | |
| minimum=1, | |
| maximum=20, | |
| default=20, | |
| field_name="气象预警聚合单批最大条数", | |
| ) | |
| ) | |
| agg["rate_limit_max_messages"] = int( | |
| ConfigValidator._clamp_number( | |
| agg.get("rate_limit_max_messages", 3), | |
| minimum=1, | |
| maximum=20, | |
| default=3, | |
| field_name="气象预警限流最大消息数", | |
| ) | |
| ) | |
| agg["rate_limit_window_seconds"] = int( | |
| ConfigValidator._clamp_number( | |
| agg.get("rate_limit_window_seconds", 900), | |
| minimum=60, | |
| maximum=3600, | |
| default=900, | |
| field_name="气象预警限流时间窗口", | |
| ) | |
| ) | |
| cfg["weather_aggregation"] = agg |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/services/config/config_validation_service.py` around lines 1177 - 1221,
Update the weather_aggregation validation in ConfigValidator so max_batch_size
uses the schema contract’s maximum of 20 instead of 50. Also verify the schema
bounds for rate_limit_max_messages and rate_limit_window_seconds, and align
their _clamp_number ranges with those declared limits.
Comment on lines
306
to
+314
| items = await client.fetch_list(limit=limit, use_cache=False) | ||
| if not isinstance(items, list): | ||
| self._consecutive_failures += 1 | ||
| self._notify_silence_fetch_completed(success=False) | ||
| return [] | ||
|
|
||
| self._consecutive_failures = 0 | ||
| self._last_success_at = time.time() | ||
| coordinator = getattr(self.service, "startup_silence", None) | ||
| if coordinator is not None: | ||
| try: | ||
| coordinator.note_poll_fetch_completed("eqsc_cenc_ir", success=True) | ||
| except Exception as exc: | ||
| logger.debug( | ||
| f"[灾害预警] EQSC CENC 烈度速报轮询通知静默协调器失败: {exc}" | ||
| ) | ||
| self._notify_silence_fetch_completed(success=True) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
不要将失败返回的空列表标记为首轮成功。
fetch_list() 在无 token、非 200 和请求异常时都会返回 []，但这里仍调用 success=True。静默协调器会因此将轮询门闩置为 ready，后续恢复时可能把首批历史数据当作新事件处理。请让客户端返回可区分的请求状态，或在此保留“结果未知/失败”状态。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/services/eqsc/eqsc_cenc_intensity_poll_service.py` around lines 306 -
314, 修复 eqsc 轮询流程中将请求失败的空列表标记为成功的问题。更新 fetch_list 与调用方，使无 token、非 200
或请求异常可被区分；在 fetch_list 返回结果未知/失败时保持失败状态并调用
_notify_silence_fetch_completed(success=False)，仅在请求确实成功时更新 _last_success_at、重置
_consecutive_failures 并标记 success=True。
Comment on lines
219
to
+236
| async def fetch_once(self, *, emit_event: bool = True) -> dict[str, Any] | None: | ||
| """抓取一轮 EQSC 海啸快照，可选投递事件。""" | ||
| client = self._ensure_client() | ||
| if client is None: | ||
| self._notify_silence_fetch_completed(success=False) | ||
| return None | ||
|
|
||
| # 轮询侧强制绕过短缓存，确保按间隔拿到最新快照； | ||
| # 客户端缓存仍可用于同间隔内的并发查询复用。 | ||
| raw = await client.fetch_latest_tsunami(use_cache=False) | ||
| if not isinstance(raw, dict) or not raw: | ||
| self._consecutive_failures += 1 | ||
| self._notify_silence_fetch_completed(success=False) | ||
| return None | ||
|
|
||
| self._consecutive_failures = 0 | ||
| self._last_success_at = time.time() | ||
| coordinator = getattr(self.service, "startup_silence", None) | ||
| if coordinator is not None: | ||
| try: | ||
| coordinator.note_poll_fetch_completed("eqsc_tsunami", success=True) | ||
| except Exception as exc: | ||
| logger.debug(f"[灾害预警] EQSC 海啸轮询通知静默协调器失败: {exc}") | ||
| self._notify_silence_fetch_completed(success=True) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
将静默门闩完成通知移到首轮轮询结束后。
Line 236 在解析和 _handle_disaster_event 前就将门闩标记为成功；超过 settle 时间后，启动快照仍在处理时可能已恢复正常推送。S-Net 的 fetch_once 还会被 fetch_for_query 调用，因此查询请求也能改变启动门闩状态。仅由 _poll_loop 在首轮 fetch_once 完整返回后报告完成，并从可复用的 fetch_once 中移除这些通知。
core/services/eqsc/eqsc_tsunami_poll_service.py#L219-L236: 将成功/失败完成通知移至首轮_poll_loop调用之后。core/services/snet/snet_poll_service.py#L199-L204: 移除fetch_once内的完成通知（包括现有成功分支）；避免fetch_for_query影响启动静默状态。
📍 Affects 2 files
core/services/eqsc/eqsc_tsunami_poll_service.py#L219-L236(this comment)core/services/snet/snet_poll_service.py#L199-L204
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/services/eqsc/eqsc_tsunami_poll_service.py` around lines 219 - 236, Move
silent-gate completion reporting out of reusable fetch_once methods: in
core/services/eqsc/eqsc_tsunami_poll_service.py lines 219-236, remove the
success/failure notifications from fetch_once and have _poll_loop report
completion only after its first fetch_once call fully finishes processing,
including _handle_disaster_event; in core/services/snet/snet_poll_service.py
lines 199-204, remove the existing fetch_once completion notifications as well
so fetch_for_query cannot affect startup silence state.
Comment on lines
+468
to
+472
| "timestampMs": 1752401400000, | ||
| "payload": { | ||
| "id": "32031241600000_20260729124523", | ||
| "headline": "铜山区气象台发布强对流黄色预警[Ⅲ级/较重]", | ||
| "effective": "2026/07/29 12:45", |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
修正示例中的 timestampMs。
timestampMs=1752401400000 对应 2025 年的时间，而同一示例的 id、effective 和说明文本都指向 2026 年 7 月 29 日 12:45。请替换为与该示例日期一致的毫秒时间戳，避免接入方据此编写错误测试。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/OpenQuakeAPI.md` around lines 468 - 472, 更新 docs/OpenQuakeAPI.md 示例中的
timestampMs，使其对应 2026 年 7 月 29 日 12:45，并与同一示例的 id、effective
和说明文本保持一致；不要修改示例中的其他字段。
Comment on lines
+489
to
+529
| # 正常区域搜索：结果较多时也走合并转发分批发送，避免单条消息过长 | ||
| if text_blocks and len(text_blocks) > 1: | ||
| try: | ||
| ok = await _send_forward_batches(text_blocks) | ||
| if ok: | ||
| await self._track_command_feature( | ||
| "command_weather_query", | ||
| { | ||
| "success": True, | ||
| "query_mode": str(result.get("query_mode") or "search"), | ||
| "is_nationwide": is_nationwide, | ||
| "result_count": int(total or 0), | ||
| "has_optional_type": bool(optional_a), | ||
| "has_optional_level": bool(optional_b), | ||
| "delivery_mode": "forward_batches", | ||
| }, | ||
| ) | ||
| return | ||
| except Exception as forward_error: | ||
| logger.warning( | ||
| f"[灾害预警] 合并转发送失败，回退文本: {forward_error}" | ||
| ) | ||
| try: | ||
| await _send_text_blocks(text_blocks, total) | ||
| await self._track_command_feature( | ||
| "command_weather_query", | ||
| { | ||
| "success": True, | ||
| "query_mode": str(result.get("query_mode") or "search"), | ||
| "is_nationwide": is_nationwide, | ||
| "result_count": int(total or 0), | ||
| "has_optional_type": bool(optional_a), | ||
| "has_optional_level": bool(optional_b), | ||
| "delivery_mode": "text_blocks", | ||
| }, | ||
| ) | ||
| return | ||
| except Exception as text_error: | ||
| logger.warning(f"[灾害预警] 文本回退发送失败: {text_error}") | ||
|
|
||
| # 结果较少时，组装文字概要 |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
全国分支失败后会二次进入该分支，可能重复发送。
全国分支（第 449-487 行）在 _send_forward_batches 抛异常且文本回退也失败时不会 return，控制流会落到这里并再次尝试 _send_forward_batches。由于 _send_forward_batches 是逐批发送、可能在中途失败，用户可能已收到前几批内容，此时重复发送会造成重复消息。建议在全国分支的最终失败路径显式 return，或用 is_nationwide 排除本分支。
另外这段与全国分支的投递+遥测逻辑几乎完全一致，建议抽成一个内部辅助函数（参数化 is_nationwide）以消除重复。
🐛 建议修复
- # 正常区域搜索：结果较多时也走合并转发分批发送，避免单条消息过长
- if text_blocks and len(text_blocks) > 1:
+ # 正常区域搜索：结果较多时也走合并转发分批发送，避免单条消息过长
+ # 全国分支已尝试过投递（含回退），此处不再重复发送
+ if not is_nationwide and text_blocks and len(text_blocks) > 1:📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| # 正常区域搜索：结果较多时也走合并转发分批发送，避免单条消息过长 | |
| if text_blocks and len(text_blocks) > 1: | |
| try: | |
| ok = await _send_forward_batches(text_blocks) | |
| if ok: | |
| await self._track_command_feature( | |
| "command_weather_query", | |
| { | |
| "success": True, | |
| "query_mode": str(result.get("query_mode") or "search"), | |
| "is_nationwide": is_nationwide, | |
| "result_count": int(total or 0), | |
| "has_optional_type": bool(optional_a), | |
| "has_optional_level": bool(optional_b), | |
| "delivery_mode": "forward_batches", | |
| }, | |
| ) | |
| return | |
| except Exception as forward_error: | |
| logger.warning( | |
| f"[灾害预警] 合并转发送失败，回退文本: {forward_error}" | |
| ) | |
| try: | |
| await _send_text_blocks(text_blocks, total) | |
| await self._track_command_feature( | |
| "command_weather_query", | |
| { | |
| "success": True, | |
| "query_mode": str(result.get("query_mode") or "search"), | |
| "is_nationwide": is_nationwide, | |
| "result_count": int(total or 0), | |
| "has_optional_type": bool(optional_a), | |
| "has_optional_level": bool(optional_b), | |
| "delivery_mode": "text_blocks", | |
| }, | |
| ) | |
| return | |
| except Exception as text_error: | |
| logger.warning(f"[灾害预警] 文本回退发送失败: {text_error}") | |
| # 结果较少时，组装文字概要 | |
| # 正常区域搜索：结果较多时也走合并转发分批发送，避免单条消息过长 | |
| # 全国分支已尝试过投递（含回退），此处不再重复发送 | |
| if not is_nationwide and text_blocks and len(text_blocks) > 1: | |
| try: | |
| ok = await _send_forward_batches(text_blocks) | |
| if ok: | |
| await self._track_command_feature( | |
| "command_weather_query", | |
| { | |
| "success": True, | |
| "query_mode": str(result.get("query_mode") or "search"), | |
| "is_nationwide": is_nationwide, | |
| "result_count": int(total or 0), | |
| "has_optional_type": bool(optional_a), | |
| "has_optional_level": bool(optional_b), | |
| "delivery_mode": "forward_batches", | |
| }, | |
| ) | |
| return | |
| except Exception as forward_error: | |
| logger.warning( | |
| f"[灾害预警] 合并转发送失败，回退文本: {forward_error}" | |
| ) | |
| try: | |
| await _send_text_blocks(text_blocks, total) | |
| await self._track_command_feature( | |
| "command_weather_query", | |
| { | |
| "success": True, | |
| "query_mode": str(result.get("query_mode") or "search"), | |
| "is_nationwide": is_nationwide, | |
| "result_count": int(total or 0), | |
| "has_optional_type": bool(optional_a), | |
| "has_optional_level": bool(optional_b), | |
| "delivery_mode": "text_blocks", | |
| }, | |
| ) | |
| return | |
| except Exception as text_error: | |
| logger.warning(f"[灾害预警] 文本回退发送失败: {text_error}") | |
| # 结果较少时，组装文字概要 |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@plugin/commands/plugin_query_command_service.py` around lines 489 - 529,
修复全国分支在 `_send_forward_batches` 和 `_send_text_blocks`
均失败后继续落入普通搜索分支的问题：在全国分支最终失败路径显式返回，避免重复投递。将全国与普通区域中重复的投递及
`_track_command_feature` 逻辑抽取为内部辅助函数，并通过 `is_nationwide`
等参数区分行为，保持现有成功与回退遥测结果不变。
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
📝 描述 / Description
feat #42
feat #109
🛠️ 改动点 / Modifications
📸 运行截图或测试结果 / Screenshots or Test Results
✅ 检查清单 / Checklist
requirements.txt文件中。/ I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added torequirements.txt.❤️ CONTRIBUTING
Summary by Sourcery
通过 OpenQuakeAPI 集成 CMA 气象预警，新增会话级天气预警聚合（可配置时间窗口与速率限制），并在多种灾害数据源上优化日志记录、过滤、轮询与展示行为。
New Features:
Bug Fixes:
Enhancements:
detailed_jma_intensity。Documentation:
Original summary in English
Summary by Sourcery
Integrate CMA meteorological warnings via OpenQuakeAPI, add session-level weather alert aggregation with configurable windows and rate limiting, and refine logging, filtering, polling, and presentation behavior across multiple disaster data sources.
New Features:
Bug Fixes:
Enhancements:
Documentation:
Summary by CodeRabbit
新功能
改进
Bug 修复