From e1d21c6ce3b44d1ecf82458106c698584ab53257 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Sun, 10 May 2026 18:53:09 +0800 Subject: [PATCH] =?UTF-8?q?=E7=AE=80=E5=8C=96=20AMOS=20=E8=8E=B7=E5=8F=96?= =?UTF-8?q?=EF=BC=9A=E4=BB=85=E6=94=AF=E6=8C=81=20RKSI=EF=BC=88=E4=BB=81?= =?UTF-8?q?=E5=B7=9D=EF=BC=89=EF=BC=8CBusan=20=E5=9B=9E=E9=80=80=E6=A0=87?= =?UTF-8?q?=E5=87=86=20METAR?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原因：AMOS 页面默认始终显示仁川 RKSI，机场切换用 JS 实现 无法通过 URL 参数或 POST form data 切换到其他机场 尝试了 GET ?icao=、?stn=、?airport= 和 POST form data 均无效 变更： - _amos_get_page 简化为仅处理 RKSI - 非 RKSI 直接返回 None，走标准 METAR 回退链 - 移除无效的 AMOS_STATION_IDS 和 POST 策略代码 - Busan 通过 aviationweather.gov METAR 正常获取温度/风/气压 - 跑道面板仅在 AMOS 数据存在时显示（即仅 Seoul/仁川） Known: Busan 无跑道级数据，但标准 METAR 仍然可用 --- src/data_collection/amos_station_sources.py | 84 +++++---------------- 1 file changed, 20 insertions(+), 64 deletions(-) diff --git a/src/data_collection/amos_station_sources.py b/src/data_collection/amos_station_sources.py index 6cad1f76..b13b5003 100644 --- a/src/data_collection/amos_station_sources.py +++ b/src/data_collection/amos_station_sources.py @@ -30,10 +30,6 @@ AMOS_AIRPORT_CODES: Dict[str, Dict[str, str]] = { }, } -AMOS_STATION_IDS: Dict[str, str] = { - "RKSI": "471080", # Incheon - "RKPK": "471530", # Gimhae (Busan) -} def _amos_safe_float(value: str | None) -> Optional[float]: @@ -136,72 +132,32 @@ class AmosStationSourceMixin: amos_cache_ttl_sec: int = 300 # 5 minutes def _amos_get_page(self, icao: str) -> Optional[str]: - """Fetch the AMOS page for a given ICAO code. + """Fetch the AMOS page. - The AMOS site uses JavaScript airport switching. We try multiple - approaches: GET with parameters, POST with form data, and cookie replay. + The AMOS site (global.amo.go.kr) always loads Incheon (RKSI) by default. + Airport switching is done via JavaScript — not via URL parameters or POST. + This means RKPK (Busan/Gimhae) cannot be fetched reliably from this endpoint. + For non-RKSI airports, we fall back to standard METAR sources. """ started = time.perf_counter() - station_id = AMOS_STATION_IDS.get(icao, "") - airport_label_ko = { - "RKSI": "인천", - "RKPK": "김해", - }.get(icao, "") - - fetch_impl = None - if hasattr(self, "session") and hasattr(self.session, "get"): - def _get(url: str) -> str: - resp = self.session.get(url, timeout=float(getattr(self, "timeout", 4.0))) - resp.raise_for_status() - return resp.text - - def _post(url: str, data: dict) -> str: - resp = self.session.post(url, data=data, timeout=float(getattr(self, "timeout", 4.0))) - resp.raise_for_status() - return resp.text - fetch_impl = (_get, _post) - else: - getter = getattr(self, "_http_get_text", None) - if not callable(getter): - return None - def _get(url: str) -> str: - return str(getter(url)) - def _post(url: str, data: dict) -> str: - return str(getter(url)) - fetch_impl = (_get, _post) - - http_get, http_post = fetch_impl + if icao != "RKSI": + logger.debug("AMOS page only supports RKSI (Incheon), not {}", icao) + return None try: - # Strategy 1: GET first to establish session, then POST to switch - _ = http_get(AMOS_BASE_URL) - for form_data in [ - {"icao": icao, "stn": station_id, "airport": airport_label_ko}, - {"stnId": station_id}, - {"code": icao}, - {"selectAirport": icao}, - ]: - try: - text = http_post(AMOS_BASE_URL, form_data) - if text and (icao in text.upper() or airport_label_ko in text): - record_source_call("amos", "page", "success", (time.perf_counter() - started) * 1000.0) - return text - except Exception: - continue - - # Strategy 2: GET with query parameters - for url in [ - f"{AMOS_BASE_URL}?icao={icao}", - f"{AMOS_BASE_URL}?stn={station_id}", - ]: - try: - text = http_get(url) - if text and (icao in text.upper() or airport_label_ko in text): - record_source_call("amos", "page", "success", (time.perf_counter() - started) * 1000.0) - return text - except Exception: - continue + getter = getattr(self, "_http_get_text", None) + if callable(getter): + text = str(getter(AMOS_BASE_URL)) + elif hasattr(self, "session"): + resp = self.session.get(AMOS_BASE_URL, timeout=float(getattr(self, "timeout", 4.0))) + resp.raise_for_status() + text = resp.text + else: + return None + if text and "RKSI" in text.upper(): + record_source_call("amos", "page", "success", (time.perf_counter() - started) * 1000.0) + return text return None except Exception as exc: logger.debug("AMOS page fetch failed icao={}: {}", icao, exc)