From 0cb8a7967bc321f0f5c56e8b883094bd5cd34df7 Mon Sep 17 00:00:00 2001 From: IceCodeNew <32576256+IceCodeNew@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:36:39 +0800 Subject: [PATCH 1/5] fix: ignore stale air quality conflicts --- README.md | 2 + docs/design.md | 2 + docs/notes.md | 8 ++ docs/requirements.md | 1 + tests/test_collection.py | 148 +++++++++++++++++++++ weather_briefing/application/collection.py | 49 ++++++- 6 files changed, 209 insertions(+), 1 deletion(-) create mode 100644 tests/test_collection.py diff --git a/README.md b/README.md index 1e9d3433..3e49d07f 100644 --- a/README.md +++ b/README.md @@ -107,6 +107,8 @@ You can set `WEATHER_PROVIDERS` to override the regional default order. Include When NEA or JMA content conflicts with Open-Meteo for the same time and region, the briefing prioritizes the latest data from the local official agency and retains conflicting sources for your verification. +Current air-quality observations from multiple sources are compared only when they are within two hours of the latest available data. Older observations are excluded from the current conflict while the source's weather forecast remains available. + ### JMA office codes See [`docs/jma-office-codes.md`](docs/jma-office-codes.md) for the forecast office codes covering all 47 prefectures and usage instructions. diff --git a/docs/design.md b/docs/design.md index 49315862..b86f2c62 100644 --- a/docs/design.md +++ b/docs/design.md @@ -135,6 +135,8 @@ JMA 没有 office code 时不会猜测东京或其他预报区。 指定目标日期时，天气、空气质量、生活指数和过敏原要选择同一天的数据。服务不支持该日期时保留明确缺失，不复用当前观测或其他日期建议。 +多个来源同时提供当前空气质量观测时，应用层以最新资料为基准，只把落后不超过两小时的观测转换为本轮模型来源文档。来源提供空气质量有效时刻时直接使用；没有提供时，使用包含该数据的天气快照更新时间进行比较。被淘汰的来源仍保留天气文档，筛选不处理指定日期的空气质量预报。 + Open-Meteo 的逐小时空气质量和花粉预报按目标日峰值生成生活建议输入。AQI 和污染物保持来源给出的标准和单位，不做跨标准换算。 ## 语言 diff --git a/docs/notes.md b/docs/notes.md index 02aa284f..a7da620b 100644 --- a/docs/notes.md +++ b/docs/notes.md @@ -16,6 +16,14 @@ 如果厂商以后提供具体花粉种类和浓度，再把它声明为结构化过敏原能力。 +## 当前空气质量冲突使用固定两小时窗口 + +空气质量会在数小时内明显变化。把相隔很久的观测并列为冲突，会让旧数值获得与最新资料相同的权重；只保留单个最新来源又会隐藏同一时段不同标准或模型之间值得核验的真实差异。 + +因此多个当前观测只在落后最新资料不超过两小时时共同进入本轮模型上下文。两小时覆盖当前按小时更新和调度的正常时间偏差，同时能排除晚间仍返回的午后数据。阈值属于产品规则，不开放为运行时配置。 + +这项选择成立的条件是主要空气质量来源仍按小时级频率更新。如果来源更新周期明显改变，或运行数据表明两小时会频繁误删有效资料，再根据观测到的延迟分布调整窗口。未来日期的预报不使用这项筛选。 + ## 默认简报语言是英文 简报语言以前是隐含行为。变成配置项后，产品级默认值选择英文，现有中国大陆示例则显式填写 `zh-CN`。 diff --git a/docs/requirements.md b/docs/requirements.md index 199de483..b6998726 100644 --- a/docs/requirements.md +++ b/docs/requirements.md @@ -53,6 +53,7 @@ - 较低精度匹配在用户确认前不能自动写入地点配置。 - 中国大陆、新加坡和日本应使用适合当地的天气信息。 - 同一时间和地区的信息冲突时，应优先采用当地权威气象机构的最新资料，并保留冲突来源供用户核验。 +- 多个来源的当前空气质量观测只有在时刻接近时才形成冲突；明显落后于最新资料的旧观测不得继续参与本轮结论。 - 用户也可以明确指定天气来源及备用顺序。 - 本地天气来源缺少完整数据时，可以与全球天气来源组合使用。 diff --git a/tests/test_collection.py b/tests/test_collection.py new file mode 100644 index 00000000..69a61ebe --- /dev/null +++ b/tests/test_collection.py @@ -0,0 +1,148 @@ +import pendulum + +from weather_briefing.application.collection import collect_weather_documents +from weather_briefing.capabilities import CapabilityName, CapabilityProviderSet, ProviderCapabilities +from weather_briefing.models import ( + AirQualitySnapshot, + AirQualityTimeKind, + 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, +) -> 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), + ) + + +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_air_quality_observations_more_than_two_hours_behind() -> 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_keeps_air_quality_observations_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_does_not_filter_air_quality_forecasts() -> 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", + } diff --git a/weather_briefing/application/collection.py b/weather_briefing/application/collection.py index b67e010e..98ea8e7c 100644 --- a/weather_briefing/application/collection.py +++ b/weather_briefing/application/collection.py @@ -4,16 +4,25 @@ import asyncio import logging +from dataclasses import replace 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 ..weather import WeatherContextProvider, fetch_weather_context, snapshot_to_documents _LOGGER = logging.getLogger("weather_briefing.service") +_AIR_QUALITY_OBSERVATION_MAX_LAG_HOURS = 2 async def collect_rss_articles( @@ -65,4 +74,42 @@ async def collect_weather_documents( ) else: snapshots = (await fetch_weather_context(provider, location.latitude, location.longitude, forecast_date),) + snapshots = _filter_stale_air_quality_observations(snapshots) return tuple(document for snapshot in snapshots for document in snapshot_to_documents(snapshot)) + + +def _filter_stale_air_quality_observations( + snapshots: tuple[WeatherContextSnapshot, ...], +) -> tuple[WeatherContextSnapshot, ...]: + """Remove observations too old to support a current air-quality conflict.""" + observation_times = tuple(_air_quality_observation_time(snapshot) for snapshot in snapshots) + available_times = tuple(value for value in observation_times if value is not None) + if len(available_times) < 2: + return snapshots + + latest_time = max(available_times) + earliest_retained_time = latest_time.subtract(hours=_AIR_QUALITY_OBSERVATION_MAX_LAG_HOURS) + filtered: list[WeatherContextSnapshot] = [] + for snapshot, observation_time in zip(snapshots, observation_times, strict=True): + air_quality = snapshot.air_quality + if observation_time is not None and observation_time < earliest_retained_time and air_quality is not None: + _LOGGER.info( + "Discarding stale air-quality observation source_id=%s comparison_time=%s " + "latest_time=%s max_lag_hours=%d", + air_quality.source_id, + observation_time.to_iso8601_string(), + latest_time.to_iso8601_string(), + _AIR_QUALITY_OBSERVATION_MAX_LAG_HOURS, + ) + filtered.append(replace(snapshot, air_quality=None)) + else: + filtered.append(snapshot) + return tuple(filtered) + + +def _air_quality_observation_time(snapshot: WeatherContextSnapshot) -> pendulum.DateTime | None: + """Return the best available time for comparing a current observation.""" + air_quality = snapshot.air_quality + if air_quality is None or air_quality.time_kind is not AirQualityTimeKind.OBSERVATION: + return None + return air_quality.effective_at or snapshot.observed_at From 9c0ea279e0ff677a76b87325bc27f3609c66269e Mon Sep 17 00:00:00 2001 From: IceCodeNew <32576256+IceCodeNew@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:35:50 +0800 Subject: [PATCH 2/5] docs: clarify air quality freshness boundary --- docs/requirements.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/requirements.md b/docs/requirements.md index b6998726..9ec16f9c 100644 --- a/docs/requirements.md +++ b/docs/requirements.md @@ -53,7 +53,7 @@ - 较低精度匹配在用户确认前不能自动写入地点配置。 - 中国大陆、新加坡和日本应使用适合当地的天气信息。 - 同一时间和地区的信息冲突时，应优先采用当地权威气象机构的最新资料，并保留冲突来源供用户核验。 -- 多个来源的当前空气质量观测只有在时刻接近时才形成冲突；明显落后于最新资料的旧观测不得继续参与本轮结论。 +- 多个来源的当前空气质量观测只有在落后最新资料不超过两小时时才形成冲突；超过两小时的旧观测不得继续参与本轮结论。 - 用户也可以明确指定天气来源及备用顺序。 - 本地天气来源缺少完整数据时，可以与全球天气来源组合使用。 From 2917a25d186ddea4ddbfdd64fb8f216aa1acb045 Mon Sep 17 00:00:00 2001 From: IceCodeNew <32576256+IceCodeNew@users.noreply.github.com> Date: Wed, 29 Jul 2026 19:33:51 +0800 Subject: [PATCH 3/5] fix: apply freshness window to current data --- README.md | 2 +- docs/design.md | 4 +- docs/notes.md | 8 -- docs/requirements.md | 6 +- tests/test_collection.py | 108 ++++++++++++++++++++- tests/test_prompts.py | 7 ++ weather_briefing/application/collection.py | 66 +++++++------ weather_briefing/data/system_prompt.txt | 5 +- 8 files changed, 155 insertions(+), 51 deletions(-) diff --git a/README.md b/README.md index 3e49d07f..356405fb 100644 --- a/README.md +++ b/README.md @@ -107,7 +107,7 @@ You can set `WEATHER_PROVIDERS` to override the regional default order. Include When NEA or JMA content conflicts with Open-Meteo for the same time and region, the briefing prioritizes the latest data from the local official agency and retains conflicting sources for your verification. -Current air-quality observations from multiple sources are compared only when they are within two hours of the latest available data. Older observations are excluded from the current conflict while the source's weather forecast remains available. +Current observations from multiple weather sources are compared only when each item is no more than two hours behind the latest available data. Older weather, air-quality, and allergen observations are excluded from the current briefing; explicitly requested forecast dates are unaffected. ### JMA office codes diff --git a/docs/design.md b/docs/design.md index b86f2c62..d1020add 100644 --- a/docs/design.md +++ b/docs/design.md @@ -135,7 +135,9 @@ JMA 没有 office code 时不会猜测东京或其他预报区。 指定目标日期时，天气、空气质量、生活指数和过敏原要选择同一天的数据。服务不支持该日期时保留明确缺失，不复用当前观测或其他日期建议。 -多个来源同时提供当前空气质量观测时，应用层以最新资料为基准，只把落后不超过两小时的观测转换为本轮模型来源文档。来源提供空气质量有效时刻时直接使用；没有提供时，使用包含该数据的天气快照更新时间进行比较。被淘汰的来源仍保留天气文档，筛选不处理指定日期的空气质量预报。 +多个来源同时提供当前资料时，应用层以最新资料为基准，只保留落后不超过两小时的当前来源文档。天气使用快照更新时间，空气质量和过敏原优先使用各自的观测时刻；一个字段过时不会连带删除同一来源仍然新鲜的其他字段。指定日期的预报不使用这项筛选。 + +未发送文章和历史快照继续保留供累计变化判断。模型输入契约要求天气、气温、降水、风力和空气质量等短时信息在落后最新适用资料超过两小时后只作为变化历史，不得写入当前结论或触发补发；预警、灾害跟踪和指定日期预报仍按各自的有效性规则判断。 Open-Meteo 的逐小时空气质量和花粉预报按目标日峰值生成生活建议输入。AQI 和污染物保持来源给出的标准和单位，不做跨标准换算。 diff --git a/docs/notes.md b/docs/notes.md index a7da620b..02aa284f 100644 --- a/docs/notes.md +++ b/docs/notes.md @@ -16,14 +16,6 @@ 如果厂商以后提供具体花粉种类和浓度，再把它声明为结构化过敏原能力。 -## 当前空气质量冲突使用固定两小时窗口 - -空气质量会在数小时内明显变化。把相隔很久的观测并列为冲突，会让旧数值获得与最新资料相同的权重；只保留单个最新来源又会隐藏同一时段不同标准或模型之间值得核验的真实差异。 - -因此多个当前观测只在落后最新资料不超过两小时时共同进入本轮模型上下文。两小时覆盖当前按小时更新和调度的正常时间偏差，同时能排除晚间仍返回的午后数据。阈值属于产品规则，不开放为运行时配置。 - -这项选择成立的条件是主要空气质量来源仍按小时级频率更新。如果来源更新周期明显改变，或运行数据表明两小时会频繁误删有效资料，再根据观测到的延迟分布调整窗口。未来日期的预报不使用这项筛选。 - ## 默认简报语言是英文 简报语言以前是隐含行为。变成配置项后，产品级默认值选择英文，现有中国大陆示例则显式填写 `zh-CN`。 diff --git a/docs/requirements.md b/docs/requirements.md index 9ec16f9c..8235b340 100644 --- a/docs/requirements.md +++ b/docs/requirements.md @@ -24,9 +24,9 @@ - 值得提醒的变化包括临近降雨、明显温度或风力变化，以及预警变化。 - 确实影响关注地点的灾害动态也应提醒。 - 普通天气复述、轻微波动和没有变化的持续预警不应打扰用户。 -- 未发送的信息要保留，并在后续判断中与新信息一起考虑。 +- 未发送的信息要保留，并在后续判断中与新信息一起考虑；天气、气温、降水、风力和空气质量等会快速过期的内容，落后最新适用资料超过两小时后只能作为变化历史，不能继续形成当前结论或单独触发补发。 - 每天最后一次检查时，如果当天还没有发送过变化提醒，则发送一条无声消息。 -- 用户手动运行时，应立即发送积压信息，不受调度窗口限制。 +- 用户手动运行时，应立即发送仍有时效性的积压信息，不受调度窗口限制。 ## 内容要求 @@ -53,7 +53,7 @@ - 较低精度匹配在用户确认前不能自动写入地点配置。 - 中国大陆、新加坡和日本应使用适合当地的天气信息。 - 同一时间和地区的信息冲突时，应优先采用当地权威气象机构的最新资料，并保留冲突来源供用户核验。 -- 多个来源的当前空气质量观测只有在落后最新资料不超过两小时时才形成冲突；超过两小时的旧观测不得继续参与本轮结论。 +- 多个来源的当前资料只有在落后最新资料不超过两小时时才参与本轮结论；超过两小时的天气、空气质量等旧资料不得继续使用。 - 用户也可以明确指定天气来源及备用顺序。 - 本地天气来源缺少完整数据时，可以与全球天气来源组合使用。 diff --git a/tests/test_collection.py b/tests/test_collection.py index 69a61ebe..21288aef 100644 --- a/tests/test_collection.py +++ b/tests/test_collection.py @@ -5,6 +5,8 @@ from weather_briefing.models import ( AirQualitySnapshot, AirQualityTimeKind, + AllergenLevel, + AllergenSnapshot, ResolvedLocation, WeatherContextSnapshot, ) @@ -39,6 +41,9 @@ def _snapshot( *, 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}", @@ -46,7 +51,24 @@ def _snapshot( 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), + 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 + ), ) @@ -92,7 +114,7 @@ def _provider_set(*snapshots: WeatherContextSnapshot) -> CapabilityProviderSet: ) -async def test_collection_drops_air_quality_observations_more_than_two_hours_behind() -> None: +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), @@ -108,7 +130,27 @@ async def test_collection_drops_air_quality_observations_more_than_two_hours_beh } -async def test_collection_keeps_air_quality_observations_at_two_hour_boundary() -> None: +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), @@ -123,7 +165,65 @@ async def test_collection_keeps_air_quality_observations_at_two_hour_boundary() } -async def test_collection_does_not_filter_air_quality_forecasts() -> None: +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( 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 98ea8e7c..1dfe8c7e 100644 --- a/weather_briefing/application/collection.py +++ b/weather_briefing/application/collection.py @@ -4,7 +4,6 @@ import asyncio import logging -from dataclasses import replace import pendulum @@ -22,7 +21,7 @@ from ..weather import WeatherContextProvider, fetch_weather_context, snapshot_to_documents _LOGGER = logging.getLogger("weather_briefing.service") -_AIR_QUALITY_OBSERVATION_MAX_LAG_HOURS = 2 +_CURRENT_DOCUMENT_MAX_LAG_HOURS = 2 async def collect_rss_articles( @@ -74,42 +73,45 @@ async def collect_weather_documents( ) else: snapshots = (await fetch_weather_context(provider, location.latitude, location.longitude, forecast_date),) - snapshots = _filter_stale_air_quality_observations(snapshots) - 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_air_quality_observations( - snapshots: tuple[WeatherContextSnapshot, ...], -) -> tuple[WeatherContextSnapshot, ...]: - """Remove observations too old to support a current air-quality conflict.""" - observation_times = tuple(_air_quality_observation_time(snapshot) for snapshot in snapshots) - available_times = tuple(value for value in observation_times if value is not None) - if len(available_times) < 2: - return snapshots - +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=_AIR_QUALITY_OBSERVATION_MAX_LAG_HOURS) - filtered: list[WeatherContextSnapshot] = [] - for snapshot, observation_time in zip(snapshots, observation_times, strict=True): - air_quality = snapshot.air_quality - if observation_time is not None and observation_time < earliest_retained_time and air_quality is not None: + 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 air-quality observation source_id=%s comparison_time=%s " - "latest_time=%s max_lag_hours=%d", - air_quality.source_id, - observation_time.to_iso8601_string(), + "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(), - _AIR_QUALITY_OBSERVATION_MAX_LAG_HOURS, + _CURRENT_DOCUMENT_MAX_LAG_HOURS, ) - filtered.append(replace(snapshot, air_quality=None)) - else: - filtered.append(snapshot) - return tuple(filtered) + return retained -def _air_quality_observation_time(snapshot: WeatherContextSnapshot) -> pendulum.DateTime | None: - """Return the best available time for comparing a current observation.""" +def _snapshot_documents_with_times( + snapshot: WeatherContextSnapshot, +) -> tuple[tuple[SourceDocument, pendulum.DateTime | None], ...]: + """Attach current observation times to documents that can expire.""" + observation_times = {snapshot.source_id: snapshot.observed_at} air_quality = snapshot.air_quality - if air_quality is None or air_quality.time_kind is not AirQualityTimeKind.OBSERVATION: - return None - return air_quality.effective_at or snapshot.observed_at + if air_quality is not None and air_quality.time_kind is AirQualityTimeKind.OBSERVATION: + observation_times[air_quality.source_id] = air_quality.effective_at or snapshot.observed_at + allergen = snapshot.allergen + if allergen is not None: + observation_times[allergen.source_id] = allergen.observed_at or snapshot.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 的文章由程序另行全文转发，不要复述或改写其正文。 From 0fa6ed7d0a4b4341b66019b93f0571afb7f3687c Mon Sep 17 00:00:00 2001 From: IceCodeNew <32576256+IceCodeNew@users.noreply.github.com> Date: Wed, 29 Jul 2026 19:43:18 +0800 Subject: [PATCH 4/5] fix: validate current observation times --- docs/design.md | 4 +- tests/test_collection.py | 43 ++++++++++++++++++++++ weather_briefing/application/collection.py | 25 +++++++++++-- 3 files changed, 66 insertions(+), 6 deletions(-) diff --git a/docs/design.md b/docs/design.md index d1020add..ef6f1635 100644 --- a/docs/design.md +++ b/docs/design.md @@ -135,9 +135,7 @@ JMA 没有 office code 时不会猜测东京或其他预报区。 指定目标日期时，天气、空气质量、生活指数和过敏原要选择同一天的数据。服务不支持该日期时保留明确缺失，不复用当前观测或其他日期建议。 -多个来源同时提供当前资料时，应用层以最新资料为基准，只保留落后不超过两小时的当前来源文档。天气使用快照更新时间，空气质量和过敏原优先使用各自的观测时刻；一个字段过时不会连带删除同一来源仍然新鲜的其他字段。指定日期的预报不使用这项筛选。 - -未发送文章和历史快照继续保留供累计变化判断。模型输入契约要求天气、气温、降水、风力和空气质量等短时信息在落后最新适用资料超过两小时后只作为变化历史，不得写入当前结论或触发补发；预警、灾害跟踪和指定日期预报仍按各自的有效性规则判断。 +当前来源文档在应用层关联用于新鲜度比较的时刻：天气使用快照更新时间，空气质量和过敏原优先使用各自的观测时刻，没有独立时刻时回退天气快照时间。应用层统一执行产品需求定义的新鲜度窗口；指定日期的预报和没有当前观测语义的文档不参与比较。 Open-Meteo 的逐小时空气质量和花粉预报按目标日峰值生成生活建议输入。AQI 和污染物保持来源给出的标准和单位，不做跨标准换算。 diff --git a/tests/test_collection.py b/tests/test_collection.py index 21288aef..28ceb7d8 100644 --- a/tests/test_collection.py +++ b/tests/test_collection.py @@ -1,4 +1,5 @@ import pendulum +import pytest from weather_briefing.application.collection import collect_weather_documents from weather_briefing.capabilities import CapabilityName, CapabilityProviderSet, ProviderCapabilities @@ -246,3 +247,45 @@ async def test_collection_does_not_filter_dated_forecast_snapshots() -> None: "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/weather_briefing/application/collection.py b/weather_briefing/application/collection.py index 1dfe8c7e..da285afe 100644 --- a/weather_briefing/application/collection.py +++ b/weather_briefing/application/collection.py @@ -18,6 +18,7 @@ ) 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") @@ -107,11 +108,29 @@ def _snapshot_documents_with_times( snapshot: WeatherContextSnapshot, ) -> tuple[tuple[SourceDocument, pendulum.DateTime | None], ...]: """Attach current observation times to documents that can expire.""" - observation_times = {snapshot.source_id: snapshot.observed_at} + 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] = air_quality.effective_at or snapshot.observed_at + 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] = allergen.observed_at or snapshot.observed_at + 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)) From 912a4a7284f5b0bd2bd04c530ccd5bb9774cf8e4 Mon Sep 17 00:00:00 2001 From: IceCodeNew <32576256+IceCodeNew@users.noreply.github.com> Date: Wed, 29 Jul 2026 21:22:39 +0800 Subject: [PATCH 5/5] Update README.md --- README.md | 2 -- 1 file changed, 2 deletions(-) diff --git a/README.md b/README.md index 356405fb..1e9d3433 100644 --- a/README.md +++ b/README.md @@ -107,8 +107,6 @@ You can set `WEATHER_PROVIDERS` to override the regional default order. Include When NEA or JMA content conflicts with Open-Meteo for the same time and region, the briefing prioritizes the latest data from the local official agency and retains conflicting sources for your verification. -Current observations from multiple weather sources are compared only when each item is no more than two hours behind the latest available data. Older weather, air-quality, and allergen observations are excluded from the current briefing; explicitly requested forecast dates are unaffected. - ### JMA office codes See [`docs/jma-office-codes.md`](docs/jma-office-codes.md) for the forecast office codes covering all 47 prefectures and usage instructions.