拆分 analysis_service：观测源配置与新鲜度工具函数移至 observation_freshness.py

抽取 _OBSERVATION_SOURCE_PROFILES、parse_utc_datetime、observation_age_min、canonical_observation_source_code、build_observation_freshness 到独立模块。analysis_service.py 加 10 行导入替代原 ~180 行内联定义。

web/analysis_service.py

@@ -35,6 +35,10 @@ from web.services.city_payloads import (
     build_city_market_scan_payload as _city_payload_market_scan,
     build_city_summary_payload as _city_payload_summary,
 )
+from web.services.observation_freshness import (
+    build_observation_freshness as _build_observation_freshness,
+    observation_age_min as _observation_age_min,
+)
 from web.services.analysis_signals import (
     _build_deviation_monitor,
     _build_taf_signal,
@@ -189,183 +193,6 @@ def _metar_is_current_local_day(
     return local_dt.strftime("%Y-%m-%d") == local_date
 
 
-_OBSERVATION_SOURCE_PROFILES: Dict[str, Dict[str, Any]] = {
-    "amos": {
-        "label": "AMOS",
-        "native_update_interval_sec": 60,
-        "fresh_window_sec": 180,
-        "expected_grace_sec": 180,
-        "stale_after_sec": 900,
-    },
-    "amsc_awos": {
-        "label": "AMSC AWOS",
-        "native_update_interval_sec": 60,
-        "fresh_window_sec": 180,
-        "expected_grace_sec": 180,
-        "stale_after_sec": 900,
-    },
-    "jma": {
-        "label": "JMA",
-        "native_update_interval_sec": 600,
-        "fresh_window_sec": 900,
-        "expected_grace_sec": 600,
-        "stale_after_sec": 2700,
-    },
-    "fmi": {
-        "label": "FMI",
-        "native_update_interval_sec": 600,
-        "fresh_window_sec": 900,
-        "expected_grace_sec": 600,
-        "stale_after_sec": 2700,
-    },
-    "knmi": {
-        "label": "KNMI",
-        "native_update_interval_sec": 600,
-        "fresh_window_sec": 900,
-        "expected_grace_sec": 600,
-        "stale_after_sec": 2700,
-    },
-    "hko": {
-        "label": "HKO",
-        "native_update_interval_sec": 600,
-        "fresh_window_sec": 900,
-        "expected_grace_sec": 600,
-        "stale_after_sec": 2700,
-    },
-    "cwa": {
-        "label": "CWA",
-        "native_update_interval_sec": 600,
-        "fresh_window_sec": 900,
-        "expected_grace_sec": 600,
-        "stale_after_sec": 2700,
-    },
-    "mgm": {
-        "label": "MGM",
-        "native_update_interval_sec": 900,
-        "fresh_window_sec": 900,
-        "expected_grace_sec": 900,
-        "stale_after_sec": 3600,
-    },
-    "metar": {
-        "label": "METAR",
-        "native_update_interval_sec": 900,
-        "fresh_window_sec": 600,
-        "expected_grace_sec": 900,
-        "stale_after_sec": 3600,
-    },
-    "noaa": {
-        "label": "NOAA",
-        "native_update_interval_sec": 900,
-        "fresh_window_sec": 600,
-        "expected_grace_sec": 900,
-        "stale_after_sec": 3600,
-    },
-    "wunderground": {
-        "label": "METAR",
-        "native_update_interval_sec": 900,
-        "fresh_window_sec": 600,
-        "expected_grace_sec": 900,
-        "stale_after_sec": 3600,
-    },
-}
-
-
-def _canonical_observation_source_code(value: Any) -> str:
-    raw = str(value or "").strip().lower()
-    if not raw:
-        return "metar"
-    if "amos" in raw:
-        return "amos"
-    if "jma" in raw:
-        return "jma"
-    if "fmi" in raw:
-        return "fmi"
-    if "knmi" in raw:
-        return "knmi"
-    if "hko" in raw:
-        return "hko"
-    if "cwa" in raw:
-        return "cwa"
-    if "mgm" in raw:
-        return "mgm"
-    if "noaa" in raw:
-        return "noaa"
-    if "wunderground" in raw or raw == "wu":
-        return "wunderground"
-    return raw
-
-
-def _observation_age_min(value: Any, now_utc: Optional[datetime] = None) -> Optional[int]:
-    obs_dt = _parse_utc_datetime(value)
-    if obs_dt is None:
-        return None
-    now = now_utc or datetime.now(timezone.utc)
-    return max(0, int((now - obs_dt).total_seconds() / 60))
-
-
-def _optional_str(value: Any) -> Optional[str]:
-    raw = str(value or "").strip()
-    return raw or None
-
-
-def _build_observation_freshness(
-    *,
-    source_code: Any,
-    source_label: Any = None,
-    observed_at: Any = None,
-    observed_at_local: Any = None,
-    ingested_at: Any = None,
-    age_min: Optional[int] = None,
-    now_utc: Optional[datetime] = None,
-) -> Dict[str, Any]:
-    code = _canonical_observation_source_code(source_code or source_label)
-    profile = _OBSERVATION_SOURCE_PROFILES.get(code) or _OBSERVATION_SOURCE_PROFILES["metar"]
-    now = now_utc or datetime.now(timezone.utc)
-    obs_dt = _parse_utc_datetime(observed_at)
-    age_sec = None
-    if age_min is not None:
-        try:
-            age_sec = max(0, int(age_min) * 60)
-        except Exception:
-            age_sec = None
-    if age_sec is None and obs_dt is not None:
-        age_sec = max(0, int((now - obs_dt).total_seconds()))
-
-    if age_sec is None:
-        status = "unknown"
-        reason = "observation_time_missing"
-    elif age_sec <= int(profile["fresh_window_sec"]):
-        status = "fresh"
-        reason = "within_native_fresh_window"
-    elif age_sec <= int(profile["native_update_interval_sec"]) + int(profile["expected_grace_sec"]):
-        status = "expected_wait"
-        reason = "within_source_expected_cadence"
-    elif age_sec <= int(profile["stale_after_sec"]):
-        status = "delayed"
-        reason = "past_expected_cadence"
-    else:
-        status = "stale"
-        reason = "past_stale_threshold"
-
-    expected_next = (
-        obs_dt + timedelta(seconds=int(profile["native_update_interval_sec"]))
-        if obs_dt is not None
-        else None
-    )
-    return {
-        "source_code": code,
-        "source_label": str(source_label or profile["label"]),
-        "observed_at": obs_dt.isoformat() if obs_dt is not None else _optional_str(observed_at),
-        "observed_at_local": _optional_str(observed_at_local),
-        "ingested_at": _optional_str(ingested_at),
-        "native_update_interval_sec": int(profile["native_update_interval_sec"]),
-        "expected_next_update_at": expected_next.isoformat() if expected_next is not None else None,
-        "freshness_status": status,
-        "freshness_reason": reason,
-        "age_sec": age_sec,
-    }
-
-
 def _record_analysis_cache_event(*, city: str, hit: bool, force_refresh: bool) -> None:
     now = datetime.now(timezone.utc).isoformat()
     with _ANALYSIS_CACHE_STATS_LOCK:

web/services/observation_freshness.py

@@ -0,0 +1,199 @@
+"""Observation source freshness profiles and helpers.
+
+Extracted from analysis_service.py to keep the god module leaner.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone, timedelta
+from typing import Any, Dict, Optional
+
+
+_OBSERVATION_SOURCE_PROFILES: Dict[str, Dict[str, Any]] = {
+    "amos": {
+        "label": "AMOS",
+        "native_update_interval_sec": 60,
+        "fresh_window_sec": 180,
+        "expected_grace_sec": 180,
+        "stale_after_sec": 900,
+    },
+    "amsc_awos": {
+        "label": "AMSC AWOS",
+        "native_update_interval_sec": 60,
+        "fresh_window_sec": 180,
+        "expected_grace_sec": 180,
+        "stale_after_sec": 900,
+    },
+    "jma": {
+        "label": "JMA",
+        "native_update_interval_sec": 600,
+        "fresh_window_sec": 900,
+        "expected_grace_sec": 600,
+        "stale_after_sec": 2700,
+    },
+    "fmi": {
+        "label": "FMI",
+        "native_update_interval_sec": 600,
+        "fresh_window_sec": 900,
+        "expected_grace_sec": 600,
+        "stale_after_sec": 2700,
+    },
+    "knmi": {
+        "label": "KNMI",
+        "native_update_interval_sec": 600,
+        "fresh_window_sec": 900,
+        "expected_grace_sec": 600,
+        "stale_after_sec": 2700,
+    },
+    "hko": {
+        "label": "HKO",
+        "native_update_interval_sec": 600,
+        "fresh_window_sec": 900,
+        "expected_grace_sec": 600,
+        "stale_after_sec": 2700,
+    },
+    "cwa": {
+        "label": "CWA",
+        "native_update_interval_sec": 600,
+        "fresh_window_sec": 900,
+        "expected_grace_sec": 600,
+        "stale_after_sec": 2700,
+    },
+    "mgm": {
+        "label": "MGM",
+        "native_update_interval_sec": 900,
+        "fresh_window_sec": 900,
+        "expected_grace_sec": 900,
+        "stale_after_sec": 3600,
+    },
+    "metar": {
+        "label": "METAR",
+        "native_update_interval_sec": 900,
+        "fresh_window_sec": 600,
+        "expected_grace_sec": 900,
+        "stale_after_sec": 3600,
+    },
+    "noaa": {
+        "label": "NOAA",
+        "native_update_interval_sec": 900,
+        "fresh_window_sec": 600,
+        "expected_grace_sec": 900,
+        "stale_after_sec": 3600,
+    },
+    "wunderground": {
+        "label": "METAR",
+        "native_update_interval_sec": 900,
+        "fresh_window_sec": 600,
+        "expected_grace_sec": 900,
+        "stale_after_sec": 3600,
+    },
+}
+
+
+def parse_utc_datetime(value: Any) -> Optional[datetime]:
+    raw = str(value or "").strip()
+    if not raw or "T" not in raw:
+        return None
+    try:
+        dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+    except Exception:
+        return None
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)
+
+
+def observation_age_min(value: Any, now_utc: Optional[datetime] = None) -> Optional[int]:
+    obs_dt = parse_utc_datetime(value)
+    if obs_dt is None:
+        return None
+    now = now_utc or datetime.now(timezone.utc)
+    return max(0, int((now - obs_dt).total_seconds() / 60))
+
+
+def canonical_observation_source_code(value: Any) -> str:
+    raw = str(value or "").strip().lower()
+    if not raw:
+        return "metar"
+    if "amos" in raw:
+        return "amos"
+    if "jma" in raw:
+        return "jma"
+    if "fmi" in raw:
+        return "fmi"
+    if "knmi" in raw:
+        return "knmi"
+    if "hko" in raw:
+        return "hko"
+    if "cwa" in raw:
+        return "cwa"
+    if "mgm" in raw:
+        return "mgm"
+    if "noaa" in raw:
+        return "noaa"
+    if "wunderground" in raw or raw == "wu":
+        return "wunderground"
+    return raw
+
+
+def optional_str(value: Any) -> Optional[str]:
+    raw = str(value or "").strip()
+    return raw or None
+
+
+def build_observation_freshness(
+    *,
+    source_code: Any,
+    source_label: Any = None,
+    observed_at: Any = None,
+    observed_at_local: Any = None,
+    ingested_at: Any = None,
+    age_min: Optional[int] = None,
+    now_utc: Optional[datetime] = None,
+) -> Dict[str, Any]:
+    code = canonical_observation_source_code(source_code or source_label)
+    profile = _OBSERVATION_SOURCE_PROFILES.get(code) or _OBSERVATION_SOURCE_PROFILES["metar"]
+    now = now_utc or datetime.now(timezone.utc)
+    obs_dt = parse_utc_datetime(observed_at)
+    age_sec = None
+    if age_min is not None:
+        try:
+            age_sec = max(0, int(age_min) * 60)
+        except Exception:
+            age_sec = None
+    if age_sec is None and obs_dt is not None:
+        age_sec = max(0, int((now - obs_dt).total_seconds()))
+
+    if age_sec is None:
+        status = "unknown"
+        reason = "observation_time_missing"
+    elif age_sec <= int(profile["fresh_window_sec"]):
+        status = "fresh"
+        reason = "within_native_fresh_window"
+    elif age_sec <= int(profile["native_update_interval_sec"]) + int(profile["expected_grace_sec"]):
+        status = "expected_wait"
+        reason = "within_source_expected_cadence"
+    elif age_sec <= int(profile["stale_after_sec"]):
+        status = "delayed"
+        reason = "past_expected_cadence"
+    else:
+        status = "stale"
+        reason = "past_stale_threshold"
+
+    expected_next = (
+        obs_dt + timedelta(seconds=int(profile["native_update_interval_sec"]))
+        if obs_dt is not None
+        else None
+    )
+    return {
+        "source_code": code,
+        "source_label": str(source_label or profile["label"]),
+        "observed_at": obs_dt.isoformat() if obs_dt is not None else optional_str(observed_at),
+        "observed_at_local": optional_str(observed_at_local),
+        "ingested_at": optional_str(ingested_at),
+        "native_update_interval_sec": int(profile["native_update_interval_sec"]),
+        "expected_next_update_at": expected_next.isoformat() if expected_next is not None else None,
+        "freshness_status": status,
+        "freshness_reason": reason,
+        "age_sec": age_sec,
+    }