feat: configure collector endpoints and health plan

This commit is contained in:
linkong
2026-04-08 10:15:33 +08:00
parent 2d43263b9e
commit 8bd9d34376
14 changed files with 673 additions and 53 deletions

View File

@@ -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:

View File

@@ -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"

View File

@@ -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:

View File

@@ -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 = []

View File

@@ -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)

View File

@@ -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)

View File

@@ -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 = []

View File

@@ -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)

View File

@@ -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",

View File

@@ -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:

View File

@@ -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)

View File

@@ -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:

View File

@@ -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.

View File

@@ -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'
____ _