linkong/planet
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

release: bump version to 0.66.1
Some checks failed
ci / backend (push) Has been cancelled
Details
ci / frontend (push) Has been cancelled
Details
ci / delivery (push) Has been cancelled
Details
release / images (push) Has been cancelled
Details

Browse Source
This commit is contained in:
rayd1o
2026-05-26 04:38:18 +08:00
parent 5bf5c73ca0
commit 887fec972e
17 changed files with 667 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

211
backend/app/services/collectors/celestrak.py
View File

@@ -16,13 +16,24 @@ from app.core.logging import get_logger
from app.core.satellite_tle import build_tle_lines_from_elements
from app.services.business_logs import emit_business_log, exception_context
from app.services.collectors.base import BaseCollector
from app.services.collectors.downloads import ResumableFileDownloader
from app.services.collectors.downloads import DownloadHTTPStatusError, ResumableFileDownloader
logger = get_logger(__name__, service="collector")
ACTIVE_GROUP = "active"
FALLBACK_GROUPS = (
"starlink",
"gps-ops",
"galileo",
"glonass",
"beidou",
"leo",
"geo",
"iridium-next",
)
FETCH_RETRY_ATTEMPTS = 3
FETCH_RETRY_BASE_DELAY_SECONDS = 0.8
CELESTRAK_NOT_UPDATED_MARKER = "GP data has not updated since your last successful"
class CelesTrakTLECollector(BaseCollector):
@@ -41,9 +52,12 @@ class CelesTrakTLECollector(BaseCollector):
return self._resolved_url or ""
def _active_url(self) -> str:
return self._group_url(ACTIVE_GROUP)
def _group_url(self, group: str) -> str:
if not self.base_url:
raise RuntimeError("CelesTrak base URL is not configured")
return f"{self.base_url}?{urlencode({'GROUP': ACTIVE_GROUP, 'FORMAT': 'json'})}"
return f"{self.base_url}?{urlencode({'GROUP': group, 'FORMAT': 'json'})}"
async def fetch(self) -> List[Dict[str, Any]]:
url = self._active_url()
@@ -76,14 +90,7 @@ class CelesTrakTLECollector(BaseCollector):
progress_callback=self._report_download_progress,
validate_existing=self._validate_json_file,
)
try:
data = self._load_active_payload(body_path)
except RuntimeError as exc:
await self._log_parse_failure(exc)
raise
for item in data:
item["_celestrak_query_group"] = ACTIVE_GROUP
item["_celestrak_source_url"] = url
data = await self._load_downloaded_payload(body_path, url)
await emit_business_log(
logger,
event="collector.celestrak.download.success",
@@ -101,6 +108,54 @@ class CelesTrakTLECollector(BaseCollector):
},
)
return data
except DownloadHTTPStatusError as exc:
if self._is_not_updated_response(exc):
cached_path = self._downloader.get_cached_file(
url,
".json",
validate_existing=self._validate_json_file,
)
if cached_path is not None:
data = await self._load_downloaded_payload(cached_path, url)
await emit_business_log(
logger,
event="collector.celestrak.download.cached_not_updated",
message="CelesTrak active satellite data has not changed; using cached download",
category="collector",
level="warning",
service="collector",
module=__name__,
context={
"collector_name": self.name,
"datasource_id": getattr(self, "_datasource_id", None),
"group": ACTIVE_GROUP,
"attempt": attempt,
"record_count": len(data),
"duration_ms": self._duration_ms(started_at),
},
)
return data
await emit_business_log(
logger,
event="collector.celestrak.download.not_updated_no_cache",
message="CelesTrak active satellite data has not changed; trying fallback groups",
category="collector",
level="warning",
service="collector",
module=__name__,
context=exception_context(
exc,
{
"collector_name": self.name,
"datasource_id": getattr(self, "_datasource_id", None),
"group": ACTIVE_GROUP,
"attempt": attempt,
"duration_ms": self._duration_ms(started_at),
},
),
)
return await self._fetch_fallback_groups(client, active_error=exc)
raise
except Exception as exc:
last_error = exc
is_final_attempt = attempt >= FETCH_RETRY_ATTEMPTS
@@ -136,6 +191,142 @@ class CelesTrakTLECollector(BaseCollector):
raise RuntimeError(f"CelesTrak active satellite download failed after retries: {last_error}")
async def _fetch_fallback_groups(
self,
client: httpx.AsyncClient,
*,
active_error: DownloadHTTPStatusError,
) -> List[Dict[str, Any]]:
started_at = perf_counter()
records_by_norad: dict[str, Dict[str, Any]] = {}
group_counts: dict[str, int] = {}
await emit_business_log(
logger,
event="collector.celestrak.fallback_groups.start",
message="CelesTrak fallback group download started",
category="collector",
level="warning",
service="collector",
module=__name__,
context={
"collector_name": self.name,
"datasource_id": getattr(self, "_datasource_id", None),
"groups": list(FALLBACK_GROUPS),
"reason": "active_not_updated_without_cache",
},
)
try:
for group in FALLBACK_GROUPS:
group_url = self._group_url(group)
try:
body_path = await self._downloader.download_file(
client,
group_url,
extension=".json",
accept="application/json",
validate_existing=self._validate_json_file,
)
except DownloadHTTPStatusError as exc:
if not self._is_not_updated_response(exc):
raise RuntimeError(f"CelesTrak fallback group '{group}' download failed: {exc}") from exc
cached_path = self._downloader.get_cached_file(
group_url,
".json",
validate_existing=self._validate_json_file,
)
if cached_path is None:
raise RuntimeError(
f"CelesTrak fallback group '{group}' has not updated and no local cached copy is available"
) from exc
body_path = cached_path
group_records = await self._load_downloaded_payload(
body_path,
group_url,
query_group=group,
constellation_group=group,
)
group_counts[group] = len(group_records)
for item in group_records:
norad_cat_id = item.get("NORAD_CAT_ID")
if norad_cat_id is None:
continue
records_by_norad.setdefault(str(norad_cat_id), item)
except Exception as exc:
await emit_business_log(
logger,
event="collector.celestrak.fallback_groups.failed",
message="CelesTrak fallback group download failed",
category="collector",
level="error",
service="collector",
module=__name__,
context=exception_context(
exc,
{
"collector_name": self.name,
"datasource_id": getattr(self, "_datasource_id", None),
"groups": list(FALLBACK_GROUPS),
"completed_groups": list(group_counts),
"duration_ms": self._duration_ms(started_at),
},
),
)
raise RuntimeError(
"CelesTrak active data has not updated since this network's last successful download, "
"no active cache is available, and fallback group mode failed. Wait until CelesTrak "
"publishes the next GP update, restore the Planet download cache, or use Space-Track."
) from active_error
records = list(records_by_norad.values())
if not records:
raise RuntimeError("CelesTrak fallback group mode produced no satellite records")
await emit_business_log(
logger,
event="collector.celestrak.fallback_groups.success",
message="CelesTrak fallback group download completed",
category="collector",
level="warning",
service="collector",
module=__name__,
context={
"collector_name": self.name,
"datasource_id": getattr(self, "_datasource_id", None),
"groups": list(FALLBACK_GROUPS),
"group_counts": group_counts,
"record_count": len(records),
"duration_ms": self._duration_ms(started_at),
},
)
return records
async def _load_downloaded_payload(
self,
body_path: Path,
url: str,
*,
query_group: str = ACTIVE_GROUP,
constellation_group: str | None = None,
) -> List[Dict[str, Any]]:
try:
data = self._load_active_payload(body_path)
except RuntimeError as exc:
await self._log_parse_failure(exc)
raise
for item in data:
item["_celestrak_query_group"] = query_group
item["_celestrak_source_url"] = url
if constellation_group:
item["_celestrak_group"] = constellation_group
return data
@staticmethod
def _is_not_updated_response(exc: DownloadHTTPStatusError) -> bool:
return exc.status_code == 403 and CELESTRAK_NOT_UPDATED_MARKER in exc.body
@staticmethod
def _duration_ms(started_at: float) -> int:
return int((perf_counter() - started_at) * 1000)

