From 8bf5da72323a3ad4fe6a0bec169e29b78f052c03 Mon Sep 17 00:00:00 2001 From: IceCodeNew <32576256+IceCodeNew@users.noreply.github.com> Date: Wed, 15 Jul 2026 01:39:24 +0800 Subject: [PATCH 1/4] refactor: remove cast() bypasses and tighten JSON boundary types MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace explicit cast() calls in weather_context.py with direct type annotations on variables assigned from response.json() (which returns Any). Change _format_qweather_lifestyle and _format_qweather_day parameter types from object to dict[str, object] so isinstance-narrowed dicts support string-key subscripting without cast. Tighten load_reference_data return type from dict[str, Any] to dict[str, object]—json.loads returns Any, isinstance narrows to dict[Any, Any], which is assignable to dict[str, object]. --- weather_briefing/geocoding.py | 6 +++--- weather_briefing/reference_data.py | 2 +- weather_briefing/weather_context.py | 33 ++++++++++++++--------------- 3 files changed, 20 insertions(+), 21 deletions(-) diff --git a/weather_briefing/geocoding.py b/weather_briefing/geocoding.py index 2986633f..b2bcfeaf 100644 --- a/weather_briefing/geocoding.py +++ b/weather_briefing/geocoding.py @@ -134,7 +134,7 @@ def __init__( async def geocode(self, location: LocationSpec) -> ResolvedLocation: async with self._lock: - result: dict[str, Any] | None = None + result: dict[str, object] | None = None for query in _nominatim_queries(location.name): delay = 1.0 - (time.monotonic() - self._last_request_at) if delay > 0: @@ -299,13 +299,13 @@ def _nominatim_queries(name: str) -> tuple[str, ...]: return tuple(dict.fromkeys((normalized, name))) -def _nominatim_result_matches(name: str, result: dict[str, Any]) -> bool: +def _nominatim_result_matches(name: str, result: dict[str, object]) -> bool: display_name = str(result.get("display_name", "")).casefold() specific_name = _specific_location_name(name) return specific_name.casefold() in display_name -def _open_meteo_result_matches(name: str, result: dict[str, Any]) -> bool: +def _open_meteo_result_matches(name: str, result: dict[str, object]) -> bool: result_description = " ".join( str(result.get(field, "")) for field in ("name", "admin1", "admin2", "admin3", "admin4", "country") ).casefold() diff --git a/weather_briefing/reference_data.py b/weather_briefing/reference_data.py index 5e491e1b..4a5871ec 100644 --- a/weather_briefing/reference_data.py +++ b/weather_briefing/reference_data.py @@ -14,7 +14,7 @@ class ReferenceDataError(RuntimeError): @cache -def load_reference_data(filename: str) -> dict[str, Any]: +def load_reference_data(filename: str) -> dict[str, object]: if PurePath(filename).name != filename or not filename.endswith(".json"): raise ReferenceDataError("Reference data filename must identify one JSON file") try: diff --git a/weather_briefing/weather_context.py b/weather_briefing/weather_context.py index 50e5bae3..95e697de 100644 --- a/weather_briefing/weather_context.py +++ b/weather_briefing/weather_context.py @@ -7,7 +7,7 @@ from collections.abc import Callable from contextlib import suppress from dataclasses import replace -from typing import Any, Protocol, cast, runtime_checkable +from typing import Any, Protocol, runtime_checkable import httpx import jwt @@ -203,7 +203,7 @@ async def fetch( raise WeatherContextError("QWeather returned no daily forecast") raise WeatherContextError(f"QWeather returned no forecast for {forecast_date}") - indices_payload: dict[str, Any] = {} + indices_payload: dict[str, object] = {} lifestyle_advice: tuple[str, ...] = () if forecast_date is None or str(forecast_date) == first_forecast_date: operation = "lifestyle indices" @@ -355,14 +355,15 @@ async def fetch( ) response.raise_for_status() payload = response.json() - daily = cast(dict[str, list[object]], payload["daily"]) + daily: dict[str, list[object]] = payload["daily"] times = daily["time"] forecast_count = min(2, len(times)) if forecast_date is None else len(times) weather_forecast = tuple(_format_open_meteo_day(daily, index) for index in range(forecast_count)) if not weather_forecast: raise WeatherContextError("Open-Meteo returned no daily forecast") + current: dict[str, object] = payload["current"] observed_at = parse_datetime_with_default_timezone( - str(cast(dict[str, object], payload["current"])["time"]), + str(current["time"]), str(payload["timezone"]), context="Open-Meteo weather update time", ) @@ -424,7 +425,7 @@ async def _fetch_air_quality_and_allergen( ) response.raise_for_status() payload = response.json() - current = cast(dict[str, Any], payload["current"]) + current: dict[str, object] = payload["current"] except (httpx.HTTPError, KeyError, TypeError, ValueError) as exc: _LOGGER.warning( "Weather API optional call failed provider=open-meteo operation=air-quality reason=%s", @@ -677,31 +678,29 @@ def _first_attribution(payload: dict[str, object]) -> str | None: return str(attributions[0]) -def _aqi_standard(index: dict[str, Any]) -> str: +def _aqi_standard(index: dict[str, object]) -> str: code = str(index["code"]) name = str(index.get("name") or code) return name if name == code else f"{name}（{code}）" -def _format_qweather_lifestyle(item: object) -> str: +def _format_qweather_lifestyle(item: dict[str, object]) -> str: if not isinstance(item, dict): raise ValueError("QWeather lifestyle index must be an object") - values = cast(dict[str, Any], item) - name = str(values["name"]) - category = str(values.get("category", "未知")) - text = str(values.get("text") or "无详细建议") + name = str(item["name"]) + category = str(item.get("category", "未知")) + text = str(item.get("text") or "无详细建议") return f"{name}（{category}）：{text}" -def _format_qweather_day(item: object) -> str: +def _format_qweather_day(item: dict[str, object]) -> str: if not isinstance(item, dict): raise ValueError("QWeather daily forecast must be an object") - values = cast(dict[str, Any], item) return ( - f"{values['fxDate']}：{values['textDay']}转{values['textNight']}，" - f"{values['tempMin']}~{values['tempMax']}℃，" - f"{values['windDirDay']}{values['windScaleDay']}级，" - f"相对湿度{values['humidity']}%，预计降水量{values['precip']}毫米" + f"{item['fxDate']}：{item['textDay']}转{item['textNight']}，" + f"{item['tempMin']}~{item['tempMax']}℃，" + f"{item['windDirDay']}{item['windScaleDay']}级，" + f"相对湿度{item['humidity']}%，预计降水量{item['precip']}毫米" ) From cfc647633f999f459b0d757cf537cddf4f0545a8 Mon Sep 17 00:00:00 2001 From: IceCodeNew <32576256+IceCodeNew@users.noreply.github.com> Date: Wed, 15 Jul 2026 01:39:29 +0800 Subject: [PATCH 2/4] refactor: replace concrete source types with Protocols for testability Add RSSFeedSource and ContextDocumentSource Protocols to sources.py, mirroring the existing LLMProvider and WeatherContextProvider pattern. BriefingService now depends on these Protocols plus a new BriefingSettings Protocol (read-only @property members) instead of the concrete Settings, RSSSource, and HTTPContextSource classes. This removes all cast(Any, ...) calls in test_service.py: test doubles now satisfy the Protocols structurally, and a _TestSettings frozen dataclass replaces SimpleNamespace. Payload-access casts are replaced with a TypeGuard-based _is_dict_list helper. --- tests/test_service.py | 260 ++++++++++++++++++++---------------- weather_briefing/service.py | 41 +++++- weather_briefing/sources.py | 9 ++ 3 files changed, 190 insertions(+), 120 deletions(-) diff --git a/tests/test_service.py b/tests/test_service.py index 475e8b6c..8b44bc50 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -1,7 +1,7 @@ import asyncio +from dataclasses import dataclass from pathlib import Path -from types import SimpleNamespace -from typing import Any, cast +from typing import TypeGuard import pendulum import pytest @@ -10,9 +10,11 @@ from weather_briefing.models import ( AirQualitySnapshot, Article, + ContextSourceConfig, FeedConfig, RenderedMessage, ResolvedLocation, + SourceDocument, Warning, WeatherContextSnapshot, ) @@ -23,8 +25,25 @@ from weather_briefing.weather_context import WeatherContextError +@dataclass(frozen=True, slots=True) +class _TestSettings: + timezone: pendulum.Timezone + feeds: tuple[FeedConfig, ...] = () + context_sources: tuple[ContextSourceConfig, ...] = () + rss_stale_hours: int = 24 + rss_failure_threshold: int = 3 + warning_retention_hours: int = 12 + history_hours: int = 48 + briefing_max_characters: int = 3500 + llm_max_attempts: int = 3 + + +def _is_dict_list(value: object) -> TypeGuard[list[dict[str, object]]]: + return isinstance(value, list) and all(isinstance(item, dict) for item in value) + + class EmptyRSSSource: - async def fetch(self, config: object) -> tuple[object, ...]: + async def fetch(self, config: FeedConfig) -> tuple[Article, ...]: raise AssertionError("No RSS feed should be requested in this test") @@ -32,17 +51,17 @@ class StaticRSSSource: def __init__(self, article: Article) -> None: self._article = article - async def fetch(self, config: object) -> tuple[Article, ...]: + async def fetch(self, config: FeedConfig) -> tuple[Article, ...]: return (self._article,) class FailingRSSSource: - async def fetch(self, config: object) -> tuple[Article, ...]: + async def fetch(self, config: FeedConfig) -> tuple[Article, ...]: raise RuntimeError("feed unavailable") class CanceledRSSSource: - async def fetch(self, config: object) -> tuple[Article, ...]: + async def fetch(self, config: FeedConfig) -> tuple[Article, ...]: raise asyncio.CancelledError @@ -61,7 +80,7 @@ async def fetch(self, latitude: float, longitude: float) -> WeatherContextSnapsh class EmptyContextSource: - async def fetch(self, config: object) -> object: + async def fetch(self, config: ContextSourceConfig) -> SourceDocument: raise AssertionError("No context source should be requested in this test") @@ -107,12 +126,15 @@ def __init__( async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dict[str, object]: self.payload = payload - context = cast(list[dict[str, object]], payload["context_documents"]) - articles = ( - *cast(list[dict[str, object]], payload["new_articles"]), - *cast(list[dict[str, object]], payload["deferred_articles"]), - ) - source_id = str((context or articles)[0]["source_id"]) + source_documents: list[dict[str, object]] = [] + context_documents = payload["context_documents"] + assert _is_dict_list(context_documents) + source_documents.extend(context_documents) + for key in ("new_articles", "deferred_articles"): + group = payload[key] + assert _is_dict_list(group) + source_documents.extend(group) + source_id = str(source_documents[0]["source_id"]) conclusion = { "text": "AQI is 42 under test-standard.", "source_ids": [source_id], @@ -178,7 +200,7 @@ async def test_forecast_uses_configured_coordinates_and_air_quality_context( tmp_path: Path, ) -> None: timezone = pendulum.timezone("Asia/Shanghai") - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(), context_sources=(), @@ -198,11 +220,11 @@ async def test_forecast_uses_configured_coordinates_and_air_quality_context( with SQLiteStateStore(tmp_path / "state.sqlite3") as state: state.save_briefing("briefing", "Earlier update", now.subtract(hours=1)) service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, EmptyRSSSource()), - cast(Any, EmptyContextSource()), + EmptyRSSSource(), + EmptyContextSource(), llm, delivery, delivery, @@ -212,13 +234,15 @@ async def test_forecast_uses_configured_coordinates_and_air_quality_context( assert weather_context.coordinates == (39.911389, 116.380556) assert llm.payload is not None - context_documents = cast(list[dict[str, str]], llm.payload["context_documents"]) + context_documents = llm.payload["context_documents"] + assert _is_dict_list(context_documents) assert len(context_documents) == 2 air_document = next(item for item in context_documents if item["source_id"] == "air-quality:test") assert air_document["url"] == "https://example.invalid/air-quality" - assert "AQI：42（标准：test-standard" in air_document["content"] - assert "PM2.5 原始浓度：12 µg/m³" in air_document["content"] - recent_briefings = cast(list[dict[str, str]], llm.payload["recent_briefings"]) + assert "AQI：42（标准：test-standard" in str(air_document["content"]) + assert "PM2.5 原始浓度：12 µg/m³" in str(air_document["content"]) + recent_briefings = llm.payload["recent_briefings"] + assert _is_dict_list(recent_briefings) assert recent_briefings == [ { "mode": "briefing", @@ -234,7 +258,7 @@ async def test_forecast_date_is_separate_from_run_time_and_reaches_weather_provi timezone = pendulum.timezone("Asia/Shanghai") run_time = pendulum.datetime(2026, 7, 13, 22, 30, tz=timezone) target_date = pendulum.date(2026, 7, 15) - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(), context_sources=(), @@ -272,11 +296,11 @@ async def fetch_for_date( with SQLiteStateStore(tmp_path / "future-forecast.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, EmptyRSSSource()), - cast(Any, EmptyContextSource()), + EmptyRSSSource(), + EmptyContextSource(), llm, delivery, delivery, @@ -292,7 +316,7 @@ async def fetch_for_date( async def test_forecast_date_is_rejected_for_briefing_mode() -> None: service = object.__new__(BriefingService) - service._settings = cast(Any, SimpleNamespace(timezone=pendulum.timezone("Asia/Shanghai"))) + service._settings = _TestSettings(timezone=pendulum.timezone("Asia/Shanghai")) with pytest.raises(ValueError, match="only supported in forecast mode"): await service.run("briefing", forecast_date=pendulum.date(2026, 7, 15)) @@ -310,7 +334,7 @@ async def test_briefing_also_uses_the_llm_provider(tmp_path: Path) -> None: published_at=pendulum.datetime(2026, 7, 12, 23, 30, tz="UTC"), content="New weather information", ) - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(FeedConfig("feed", "Weather feed", "https://example.invalid/rss"),), context_sources=(), @@ -327,11 +351,11 @@ async def test_briefing_also_uses_the_llm_provider(tmp_path: Path) -> None: with SQLiteStateStore(tmp_path / "state.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, StaticRSSSource(article)), - cast(Any, EmptyContextSource()), + StaticRSSSource(article), + EmptyContextSource(), llm, delivery, delivery, @@ -340,7 +364,9 @@ async def test_briefing_also_uses_the_llm_provider(tmp_path: Path) -> None: assert llm.payload is not None assert llm.payload["mode"] == "briefing" - assert cast(list[dict[str, object]], llm.payload["new_articles"])[0]["source_id"] == "article-id" + new_articles = llm.payload["new_articles"] + assert _is_dict_list(new_articles) + assert new_articles[0]["source_id"] == "article-id" assert len(publisher.messages) == 1 @@ -348,7 +374,7 @@ async def test_briefing_api_only_update_can_be_remembered_without_delivery( tmp_path: Path, ) -> None: timezone = pendulum.timezone("Asia/Shanghai") - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(), context_sources=(), @@ -367,11 +393,11 @@ async def test_briefing_api_only_update_can_be_remembered_without_delivery( with SQLiteStateStore(tmp_path / "state.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, EmptyRSSSource()), - cast(Any, EmptyContextSource()), + EmptyRSSSource(), + EmptyContextSource(), llm, delivery, delivery, @@ -410,7 +436,7 @@ async def test_unchanged_active_warning_does_not_force_briefing_delivery(tmp_pat (article.id,), now, ) - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(), context_sources=(), @@ -449,11 +475,11 @@ async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dic state.save_articles((article,), now) state.update_warnings((warning,), (), now, {article.id}) service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, EmptyRSSSource()), - cast(Any, EmptyContextSource()), + EmptyRSSSource(), + EmptyContextSource(), UnchangedWarningLLM(), delivery, delivery, @@ -479,7 +505,7 @@ async def test_unpublished_article_is_included_until_a_later_briefing_is_publish published_at=now, content="A small change that may become relevant later", ) - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(FeedConfig("feed", "Weather feed", "https://example.invalid/rss"),), context_sources=(), @@ -497,10 +523,11 @@ def __init__(self) -> None: async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dict[str, object]: self.payloads.append(payload) - sources = ( - *cast(list[dict[str, object]], payload["new_articles"]), - *cast(list[dict[str, object]], payload["deferred_articles"]), - ) + sources: list[dict[str, object]] = [] + for key in ("new_articles", "deferred_articles"): + group = payload[key] + assert _is_dict_list(group) + sources.extend(group) source_id = str(sources[0]["source_id"]) return { "headline": "Accumulated update", @@ -519,11 +546,11 @@ async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dic with SQLiteStateStore(tmp_path / "deferred.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, StaticRSSSource(article)), - cast(Any, EmptyContextSource()), + StaticRSSSource(article), + EmptyContextSource(), llm, delivery, delivery, @@ -535,10 +562,14 @@ async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dic assert state.pending_articles() == () assert state.known_article_ids((article.id,)) == {article.id} - assert cast(list[dict[str, object]], llm.payloads[0]["new_articles"])[0]["source_id"] == article.id + first_new = llm.payloads[0]["new_articles"] + assert _is_dict_list(first_new) + assert first_new[0]["source_id"] == article.id assert llm.payloads[0]["deferred_articles"] == [] assert llm.payloads[1]["new_articles"] == [] - assert cast(list[dict[str, object]], llm.payloads[1]["deferred_articles"])[0]["source_id"] == article.id + second_deferred = llm.payloads[1]["deferred_articles"] + assert _is_dict_list(second_deferred) + assert second_deferred[0]["source_id"] == article.id assert len(publisher.messages) == 1 @@ -557,7 +588,7 @@ async def test_forced_briefing_publishes_deferred_information_and_clears_pending content="The temperature was 31 C at 15:00", is_verbatim=True, ) - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(FeedConfig("feed", "Weather feed", "https://example.invalid/rss"),), context_sources=(), @@ -574,11 +605,11 @@ async def test_forced_briefing_publishes_deferred_information_and_clears_pending with SQLiteStateStore(tmp_path / "forced.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, StaticRSSSource(article)), - cast(Any, EmptyContextSource()), + StaticRSSSource(article), + EmptyContextSource(), llm, delivery, delivery, @@ -597,7 +628,8 @@ async def test_forced_briefing_publishes_deferred_information_and_clears_pending assert state.pending_articles() == () assert llm.payload is not None - deferred = cast(list[dict[str, object]], llm.payload["deferred_articles"]) + deferred = llm.payload["deferred_articles"] + assert _is_dict_list(deferred) assert deferred[0]["source_id"] == article.id assert publisher.messages[0] == (RenderedMessage(body, len(body)), True, True) assert len(publisher.messages) == 2 @@ -607,7 +639,7 @@ async def test_forced_briefing_publishes_deferred_information_and_clears_pending async def test_final_window_keeps_worthy_briefing_notifications_enabled(tmp_path: Path) -> None: timezone = pendulum.timezone("Asia/Shanghai") now = pendulum.datetime(2026, 7, 13, 23, tz=timezone) - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(), context_sources=(), @@ -623,11 +655,11 @@ async def test_final_window_keeps_worthy_briefing_notifications_enabled(tmp_path with SQLiteStateStore(tmp_path / "worthy-final.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, EmptyRSSSource()), - cast(Any, EmptyContextSource()), + EmptyRSSSource(), + EmptyContextSource(), RecordingLLM(should_publish=True), delivery, delivery, @@ -658,7 +690,7 @@ async def test_service_rejects_mode_specific_llm_contract_violations( message: str, ) -> None: timezone = pendulum.timezone("Asia/Shanghai") - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(), context_sources=(), @@ -676,11 +708,11 @@ async def test_service_rejects_mode_specific_llm_contract_violations( with SQLiteStateStore(tmp_path / f"{kind}.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, EmptyRSSSource()), - cast(Any, EmptyContextSource()), + EmptyRSSSource(), + EmptyContextSource(), llm, delivery, ops_delivery, @@ -697,7 +729,7 @@ async def test_task_failure_alert_is_sent_only_on_first_consecutive_failure( tmp_path: Path, ) -> None: timezone = pendulum.timezone("Asia/Shanghai") - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(), context_sources=(), @@ -714,11 +746,11 @@ async def test_task_failure_alert_is_sent_only_on_first_consecutive_failure( with SQLiteStateStore(tmp_path / "failure.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, EmptyRSSSource()), - cast(Any, EmptyContextSource()), + EmptyRSSSource(), + EmptyContextSource(), RecordingLLM(), delivery, delivery, @@ -743,7 +775,7 @@ async def test_task_failure_alert_delivery_failure_is_retried( caplog: pytest.LogCaptureFixture, ) -> None: timezone = pendulum.timezone("Asia/Shanghai") - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(), context_sources=(), @@ -760,11 +792,11 @@ async def test_task_failure_alert_delivery_failure_is_retried( with SQLiteStateStore(tmp_path / "failure-alert.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, EmptyRSSSource()), - cast(Any, EmptyContextSource()), + EmptyRSSSource(), + EmptyContextSource(), RecordingLLM(), delivery, DeliveryProvider(PlainTextRenderer(), ops_publisher), @@ -803,7 +835,7 @@ async def test_forecast_publishes_verbatim_articles(tmp_path: Path, caplog) -> N content="Raw forecast", is_verbatim=True, ) - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(FeedConfig("feed", "Feed", "https://example.invalid/rss"),), context_sources=(), @@ -819,11 +851,11 @@ async def test_forecast_publishes_verbatim_articles(tmp_path: Path, caplog) -> N with caplog.at_level("DEBUG"), SQLiteStateStore(tmp_path / "v.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, StaticRSSSource(verbatim)), - cast(Any, EmptyContextSource()), + StaticRSSSource(verbatim), + EmptyContextSource(), RecordingLLM(), delivery, delivery, @@ -843,7 +875,7 @@ async def test_forecast_publishes_verbatim_articles(tmp_path: Path, caplog) -> N async def test_run_returns_none_when_no_content_and_no_warnings(tmp_path: Path) -> None: timezone = pendulum.timezone("Asia/Shanghai") now = pendulum.datetime(2026, 7, 13, 9, tz=timezone) - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(), context_sources=(), @@ -860,11 +892,11 @@ async def test_run_returns_none_when_no_content_and_no_warnings(tmp_path: Path) with SQLiteStateStore(tmp_path / "empty.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, EmptyRSSSource()), - cast(Any, EmptyContextSource()), + EmptyRSSSource(), + EmptyContextSource(), llm, delivery, delivery, @@ -887,7 +919,7 @@ async def test_stale_feed_triggers_ops_alert(tmp_path: Path) -> None: published_at=yesterday, content="content", ) - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(FeedConfig("feed", "Feed", "https://example.invalid/rss"),), context_sources=(), @@ -907,11 +939,11 @@ async def test_stale_feed_triggers_ops_alert(tmp_path: Path) -> None: with SQLiteStateStore(tmp_path / "stale.sqlite3") as state: state.record_source_check("feed", yesterday, yesterday) service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, StaticRSSSource(article)), - cast(Any, EmptyContextSource()), + StaticRSSSource(article), + EmptyContextSource(), llm, delivery, ops_delivery, @@ -961,7 +993,7 @@ async def test_llm_retry_on_validation_failure(tmp_path: Path) -> None: published_at=now, content="content", ) - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(FeedConfig("feed", "Feed", "https://example.invalid/rss"),), context_sources=(), @@ -978,11 +1010,11 @@ async def test_llm_retry_on_validation_failure(tmp_path: Path) -> None: with SQLiteStateStore(tmp_path / "retry.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, StaticRSSSource(article)), - cast(Any, EmptyContextSource()), + StaticRSSSource(article), + EmptyContextSource(), llm, delivery, delivery, @@ -996,7 +1028,7 @@ async def test_llm_retry_on_validation_failure(tmp_path: Path) -> None: async def test_briefing_exceeding_character_limit_is_rejected(tmp_path: Path) -> None: timezone = pendulum.timezone("Asia/Shanghai") now = pendulum.datetime(2026, 7, 13, 9, tz=timezone) - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(), context_sources=(), @@ -1025,11 +1057,11 @@ async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dic with SQLiteStateStore(tmp_path / "long.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, EmptyRSSSource()), - cast(Any, EmptyContextSource()), + EmptyRSSSource(), + EmptyContextSource(), LongLLM(), delivery, delivery, @@ -1051,7 +1083,7 @@ async def test_is_forecast_article_returns_false_for_unknown_feed(tmp_path: Path published_at=now.subtract(days=1), content="content", ) - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(FeedConfig("known-feed", "Known", "https://example.invalid/rss"),), context_sources=(), @@ -1068,11 +1100,11 @@ async def test_is_forecast_article_returns_false_for_unknown_feed(tmp_path: Path with SQLiteStateStore(tmp_path / "unknown.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, StaticRSSSource(article)), - cast(Any, EmptyContextSource()), + StaticRSSSource(article), + EmptyContextSource(), llm, delivery, delivery, @@ -1088,7 +1120,7 @@ async def test_rss_failure_does_not_crash_forecast_with_weather_context( tmp_path: Path, ) -> None: timezone = pendulum.timezone("Asia/Shanghai") - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(FeedConfig("failing-feed", "Failing", "https://example.invalid/feed"),), context_sources=(), @@ -1106,11 +1138,11 @@ async def test_rss_failure_does_not_crash_forecast_with_weather_context( with SQLiteStateStore(tmp_path / "rss-fail.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, FailingRSSSource()), - cast(Any, EmptyContextSource()), + FailingRSSSource(), + EmptyContextSource(), llm, delivery, delivery, @@ -1126,7 +1158,7 @@ async def test_rss_cancellation_aborts_task_without_recording_failure( tmp_path: Path, ) -> None: timezone = pendulum.timezone("Asia/Shanghai") - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(FeedConfig("canceled-feed", "Canceled", "https://example.invalid/feed"),), context_sources=(), @@ -1143,11 +1175,11 @@ async def test_rss_cancellation_aborts_task_without_recording_failure( with SQLiteStateStore(tmp_path / "rss-canceled.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, CanceledRSSSource()), - cast(Any, EmptyContextSource()), + CanceledRSSSource(), + EmptyContextSource(), RecordingLLM(), delivery, delivery, @@ -1166,7 +1198,7 @@ async def test_rss_cancellation_records_other_completed_feed_results( tmp_path: Path, ) -> None: timezone = pendulum.timezone("Asia/Shanghai") - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=( FeedConfig("canceled-feed", "Canceled", "https://example.invalid/canceled"), @@ -1188,11 +1220,11 @@ async def test_rss_cancellation_records_other_completed_feed_results( with SQLiteStateStore(tmp_path / "rss-canceled-results.sqlite3") as state: state.record_rss_fetch_failure("recovered-feed") service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, MixedOutcomeRSSSource()), - cast(Any, EmptyContextSource()), + MixedOutcomeRSSSource(), + EmptyContextSource(), RecordingLLM(), delivery, delivery, @@ -1210,7 +1242,7 @@ async def test_rss_failure_alert_is_sent_after_threshold( tmp_path: Path, ) -> None: timezone = pendulum.timezone("Asia/Shanghai") - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(FeedConfig("fail-feed", "Failing", "https://example.invalid/feed"),), context_sources=(), @@ -1230,11 +1262,11 @@ async def test_rss_failure_alert_is_sent_after_threshold( with SQLiteStateStore(tmp_path / "rss-alert.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, FailingRSSSource()), - cast(Any, EmptyContextSource()), + FailingRSSSource(), + EmptyContextSource(), llm, delivery, ops_delivery, @@ -1259,7 +1291,7 @@ async def test_failed_rss_alert_delivery_is_retried( caplog: pytest.LogCaptureFixture, ) -> None: timezone = pendulum.timezone("Asia/Shanghai") - settings = SimpleNamespace( + settings = _TestSettings( timezone=timezone, feeds=(FeedConfig("fail-feed", "Failing", "https://example.invalid/feed"),), context_sources=(), @@ -1278,11 +1310,11 @@ async def test_failed_rss_alert_delivery_is_retried( with SQLiteStateStore(tmp_path / "rss-alert-retry.sqlite3") as state: service = BriefingService( - cast(Any, settings), + settings, _location(), state, - cast(Any, FailingRSSSource()), - cast(Any, EmptyContextSource()), + FailingRSSSource(), + EmptyContextSource(), RecordingLLM(), delivery, ops_delivery, diff --git a/weather_briefing/service.py b/weather_briefing/service.py index c2bbdec4..99486b34 100644 --- a/weather_briefing/service.py +++ b/weather_briefing/service.py @@ -3,15 +3,15 @@ import asyncio import logging from collections.abc import Callable +from typing import Protocol import pendulum -from .config import Settings from .llm import LLMError, LLMProvider, parse_result -from .models import Article, BriefingResult, ResolvedLocation, SourceDocument, Warning +from .models import Article, BriefingResult, ContextSourceConfig, FeedConfig, ResolvedLocation, SourceDocument, Warning from .prompts import SYSTEM_PROMPT from .publishers import DeliveryProvider -from .sources import HTTPContextSource, RSSSource +from .sources import ContextDocumentSource, RSSFeedSource from .state import SQLiteStateStore from .time_utils import require_aware_datetime from .weather_context import WeatherContextProvider, fetch_weather_context, snapshot_to_documents @@ -19,14 +19,43 @@ _LOGGER = logging.getLogger("weather_briefing.service") +class BriefingSettings(Protocol): + @property + def timezone(self) -> pendulum.Timezone: ... + + @property + def feeds(self) -> tuple[FeedConfig, ...]: ... + + @property + def context_sources(self) -> tuple[ContextSourceConfig, ...]: ... + + @property + def rss_stale_hours(self) -> int: ... + + @property + def rss_failure_threshold(self) -> int: ... + + @property + def warning_retention_hours(self) -> int: ... + + @property + def history_hours(self) -> int: ... + + @property + def briefing_max_characters(self) -> int: ... + + @property + def llm_max_attempts(self) -> int: ... + + class BriefingService: def __init__( self, - settings: Settings, + settings: BriefingSettings, location: ResolvedLocation, state: SQLiteStateStore, - rss_source: RSSSource, - context_source: HTTPContextSource, + rss_source: RSSFeedSource, + context_source: ContextDocumentSource, llm: LLMProvider, delivery: DeliveryProvider, ops_delivery: DeliveryProvider, diff --git a/weather_briefing/sources.py b/weather_briefing/sources.py index 1254a67b..85e1e419 100644 --- a/weather_briefing/sources.py +++ b/weather_briefing/sources.py @@ -5,6 +5,7 @@ import logging import random from time import struct_time +from typing import Protocol import feedparser import httpx @@ -21,6 +22,14 @@ class SourceFetchError(RuntimeError): """Raised after a source exhausts all retry attempts.""" +class RSSFeedSource(Protocol): + async def fetch(self, config: FeedConfig) -> tuple[Article, ...]: ... + + +class ContextDocumentSource(Protocol): + async def fetch(self, config: ContextSourceConfig) -> SourceDocument: ... + + def _entry_time(entry: feedparser.FeedParserDict) -> pendulum.DateTime | None: parsed: struct_time | None = entry.get("published_parsed") or entry.get("updated_parsed") if parsed is None: From e1f99253d8834b6bff917ba5f6507109e5e9194c Mon Sep 17 00:00:00 2001 From: IceCodeNew <32576256+IceCodeNew@users.noreply.github.com> Date: Wed, 15 Jul 2026 03:23:21 +0800 Subject: [PATCH 3/4] fix: remove redundant isinstance guards from QWeather format helpers The _format_qweather_lifestyle and _format_qweather_day helpers now accept dict[str, object] parameters. The runtime isinstance(item, dict) guards are redundant since callers already filter items at the JSON boundary (weather_context.py:198-199) or validate the response payload before iteration (weather_context.py:220-226). Non-dict items now raise TypeError (caught and wrapped by line 238) instead of ValueError. --- tests/test_weather_context.py | 4 ++-- weather_briefing/weather_context.py | 4 ---- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/tests/test_weather_context.py b/tests/test_weather_context.py index 3190164c..89b6f5ff 100644 --- a/tests/test_weather_context.py +++ b/tests/test_weather_context.py @@ -1071,7 +1071,7 @@ def handler(request: httpx.Request) -> httpx.Response: raise AssertionError(f"Unexpected request: {request.url}") async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: - with pytest.raises(WeatherContextError, match="lifestyle indices failed: ValueError"): + with pytest.raises(WeatherContextError, match="lifestyle indices failed: TypeError"): await QWeatherProvider( client, authenticator=StaticAuthenticator(), @@ -1096,7 +1096,7 @@ def handler(request: httpx.Request) -> httpx.Response: raise AssertionError(f"Unexpected request: {request.url}") async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: - with pytest.raises(WeatherContextError, match="weather forecast failed: ValueError"): + with pytest.raises(WeatherContextError, match="weather forecast failed: TypeError"): await QWeatherProvider( client, authenticator=StaticAuthenticator(), diff --git a/weather_briefing/weather_context.py b/weather_briefing/weather_context.py index 95e697de..f6f066dd 100644 --- a/weather_briefing/weather_context.py +++ b/weather_briefing/weather_context.py @@ -685,8 +685,6 @@ def _aqi_standard(index: dict[str, object]) -> str: def _format_qweather_lifestyle(item: dict[str, object]) -> str: - if not isinstance(item, dict): - raise ValueError("QWeather lifestyle index must be an object") name = str(item["name"]) category = str(item.get("category", "未知")) text = str(item.get("text") or "无详细建议") @@ -694,8 +692,6 @@ def _format_qweather_lifestyle(item: dict[str, object]) -> str: def _format_qweather_day(item: dict[str, object]) -> str: - if not isinstance(item, dict): - raise ValueError("QWeather daily forecast must be an object") return ( f"{item['fxDate']}：{item['textDay']}转{item['textNight']}，" f"{item['tempMin']}~{item['tempMax']}℃，" From 111039c3e8e3f743f357027c87dbb3e3804b3665 Mon Sep 17 00:00:00 2001 From: IceCodeNew <32576256+IceCodeNew@users.noreply.github.com> Date: Wed, 15 Jul 2026 03:26:35 +0800 Subject: [PATCH 4/4] fix: tighten TypeGuard and replace str() cast in tests - _is_dict_list TypeGuard now verifies string keys, making the narrowing to list[dict[str, object]] honest - Replace str(air_document["content"]) with explicit isinstance(content, str) assertion before substring checks --- tests/test_service.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/tests/test_service.py b/tests/test_service.py index 8b44bc50..a0e64769 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -39,7 +39,9 @@ class _TestSettings: def _is_dict_list(value: object) -> TypeGuard[list[dict[str, object]]]: - return isinstance(value, list) and all(isinstance(item, dict) for item in value) + return isinstance(value, list) and all( + isinstance(item, dict) and all(isinstance(k, str) for k in item) for item in value + ) class EmptyRSSSource: @@ -239,8 +241,10 @@ async def test_forecast_uses_configured_coordinates_and_air_quality_context( assert len(context_documents) == 2 air_document = next(item for item in context_documents if item["source_id"] == "air-quality:test") assert air_document["url"] == "https://example.invalid/air-quality" - assert "AQI：42（标准：test-standard" in str(air_document["content"]) - assert "PM2.5 原始浓度：12 µg/m³" in str(air_document["content"]) + content = air_document["content"] + assert isinstance(content, str) + assert "AQI：42（标准：test-standard" in content + assert "PM2.5 原始浓度：12 µg/m³" in content recent_briefings = llm.payload["recent_briefings"] assert _is_dict_list(recent_briefings) assert recent_briefings == [