diff --git a/docs/design.md b/docs/design.md index 49315862..ef6f1635 100644 --- a/docs/design.md +++ b/docs/design.md @@ -135,6 +135,8 @@ JMA 没有 office code 时不会猜测东京或其他预报区。 指定目标日期时，天气、空气质量、生活指数和过敏原要选择同一天的数据。服务不支持该日期时保留明确缺失，不复用当前观测或其他日期建议。 +当前来源文档在应用层关联用于新鲜度比较的时刻：天气使用快照更新时间，空气质量和过敏原优先使用各自的观测时刻，没有独立时刻时回退天气快照时间。应用层统一执行产品需求定义的新鲜度窗口；指定日期的预报和没有当前观测语义的文档不参与比较。 + Open-Meteo 的逐小时空气质量和花粉预报按目标日峰值生成生活建议输入。AQI 和污染物保持来源给出的标准和单位，不做跨标准换算。 ## 语言 diff --git a/docs/requirements.md b/docs/requirements.md index 199de483..8235b340 100644 --- a/docs/requirements.md +++ b/docs/requirements.md @@ -24,9 +24,9 @@ - 值得提醒的变化包括临近降雨、明显温度或风力变化，以及预警变化。 - 确实影响关注地点的灾害动态也应提醒。 - 普通天气复述、轻微波动和没有变化的持续预警不应打扰用户。 -- 未发送的信息要保留，并在后续判断中与新信息一起考虑。 +- 未发送的信息要保留，并在后续判断中与新信息一起考虑；天气、气温、降水、风力和空气质量等会快速过期的内容，落后最新适用资料超过两小时后只能作为变化历史，不能继续形成当前结论或单独触发补发。 - 每天最后一次检查时，如果当天还没有发送过变化提醒，则发送一条无声消息。 -- 用户手动运行时，应立即发送积压信息，不受调度窗口限制。 +- 用户手动运行时，应立即发送仍有时效性的积压信息，不受调度窗口限制。 ## 内容要求 @@ -53,6 +53,7 @@ - 较低精度匹配在用户确认前不能自动写入地点配置。 - 中国大陆、新加坡和日本应使用适合当地的天气信息。 - 同一时间和地区的信息冲突时，应优先采用当地权威气象机构的最新资料，并保留冲突来源供用户核验。 +- 多个来源的当前资料只有在落后最新资料不超过两小时时才参与本轮结论；超过两小时的天气、空气质量等旧资料不得继续使用。 - 用户也可以明确指定天气来源及备用顺序。 - 本地天气来源缺少完整数据时，可以与全球天气来源组合使用。 diff --git a/tests/test_collection.py b/tests/test_collection.py new file mode 100644 index 00000000..28ceb7d8 --- /dev/null +++ b/tests/test_collection.py @@ -0,0 +1,291 @@ +import pendulum +import pytest + +from weather_briefing.application.collection import collect_weather_documents +from weather_briefing.capabilities import CapabilityName, CapabilityProviderSet, ProviderCapabilities +from weather_briefing.models import ( + AirQualitySnapshot, + AirQualityTimeKind, + AllergenLevel, + AllergenSnapshot, + ResolvedLocation, + WeatherContextSnapshot, +) + + +def _air_quality( + source: str, + *, + effective_at: pendulum.DateTime | None, + time_kind: AirQualityTimeKind = AirQualityTimeKind.OBSERVATION, +) -> AirQualitySnapshot: + return AirQualitySnapshot( + source_id=f"air-quality:{source}", + source_name=source, + source_url=f"https://example.invalid/{source}/air-quality", + effective_at=effective_at, + time_kind=time_kind, + aqi=50, + aqi_display="50", + aqi_standard="test", + pm25_aqi=None, + pm25_concentration=10, + pm25_unit="μg/m³", + category="good", + health_guidance="normal activity", + ) + + +def _snapshot( + source: str, + observed_at: pendulum.DateTime, + *, + air_quality_effective_at: pendulum.DateTime | None, + time_kind: AirQualityTimeKind = AirQualityTimeKind.OBSERVATION, + include_air_quality: bool = True, + include_allergen: bool = False, + allergen_observed_at: pendulum.DateTime | None = None, +) -> WeatherContextSnapshot: + return WeatherContextSnapshot( + source_id=f"weather:{source}", + source_name=source, + source_url=f"https://example.invalid/{source}/weather", + observed_at=observed_at, + weather_forecast=("forecast",), + air_quality=( + _air_quality(source, effective_at=air_quality_effective_at, time_kind=time_kind) + if include_air_quality + else None + ), + allergen=( + AllergenSnapshot( + source_id=f"allergen:{source}", + source_name=source, + source_url=f"https://example.invalid/{source}/allergen", + observed_at=allergen_observed_at, + levels=(AllergenLevel("pollen", "low", 1.0),), + overall_category="low", + health_guidance="normal activity", + ) + if include_allergen + else None + ), + ) + + +class _Provider: + def __init__(self, snapshot: WeatherContextSnapshot) -> None: + self._snapshot = snapshot + + async def fetch(self, latitude: float, longitude: float) -> WeatherContextSnapshot: + return self._snapshot + + async def fetch_for_date( + self, + latitude: float, + longitude: float, + forecast_date: pendulum.Date, + ) -> WeatherContextSnapshot: + return self._snapshot + + +def _provider_set(*snapshots: WeatherContextSnapshot) -> CapabilityProviderSet: + metadata = ProviderCapabilities( + provider_id="test", + provider_name="Test", + capabilities=frozenset({CapabilityName.WEATHER, CapabilityName.AIR_QUALITY}), + ) + return CapabilityProviderSet( + weather=_Provider(snapshots[0]), + weather_metadata=metadata, + supplements=tuple(_Provider(snapshot) for snapshot in snapshots[1:]), + supplement_metadata=tuple(metadata for _ in snapshots[1:]), + ) + + +_LOCATION = ResolvedLocation( + id="test", + name="Test", + latitude=39.9, + longitude=116.3, + country_code="CN", + administrative_area="Beijing", + timezone="Asia/Shanghai", + is_mainland_china=True, +) + + +async def test_collection_drops_only_current_documents_that_are_stale() -> None: + latest = pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai") + provider = _provider_set( + _snapshot("qweather", latest, air_quality_effective_at=None), + _snapshot("open-meteo", latest, air_quality_effective_at=latest.subtract(hours=8)), + ) + + documents = await collect_weather_documents(provider, _LOCATION, None) + + assert {document.id for document in documents} == { + "weather:qweather", + "air-quality:qweather", + "weather:open-meteo", + } + + +async def test_collection_drops_stale_weather_snapshot_without_air_quality() -> None: + latest = pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai") + provider = _provider_set( + _snapshot("qweather", latest, air_quality_effective_at=None), + _snapshot( + "open-meteo", + latest.subtract(hours=8), + air_quality_effective_at=None, + include_air_quality=False, + ), + ) + + documents = await collect_weather_documents(provider, _LOCATION, None) + + assert {document.id for document in documents} == { + "weather:qweather", + "air-quality:qweather", + } + + +async def test_collection_keeps_current_documents_at_two_hour_boundary() -> None: + latest = pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai") + provider = _provider_set( + _snapshot("qweather", latest, air_quality_effective_at=None), + _snapshot("open-meteo", latest, air_quality_effective_at=latest.subtract(hours=2)), + ) + + documents = await collect_weather_documents(provider, _LOCATION, None) + + assert {document.id for document in documents if document.id.startswith("air-quality:")} == { + "air-quality:qweather", + "air-quality:open-meteo", + } + + +async def test_collection_filters_allergen_by_its_own_observation_time() -> None: + latest = pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai") + provider = _provider_set( + _snapshot("qweather", latest, air_quality_effective_at=None), + _snapshot( + "open-meteo", + latest, + air_quality_effective_at=latest, + include_allergen=True, + allergen_observed_at=latest.subtract(hours=8), + ), + ) + + documents = await collect_weather_documents(provider, _LOCATION, None) + + assert "allergen:open-meteo" not in {document.id for document in documents} + assert "weather:open-meteo" in {document.id for document in documents} + + +async def test_collection_uses_weather_time_when_allergen_has_no_observation_time() -> None: + latest = pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai") + provider = _provider_set( + _snapshot( + "qweather", + latest, + air_quality_effective_at=None, + include_allergen=True, + ), + ) + + documents = await collect_weather_documents(provider, _LOCATION, None) + + assert "allergen:qweather" in {document.id for document in documents} + + +async def test_collection_keeps_forecast_document_while_filtering_current_data() -> None: + latest = pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai") + provider = _provider_set( + _snapshot( + "qweather", + latest, + air_quality_effective_at=latest.subtract(hours=8), + time_kind=AirQualityTimeKind.FORECAST, + ), + _snapshot( + "open-meteo", + latest.subtract(hours=8), + air_quality_effective_at=None, + include_air_quality=False, + ), + ) + + documents = await collect_weather_documents(provider, _LOCATION, None) + + assert "air-quality:qweather" in {document.id for document in documents} + assert "weather:open-meteo" not in {document.id for document in documents} + + +async def test_collection_does_not_filter_dated_forecast_snapshots() -> None: + latest = pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai") + provider = _provider_set( + _snapshot( + "qweather", + latest, + air_quality_effective_at=latest, + time_kind=AirQualityTimeKind.FORECAST, + ), + _snapshot( + "open-meteo", + latest, + air_quality_effective_at=latest.subtract(hours=8), + time_kind=AirQualityTimeKind.FORECAST, + ), + ) + + documents = await collect_weather_documents(provider, _LOCATION, pendulum.date(2026, 7, 28)) + + assert {document.id for document in documents if document.id.startswith("air-quality:")} == { + "air-quality:qweather", + "air-quality:open-meteo", + } + + +@pytest.mark.parametrize( + ("snapshot", "expected_context"), + ( + ( + _snapshot( + "qweather", + pendulum.datetime(2026, 7, 27, 21, tz=None), + air_quality_effective_at=None, + ), + "Weather snapshot weather:qweather observation time", + ), + ( + _snapshot( + "qweather", + pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai"), + air_quality_effective_at=pendulum.datetime(2026, 7, 27, 21, tz=None), + ), + "Air-quality snapshot air-quality:qweather observation time", + ), + ( + _snapshot( + "qweather", + pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai"), + air_quality_effective_at=None, + include_allergen=True, + allergen_observed_at=pendulum.datetime(2026, 7, 27, 21, tz=None), + ), + "Allergen snapshot allergen:qweather observation time", + ), + ), +) +async def test_collection_rejects_ambiguous_current_observation_times( + snapshot: WeatherContextSnapshot, + expected_context: str, +) -> None: + with pytest.raises( + ValueError, + match=rf"^{expected_context} must include explicit timezone information$", + ): + await collect_weather_documents(_provider_set(snapshot), _LOCATION, None) diff --git a/tests/test_prompts.py b/tests/test_prompts.py index 22c33adc..8d571b7f 100644 --- a/tests/test_prompts.py +++ b/tests/test_prompts.py @@ -37,6 +37,13 @@ def test_prompt_uses_actionable_publication_threshold() -> None: assert "service_status 类型" in NOTIFICATION_POLICY +def test_prompt_does_not_publish_expired_deferred_weather() -> None: + assert "落后最新适用资料超过两小时的积压内容" in SYSTEM_PROMPT + assert "不得写入当前结论，也不得单独触发发布" in SYSTEM_PROMPT + assert "恰好两小时仍可保留" in SYSTEM_PROMPT + assert "有效预警、灾害跟踪和指定日期预报仍按各自的有效性规则判断" in SYSTEM_PROMPT + + def test_prompt_separates_advice_and_avoids_repetition() -> None: assert "过敏原信息只能放入 advice" in SYSTEM_PROMPT assert "不得使用“原始浓度”" in SYSTEM_PROMPT diff --git a/weather_briefing/application/collection.py b/weather_briefing/application/collection.py index b67e010e..da285afe 100644 --- a/weather_briefing/application/collection.py +++ b/weather_briefing/application/collection.py @@ -8,12 +8,21 @@ import pendulum from ..capabilities import CapabilityProviderSet -from ..models import Article, FeedConfig, ResolvedLocation, SourceDocument +from ..models import ( + AirQualityTimeKind, + Article, + FeedConfig, + ResolvedLocation, + SourceDocument, + WeatherContextSnapshot, +) from ..sources import RSSFeedSource from ..state import SQLiteStateStore +from ..time_utils import require_aware_datetime from ..weather import WeatherContextProvider, fetch_weather_context, snapshot_to_documents _LOGGER = logging.getLogger("weather_briefing.service") +_CURRENT_DOCUMENT_MAX_LAG_HOURS = 2 async def collect_rss_articles( @@ -65,4 +74,63 @@ async def collect_weather_documents( ) else: snapshots = (await fetch_weather_context(provider, location.latitude, location.longitude, forecast_date),) - return tuple(document for snapshot in snapshots for document in snapshot_to_documents(snapshot)) + timed_documents = tuple(item for snapshot in snapshots for item in _snapshot_documents_with_times(snapshot)) + if forecast_date is not None: + return tuple(document for document, _ in timed_documents) + return _filter_stale_current_documents(timed_documents) + + +def _filter_stale_current_documents( + timed_documents: tuple[tuple[SourceDocument, pendulum.DateTime | None], ...], +) -> tuple[SourceDocument, ...]: + """Remove current documents too old to support a comparison.""" + available_times = tuple(observed_at for _, observed_at in timed_documents if observed_at is not None) + latest_time = max(available_times) + earliest_retained_time = latest_time.subtract(hours=_CURRENT_DOCUMENT_MAX_LAG_HOURS) + retained = tuple( + document + for document, observed_at in timed_documents + if observed_at is None or observed_at >= earliest_retained_time + ) + for document, observed_at in timed_documents: + if observed_at is not None and observed_at < earliest_retained_time: + _LOGGER.info( + "Discarding stale current document source_id=%s observed_at=%s latest_time=%s max_lag_hours=%d", + document.id, + observed_at.to_iso8601_string(), + latest_time.to_iso8601_string(), + _CURRENT_DOCUMENT_MAX_LAG_HOURS, + ) + return retained + + +def _snapshot_documents_with_times( + snapshot: WeatherContextSnapshot, +) -> tuple[tuple[SourceDocument, pendulum.DateTime | None], ...]: + """Attach current observation times to documents that can expire.""" + weather_observed_at = require_aware_datetime( + snapshot.observed_at, + context=f"Weather snapshot {snapshot.source_id} observation time", + ) + observation_times = {snapshot.source_id: weather_observed_at} + air_quality = snapshot.air_quality + if air_quality is not None and air_quality.time_kind is AirQualityTimeKind.OBSERVATION: + observation_times[air_quality.source_id] = ( + require_aware_datetime( + air_quality.effective_at, + context=f"Air-quality snapshot {air_quality.source_id} observation time", + ) + if air_quality.effective_at is not None + else weather_observed_at + ) + allergen = snapshot.allergen + if allergen is not None: + observation_times[allergen.source_id] = ( + require_aware_datetime( + allergen.observed_at, + context=f"Allergen snapshot {allergen.source_id} observation time", + ) + if allergen.observed_at is not None + else weather_observed_at + ) + return tuple((document, observation_times.get(document.id)) for document in snapshot_to_documents(snapshot)) diff --git a/weather_briefing/data/system_prompt.txt b/weather_briefing/data/system_prompt.txt index b1298eb1..387006f0 100644 --- a/weather_briefing/data/system_prompt.txt +++ b/weather_briefing/data/system_prompt.txt @@ -58,7 +58,8 @@ headline、conclusions、active_warnings、disaster_tracking 和 advice 之间 recent_context_documents 的 history_role 分别标识各来源的最新值、保留窗口基线和最近变化节点。 content_compacted=true 表示 content 是 adapter 从完整历史快照生成的确定性摘要， 只能按其明确提供的信息比较，不得补全被省略的细节。 -对气温、降水、风力、空气质量、短时预报等会过期的信息，始终以时间最新且仍适用于当前时刻的来源为准； -不得因为旧信息尚未发送就保留已经被较新快照取代的数值或结论。 +对天气、气温、降水、风力、空气质量和短时预报等会快速过期的信息，始终以时间最新且仍适用于当前时刻的来源为准； +落后最新适用资料超过两小时的积压内容只能用于判断变化历史，不得写入当前结论，也不得单独触发发布；恰好两小时仍可保留。 +不得因为旧信息尚未发送就保留已经被较新快照取代的数值或结论。有效预警、灾害跟踪和指定日期预报仍按各自的有效性规则判断。 active_warnings 应包含仍有效的预警以维持状态。 标题被标为 verbatim 的文章由程序另行全文转发，不要复述或改写其正文。