From 43c59a9ea21d0ccd1d9286f805701498457a599f Mon Sep 17 00:00:00 2001 From: IceCodeNew <32576256+IceCodeNew@users.noreply.github.com> Date: Tue, 14 Jul 2026 10:52:55 +0800 Subject: [PATCH] feat: isolate RSS failures from briefing runs --- docs/design.md | 18 +- docs/requirements.md | 4 +- env.example | 2 +- tests/test_cli.py | 2 +- tests/test_config.py | 2 +- tests/test_service.py | 352 +++++++++++++++++++++++++++++++++--- tests/test_state.py | 53 ++++++ weather_briefing/config.py | 4 +- weather_briefing/service.py | 71 ++++++-- weather_briefing/state.py | 76 ++++++++ 10 files changed, 542 insertions(+), 42 deletions(-) diff --git a/docs/design.md b/docs/design.md index e6571345..2da7797d 100644 --- a/docs/design.md +++ b/docs/design.md @@ -81,7 +81,23 @@ SQLite 没有原生日期时间类型，状态存储需要直接对 TEXT 做范 ## 失败语义 -任一已配置 RSS 源重试耗尽会终止任务并增加失败计数。失败计数首次达到阈值时通过投递 provider 发送一次运维提醒，成功任务会重置计数。无更新检测按“最后一次看见该源任意文章”计算，不因当天过滤而丢失活跃信号；同一轮长期无更新只提醒一次，看到更新后才重新开放下一次陈旧提醒。没有 RSS 配置时不会创建 RSS 请求或健康告警，天气 API 继续驱动每小时任务。 +### RSS 失败 + +RSS 是可选补充源，其失败不影响任务成功率；天气 API 是主要信息来源。 + +**获取失败** — 单个 RSS 源经可配置次数重试（默认 3 次，间隔 3–5 秒）后仍无法获取或解析。行为：记录警告日志，本次运行继续使用已成功获取的 RSS 内容和天气 API 数据。 + +**长期无更新** — RSS 源在配置的小时数（默认 24）内未曾见到任何新文章。判定基准为 `source_health` 表中记录的最后一次文章时间，不受当天本地日期筛选影响。 + +**告警** — 连续获取失败达到可配置阈值（`RSS_FAILURE_THRESHOLD`，默认 3）时向运维渠道发送告警；成功记录告警状态后，同一失败周期不再重复告警，源恢复后重置计数器并重新开放告警。长期无更新采用相同规则，看到新文章后重新开放。两类 RSS 健康告警均为至少一次投递：投递或状态写入失败只记录日志并在后续任务中重试，不得终止简报任务；投递成功但状态写入失败时可能产生内容相同的重复告警，接收端应按告警类型、源集合和失败周期去重。 + +未配置任何 RSS 源时不创建 RSS 请求或健康告警，天气 API 独立驱动每小时和每日任务。 + +### 任务失败 + +**触发条件**：天气上下文获取失败（所有 provider 均不可用）、辅助上下文源获取失败（例如 `HTTPContextSource.fetch()` 抛出 `SourceFetchError`）、LLM 调用或输出校验失败、配置校验失败。 + +**行为**：终止当前任务，并立即通过投递 provider 发送一次性运维提醒。投递失败时在后续任务失败中继续重试；成功投递后，同一失败周期不再重复告警。成功的任务重置计数和告警状态。RSS 获取失败不计入任务失败计数。 小时 LLM 结果包含布尔字段 `should_publish`。模型比较当前及历史 API 快照，仅在降雨、显著天气变化、预警或灾害动态值得打扰时设为真；活动预警不允许与 false 同时出现。false 结果不投递消息，但当前快照、文章去重和预警状态仍持久化。 diff --git a/docs/requirements.md b/docs/requirements.md index de8faedc..8860f822 100644 --- a/docs/requirements.md +++ b/docs/requirements.md @@ -34,8 +34,8 @@ ## 可靠性 1. RSS 请求失败后随机等待 3–5 秒并重试；重试次数可配置。 -2. 重试耗尽则本次任务失败。 -3. 连续三次任务失败，或任一已配置 RSS 源超过 24 小时没有新文章，向运维通知渠道报警。未配置 RSS 不属于故障。 +2. 重试耗尽则记录警告日志并继续任务；RSS 是可选补充源，其获取失败不终止任务，但连续失败达到阈值时独立发出运维告警。连续失败和长期无更新告警均采用至少一次投递；投递或状态写入失败时记录日志并在后续任务中重试，不得使简报任务失败，接收端应按告警类型、源集合和失败周期去重。 +3. 任务发生非 RSS 错误时立即向运维通知渠道报警；投递失败时在后续任务失败中重试，成功投递后同一轮失败不重复告警，任务成功后重新开放。RSS 获取连续失败达到可配置阈值（默认 3）时独立报警，与任务失败隔离。任一已配置 RSS 源超过配置的时长（默认 24 小时）没有新文章也触发报警。未配置 RSS 不属于故障。 ## 架构与隐私 diff --git a/env.example b/env.example index d7b4d135..7f8c2211 100644 --- a/env.example +++ b/env.example @@ -75,6 +75,6 @@ BRIEFING_MAX_CHARACTERS=3500 # Optional state and reliability tuning. BRIEFING_STATE_PATH=state/weather.sqlite3 -TASK_FAILURE_THRESHOLD=3 +RSS_FAILURE_THRESHOLD=3 WARNING_RETENTION_HOURS=12 HISTORY_HOURS=48 diff --git a/tests/test_cli.py b/tests/test_cli.py index 7219bf94..84b0ab8c 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -330,7 +330,7 @@ def _make_fake_settings(**overrides: object) -> object: "rss_retry_min_seconds": 3, "rss_retry_max_seconds": 5, "rss_stale_hours": 24, - "task_failure_threshold": 3, + "rss_failure_threshold": 3, "warning_retention_hours": 12, "history_hours": 48, "briefing_max_characters": 3500, diff --git a/tests/test_config.py b/tests/test_config.py index a49a13fe..5e183e0f 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -223,8 +223,8 @@ def test_debug_accepts_truthy_values_with_optional_outer_quotes(monkeypatch, val "name", ( "HTTP_TIMEOUT_SECONDS", + "RSS_FAILURE_THRESHOLD", "RSS_STALE_HOURS", - "TASK_FAILURE_THRESHOLD", "WARNING_RETENTION_HOURS", "HISTORY_HOURS", ), diff --git a/tests/test_service.py b/tests/test_service.py index 293b1952..518c71c6 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -1,3 +1,4 @@ +import asyncio from pathlib import Path from types import SimpleNamespace from typing import Any, cast @@ -18,6 +19,7 @@ from weather_briefing.render import PlainTextRenderer from weather_briefing.service import BriefingService from weather_briefing.state import SQLiteStateStore +from weather_briefing.weather_context import WeatherContextError class EmptyRSSSource: @@ -38,6 +40,25 @@ async def fetch(self, config: object) -> tuple[Article, ...]: raise RuntimeError("feed unavailable") +class CanceledRSSSource: + async def fetch(self, config: object) -> tuple[Article, ...]: + raise asyncio.CancelledError + + +class MixedOutcomeRSSSource: + async def fetch(self, config: FeedConfig) -> tuple[Article, ...]: + if config.id.startswith("canceled-"): + raise asyncio.CancelledError + if config.id == "failing-feed": + raise RuntimeError("feed unavailable") + return () + + +class FailingWeatherContextProvider: + async def fetch(self, latitude: float, longitude: float) -> WeatherContextSnapshot: + raise WeatherContextError("weather context unavailable") + + class EmptyContextSource: async def fetch(self, config: object) -> object: raise AssertionError("No context source should be requested in this test") @@ -112,6 +133,18 @@ async def publish(self, message: RenderedMessage, *, single_message: bool = Fals self.messages.append((message, single_message)) +class FailOncePublisher(RecordingPublisher): + def __init__(self) -> None: + super().__init__() + self.attempts = 0 + + async def publish(self, message: RenderedMessage, *, single_message: bool = False) -> None: + self.attempts += 1 + if self.attempts == 1: + raise RuntimeError("delivery unavailable") + await super().publish(message, single_message=single_message) + + def _location() -> ResolvedLocation: return ResolvedLocation( id="test", @@ -133,8 +166,8 @@ async def test_daily_briefing_uses_configured_coordinates_and_air_quality_contex timezone=timezone, feeds=(), context_sources=(), - task_failure_threshold=3, rss_stale_hours=24, + rss_failure_threshold=3, warning_retention_hours=12, history_hours=48, briefing_max_characters=3500, @@ -188,8 +221,8 @@ async def test_hourly_briefing_also_uses_the_llm_provider(tmp_path: Path) -> Non timezone=timezone, feeds=(FeedConfig("feed", "Weather feed", "https://example.invalid/rss"),), context_sources=(), - task_failure_threshold=3, rss_stale_hours=24, + rss_failure_threshold=3, warning_retention_hours=12, history_hours=48, briefing_max_characters=3500, @@ -226,8 +259,8 @@ async def test_hourly_api_only_update_can_be_remembered_without_delivery( timezone=timezone, feeds=(), context_sources=(), - task_failure_threshold=3, rss_stale_hours=24, + rss_failure_threshold=3, warning_retention_hours=12, history_hours=48, briefing_max_characters=3500, @@ -282,8 +315,8 @@ async def test_service_rejects_mode_specific_llm_contract_violations( timezone=timezone, feeds=(), context_sources=(), - task_failure_threshold=3, rss_stale_hours=24, + rss_failure_threshold=3, warning_retention_hours=12, history_hours=48, briefing_max_characters=3500, @@ -291,6 +324,8 @@ async def test_service_rejects_mode_specific_llm_contract_violations( ) publisher = RecordingPublisher() delivery = DeliveryProvider(PlainTextRenderer(), publisher) + ops_publisher = RecordingPublisher() + ops_delivery = DeliveryProvider(PlainTextRenderer(), ops_publisher) with SQLiteStateStore(tmp_path / f"{kind}.sqlite3") as state: service = BriefingService( @@ -301,7 +336,7 @@ async def test_service_rejects_mode_specific_llm_contract_violations( cast(Any, EmptyContextSource()), llm, delivery, - delivery, + ops_delivery, StaticWeatherContextProvider(), ) with pytest.raises(LLMError, match="validation failed") as error: @@ -311,16 +346,16 @@ async def test_service_rejects_mode_specific_llm_contract_violations( assert publisher.messages == [] -async def test_failure_alert_is_sent_only_when_threshold_is_first_reached( +async def test_task_failure_alert_is_sent_only_on_first_consecutive_failure( tmp_path: Path, ) -> None: timezone = pendulum.timezone("Asia/Shanghai") settings = SimpleNamespace( timezone=timezone, - feeds=(FeedConfig("feed", "Feed", "https://example.invalid/feed"),), + feeds=(), context_sources=(), - task_failure_threshold=3, rss_stale_hours=24, + rss_failure_threshold=3, warning_retention_hours=12, history_hours=48, briefing_max_characters=3500, @@ -335,19 +370,77 @@ async def test_failure_alert_is_sent_only_when_threshold_is_first_reached( cast(Any, settings), _location(), state, - cast(Any, FailingRSSSource()), + cast(Any, EmptyRSSSource()), cast(Any, EmptyContextSource()), RecordingLLM(), delivery, delivery, + FailingWeatherContextProvider(), ) - for attempt in range(4): - with pytest.raises(RuntimeError, match="feed unavailable") as error: - await service.run("hourly", now.add(hours=attempt)) - assert error.value.__notes__ == [f"Briefing run failed after {attempt + 1} consecutive failure(s)"] + # First failure: alert fires immediately + with pytest.raises(WeatherContextError, match="weather context unavailable") as error: + await service.run("hourly", now) + assert error.value.__notes__ == ["Briefing run failed"] + assert len(publisher.messages) == 1 + assert "任务执行失败" in publisher.messages[0][0].body - assert len(publisher.messages) == 1 - assert "连续失败 3 次" in publisher.messages[0][0].body + # Second consecutive failure: no new alert + with pytest.raises(WeatherContextError, match="weather context unavailable") as error: + await service.run("hourly", now.add(hours=1)) + assert error.value.__notes__ == ["Briefing run failed"] + assert len(publisher.messages) == 1 + + +async def test_task_failure_alert_delivery_failure_is_retried( + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + timezone = pendulum.timezone("Asia/Shanghai") + settings = SimpleNamespace( + timezone=timezone, + feeds=(), + context_sources=(), + rss_stale_hours=24, + rss_failure_threshold=3, + warning_retention_hours=12, + history_hours=48, + briefing_max_characters=3500, + llm_max_attempts=1, + ) + ops_publisher = FailOncePublisher() + delivery = DeliveryProvider(PlainTextRenderer(), RecordingPublisher()) + now = pendulum.datetime(2026, 7, 13, 9, tz=timezone) + + with SQLiteStateStore(tmp_path / "failure-alert.sqlite3") as state: + service = BriefingService( + cast(Any, settings), + _location(), + state, + cast(Any, EmptyRSSSource()), + cast(Any, EmptyContextSource()), + RecordingLLM(), + delivery, + DeliveryProvider(PlainTextRenderer(), ops_publisher), + FailingWeatherContextProvider(), + ) + + with ( + caplog.at_level("ERROR", logger="weather_briefing.service"), + pytest.raises(WeatherContextError, match="weather context unavailable") as error, + ): + await service.run("hourly", now) + + with pytest.raises(WeatherContextError, match="weather context unavailable"): + await service.run("hourly", now.add(hours=1)) + assert len(ops_publisher.messages) == 1 + assert "任务执行失败" in ops_publisher.messages[0][0].body + + with pytest.raises(WeatherContextError, match="weather context unavailable"): + await service.run("hourly", now.add(hours=2)) + assert len(ops_publisher.messages) == 1 + + assert error.value.__notes__ == ["Briefing run failed"] + assert "Failed to publish or record briefing failure alert" in caplog.text async def test_daily_briefing_publishes_verbatim_articles(tmp_path: Path) -> None: @@ -367,8 +460,8 @@ async def test_daily_briefing_publishes_verbatim_articles(tmp_path: Path) -> Non timezone=timezone, feeds=(FeedConfig("feed", "Feed", "https://example.invalid/rss"),), context_sources=(), - task_failure_threshold=3, rss_stale_hours=24, + rss_failure_threshold=3, warning_retention_hours=12, history_hours=48, briefing_max_characters=3500, @@ -402,8 +495,8 @@ async def test_run_returns_none_when_no_content_and_no_warnings(tmp_path: Path) timezone=timezone, feeds=(), context_sources=(), - task_failure_threshold=3, rss_stale_hours=24, + rss_failure_threshold=3, warning_retention_hours=12, history_hours=48, briefing_max_characters=3500, @@ -446,8 +539,8 @@ async def test_stale_feed_triggers_ops_alert(tmp_path: Path) -> None: timezone=timezone, feeds=(FeedConfig("feed", "Feed", "https://example.invalid/rss"),), context_sources=(), - task_failure_threshold=3, rss_stale_hours=1, + rss_failure_threshold=3, warning_retention_hours=12, history_hours=48, briefing_max_characters=3500, @@ -520,8 +613,8 @@ async def test_llm_retry_on_validation_failure(tmp_path: Path) -> None: timezone=timezone, feeds=(FeedConfig("feed", "Feed", "https://example.invalid/rss"),), context_sources=(), - task_failure_threshold=3, rss_stale_hours=24, + rss_failure_threshold=3, warning_retention_hours=12, history_hours=48, briefing_max_characters=3500, @@ -555,8 +648,8 @@ async def test_briefing_exceeding_character_limit_is_rejected(tmp_path: Path) -> timezone=timezone, feeds=(), context_sources=(), - task_failure_threshold=3, rss_stale_hours=24, + rss_failure_threshold=3, warning_retention_hours=12, history_hours=48, briefing_max_characters=10, @@ -610,8 +703,8 @@ async def test_is_forecast_article_returns_false_for_unknown_feed(tmp_path: Path timezone=timezone, feeds=(FeedConfig("known-feed", "Known", "https://example.invalid/rss"),), context_sources=(), - task_failure_threshold=3, rss_stale_hours=24, + rss_failure_threshold=3, warning_retention_hours=12, history_hours=48, briefing_max_characters=3500, @@ -637,3 +730,220 @@ async def test_is_forecast_article_returns_false_for_unknown_feed(tmp_path: Path assert body is None assert llm.payload is None assert publisher.messages == [] + + +async def test_rss_failure_does_not_crash_daily_task_with_weather_context( + tmp_path: Path, +) -> None: + timezone = pendulum.timezone("Asia/Shanghai") + settings = SimpleNamespace( + timezone=timezone, + feeds=(FeedConfig("failing-feed", "Failing", "https://example.invalid/feed"),), + context_sources=(), + rss_stale_hours=24, + rss_failure_threshold=3, + warning_retention_hours=12, + history_hours=48, + briefing_max_characters=3500, + llm_max_attempts=3, + ) + llm = RecordingLLM() + publisher = RecordingPublisher() + delivery = DeliveryProvider(PlainTextRenderer(), publisher) + now = pendulum.datetime(2026, 7, 13, 8, tz=timezone) + + with SQLiteStateStore(tmp_path / "rss-fail.sqlite3") as state: + service = BriefingService( + cast(Any, settings), + _location(), + state, + cast(Any, FailingRSSSource()), + cast(Any, EmptyContextSource()), + llm, + delivery, + delivery, + StaticWeatherContextProvider(), + ) + body = await service.run("daily", now) + + assert body is not None + assert len(publisher.messages) == 1 + + +async def test_rss_cancellation_aborts_task_without_recording_failure( + tmp_path: Path, +) -> None: + timezone = pendulum.timezone("Asia/Shanghai") + settings = SimpleNamespace( + timezone=timezone, + feeds=(FeedConfig("canceled-feed", "Canceled", "https://example.invalid/feed"),), + context_sources=(), + rss_stale_hours=24, + rss_failure_threshold=1, + warning_retention_hours=12, + history_hours=48, + briefing_max_characters=3500, + llm_max_attempts=1, + ) + publisher = RecordingPublisher() + delivery = DeliveryProvider(PlainTextRenderer(), publisher) + now = pendulum.datetime(2026, 7, 13, 9, tz=timezone) + + with SQLiteStateStore(tmp_path / "rss-canceled.sqlite3") as state: + service = BriefingService( + cast(Any, settings), + _location(), + state, + cast(Any, CanceledRSSSource()), + cast(Any, EmptyContextSource()), + RecordingLLM(), + delivery, + delivery, + StaticWeatherContextProvider(), + ) + + with pytest.raises(asyncio.CancelledError): + await service.run("hourly", now) + + assert state.rss_sources_requiring_failure_alert(("canceled-feed",), 1) == [] + + assert publisher.messages == [] + + +async def test_rss_cancellation_records_other_completed_feed_results( + tmp_path: Path, +) -> None: + timezone = pendulum.timezone("Asia/Shanghai") + settings = SimpleNamespace( + timezone=timezone, + feeds=( + FeedConfig("canceled-feed", "Canceled", "https://example.invalid/canceled"), + FeedConfig("canceled-other", "Also canceled", "https://example.invalid/canceled-other"), + FeedConfig("recovered-feed", "Recovered", "https://example.invalid/recovered"), + FeedConfig("failing-feed", "Failing", "https://example.invalid/failing"), + ), + context_sources=(), + rss_stale_hours=24, + rss_failure_threshold=1, + warning_retention_hours=12, + history_hours=48, + briefing_max_characters=3500, + llm_max_attempts=1, + ) + delivery = DeliveryProvider(PlainTextRenderer(), RecordingPublisher()) + now = pendulum.datetime(2026, 7, 13, 9, tz=timezone) + + with SQLiteStateStore(tmp_path / "rss-canceled-results.sqlite3") as state: + state.record_rss_fetch_failure("recovered-feed") + service = BriefingService( + cast(Any, settings), + _location(), + state, + cast(Any, MixedOutcomeRSSSource()), + cast(Any, EmptyContextSource()), + RecordingLLM(), + delivery, + delivery, + StaticWeatherContextProvider(), + ) + + with pytest.raises(asyncio.CancelledError): + await service.run("hourly", now) + + assert state.rss_sources_requiring_failure_alert(("recovered-feed",), 1) == [] + assert state.rss_sources_requiring_failure_alert(("failing-feed",), 1) == ["failing-feed"] + + +async def test_rss_failure_alert_is_sent_after_threshold( + tmp_path: Path, +) -> None: + timezone = pendulum.timezone("Asia/Shanghai") + settings = SimpleNamespace( + timezone=timezone, + feeds=(FeedConfig("fail-feed", "Failing", "https://example.invalid/feed"),), + context_sources=(), + rss_stale_hours=24, + rss_failure_threshold=2, + warning_retention_hours=12, + history_hours=48, + briefing_max_characters=3500, + llm_max_attempts=1, + ) + llm = RecordingLLM() + ops_publisher = RecordingPublisher() + ops_delivery = DeliveryProvider(PlainTextRenderer(), ops_publisher) + publisher = RecordingPublisher() + delivery = DeliveryProvider(PlainTextRenderer(), publisher) + now = pendulum.datetime(2026, 7, 13, 9, tz=timezone) + + with SQLiteStateStore(tmp_path / "rss-alert.sqlite3") as state: + service = BriefingService( + cast(Any, settings), + _location(), + state, + cast(Any, FailingRSSSource()), + cast(Any, EmptyContextSource()), + llm, + delivery, + ops_delivery, + StaticWeatherContextProvider(), + ) + # First failure: no alert yet (threshold is 2) + await service.run("hourly", now) + assert ops_publisher.messages == [] + + # Second failure: alert should trigger + await service.run("hourly", now.add(hours=1)) + assert len(ops_publisher.messages) == 1 + assert "持续获取失败" in ops_publisher.messages[0][0].body + + # Third failure: no new alert (already alerted) + await service.run("hourly", now.add(hours=2)) + assert len(ops_publisher.messages) == 1 + + +async def test_failed_rss_alert_delivery_is_retried( + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + timezone = pendulum.timezone("Asia/Shanghai") + settings = SimpleNamespace( + timezone=timezone, + feeds=(FeedConfig("fail-feed", "Failing", "https://example.invalid/feed"),), + context_sources=(), + rss_stale_hours=24, + rss_failure_threshold=1, + warning_retention_hours=12, + history_hours=48, + briefing_max_characters=3500, + llm_max_attempts=1, + ) + ops_publisher = FailOncePublisher() + ops_delivery = DeliveryProvider(PlainTextRenderer(), ops_publisher) + publisher = RecordingPublisher() + delivery = DeliveryProvider(PlainTextRenderer(), publisher) + now = pendulum.datetime(2026, 7, 13, 9, tz=timezone) + + with SQLiteStateStore(tmp_path / "rss-alert-retry.sqlite3") as state: + service = BriefingService( + cast(Any, settings), + _location(), + state, + cast(Any, FailingRSSSource()), + cast(Any, EmptyContextSource()), + RecordingLLM(), + delivery, + ops_delivery, + StaticWeatherContextProvider(), + ) + + with caplog.at_level("ERROR", logger="weather_briefing.service"): + await service.run("hourly", now) + assert state.rss_sources_requiring_failure_alert(("fail-feed",), 1) == ["fail-feed"] + assert len(publisher.messages) == 1 + + await service.run("hourly", now.add(hours=1)) + assert state.rss_sources_requiring_failure_alert(("fail-feed",), 1) == [] + + assert "Failed to publish or record RSS health alert" in caplog.text + assert any("持续获取失败" in message.body for message, _ in ops_publisher.messages) diff --git a/tests/test_state.py b/tests/test_state.py index 77bb7f95..68ece135 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -31,6 +31,15 @@ def test_new_empty_source_is_not_immediately_stale(tmp_path: Path) -> None: assert state.stale_sources(("source",), now, 24) == [] +def test_failed_source_check_preserves_last_article_time(tmp_path: Path) -> None: + now = pendulum.datetime(2026, 7, 13, 9, tz="Asia/Shanghai") + with SQLiteStateStore(tmp_path / "state.db") as state: + state.record_source_check("source", now, now.subtract(hours=25)) + state.record_source_check("source", now.add(hours=1), None) + + assert state.stale_sources(("source",), now.add(hours=2), 24) == ["source"] + + def test_stale_source_alert_is_sent_once_until_a_new_article_arrives(tmp_path: Path) -> None: first_article = pendulum.datetime(2026, 7, 11, 8, tz="UTC") stale_check = pendulum.datetime(2026, 7, 13, 9, tz="Asia/Shanghai") @@ -113,6 +122,39 @@ def test_mark_stale_sources_alerted_with_empty_ids_is_noop(tmp_path: Path) -> No state.mark_stale_sources_alerted((), now) +def test_mark_rss_failure_alerted_with_empty_ids_is_noop(tmp_path: Path) -> None: + now = pendulum.datetime(2026, 7, 13, 9, tz="Asia/Shanghai") + with SQLiteStateStore(tmp_path / "state.db") as state: + state.mark_rss_failure_alerted((), now) + + +def test_rss_failure_alert_is_suppressed_until_fetch_success(tmp_path: Path) -> None: + now = pendulum.datetime(2026, 7, 13, 9, tz="Asia/Shanghai") + with SQLiteStateStore(tmp_path / "state.db") as state: + assert state.record_rss_fetch_failure("source") == 1 + assert state.rss_sources_requiring_failure_alert(("source",), 2) == [] + + assert state.record_rss_fetch_failure("source") == 2 + assert state.rss_sources_requiring_failure_alert(("source",), 2) == ["source"] + state.mark_rss_failure_alerted(("source",), now) + assert state.rss_sources_requiring_failure_alert(("source",), 2) == [] + + assert state.record_rss_fetch_failure("source") == 3 + assert state.rss_sources_requiring_failure_alert(("source",), 2) == [] + + state.record_rss_fetch_success("source") + assert state.record_rss_fetch_failure("source") == 1 + assert state.rss_sources_requiring_failure_alert(("source",), 1) == ["source"] + + +def test_rss_failure_alert_sources_are_sorted(tmp_path: Path) -> None: + with SQLiteStateStore(tmp_path / "state.db") as state: + state.record_rss_fetch_failure("beta") + state.record_rss_fetch_failure("alpha") + + assert state.rss_sources_requiring_failure_alert(("beta", "alpha"), 1) == ["alpha", "beta"] + + def test_record_failure_increments_consecutive_count(tmp_path: Path) -> None: with SQLiteStateStore(tmp_path / "state.db") as state: assert state.record_failure() == 1 @@ -128,6 +170,17 @@ def test_record_success_resets_failure_count(tmp_path: Path) -> None: assert state.record_failure() == 1 +def test_record_success_reopens_task_failure_alert(tmp_path: Path) -> None: + now = pendulum.datetime(2026, 7, 13, 9, tz="Asia/Shanghai") + with SQLiteStateStore(tmp_path / "state.db") as state: + assert state.task_failure_requires_alert() + state.mark_task_failure_alerted(now) + assert not state.task_failure_requires_alert() + + state.record_success() + assert state.task_failure_requires_alert() + + def test_parse_time_rejects_invalid_format() -> None: import pytest diff --git a/weather_briefing/config.py b/weather_briefing/config.py index 69deb26d..f11a43eb 100644 --- a/weather_briefing/config.py +++ b/weather_briefing/config.py @@ -196,7 +196,7 @@ class Settings: rss_retry_min_seconds: float rss_retry_max_seconds: float rss_stale_hours: int - task_failure_threshold: int + rss_failure_threshold: int warning_retention_hours: int history_hours: int briefing_max_characters: int @@ -327,7 +327,7 @@ def from_env(cls) -> Settings: rss_retry_min_seconds=retry_min, rss_retry_max_seconds=retry_max, rss_stale_hours=_positive_integer("RSS_STALE_HOURS", 24), - task_failure_threshold=_positive_integer("TASK_FAILURE_THRESHOLD", 3), + rss_failure_threshold=_positive_integer("RSS_FAILURE_THRESHOLD", 3), warning_retention_hours=_positive_integer("WARNING_RETENTION_HOURS", 12), history_hours=_positive_integer("HISTORY_HOURS", 48), briefing_max_characters=briefing_max_characters, diff --git a/weather_briefing/service.py b/weather_briefing/service.py index c1bf331d..54532ccc 100644 --- a/weather_briefing/service.py +++ b/weather_briefing/service.py @@ -47,13 +47,17 @@ async def run(self, kind: str, now: pendulum.DateTime | None = None) -> str | No try: body = await self._run(kind, current_time) except Exception as exc: - failure_count = self._state.record_failure() - exc.add_note(f"Briefing run failed after {failure_count} consecutive failure(s)") - if failure_count == self._settings.task_failure_threshold: - await self._ops_delivery.publish_alert( - "天气简报任务连续失败", - f"任务已连续失败 {failure_count} 次，请检查运行日志和私密源配置。", - ) + self._state.record_failure() + exc.add_note("Briefing run failed") + try: + if self._state.task_failure_requires_alert(): + await self._ops_delivery.publish_alert( + "天气简报任务执行失败", + "任务执行失败，请检查运行日志、天气 API 及私密源配置。", + ) + self._state.mark_task_failure_alerted(current_time) + except Exception: + _LOGGER.exception("Failed to publish or record briefing failure alert") raise self._state.record_success() return body @@ -63,12 +67,41 @@ async def _run(self, kind: str, now: pendulum.DateTime) -> str | None: feed for feed in self._settings.feeds if not feed.location_ids or self._location.id in feed.location_ids ) _LOGGER.debug("Fetching %d RSS feed(s)", len(feeds)) - fetched = await asyncio.gather(*(self._rss_source.fetch(config) for config in feeds)) + results = await asyncio.gather( + *(self._rss_source.fetch(config) for config in feeds), + return_exceptions=True, + ) + fetched: list[tuple[Article, ...]] = [] + pending_cancellation: BaseException | None = None + for result, config in zip(results, feeds, strict=True): + if isinstance(result, BaseException): + if not isinstance(result, Exception): + if pending_cancellation is None: + pending_cancellation = result + continue + _LOGGER.warning("RSS source %s failed: %s", config.id, result) + fetched.append(()) + self._state.record_source_check(config.id, now, None) + self._state.record_rss_fetch_failure(config.id) + else: + fetched.append(result) + latest_at = max((article.published_at for article in result), default=None) + self._state.record_source_check(config.id, now, latest_at) + self._state.record_rss_fetch_success(config.id) + if pending_cancellation is not None: + raise pending_cancellation all_articles = tuple(article for group in fetched for article in group) _LOGGER.info("Fetched %d article(s) from %d feed(s)", len(all_articles), len(fetched)) - for config, articles in zip(feeds, fetched, strict=True): - latest_at = max((article.published_at for article in articles), default=None) - self._state.record_source_check(config.id, now, latest_at) + rss_failure_alert_ids = self._state.rss_sources_requiring_failure_alert( + tuple(config.id for config in feeds), + self._settings.rss_failure_threshold, + ) + if rss_failure_alert_ids: + await self._publish_rss_health_alert( + "天气 RSS 源持续获取失败", + f"以下 RSS 源已连续失败 {self._settings.rss_failure_threshold} 次：{', '.join(rss_failure_alert_ids)}", + lambda: self._state.mark_rss_failure_alerted(tuple(rss_failure_alert_ids), now), + ) stale = self._state.stale_sources_requiring_alert( tuple(config.id for config in feeds), now, @@ -76,11 +109,11 @@ async def _run(self, kind: str, now: pendulum.DateTime) -> str | None: ) if stale: _LOGGER.warning("Stale RSS source(s): %s", ", ".join(stale)) - await self._ops_delivery.publish_alert( + await self._publish_rss_health_alert( "天气 RSS 源长时间无更新", f"以下源超过 {self._settings.rss_stale_hours} 小时无新文章：{', '.join(stale)}", + lambda: self._state.mark_stale_sources_alerted(tuple(stale), now), ) - self._state.mark_stale_sources_alerted(tuple(stale), now) local_now = now.in_timezone(self._settings.timezone) today_start = local_now.start_of("day") tomorrow_start = today_start.add(days=1) @@ -184,6 +217,18 @@ def validate_length(candidate: BriefingResult) -> None: ) return message.body + async def _publish_rss_health_alert( + self, + title: str, + body: str, + mark_alerted: Callable[[], None], + ) -> None: + try: + await self._ops_delivery.publish_alert(title, body) + mark_alerted() + except Exception: + _LOGGER.exception("Failed to publish or record RSS health alert") + def _save_result_state( self, kind: str, diff --git a/weather_briefing/state.py b/weather_briefing/state.py index 01c43655..54989d55 100644 --- a/weather_briefing/state.py +++ b/weather_briefing/state.py @@ -56,6 +56,14 @@ def _initialize(self) -> None: CREATE TABLE IF NOT EXISTS task_health ( singleton INTEGER PRIMARY KEY CHECK (singleton = 1), consecutive_failures INTEGER NOT NULL ); + CREATE TABLE IF NOT EXISTS task_failure_alert ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), alerted_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS rss_failure_tracker ( + source_id TEXT PRIMARY KEY, + consecutive_failures INTEGER NOT NULL DEFAULT 0, + failure_alerted_at TEXT + ); INSERT OR IGNORE INTO task_health(singleton, consecutive_failures) VALUES (1, 0); """ ) @@ -308,6 +316,7 @@ def update_warnings( def record_success(self) -> None: self._connection.execute("UPDATE task_health SET consecutive_failures = 0 WHERE singleton = 1") + self._connection.execute("DELETE FROM task_failure_alert WHERE singleton = 1") self._connection.commit() def record_failure(self) -> int: @@ -318,6 +327,73 @@ def record_failure(self) -> int: row = self._connection.execute("SELECT consecutive_failures FROM task_health WHERE singleton = 1").fetchone() return int(row["consecutive_failures"]) + def task_failure_requires_alert(self) -> bool: + row = self._connection.execute("SELECT 1 FROM task_failure_alert WHERE singleton = 1").fetchone() + return row is None + + def mark_task_failure_alerted(self, alerted_at: pendulum.DateTime) -> None: + self._connection.execute( + "INSERT OR REPLACE INTO task_failure_alert(singleton, alerted_at) VALUES (1, ?)", + (_storage_time(alerted_at),), + ) + self._connection.commit() + + def record_rss_fetch_failure(self, source_id: str) -> int: + self._connection.execute( + """INSERT INTO rss_failure_tracker(source_id, consecutive_failures, failure_alerted_at) + VALUES (?, 1, NULL) + ON CONFLICT(source_id) DO UPDATE SET + consecutive_failures = consecutive_failures + 1""", + (source_id,), + ) + self._connection.commit() + row = self._connection.execute( + "SELECT consecutive_failures FROM rss_failure_tracker WHERE source_id = ?", + (source_id,), + ).fetchone() + return int(row["consecutive_failures"]) + + def record_rss_fetch_success(self, source_id: str) -> None: + self._connection.execute( + "DELETE FROM rss_failure_tracker WHERE source_id = ?", + (source_id,), + ) + self._connection.commit() + + def rss_sources_requiring_failure_alert( + self, + source_ids: tuple[str, ...], + threshold: int, + ) -> list[str]: + if not source_ids: + return [] + placeholders = ",".join("?" for _ in source_ids) + rows = self._connection.execute( + f"""SELECT source_id + FROM rss_failure_tracker + WHERE source_id IN ({placeholders}) + AND consecutive_failures >= ? + AND failure_alerted_at IS NULL + ORDER BY source_id""", # noqa: S608 + (*source_ids, threshold), + ) + return [str(row["source_id"]) for row in rows] + + def mark_rss_failure_alerted( + self, + source_ids: tuple[str, ...], + alerted_at: pendulum.DateTime, + ) -> None: + if not source_ids: + return + placeholders = ",".join("?" for _ in source_ids) + self._connection.execute( + f"UPDATE rss_failure_tracker SET failure_alerted_at = ? " # noqa: S608 + f"WHERE source_id IN ({placeholders})", + (_storage_time(alerted_at), *source_ids), + ) + self._connection.commit() + def _parse_time(value: str) -> pendulum.DateTime: if not _STORAGE_TIME_PATTERN.fullmatch(value):