按星座分组名称拉取 TLE 条目。 首先尝试从 Celestrak 在线拉取;若拉取失败或超时,则回退到本地缓存。 若既无网络也无缓存,返回空列表并记录警告。 参数 ---- group : str 星座/分组标识,例如 ``"starlink"``、``"gps-ops"``、``"stations"``。 支持别名映射(见 ``_GROUP_ALIASES``)。 force_refresh : bool 为 True 时即使缓存未过期也强制拉取,默认 False。 返回
(
group: str,
force_refresh: bool = False,
)
| 216 | # --------------------------------------------------------------------------- |
| 217 | |
| 218 | def fetch_group( |
| 219 | group: str, |
| 220 | force_refresh: bool = False, |
| 221 | ) -> List[TleEntry]: |
| 222 | """按星座分组名称拉取 TLE 条目。 |
| 223 | |
| 224 | 首先尝试从 Celestrak 在线拉取;若拉取失败或超时,则回退到本地缓存。 |
| 225 | 若既无网络也无缓存,返回空列表并记录警告。 |
| 226 | |
| 227 | 参数 |
| 228 | ---- |
| 229 | group : str |
| 230 | 星座/分组标识,例如 ``"starlink"``、``"gps-ops"``、``"stations"``。 |
| 231 | 支持别名映射(见 ``_GROUP_ALIASES``)。 |
| 232 | force_refresh : bool |
| 233 | 为 True 时即使缓存未过期也强制拉取,默认 False。 |
| 234 | |
| 235 | 返回 |
| 236 | ---- |
| 237 | list of TleEntry |
| 238 | 每条记录为 (name, line1, line2, epoch_utc) 元组。 |
| 239 | """ |
| 240 | canonical = _GROUP_ALIASES.get(group, group) |
| 241 | |
| 242 | # 检查缓存新鲜度 |
| 243 | if not force_refresh and _cache_is_fresh(canonical): |
| 244 | logger.debug("使用新鲜 TLE 缓存: group=%s", canonical) |
| 245 | cached = _read_cache(canonical) |
| 246 | if cached: |
| 247 | return _parse_tle_text(cached) |
| 248 | |
| 249 | # 尝试在线拉取 |
| 250 | url = _CELESTRAK_TLE_URL.format(group=canonical) |
| 251 | try: |
| 252 | raw_text = _http_get(url, timeout=_FETCH_TIMEOUT) |
| 253 | if raw_text and len(raw_text.strip()) > 0: |
| 254 | _write_cache(canonical, raw_text) |
| 255 | return _parse_tle_text(raw_text) |
| 256 | else: |
| 257 | logger.warning("Celestrak 返回空响应: group=%s url=%s", canonical, url) |
| 258 | except urllib.error.URLError as exc: |
| 259 | logger.warning("Celestrak 拉取失败 (URLError): group=%s, error=%s", canonical, exc) |
| 260 | except Exception as exc: # pragma: no cover |
| 261 | logger.warning("Celestrak 拉取异常: group=%s, error=%s", canonical, exc) |
| 262 | |
| 263 | # 回退到缓存 |
| 264 | cached = _read_cache(canonical) |
| 265 | if cached: |
| 266 | logger.info("网络不可用,使用离线 TLE 缓存: group=%s", canonical) |
| 267 | return _parse_tle_text(cached) |
| 268 | |
| 269 | logger.warning("无法获取 TLE:在线拉取失败且无本地缓存: group=%s", canonical) |
| 270 | return [] |
| 271 | |
| 272 | |
| 273 | def fetch_by_catnr(catnr: int, force_refresh: bool = False) -> List[TleEntry]: |
no test coverage detected