53
backend/app/services/collectors/downloads.py
View File

@@ -4,7 +4,7 @@ from __future__ import annotations
import hashlib
import json
import tempfile
import os
import time
from datetime import UTC, datetime
from pathlib import Path
@@ -17,6 +17,31 @@ ProgressCallback = Callable[[int, int | None], Awaitable[None]]
ValidateCallback = Callable[[Path], bool]
class DownloadHTTPStatusError(RuntimeError):
"""HTTP status error that keeps the upstream response body for caller-specific handling."""
def __init__(self, *, url: str, status_code: int, body: str) -> None:
self.url = url
self.status_code = status_code
self.body = body
preview = body.strip().replace("\r", " ").replace("\n", " ")[:240]
suffix = f": {preview}" if preview else ""
super().__init__(f"HTTP {status_code} while downloading {url}{suffix}")
def default_download_cache_root() -> Path:
configured = os.getenv("PLANET_DOWNLOAD_CACHE_DIR")
if configured:
return Path(configured).expanduser()
planet_cache = os.getenv("PLANET_CACHE_DIR")
if planet_cache:
return Path(planet_cache).expanduser() / "downloads"
xdg_cache = os.getenv("XDG_CACHE_HOME")
if xdg_cache:
return Path(xdg_cache).expanduser() / "planet" / "downloads"
return Path.home() / ".cache" / "planet" / "downloads"
class ResumableFileDownloader:
"""Download files with cache validators and byte-range resume support."""
@@ -26,8 +51,9 @@ class ResumableFileDownloader:
cache_namespace: str,
user_agent: str = "Planet-Intelligence-System/1.0 (Python/collector)",
default_accept: str = "*/*",
cache_root: Path | None = None,
) -> None:
self._cache_dir = Path(tempfile.gettempdir()) / "planet-download-cache" / cache_namespace
self._cache_dir = (cache_root or default_download_cache_root()) / cache_namespace
self._user_agent = user_agent
self._default_accept = default_accept
@@ -43,6 +69,25 @@ class ResumableFileDownloader:
meta_path = self._cache_dir / f"{key}.meta.json"
return final_path, part_path, meta_path
def cached_file_path(self, url: str, extension: str) -> Path:
final_path, _, _ = self._cache_paths(url, extension)
return final_path
def get_cached_file(
self,
url: str,
extension: str,
*,
validate_existing: ValidateCallback | None = None,
) -> Path | None:
final_path = self.cached_file_path(url, extension)
if not final_path.exists():
return None
if validate_existing and not validate_existing(final_path):
final_path.unlink(missing_ok=True)
return None
return final_path
@staticmethod
def _load_meta(meta_path: Path) -> dict[str, Any]:
if not meta_path.exists():
@@ -140,7 +185,9 @@ class ResumableFileDownloader:
if progress_callback and expected_size and expected_size > 0:
await progress_callback(expected_size, expected_size)
return final_path
response.raise_for_status()
if response.status_code >= 400:
body = (await response.aread()).decode("utf-8", errors="replace")
raise DownloadHTTPStatusError(url=url, status_code=response.status_code, body=body)
if response.status_code == 206 and resume_from > 0:
mode = "ab"