From 8bd9d3437667935693407d5273bb0c84503f386f Mon Sep 17 00:00:00 2001 From: linkong Date: Wed, 8 Apr 2026 10:15:33 +0800 Subject: [PATCH] feat: configure collector endpoints and health plan --- backend/app/core/data_sources.py | 16 +- backend/app/core/data_sources.yaml | 34 ++ backend/app/services/collectors/celestrak.py | 4 +- backend/app/services/collectors/cloudflare.py | 30 ++ backend/app/services/collectors/epoch_ai.py | 2 +- .../app/services/collectors/fao_landing.py | 4 +- .../app/services/collectors/huggingface.py | 36 ++ backend/app/services/collectors/peeringdb.py | 35 +- backend/app/services/collectors/ris_live.py | 2 +- backend/app/services/collectors/spacetrack.py | 51 +- .../app/services/collectors/telegeography.py | 15 +- backend/app/services/collectors/top500.py | 7 +- docs/datasource-health-plan.md | 486 ++++++++++++++++++ planet.sh | 4 +- 14 files changed, 673 insertions(+), 53 deletions(-) create mode 100644 docs/datasource-health-plan.md diff --git a/backend/app/core/data_sources.py b/backend/app/core/data_sources.py index 0ec35942..4246052e 100644 --- a/backend/app/core/data_sources.py +++ b/backend/app/core/data_sources.py @@ -11,6 +11,7 @@ COLLECTOR_URL_KEYS = { "fao_landing_points": "fao.landing_point_url", "telegeography_cables": "telegeography.cable_url", "telegeography_landing": "telegeography.landing_point_url", + "telegeography_systems": "telegeography.cable_url", "huggingface_models": "huggingface.models_url", "huggingface_datasets": "huggingface.datasets_url", "huggingface_spaces": "huggingface.spaces_url", @@ -23,6 +24,7 @@ COLLECTOR_URL_KEYS = { "top500": "top500.url", "epoch_ai_gpu": "epoch_ai.gpu_clusters_url", "spacetrack_tle": "spacetrack.tle_query_url", + "celestrak_tle": "celestrak.base_url", "ris_live_bgp": "ris_live.url", "bgpstream_bgp": "bgpstream.url", "iptoasn_prefix_geo": "iptoasn.combined_url", @@ -41,18 +43,22 @@ class DataSourcesConfig: with open(config_path, "r") as f: self._yaml_config = yaml.safe_load(f) or {} - def get_yaml_url(self, collector_name: str) -> str: - key = COLLECTOR_URL_KEYS.get(collector_name, "") + def get_yaml_value(self, key: str): if not key: - return "" + return None parts = key.split(".") value = self._yaml_config for part in parts: if isinstance(value, dict): - value = value.get(part, "") + value = value.get(part) else: - return "" + return None + return value + + def get_yaml_url(self, collector_name: str) -> str: + key = COLLECTOR_URL_KEYS.get(collector_name, "") + value = self.get_yaml_value(key) return value if isinstance(value, str) else "" async def get_url(self, collector_name: str, db) -> str: diff --git a/backend/app/core/data_sources.yaml b/backend/app/core/data_sources.yaml index e1261be8..c4223f74 100644 --- a/backend/app/core/data_sources.yaml +++ b/backend/app/core/data_sources.yaml @@ -2,53 +2,87 @@ # All external data source URLs should be configured here arcgis: + # ArcGIS 海缆 GeoJSON 查询接口 cable_url: "https://services.arcgis.com/6DIQcwlPy8knb6sg/ArcGIS/rest/services/SubmarineCables/FeatureServer/2/query" + # ArcGIS 登陆点 GeoJSON 查询接口 landing_point_url: "https://services.arcgis.com/6DIQcwlPy8knb6sg/ArcGIS/rest/services/SubmarineCables/FeatureServer/1/query" + # ArcGIS 海缆与登陆点关联关系查询接口 cable_landing_relation_url: "https://services.arcgis.com/6DIQcwlPy8knb6sg/ArcGIS/rest/services/SubmarineCables/FeatureServer/3/query" fao: + # FAO 登陆点 CSV 下载地址 landing_point_url: "https://data.apps.fao.org/catalog/dataset/1b75ff21-92f2-4b96-9b7b-98e8aa65ad5d/resource/b6071077-d1d4-4e97-aa00-42e902847c87/download/landing-point-geo.csv" telegeography: + # TeleGeography 海缆/系统主数据源，当前使用 GitHub 镜像 JSON cable_url: "https://raw.githubusercontent.com/lintaojlu/submarine_cable_information/main/cable.json" + # TeleGeography 登陆点主数据源，当前使用 GitHub 镜像 JSON landing_point_url: "https://raw.githubusercontent.com/lintaojlu/submarine_cable_information/main/landing_point.json" + # TeleGeography 历史 API 存档，用于 cable collector 的 fallback + archived_cable_url: "https://web.archive.org/web/2024/https://www.submarinecablemap.com/api/v3/cable" + # TeleGeography 官网页面，用于 cable collector 的最终 HTML 抓取 fallback + live_map_url: "https://www.submarinecablemap.com" huggingface: + # Hugging Face 模型目录 API models_url: "https://huggingface.co/api/models" + # Hugging Face 数据集目录 API datasets_url: "https://huggingface.co/api/datasets" + # Hugging Face Spaces 目录 API spaces_url: "https://huggingface.co/api/spaces" cloudflare: + # Cloudflare Radar 设备类型摘要接口 radar_device_url: "https://api.cloudflare.com/client/v4/radar/http/summary/device_type" + # Cloudflare Radar 请求量时间序列接口 radar_traffic_url: "https://api.cloudflare.com/client/v4/radar/http/timeseries/requests" + # Cloudflare Radar 热点地理位置接口 radar_top_locations_url: "https://api.cloudflare.com/client/v4/radar/http/top/locations" peeringdb: + # PeeringDB IXP API ixp_url: "https://www.peeringdb.com/api/ix" + # PeeringDB Network API network_url: "https://www.peeringdb.com/api/net" + # PeeringDB Facility API facility_url: "https://www.peeringdb.com/api/fac" top500: + # TOP500 榜单页面，用于主表抓取 url: "https://top500.org/lists/top500/list/2025/11/" + # TOP500 站点根地址，用于拼详情页链接 + base_url: "https://top500.org" epoch_ai: + # Epoch AI GPU Cluster 页面 gpu_clusters_url: "https://epoch.ai/data/gpu-clusters" spacetrack: + # Space-Track 站点根地址，用于首页访问和登录地址推导 base_url: "https://www.space-track.org" + # Space-Track TLE 主查询接口 tle_query_url: "https://www.space-track.org/basicspacedata/query/class/gp/orderby/EPOCH%20desc/limit/1000/format/json" +celestrak: + # CelesTrak TLE 基础接口，collector 会在其后拼接 GROUP / FORMAT 参数 + base_url: "https://celestrak.org/NORAD/elements/gp.php" + ris_live: + # RIPE RIS Live 流式订阅地址 url: "https://ris-live.ripe.net/v1/stream/?format=json&client=planet-ris-live" bgpstream: + # CAIDA BGPStream Broker API url: "https://broker.bgpstream.caida.org/v2" iptoasn: + # IPtoASN prefix geography 合并数据下载地址 combined_url: "https://iptoasn.com/data/ip2asn-combined.tsv.gz" opengeofeed: + # OpenGeoFeed 公共 geofeed CSV public_csv_url: "https://opengeofeed.org/feed/public.csv" nro: + # NRO delegated stats 下载地址 delegated_stats_url: "https://ftp.ripe.net/pub/stats/ripencc/nro-stats/latest/nro-delegated-stats" diff --git a/backend/app/services/collectors/celestrak.py b/backend/app/services/collectors/celestrak.py index e6c5f749..6c82d4a2 100644 --- a/backend/app/services/collectors/celestrak.py +++ b/backend/app/services/collectors/celestrak.py @@ -21,7 +21,7 @@ class CelesTrakTLECollector(BaseCollector): @property def base_url(self) -> str: - return "https://celestrak.org/NORAD/elements/gp.php" + return self._resolved_url or "" async def fetch(self) -> List[Dict[str, Any]]: satellite_groups = [ @@ -40,7 +40,7 @@ class CelesTrakTLECollector(BaseCollector): async with httpx.AsyncClient(timeout=120.0) as client: for group in satellite_groups: try: - url = f"https://celestrak.org/NORAD/elements/gp.php?GROUP={group}&FORMAT=json" + url = f"{self.base_url}?GROUP={group}&FORMAT=json" response = await client.get(url) if response.status_code == 200: diff --git a/backend/app/services/collectors/cloudflare.py b/backend/app/services/collectors/cloudflare.py index 4eb24015..bb7ab4cf 100644 --- a/backend/app/services/collectors/cloudflare.py +++ b/backend/app/services/collectors/cloudflare.py @@ -39,6 +39,16 @@ class CloudflareRadarDeviceCollector(HTTPCollector): if CLOUDFLARE_API_TOKEN: self.headers["Authorization"] = f"Bearer {CLOUDFLARE_API_TOKEN}" + @property + def request_url(self) -> str: + return self._resolved_url or self.base_url + + async def fetch(self) -> List[Dict[str, Any]]: + async with httpx.AsyncClient(timeout=60.0) as client: + response = await client.get(self.request_url, headers=self.headers) + response.raise_for_status() + return self.parse_response(response.json()) + def parse_response(self, response: Dict[str, Any]) -> List[Dict[str, Any]]: """Parse Cloudflare Radar device type response""" data = [] @@ -87,6 +97,16 @@ class CloudflareRadarTrafficCollector(HTTPCollector): if CLOUDFLARE_API_TOKEN: self.headers["Authorization"] = f"Bearer {CLOUDFLARE_API_TOKEN}" + @property + def request_url(self) -> str: + return self._resolved_url or self.base_url + + async def fetch(self) -> List[Dict[str, Any]]: + async with httpx.AsyncClient(timeout=60.0) as client: + response = await client.get(self.request_url, headers=self.headers) + response.raise_for_status() + return self.parse_response(response.json()) + def parse_response(self, response: Dict[str, Any]) -> List[Dict[str, Any]]: """Parse Cloudflare Radar traffic timeseries response""" data = [] @@ -135,6 +155,16 @@ class CloudflareRadarTopASCollector(HTTPCollector): if CLOUDFLARE_API_TOKEN: self.headers["Authorization"] = f"Bearer {CLOUDFLARE_API_TOKEN}" + @property + def request_url(self) -> str: + return self._resolved_url or self.base_url + + async def fetch(self) -> List[Dict[str, Any]]: + async with httpx.AsyncClient(timeout=60.0) as client: + response = await client.get(self.request_url, headers=self.headers) + response.raise_for_status() + return self.parse_response(response.json()) + def parse_response(self, response: Dict[str, Any]) -> List[Dict[str, Any]]: """Parse Cloudflare Radar top locations response""" data = [] diff --git a/backend/app/services/collectors/epoch_ai.py b/backend/app/services/collectors/epoch_ai.py index 1fa813d6..b7b0c1a7 100644 --- a/backend/app/services/collectors/epoch_ai.py +++ b/backend/app/services/collectors/epoch_ai.py @@ -23,7 +23,7 @@ class EpochAIGPUCollector(BaseCollector): async def fetch(self) -> List[Dict[str, Any]]: """Fetch Epoch AI GPU clusters data from webpage""" - url = "https://epoch.ai/data/gpu-clusters" + url = self._resolved_url or "" async with httpx.AsyncClient(timeout=60.0) as client: response = await client.get(url) diff --git a/backend/app/services/collectors/fao_landing.py b/backend/app/services/collectors/fao_landing.py index f8c38570..2df6effa 100644 --- a/backend/app/services/collectors/fao_landing.py +++ b/backend/app/services/collectors/fao_landing.py @@ -18,11 +18,9 @@ class FAOLandingPointCollector(BaseCollector): frequency_hours = 168 data_type = "landing_point" - csv_url = "https://data.apps.fao.org/catalog/dataset/1b75ff21-92f2-4b96-9b7b-98e8aa65ad5d/resource/b6071077-d1d4-4e97-aa00-42e902847c87/download/landing-point-geo.csv" - async def fetch(self) -> List[Dict[str, Any]]: async with httpx.AsyncClient(timeout=60.0) as client: - response = await client.get(self.csv_url) + response = await client.get(self._resolved_url or "") response.raise_for_status() return self.parse_csv(response.text) diff --git a/backend/app/services/collectors/huggingface.py b/backend/app/services/collectors/huggingface.py index d4c0af55..fc012dfa 100644 --- a/backend/app/services/collectors/huggingface.py +++ b/backend/app/services/collectors/huggingface.py @@ -21,6 +21,18 @@ class HuggingFaceModelCollector(HTTPCollector): data_type = "model" base_url = "https://huggingface.co/api/models" + @property + def request_url(self) -> str: + return self._resolved_url or self.base_url + + async def fetch(self) -> List[Dict[str, Any]]: + from httpx import AsyncClient + + async with AsyncClient(timeout=60.0) as client: + response = await client.get(self.request_url, headers=self.headers) + response.raise_for_status() + return self.parse_response(response.json()) + def parse_response(self, response: Dict[str, Any]) -> List[Dict[str, Any]]: """Parse Hugging Face models API response""" data = [] @@ -63,6 +75,18 @@ class HuggingFaceDatasetCollector(HTTPCollector): data_type = "dataset" base_url = "https://huggingface.co/api/datasets" + @property + def request_url(self) -> str: + return self._resolved_url or self.base_url + + async def fetch(self) -> List[Dict[str, Any]]: + from httpx import AsyncClient + + async with AsyncClient(timeout=60.0) as client: + response = await client.get(self.request_url, headers=self.headers) + response.raise_for_status() + return self.parse_response(response.json()) + def parse_response(self, response: Dict[str, Any]) -> List[Dict[str, Any]]: """Parse Hugging Face datasets API response""" data = [] @@ -104,6 +128,18 @@ class HuggingFaceSpacesCollector(HTTPCollector): data_type = "space" base_url = "https://huggingface.co/api/spaces" + @property + def request_url(self) -> str: + return self._resolved_url or self.base_url + + async def fetch(self) -> List[Dict[str, Any]]: + from httpx import AsyncClient + + async with AsyncClient(timeout=60.0) as client: + response = await client.get(self.request_url, headers=self.headers) + response.raise_for_status() + return self.parse_response(response.json()) + def parse_response(self, response: Dict[str, Any]) -> List[Dict[str, Any]]: """Parse Hugging Face Spaces API response""" data = [] diff --git a/backend/app/services/collectors/peeringdb.py b/backend/app/services/collectors/peeringdb.py index 5510095c..e9ae2819 100644 --- a/backend/app/services/collectors/peeringdb.py +++ b/backend/app/services/collectors/peeringdb.py @@ -16,6 +16,7 @@ from typing import Dict, Any, List from datetime import UTC, datetime import httpx +from urllib.parse import urlencode from app.services.collectors.base import HTTPCollector @@ -38,9 +39,13 @@ class PeeringDBIXPCollector(HTTPCollector): "User-Agent": "Planet-Intelligence-System/1.0 (Python/collector)", "Accept": "application/json", } - # API key is added to URL as query parameter - if PEERINGDB_API_KEY: - self.base_url = f"{self.base_url}?key={PEERINGDB_API_KEY}" + @property + def request_url(self) -> str: + base = self._resolved_url or self.base_url + if not PEERINGDB_API_KEY: + return base + separator = "&" if "?" in base else "?" + return f"{base}{separator}{urlencode({'key': PEERINGDB_API_KEY})}" async def fetch_with_retry( self, max_retries: int = 3, base_delay: float = 2.0 @@ -51,7 +56,7 @@ class PeeringDBIXPCollector(HTTPCollector): for attempt in range(max_retries): try: async with httpx.AsyncClient(timeout=60.0) as client: - response = await client.get(self.base_url, headers=self.headers) + response = await client.get(self.request_url, headers=self.headers) if response.status_code == 429: # Rate limited - wait and retry with exponential backoff @@ -141,8 +146,13 @@ class PeeringDBNetworkCollector(HTTPCollector): "User-Agent": "Planet-Intelligence-System/1.0 (Python/collector)", "Accept": "application/json", } - if PEERINGDB_API_KEY: - self.base_url = f"{self.base_url}?key={PEERINGDB_API_KEY}" + @property + def request_url(self) -> str: + base = self._resolved_url or self.base_url + if not PEERINGDB_API_KEY: + return base + separator = "&" if "?" in base else "?" + return f"{base}{separator}{urlencode({'key': PEERINGDB_API_KEY})}" async def fetch_with_retry( self, max_retries: int = 3, base_delay: float = 2.0 @@ -153,7 +163,7 @@ class PeeringDBNetworkCollector(HTTPCollector): for attempt in range(max_retries): try: async with httpx.AsyncClient(timeout=60.0) as client: - response = await client.get(self.base_url, headers=self.headers) + response = await client.get(self.request_url, headers=self.headers) if response.status_code == 429: delay = base_delay * (2**attempt) @@ -244,8 +254,13 @@ class PeeringDBFacilityCollector(HTTPCollector): "User-Agent": "Planet-Intelligence-System/1.0 (Python/collector)", "Accept": "application/json", } - if PEERINGDB_API_KEY: - self.base_url = f"{self.base_url}?key={PEERINGDB_API_KEY}" + @property + def request_url(self) -> str: + base = self._resolved_url or self.base_url + if not PEERINGDB_API_KEY: + return base + separator = "&" if "?" in base else "?" + return f"{base}{separator}{urlencode({'key': PEERINGDB_API_KEY})}" async def fetch_with_retry( self, max_retries: int = 3, base_delay: float = 2.0 @@ -256,7 +271,7 @@ class PeeringDBFacilityCollector(HTTPCollector): for attempt in range(max_retries): try: async with httpx.AsyncClient(timeout=60.0) as client: - response = await client.get(self.base_url, headers=self.headers) + response = await client.get(self.request_url, headers=self.headers) if response.status_code == 429: delay = base_delay * (2**attempt) diff --git a/backend/app/services/collectors/ris_live.py b/backend/app/services/collectors/ris_live.py index 38da086a..e39e9849 100644 --- a/backend/app/services/collectors/ris_live.py +++ b/backend/app/services/collectors/ris_live.py @@ -33,7 +33,7 @@ class RISLiveCollector(BaseCollector): def _fetch_via_stream(self) -> list[dict[str, Any]]: events: list[dict[str, Any]] = [] - stream_url = "https://ris-live.ripe.net/v1/stream/?format=json&client=planet-ris-live" + stream_url = self._resolved_url or "" subscribe = json.dumps( { "host": "rrc00", diff --git a/backend/app/services/collectors/spacetrack.py b/backend/app/services/collectors/spacetrack.py index bc960f6b..3104f083 100644 --- a/backend/app/services/collectors/spacetrack.py +++ b/backend/app/services/collectors/spacetrack.py @@ -7,6 +7,7 @@ API documentation: https://www.space-track.org/documentation import json from typing import Dict, Any, List import httpx +from urllib.parse import urlparse from app.services.collectors.base import BaseCollector from app.core.data_sources import get_data_sources_config @@ -21,12 +22,30 @@ class SpaceTrackTLECollector(BaseCollector): data_type = "satellite_tle" @property - def base_url(self) -> str: + def query_url(self) -> str: config = get_data_sources_config() if self._resolved_url: return self._resolved_url return config.get_yaml_url("spacetrack_tle") + @property + def site_root(self) -> str: + config = get_data_sources_config() + configured_root = config.get_yaml_value("spacetrack.base_url") + if isinstance(configured_root, str) and configured_root: + return configured_root.rstrip("/") + + parsed = urlparse(self.query_url) + return f"{parsed.scheme}://{parsed.netloc}".rstrip("/") + + @property + def login_url(self) -> str: + return f"{self.site_root}/ajaxauth/login" + + @property + def probe_url(self) -> str: + return f"{self.site_root}/basicspacedata/query/class/gp/NORAD_CAT_ID/25544/format/json" + async def fetch(self) -> List[Dict[str, Any]]: from app.core.config import settings @@ -47,13 +66,13 @@ class SpaceTrackTLECollector(BaseCollector): "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "application/json, text/html, */*", "Accept-Language": "en-US,en;q=0.9", - "Referer": "https://www.space-track.org/", + "Referer": f"{self.site_root}/", }, ) as client: - await client.get("https://www.space-track.org/") + await client.get(f"{self.site_root}/") login_response = await client.post( - "https://www.space-track.org/ajaxauth/login", + self.login_url, data={ "identity": username, "password": password, @@ -69,7 +88,7 @@ class SpaceTrackTLECollector(BaseCollector): timeout=120.0, follow_redirects=True, ) as alt_client: - await alt_client.get("https://www.space-track.org/") + await alt_client.get(f"{self.site_root}/") form_data = { "username": username, @@ -77,7 +96,7 @@ class SpaceTrackTLECollector(BaseCollector): "query": "class/gp/NORAD_CAT_ID/25544/format/json", } alt_login = await alt_client.post( - "https://www.space-track.org/ajaxauth/login", + self.login_url, data={ "identity": username, "password": password, @@ -86,9 +105,7 @@ class SpaceTrackTLECollector(BaseCollector): print(f"SPACETRACK: Alt login status: {alt_login.status_code}") if alt_login.status_code == 200: - tle_response = await alt_client.get( - "https://www.space-track.org/basicspacedata/query/class/gp/NORAD_CAT_ID/25544/format/json" - ) + tle_response = await alt_client.get(self.probe_url) if tle_response.status_code == 200: data = tle_response.json() print(f"SPACETRACK: Received {len(data)} records via alt method") @@ -98,9 +115,7 @@ class SpaceTrackTLECollector(BaseCollector): print(f"SPACETRACK: Login failed, using sample data") return self._get_sample_data() - tle_response = await client.get( - "https://www.space-track.org/basicspacedata/query/class/gp/NORAD_CAT_ID/25544/format/json" - ) + tle_response = await client.get(self.probe_url) print(f"SPACETRACK: TLE query status: {tle_response.status_code}") if tle_response.status_code != 200: @@ -127,11 +142,11 @@ class SpaceTrackTLECollector(BaseCollector): }, ) as client: # First, visit the main page to get any cookies - await client.get("https://www.space-track.org/") + await client.get(f"{self.site_root}/") # Login to get session cookie login_response = await client.post( - "https://www.space-track.org/ajaxauth/login", + self.login_url, data={ "identity": username, "password": password, @@ -146,13 +161,7 @@ class SpaceTrackTLECollector(BaseCollector): return self._get_sample_data() # Query for TLE data (get first 1000 satellites) - tle_response = await client.get( - "https://www.space-track.org/basicspacedata/query" - "/class/gp" - "/orderby/EPOCH%20desc" - "/limit/1000" - "/format/json" - ) + tle_response = await client.get(self.query_url) print(f"SPACETRACK: TLE query status: {tle_response.status_code}") if tle_response.status_code != 200: diff --git a/backend/app/services/collectors/telegeography.py b/backend/app/services/collectors/telegeography.py index b3bd7c72..d6af4e21 100644 --- a/backend/app/services/collectors/telegeography.py +++ b/backend/app/services/collectors/telegeography.py @@ -11,6 +11,7 @@ from datetime import UTC, datetime from bs4 import BeautifulSoup import httpx +from app.core.data_sources import get_data_sources_config from app.services.collectors.base import BaseCollector @@ -24,15 +25,17 @@ class TeleGeographyCableCollector(BaseCollector): async def fetch(self) -> List[Dict[str, Any]]: """Fetch submarine cable data from Wayback Machine""" + config = get_data_sources_config() # Try multiple data sources sources = [ - # Wayback Machine archive of TeleGeography - "https://web.archive.org/web/2024/https://www.submarinecablemap.com/api/v3/cable", - # Alternative: Try scraping the page - "https://www.submarinecablemap.com", + self._resolved_url or "", + str(config.get_yaml_value("telegeography.archived_cable_url") or ""), + str(config.get_yaml_value("telegeography.live_map_url") or ""), ] for url in sources: + if not url: + continue try: async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as client: response = await client.get(url) @@ -161,7 +164,7 @@ class TeleGeographyLandingPointCollector(BaseCollector): async def fetch(self) -> List[Dict[str, Any]]: """Fetch landing point data from GitHub mirror""" - url = "https://raw.githubusercontent.com/lintaojlu/submarine_cable_information/main/landing_point.json" + url = self._resolved_url or "" async with httpx.AsyncClient(timeout=60.0) as client: response = await client.get(url) @@ -225,7 +228,7 @@ class TeleGeographyCableSystemCollector(BaseCollector): async def fetch(self) -> List[Dict[str, Any]]: """Fetch cable system data""" - url = "https://raw.githubusercontent.com/lintaojlu/submarine_cable_information/main/cable.json" + url = self._resolved_url or "" async with httpx.AsyncClient(timeout=60.0) as client: response = await client.get(url) diff --git a/backend/app/services/collectors/top500.py b/backend/app/services/collectors/top500.py index 9e21937d..547017d8 100644 --- a/backend/app/services/collectors/top500.py +++ b/backend/app/services/collectors/top500.py @@ -10,6 +10,7 @@ from typing import Dict, Any, List from bs4 import BeautifulSoup import httpx +from app.core.data_sources import get_data_sources_config from app.services.collectors.base import BaseCollector @@ -22,7 +23,7 @@ class TOP500Collector(BaseCollector): async def fetch(self) -> List[Dict[str, Any]]: """Fetch TOP500 list data and enrich each row with detail-page metadata.""" - url = "https://top500.org/lists/top500/list/2025/11/" + url = self._resolved_url or "" async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as client: response = await client.get(url) @@ -48,11 +49,13 @@ class TOP500Collector(BaseCollector): return await asyncio.gather(*(enrich(entry) for entry in entries)) def _extract_system_fields(self, system_cell) -> Dict[str, str]: + config = get_data_sources_config() + top500_base_url = config.get_yaml_value("top500.base_url") or "https://top500.org" link = system_cell.find("a") system_name = link.get_text(" ", strip=True) if link else system_cell.get_text(" ", strip=True) detail_url = "" if link and link.get("href"): - detail_url = f"https://top500.org{link.get('href')}" + detail_url = f"{str(top500_base_url).rstrip('/')}{link.get('href')}" manufacturer = "" if link and link.next_sibling: diff --git a/docs/datasource-health-plan.md b/docs/datasource-health-plan.md new file mode 100644 index 00000000..19946406 --- /dev/null +++ b/docs/datasource-health-plan.md @@ -0,0 +1,486 @@ +# Datasource Health Plan + +## Overview + +This document defines a phased plan for datasource health governance. + +The goal is to make collectors observable, diagnosable, and recoverable when upstream APIs change, while avoiding unsafe automatic mutation of repository defaults. + +The key principle is: + +- do not let runtime automation rewrite repository default config + +Instead, split responsibilities across: + +- default config +- runtime overrides +- health check records +- agent-generated repair proposals + + +## Problem Statement + +Collectors currently depend on third-party APIs, data downloads, mirrored JSON files, archive links, and web pages. + +These upstream dependencies can fail in several ways: + +- endpoint becomes unreachable +- endpoint still responds but schema changes +- content-type changes +- website shuts down or moves +- mirror link disappears +- HTML structure changes and scraping fails +- endpoint requires a new path or new host + +We want a system that can: + +- detect datasource health degradation early +- identify likely cause +- search for updated endpoints when reasonable +- apply safe runtime fixes without polluting default repo config +- preserve auditability and rollback + + +## Design Principles + +1. Default config is stable + +- `backend/app/core/data_sources.yaml` remains the repository baseline. +- It should be changed intentionally through normal development flow, not by autonomous runtime agents. + +2. Runtime fixes are isolated + +- Emergency or adaptive fixes should live in a runtime override layer. +- Overrides should be reversible and auditable. + +3. Deterministic checks come first + +- Use normal programmatic health checks before using LLMs. +- Only call an agent when deterministic checks indicate a meaningful failure. + +4. Agents suggest before they mutate + +- Agents should produce proposals with evidence and confidence. +- Application of a proposal should be controlled by policy. + +5. Every repair is attributable + +- Store what changed, why, who or what suggested it, and when it was applied. + + +## Configuration Layers + +Recommended runtime precedence: + +1. datasource endpoint override +2. datasource DB endpoint override +3. repository default YAML +4. collector internal fallback logic + +Definitions: + +- repository default YAML: + - `backend/app/core/data_sources.yaml` + - versioned baseline +- datasource DB endpoint override: + - existing `DataSourceConfig.endpoint` + - current runtime override entrypoint +- datasource endpoint override: + - a dedicated new override table + - used for health-repair and proposal application +- collector internal fallback logic: + - final defensive fallback + - should be minimized over time + + +## Recommended Architecture + +### 1. Deterministic Health Checks + +Each collector gets a health profile with checks such as: + +- endpoint resolves +- HTTP request succeeds +- status code is acceptable +- content-type is expected +- body parses successfully +- minimum structural fields exist +- sample item count is plausible +- latency is within threshold + +Output states: + +- `healthy` +- `degraded` +- `failed` +- `schema_changed` +- `rate_limited` +- `auth_required` + + +### 2. Agent-Assisted Repair Discovery + +Only triggered when deterministic health checks fail or return suspicious structure. + +Agent responsibilities: + +- search for current official endpoint or replacement path +- inspect likely upstream documentation or landing pages +- compare candidate endpoint output to collector expectations +- produce a repair proposal with confidence and evidence + +Agent should not directly modify repository defaults. + + +### 3. Safe Runtime Repair Application + +Repair proposals can be: + +- reviewed manually +- auto-applied only under strict low-risk policy + +Auto-apply should be limited to cases like: + +- same trusted domain +- highly similar response structure +- repeated successful verification +- confidence above threshold + + +## Phased Delivery Plan + +## Phase 1: Deterministic Health MVP + +Goal: + +- build health observability without automated repair + +Scope: + +- datasource health check task runner +- datasource health result persistence +- endpoint reachability + parse checks +- dashboard or API visibility into health status + +Deliverables: + +- health check service +- health check record table +- status endpoint +- scheduled or manual check trigger + +No agent usage yet. + + +## Phase 2: Agent Repair Proposals + +Goal: + +- let agent investigate failing sources and propose updated endpoints + +Scope: + +- invoke agent only when datasource health is `failed` or `schema_changed` +- web search + page inspection +- candidate endpoint extraction +- proposal persistence + +Deliverables: + +- repair proposal schema +- proposal generation pipeline +- confidence and evidence model +- operator review view or API + +Still no automatic config mutation. + + +## Phase 3: Runtime Overrides + +Goal: + +- allow approved proposals to take effect safely at runtime + +Scope: + +- add dedicated override storage +- runtime resolution prefers override over default config +- proposal application writes override only + +Deliverables: + +- endpoint override table +- override-aware resolution logic +- apply/reject endpoints +- rollback endpoint + +Repository default YAML remains untouched. + + +## Phase 4: Limited Auto-Apply + +Goal: + +- safely automate a narrow slice of low-risk repairs + +Scope: + +- policy engine for auto-apply +- same-domain or trusted-domain checks +- structure validation +- staged verification after apply + +Deliverables: + +- auto-apply rules +- audit logs +- automatic post-apply health verification +- auto-disable or rollback on regression + + +## Data Model Draft + +### datasource_health_checks + +Purpose: + +- store each health evaluation result + +Suggested fields: + +- `id` +- `datasource_id` +- `collector_name` +- `endpoint_checked` +- `status` +- `http_status` +- `content_type` +- `latency_ms` +- `sample_count` +- `error_message` +- `details` +- `checked_at` + +`details` can store structured diagnostic data such as: + +- parsed fields +- schema mismatch summary +- retry count +- exception class + + +### datasource_repair_proposals + +Purpose: + +- store agent-generated repair suggestions + +Suggested fields: + +- `id` +- `datasource_id` +- `collector_name` +- `old_endpoint` +- `candidate_endpoint` +- `reason` +- `confidence` +- `evidence_urls` +- `evidence_summary` +- `status` +- `created_by` +- `created_at` +- `reviewed_at` + +Suggested `status` values: + +- `proposed` +- `approved` +- `rejected` +- `applied` +- `expired` + + +### datasource_endpoint_overrides + +Purpose: + +- runtime endpoint override layer + +Suggested fields: + +- `id` +- `datasource_id` +- `collector_name` +- `endpoint` +- `reason` +- `source` +- `proposal_id` +- `enabled` +- `created_at` +- `updated_at` + +Suggested `source` values: + +- `manual` +- `health-agent` +- `migration` + + +## API Draft + +### Health + +- `GET /api/v1/datasources/health` +- `GET /api/v1/datasources/{id}/health` +- `POST /api/v1/datasources/{id}/health-check` +- `POST /api/v1/datasources/health-check-all` + +### Repair proposals + +- `GET /api/v1/datasources/{id}/repair-proposals` +- `POST /api/v1/datasources/{id}/repair-proposals/generate` +- `POST /api/v1/datasources/{id}/repair-proposals/{proposal_id}/approve` +- `POST /api/v1/datasources/{id}/repair-proposals/{proposal_id}/reject` +- `POST /api/v1/datasources/{id}/repair-proposals/{proposal_id}/apply` + +### Overrides + +- `GET /api/v1/datasources/{id}/overrides` +- `POST /api/v1/datasources/{id}/overrides` +- `PUT /api/v1/datasources/{id}/overrides/{override_id}` +- `DELETE /api/v1/datasources/{id}/overrides/{override_id}` + + +## Agent Contract Draft + +When deterministic health fails, the agent should receive: + +- datasource name +- collector name +- current endpoint +- current failure mode +- expected response shape summary +- known trusted domains + +Expected output: + +```json +{ + "status": "proposal", + "candidate_endpoint": "https://example.com/api/v2/data", + "confidence": 0.86, + "reason": "Official docs now point to v2 endpoint", + "evidence_urls": [ + "https://example.com/docs/api", + "https://example.com/changelog" + ], + "notes": "Response shape appears compatible after light field remapping" +} +``` + +The agent should never output "rewrite the default yaml" as its primary action. + + +## Risk Analysis + +### Risk: wrong endpoint chosen by agent + +Mitigation: + +- use trusted-domain allowlists +- require evidence URLs +- require confidence threshold +- add manual review for medium-risk sources + + +### Risk: endpoint responds but schema silently changed + +Mitigation: + +- deterministic schema checks +- parse and sample validation +- content-type checks +- collector-specific required fields + + +### Risk: automatic runtime override causes hidden drift + +Mitigation: + +- store all overrides explicitly +- mark source of override +- keep default YAML unchanged +- expose active overrides in API/UI + + +### Risk: persistent bad override breaks data collection + +Mitigation: + +- allow rollback +- keep parent/default endpoint visible +- re-run verification after apply +- auto-disable override on repeated failure + + +## Operational Policy Recommendations + +1. Do not auto-apply for high-value or high-fragility sources initially. + +2. Use manual approval for: + +- scraped HTML sources +- unofficial mirrors +- sources with auth or rate-limit complexity +- sources with legal or trust ambiguity + +3. Allow auto-apply only for: + +- same-domain version bumps +- obvious official migration paths +- repeated passing verification + +4. Expose health + proposal + override state together in one operator view. + + +## Suggested Implementation Order + +1. Phase 1 +- health result table +- deterministic checks +- API and UI visibility + +2. Phase 2 +- proposal table +- agent prompt/output contract +- proposal generation job + +3. Phase 3 +- runtime override table +- resolver precedence update +- apply/reject endpoints + +4. Phase 4 +- auto-apply rules +- rollback policy +- operator automation + + +## Out Of Scope For The First Iteration + +- direct automatic mutation of repository default YAML +- automatic git commits by repair agents +- unrestricted autonomous endpoint replacement +- fully generalized schema remapping engine + + +## Recommended First Milestone + +The first milestone should be: + +- deterministic datasource health checks +- persisted results +- manual visibility +- no automatic repair + +This gives immediate operational value with low risk, and prepares clean inputs for the later agent phase. diff --git a/planet.sh b/planet.sh index d2badfe9..57d4318b 100755 --- a/planet.sh +++ b/planet.sh @@ -129,7 +129,7 @@ print_splash() { |___|_| |_|\__\___|_|_|_|\__, |\___|_| |_|\__| |___/ EOF - printf "%b\n" "$NC" + # printf "%b\n" "$NC" printf "%b" "$BLUE" cat <<'EOF' ____ _ _ @@ -138,7 +138,7 @@ EOF | __/| | (_| | | | | __/| |_ |_| |_|\__,_|_| |_|\___| \__| EOF - printf "%b\n" "$NC" + # printf "%b\n" "$NC" printf "%b" "$WHITE" cat <<'EOF' ____ _