diff --git a/TODO.md b/TODO.md index 2a968fcd..92a3a977 100644 --- a/TODO.md +++ b/TODO.md @@ -22,3 +22,5 @@ - [ ] 可选优化（非必做）：将 BGP incident/collector 标点改为 HTML marker（参考 worldmonitor 的 `htmlElementsData` 思路），实现近乎固定屏幕尺寸与更高密度可点击性 - [ ] 保持 Earth 当前这批纯个人偏好设置继续走本地持久化：`旋转模式`、HUD 面板显示/隐藏、`地形透明度` 暂不升级到后端系统设置，避免把设备级偏好过早做成全局配置 - [ ] 如果后续明确需要“账号级同步 Earth 偏好”，再单独设计 `Earth user preferences`：优先按用户维度而不是全局系统设置保存，并规划 `localStorage -> backend` 的平滑迁移策略 +- [ ] 把 Earth 态势新闻源从 [earth_news.py](/home/ray/dev/linkong/planet/backend/app/services/earth_news.py) 的硬编码列表抽成可配置目录，优先保持当前“实时聚合”链路不变，只先解决新闻源不可配置的问题 +- [ ] 为 Earth 态势新闻设计后续采集器化方案：明确新闻数据模型、去重策略、区域映射、过期清理和 Earth/AI 复用方式，再决定何时把新闻从实时抓取升级成正式 collector diff --git a/VERSION b/VERSION index 9eb2aa3f..be386c9e 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.32.0 +0.33.0 diff --git a/backend/app/api/v1/datasources.py b/backend/app/api/v1/datasources.py index 08b8b154..fedcad0c 100644 --- a/backend/app/api/v1/datasources.py +++ b/backend/app/api/v1/datasources.py @@ -420,6 +420,7 @@ async def list_datasources( collector_list.append( { "id": datasource.id, + "source": datasource.source, "name": datasource.name, "module": datasource.module, "priority": datasource.priority, diff --git a/backend/app/core/data_sources.py b/backend/app/core/data_sources.py index 4246052e..8a2a7669 100644 --- a/backend/app/core/data_sources.py +++ b/backend/app/core/data_sources.py @@ -30,6 +30,7 @@ COLLECTOR_URL_KEYS = { "iptoasn_prefix_geo": "iptoasn.combined_url", "opengeofeed_prefix_geo": "opengeofeed.public_csv_url", "nro_delegated_prefix_geo": "nro.delegated_stats_url", + "news_live_streams": "news_live_streams.channels_url", } diff --git a/backend/app/core/data_sources.yaml b/backend/app/core/data_sources.yaml index c4223f74..17e5658e 100644 --- a/backend/app/core/data_sources.yaml +++ b/backend/app/core/data_sources.yaml @@ -86,3 +86,11 @@ opengeofeed: nro: # NRO delegated stats 下载地址 delegated_stats_url: "https://ftp.ripe.net/pub/stats/ripencc/nro-stats/latest/nro-delegated-stats" + +news_live_streams: + # IPTV-org 频道元数据 JSON + channels_url: "https://iptv-org.github.io/api/channels.json" + # IPTV-org 频道播放流 JSON + streams_url: "https://iptv-org.github.io/api/streams.json" + # IPTV-org 台标 JSON + logos_url: "https://iptv-org.github.io/api/logos.json" diff --git a/backend/app/services/collectors/news_live_streams.py b/backend/app/services/collectors/news_live_streams.py index 84644605..f3b89e99 100644 --- a/backend/app/services/collectors/news_live_streams.py +++ b/backend/app/services/collectors/news_live_streams.py @@ -1,10 +1,16 @@ from __future__ import annotations +import asyncio +import base64 from datetime import UTC, datetime from typing import Any +from urllib.parse import urlparse import httpx +from sqlalchemy import select +from app.core.data_sources import get_data_sources_config +from app.models.datasource_config import DataSourceConfig from app.services.collectors.base import BaseCollector @@ -18,52 +24,537 @@ class NewsLiveStreamsCollector(BaseCollector): data_type = "news_live_stream" fail_on_empty = False + DEFAULT_TIMEOUT = 45.0 + DEFAULT_HEADERS = { + "User-Agent": "Planet-Intelligence-System/1.0 (Python/collector)", + "Accept": "application/json", + } + RESPONSE_CANDIDATE_KEYS = ("sources", "streams", "channels", "items", "results", "data") + DEFAULT_ADAPTER = "iptv_org" + DEFAULT_IPTV_ORG_STREAMS_URL = "https://iptv-org.github.io/api/streams.json" + DEFAULT_IPTV_ORG_LOGOS_URL = "https://iptv-org.github.io/api/logos.json" + DEFAULT_IPTV_ORG_NEWS_CATEGORIES = ("news", "business", "weather") + DEFAULT_IPTV_ORG_EXCLUDE_CATEGORIES = ("music", "sports", "kids", "entertainment") + DEFAULT_IPTV_ORG_MAX_SOURCES = 120 + async def fetch(self) -> list[dict[str, Any]]: request_url = (self._resolved_url or "").strip() if not request_url: return [] - async with httpx.AsyncClient(timeout=45.0, follow_redirects=True) as client: - response = await client.get( + datasource_config = await self._load_datasource_config() + effective_config = self._get_effective_config(datasource_config) + adapter = str(effective_config.get("adapter") or "").strip().lower() + if adapter == "iptv_org": + return await self._fetch_iptv_org(request_url, effective_config) + + request_headers = self._build_request_headers(datasource_config) + request_config = self._get_request_config(datasource_config) + request_params = self._build_request_params(datasource_config) + request_json = self._build_request_json_body(datasource_config) + request_data = self._build_request_form_body(datasource_config) + timeout = self._get_timeout(datasource_config) + + async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client: + response = await client.request( + request_config["method"], request_url, - headers={ - "User-Agent": "Planet-Intelligence-System/1.0 (Python/collector)", - "Accept": "application/json", - }, + headers=request_headers, + params=request_params or None, + json=request_json, + data=request_data, ) response.raise_for_status() - return self.parse_response(response.json()) + return self.parse_response( + response.json(), + response_path=request_config["response_path"], + ) - def parse_response(self, response: Any) -> list[dict[str, Any]]: - if isinstance(response, dict): - candidates = response.get("sources") or response.get("streams") or response.get("data") or [] - elif isinstance(response, list): - candidates = response + async def _load_datasource_config(self) -> DataSourceConfig | None: + if not self._db_session: + return None + + result = await self._db_session.execute( + select(DataSourceConfig) + .where(DataSourceConfig.name == self.name) + .where(DataSourceConfig.is_active.is_(True)) + .limit(1) + ) + return result.scalar_one_or_none() + + def _get_effective_config(self, datasource_config: DataSourceConfig | None) -> dict[str, Any]: + payload = dict(datasource_config.config or {}) if datasource_config else {} + if payload: + return payload + + yaml_config = get_data_sources_config() + return { + "adapter": self.DEFAULT_ADAPTER, + "streams_url": yaml_config.get_yaml_value("news_live_streams.streams_url") + or self.DEFAULT_IPTV_ORG_STREAMS_URL, + "logos_url": yaml_config.get_yaml_value("news_live_streams.logos_url") + or self.DEFAULT_IPTV_ORG_LOGOS_URL, + "news_categories": list(self.DEFAULT_IPTV_ORG_NEWS_CATEGORIES), + "exclude_categories": list(self.DEFAULT_IPTV_ORG_EXCLUDE_CATEGORIES), + "max_sources": self.DEFAULT_IPTV_ORG_MAX_SOURCES, + } + + def _get_request_config(self, datasource_config: DataSourceConfig | None) -> dict[str, Any]: + payload = self._get_effective_config(datasource_config) + raw_method = payload.get("method") or payload.get("request_method") or "GET" + method = str(raw_method).strip().upper() or "GET" + if method not in {"GET", "POST"}: + method = "GET" + + response_path = payload.get("response_path") or payload.get("payload_path") or payload.get("items_path") + if isinstance(response_path, str): + response_path = response_path.strip() else: - candidates = [] + response_path = None + + return { + "method": method, + "response_path": response_path or None, + } + + def _get_timeout(self, datasource_config: DataSourceConfig | None) -> float: + payload = self._get_effective_config(datasource_config) + try: + return float(payload.get("timeout", self.DEFAULT_TIMEOUT)) + except (TypeError, ValueError): + return self.DEFAULT_TIMEOUT + + def _build_request_headers(self, datasource_config: DataSourceConfig | None) -> dict[str, str]: + headers = dict(self.DEFAULT_HEADERS) + if datasource_config: + headers.update(self._normalize_headers(datasource_config.headers)) + headers.update(self._build_auth_headers(datasource_config)) + return headers + + def _build_request_params(self, datasource_config: DataSourceConfig | None) -> dict[str, Any]: + params: dict[str, Any] = {} + if not datasource_config: + return params + + payload = datasource_config.config or {} + candidate = payload.get("params") or payload.get("query_params") + if isinstance(candidate, dict): + params.update(candidate) + + if datasource_config.auth_type == "api_key": + auth_config = datasource_config.auth_config or {} + if str(auth_config.get("in") or auth_config.get("location") or "header").lower() == "query": + api_key = auth_config.get("api_key") + key_name = auth_config.get("key_name") or auth_config.get("param_name") or "api_key" + if api_key and key_name: + params[str(key_name)] = api_key + + return params + + def _build_request_json_body(self, datasource_config: DataSourceConfig | None) -> Any: + if not datasource_config: + return None + + payload = datasource_config.config or {} + body = payload.get("json_body") + if body is None and str(payload.get("body_type") or "").lower() in {"json", ""}: + candidate = payload.get("body") + if isinstance(candidate, (dict, list)): + body = candidate + return body + + def _build_request_form_body(self, datasource_config: DataSourceConfig | None) -> Any: + if not datasource_config: + return None + + payload = datasource_config.config or {} + form_body = payload.get("form_body") + if form_body is not None: + return form_body + + if str(payload.get("body_type") or "").lower() == "form": + candidate = payload.get("body") + if isinstance(candidate, dict): + return candidate + return None + + def _normalize_headers(self, headers: Any) -> dict[str, str]: + if not isinstance(headers, dict): + return {} + normalized: dict[str, str] = {} + for key, value in headers.items(): + header_name = str(key).strip() + if not header_name or value is None: + continue + normalized[header_name] = str(value) + return normalized + + def _build_auth_headers(self, datasource_config: DataSourceConfig | None) -> dict[str, str]: + if not datasource_config: + return {} + + auth_type = str(datasource_config.auth_type or "none").lower() + auth_config = datasource_config.auth_config or {} + if auth_type == "bearer" and auth_config.get("token"): + return {"Authorization": f"Bearer {auth_config['token']}"} + + if auth_type == "api_key" and auth_config.get("api_key"): + location = str(auth_config.get("in") or auth_config.get("location") or "header").lower() + if location == "query": + return {} + key_name = auth_config.get("key_name") or "X-API-Key" + return {str(key_name): str(auth_config["api_key"])} + + if auth_type == "basic": + username = str(auth_config.get("username") or "") + password = str(auth_config.get("password") or "") + encoded = base64.b64encode(f"{username}:{password}".encode()).decode() + return {"Authorization": f"Basic {encoded}"} + + return {} + + def _extract_candidates(self, response: Any, response_path: str | None) -> list[Any]: + if response_path: + extracted = self._extract_from_path(response, response_path) + if isinstance(extracted, list): + return extracted + if isinstance(extracted, dict): + for key in self.RESPONSE_CANDIDATE_KEYS: + nested = extracted.get(key) + if isinstance(nested, list): + return nested + return [extracted] + + if isinstance(response, dict): + for key in self.RESPONSE_CANDIDATE_KEYS: + nested = response.get(key) + if isinstance(nested, list): + return nested + return [] + + if isinstance(response, list): + return response + return [] + + def _extract_from_path(self, payload: Any, path: str) -> Any: + current = payload + for segment in (part.strip() for part in path.split(".") if part.strip()): + if isinstance(current, dict): + current = current.get(segment) + continue + if isinstance(current, list): + try: + current = current[int(segment)] + except (TypeError, ValueError, IndexError): + return None + continue + return None + return current + + def _infer_source_type(self, item: dict[str, Any]) -> str: + explicit = str(item.get("source_type") or item.get("type") or "").strip().lower() + if explicit in {"iframe", "hls", "video", "external", "youtube"}: + return explicit + + youtube_video_id = self._clean_text( + item.get("youtube_video_id") + or item.get("video_id") + or item.get("youtubeVideoId") + ) + youtube_channel = self._clean_text(item.get("youtube_channel") or item.get("channel_handle")) + embed_url = self._clean_url(item.get("embed_url") or item.get("embed") or item.get("page_url")) + stream_url = self._clean_url(item.get("stream_url") or item.get("stream") or item.get("playback_url") or item.get("hls_url")) + homepage_url = self._clean_url(item.get("homepage_url") or item.get("source_url") or item.get("website")) + + if youtube_video_id or youtube_channel: + return "youtube" + if stream_url.endswith(".m3u8"): + return "hls" + if stream_url: + return "video" + if embed_url: + parsed = urlparse(embed_url) + if "youtube.com" in (parsed.netloc or "") or "youtu.be" in (parsed.netloc or ""): + return "youtube" + return "iframe" + if homepage_url: + return "external" + return "iframe" + + def _parse_enabled(self, item: dict[str, Any]) -> bool: + if "is_enabled" in item: + return self._to_bool(item.get("is_enabled"), default=True) + if "enabled" in item: + return self._to_bool(item.get("enabled"), default=True) + if "active" in item: + return self._to_bool(item.get("active"), default=True) + if "status" in item: + status = str(item.get("status") or "").strip().lower() + if status in {"disabled", "inactive", "offline"}: + return False + if status in {"enabled", "active", "online", "live"}: + return True + return True + + def _to_bool(self, value: Any, *, default: bool) -> bool: + if isinstance(value, bool): + return value + if value in (None, ""): + return default + if isinstance(value, str): + lowered = value.strip().lower() + if lowered in {"1", "true", "yes", "on", "enabled", "active", "online", "live"}: + return True + if lowered in {"0", "false", "no", "off", "disabled", "inactive", "offline"}: + return False + return bool(value) + + def _clean_text(self, value: Any) -> str: + if value is None: + return "" + return str(value).strip() + + def _clean_url(self, value: Any) -> str: + text = self._clean_text(value) + if not text: + return "" + parsed = urlparse(text) + if parsed.scheme and parsed.scheme not in {"http", "https"}: + return "" + if parsed.scheme and not parsed.netloc: + return "" + return text + + async def _fetch_iptv_org(self, channels_url: str, collector_config: dict[str, Any]) -> list[dict[str, Any]]: + streams_url = self._clean_url(collector_config.get("streams_url")) or self.DEFAULT_IPTV_ORG_STREAMS_URL + logos_url = self._clean_url(collector_config.get("logos_url")) or self.DEFAULT_IPTV_ORG_LOGOS_URL + news_categories = { + self._clean_text(value).lower() + for value in (collector_config.get("news_categories") or self.DEFAULT_IPTV_ORG_NEWS_CATEGORIES) + if self._clean_text(value) + } + exclude_categories = { + self._clean_text(value).lower() + for value in (collector_config.get("exclude_categories") or self.DEFAULT_IPTV_ORG_EXCLUDE_CATEGORIES) + if self._clean_text(value) + } + try: + max_sources = int(collector_config.get("max_sources", self.DEFAULT_IPTV_ORG_MAX_SOURCES)) + except (TypeError, ValueError): + max_sources = self.DEFAULT_IPTV_ORG_MAX_SOURCES + + timeout = self.DEFAULT_TIMEOUT + try: + timeout = float(collector_config.get("timeout", self.DEFAULT_TIMEOUT)) + except (TypeError, ValueError): + timeout = self.DEFAULT_TIMEOUT + + async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client: + channels_payload, streams_payload, logos_payload = await self._gather_iptv_org_payloads( + client, + channels_url, + streams_url, + logos_url, + ) + + channels = channels_payload if isinstance(channels_payload, list) else [] + streams = streams_payload if isinstance(streams_payload, list) else [] + logos = logos_payload if isinstance(logos_payload, list) else [] + + logo_by_channel = { + self._clean_text(item.get("channel")): self._clean_url(item.get("url")) + for item in logos + if isinstance(item, dict) and self._clean_text(item.get("channel")) and self._clean_url(item.get("url")) + } + + streams_by_channel: dict[str, list[dict[str, Any]]] = {} + for stream in streams: + if not isinstance(stream, dict): + continue + channel_id = self._clean_text(stream.get("channel")) + if not channel_id: + continue + streams_by_channel.setdefault(channel_id, []).append(stream) + + normalized: list[dict[str, Any]] = [] + for channel in channels: + if not isinstance(channel, dict): + continue + + categories = [ + self._clean_text(value).lower() + for value in (channel.get("categories") or []) + if self._clean_text(value) + ] + if news_categories and not any(category in news_categories for category in categories): + continue + if exclude_categories and any(category in exclude_categories for category in categories): + continue + if channel.get("is_nsfw") is True: + continue + if channel.get("closed"): + continue + + channel_id = self._clean_text(channel.get("id")) + if not channel_id: + continue + + stream = self._pick_iptv_org_stream(streams_by_channel.get(channel_id) or []) + if not stream: + continue + + stream_url = self._clean_url(stream.get("url")) + if not stream_url: + continue + + name = self._clean_text(channel.get("name")) or channel_id + notes_parts = [ + f"Imported from IPTV-org catalog ({channel_id})", + f"Categories: {', '.join(categories)}" if categories else "", + f"Quality: {self._clean_text(stream.get('quality'))}" if self._clean_text(stream.get("quality")) else "", + ] + metadata = { + "provider": self._clean_text(channel.get("network")) or "IPTV-org", + "region": self._clean_text(channel.get("country")) or "Global", + "language": "und", + "source_type": "hls" if stream_url.endswith(".m3u8") else "video", + "embed_url": "", + "stream_url": stream_url, + "homepage_url": self._clean_url(channel.get("website")), + "poster_url": logo_by_channel.get(channel_id, ""), + "youtube_video_id": "", + "youtube_channel": "", + "sort_order": 400 + len(normalized), + "notes": "; ".join(part for part in notes_parts if part), + "is_enabled": True, + "collector_adapter": "iptv_org", + "channel_id": channel_id, + "categories": categories, + "quality": self._clean_text(stream.get("quality")), + "stream_label": self._clean_text(stream.get("label") or stream.get("title")), + "stream_referrer": self._clean_text(stream.get("referrer")), + "stream_user_agent": self._clean_text(stream.get("user_agent")), + } + + normalized.append( + { + "source_id": channel_id, + "name": name, + "description": metadata["notes"], + "metadata": metadata, + "reference_date": datetime.now(UTC).isoformat(), + } + ) + if len(normalized) >= max_sources: + break + + return normalized + + async def _gather_iptv_org_payloads( + self, + client: httpx.AsyncClient, + channels_url: str, + streams_url: str, + logos_url: str, + ) -> tuple[Any, Any, Any]: + headers = dict(self.DEFAULT_HEADERS) + channels_payload, streams_payload, logos_payload = await asyncio.gather( + client.get(channels_url, headers=headers), + client.get(streams_url, headers=headers), + client.get(logos_url, headers=headers), + ) + channels_payload.raise_for_status() + streams_payload.raise_for_status() + logos_payload.raise_for_status() + return channels_payload.json(), streams_payload.json(), logos_payload.json() + + def _pick_iptv_org_stream(self, streams: list[dict[str, Any]]) -> dict[str, Any] | None: + if not streams: + return None + + def score(stream: dict[str, Any]) -> tuple[int, int]: + url = self._clean_url(stream.get("url")) + quality = self._clean_text(stream.get("quality")).lower() + quality_score = 0 + if quality.endswith("p"): + try: + quality_score = int(quality[:-1]) + except ValueError: + quality_score = 0 + stream_score = 1000 if url.endswith(".m3u8") else 0 + return stream_score, quality_score + + sorted_streams = sorted(streams, key=score, reverse=True) + return sorted_streams[0] + + def parse_response(self, response: Any, *, response_path: str | None = None) -> list[dict[str, Any]]: + candidates = self._extract_candidates(response, response_path) normalized: list[dict[str, Any]] = [] for index, item in enumerate(candidates): if not isinstance(item, dict): continue - stream_id = item.get("id") or item.get("source_id") or item.get("slug") or f"news-live-{index + 1}" - name = str(item.get("name") or item.get("title") or f"News Live {index + 1}").strip() + stream_id = ( + item.get("id") + or item.get("source_id") + or item.get("slug") + or item.get("channel_id") + or item.get("code") + or f"news-live-{index + 1}" + ) + name = self._clean_text( + item.get("name") + or item.get("title") + or item.get("channel") + or item.get("display_name") + or f"News Live {index + 1}" + ) if not name: continue + source_type = self._infer_source_type(item) + stream_url = self._clean_url( + item.get("stream_url") + or item.get("stream") + or item.get("playback_url") + or item.get("hls_url") + or item.get("m3u8_url") + ) + embed_url = self._clean_url( + item.get("embed_url") + or item.get("embed") + or item.get("page_url") + or (item.get("url") if source_type == "iframe" else "") + ) + homepage_url = self._clean_url( + item.get("homepage_url") + or item.get("source_url") + or item.get("website") + or item.get("url") + ) metadata = { - "provider": item.get("provider") or item.get("publisher") or "Collector", - "region": item.get("region") or item.get("country") or "Global", - "language": item.get("language") or "und", - "source_type": item.get("source_type") or "iframe", - "embed_url": item.get("embed_url") or item.get("url") or "", - "stream_url": item.get("stream_url") or "", - "homepage_url": item.get("homepage_url") or item.get("source_url") or "", - "poster_url": item.get("poster_url") or "", + "provider": self._clean_text(item.get("provider") or item.get("publisher") or item.get("network")) or "Collector", + "region": self._clean_text(item.get("region") or item.get("country") or item.get("market")) or "Global", + "language": self._clean_text(item.get("language") or item.get("lang") or item.get("locale")) or "und", + "source_type": source_type, + "embed_url": embed_url, + "stream_url": stream_url, + "homepage_url": homepage_url, + "poster_url": self._clean_url(item.get("poster_url") or item.get("thumbnail_url") or item.get("logo_url")), + "youtube_video_id": self._clean_text( + item.get("youtube_video_id") + or item.get("video_id") + or item.get("youtubeVideoId") + ), + "youtube_channel": self._clean_text( + item.get("youtube_channel") + or item.get("channel_handle") + or item.get("youtubeChannel") + ), "sort_order": item.get("sort_order", 200 + index), - "notes": item.get("notes") or item.get("description") or "", - "is_enabled": item.get("is_enabled", True), + "notes": self._clean_text(item.get("notes") or item.get("description") or item.get("summary")), + "is_enabled": self._parse_enabled(item), } normalized.append( @@ -72,7 +563,7 @@ class NewsLiveStreamsCollector(BaseCollector): "name": name, "description": metadata["notes"], "metadata": metadata, - "reference_date": item.get("reference_date", datetime.now(UTC).isoformat()), + "reference_date": item.get("reference_date") or datetime.now(UTC).isoformat(), } ) diff --git a/backend/app/services/tv_streams.py b/backend/app/services/tv_streams.py index 7e28b3e1..a6d83d91 100644 --- a/backend/app/services/tv_streams.py +++ b/backend/app/services/tv_streams.py @@ -17,7 +17,7 @@ TV_LIVE_SOURCE_COLLECTOR = "news_live_streams" TV_LIVE_SOURCE_DATA_TYPE = "news_live_stream" DEFAULT_TV_SETTINGS = { - "default_source_id": DEFAULT_TV_SOURCE_ID, + "default_source_id": DEFAULT_TV_SOURCE_ID, "auto_fallback": True, "sources": [ { @@ -362,7 +362,7 @@ def _build_collected_tv_source(record: CollectedData, index: int) -> dict[str, A "sort_order": metadata.get("sort_order", 200 + index), "collector_source": record.source, "notes": record.description or metadata.get("notes") or "", - "updated_at": to_iso8601_utc(record.updated_at or record.reference_date or datetime.now(UTC)), + "updated_at": to_iso8601_utc(record.collected_at or record.reference_date or datetime.now(UTC)), }, index=index, ) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index becbda1b..cdea90d8 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -8,6 +8,23 @@ This project follows the repository versioning rule: - `improvement` -> `+0.0.1`（bugfix + 小功能混合） - `bugfix` -> `+0.0.1` +## [0.33.0] — 2026-04-22 + +### ✨ Highlights +- `news_live_streams` 采集器默认接入 `iptv-org` 频道目录，并将采集结果稳定并入 Earth TV 直播源列表 +- 数据源页支持直接编辑内置数据源 override，并为内置源提供一键恢复默认配置入口 + +### 🔧 Improvements +- `News Live Streams` 现在作为可直接触发的内置默认数据源提供，无需先手工补 override 才能采集 +- TV 播放源菜单会直接区分 `[内置]` 和 `[采集]` 来源，频道来源信息也会同步展示 +- 新增 [earth-news-source-configuration-and-collector-plan.md](/home/ray/dev/linkong/planet/docs/plans/earth-news-source-configuration-and-collector-plan.md)，正式规划 Earth 态势新闻源配置化与后续采集器化路线 + +### 🐛 Fixes +- 修复 `news_live_streams` 采集完成后 `/api/v1/tv/streams` 因读取不存在的 `updated_at` 字段而导致默认频道全部消失的问题 +- 修复内置数据源操作列按钮显示不全，以及编辑抽屉中多个 `Collapse` 紧贴的问题 + +--- + ## [0.32.0] — 2026-04-22 ### ✨ Highlights diff --git a/docs/plans/README.md b/docs/plans/README.md index fd8b1424..2062a1c2 100644 --- a/docs/plans/README.md +++ b/docs/plans/README.md @@ -20,6 +20,7 @@ - [earth-predicted-orbit-plan.md](/home/ray/dev/linkong/planet/docs/plans/earth-predicted-orbit-plan.md) - [earth-webgl-instancing-satellites-plan.md](/home/ray/dev/linkong/planet/docs/plans/earth-webgl-instancing-satellites-plan.md) - [earth-real-terrain-plan.md](/home/ray/dev/linkong/planet/docs/plans/earth-real-terrain-plan.md) +- [earth-news-source-configuration-and-collector-plan.md](/home/ray/dev/linkong/planet/docs/plans/earth-news-source-configuration-and-collector-plan.md) - [frontend-ai-playground-development-plan.md](/home/ray/dev/linkong/planet/docs/plans/frontend-ai-playground-development-plan.md) - [ue5-mvp-fused-plan.md](/home/ray/dev/linkong/planet/docs/plans/ue5-mvp-fused-plan.md) diff --git a/docs/plans/earth-news-source-configuration-and-collector-plan.md b/docs/plans/earth-news-source-configuration-and-collector-plan.md new file mode 100644 index 00000000..7c3ab44d --- /dev/null +++ b/docs/plans/earth-news-source-configuration-and-collector-plan.md @@ -0,0 +1,156 @@ +# Earth News Source Configuration And Collector Plan + +## Why + +当前 Earth 的“态势新闻”由 [earth_news.py](/home/ray/dev/linkong/planet/backend/app/services/earth_news.py) 直接在请求时抓取 RSS / Google News feed，再按当前地球视角中心区域聚合返回。 + +这条链已经可用，但存在两个明显限制： + +- 新闻源写死在代码里，不能像 TV 直播源一样从后台维护 +- 新闻并未进入统一采集体系，没有采集状态、失败监控、历史数据和后续 AI 复用能力 + +因此这块更合理的路线不是一步到位重写，而是分阶段推进： + +1. 先做“新闻源配置化” +2. 再做“新闻采集器化” + +## Current State + +当前实现分布在： + +- 新闻接口 + - [news.py](/home/ray/dev/linkong/planet/backend/app/api/v1/news.py) +- 实时聚合逻辑 + - [earth_news.py](/home/ray/dev/linkong/planet/backend/app/services/earth_news.py) +- 前端消费 + - [news.js](/home/ray/dev/linkong/planet/frontend/public/earth/js/news.js) + +当前新闻源包含： + +- `BBC World` RSS +- `DW Top Stories` RSS +- 按区域关键词拼出来的 `Google News RSS` + - `Global` + - `Americas` + - `Europe` + - `Middle East / Africa` + - `Asia Pacific` + +当前不是采集器，也不落库，只做内存缓存。 + +## Phase 1: Source Configuration + +### Goal + +把 `NEWS_FEED_SOURCES` 从硬编码列表升级成可配置新闻源目录，但继续保留当前“实时聚合”的工作方式。 + +### Scope + +- 为 Earth news 建立独立配置结构 +- 支持后台维护 feed 源 +- 支持启用/禁用、优先级、区域、源类型 +- 保持现有 `/api/v1/news/earth-feed` 输出协议不变 + +### Proposed Shape + +建议配置字段至少包括： + +- `id` +- `name` +- `region` +- `feed_url` +- `homepage_url` +- `source_type` +- `priority` +- `is_enabled` +- 可选 `query_profile` +- 可选 `language` +- 可选 `notes` + +### Suggested Storage + +优先走系统设置或单独的 news source settings payload，而不是先建复杂新表。 + +推荐原因： + +- 改动小 +- 易上线 +- 和当前 TV settings 维护体验更接近 +- 先解决“写死在代码里”的问题 + +### Non-goals + +这一阶段不做： + +- 新闻入库 +- 新闻历史回看 +- 新闻采集任务监控 +- 新闻去重流水线 + +## Phase 2: News Collectorization + +### Goal + +把“态势新闻”升级为真正的采集器链路，使其进入采集系统和数据层。 + +### Scope + +- 新增专用 news collector +- 按配置源定时采集 RSS / feed +- 做标题/链接级去重 +- 建立统一新闻记录模型 +- 为 Earth、控制台、AI 研判复用同一份新闻数据 + +### Benefits + +- 有采集状态 +- 有失败监控 +- 有历史缓存 +- 可以做时间轴 / 区域新闻基线 +- 可以作为 AI 引用证据 + +### Required Design Work + +需要提前明确： + +- 新闻数据模型 +- 去重策略 +- 过期清理策略 +- 区域映射策略 +- 聚合排序策略 +- 新闻与 Earth 当前视角/区域的关联方式 + +### Candidate Output Model + +至少应包含： + +- `source_id` +- `headline` +- `summary` +- `url` +- `publisher` +- `region` +- `published_at` +- `language` +- `tags` +- `raw_feed_source` +- `reference_date` + +## Recommended Order + +推荐执行顺序： + +1. 先完成 Phase 1 配置化 +2. 保持 Earth 继续实时聚合，但改为读取配置源 +3. 等新闻源稳定后，再设计 Phase 2 的 collector / storage / dedupe + +## Decision + +当前结论： + +- TV 直播源：优先采集器化 +- 态势新闻：优先配置化，再采集器化 + +## Source Note + +This plan is newly created for the Planet repo to separate the short-term "configurable source directory" work from the longer-term "collectorized news pipeline" work. diff --git a/docs/technical/earth-news-live-streams-collector-format.md b/docs/technical/earth-news-live-streams-collector-format.md index 5233b97e..2883e4d7 100644 --- a/docs/technical/earth-news-live-streams-collector-format.md +++ b/docs/technical/earth-news-live-streams-collector-format.md @@ -95,3 +95,93 @@ - 手工配置源 - `news_live_streams` 采集器采集源 - 当前默认兜底源为 `CCTV-4 中文国际` +- `news_live_streams` 在未配置 override 时，默认使用 `iptv-org`： + - `channels.json` + - `streams.json` + - `logos.json` + 并自动筛出新闻类频道目录 + +## 采集器配置方式 + +`news_live_streams` 不需要单独新页面，直接复用现有数据源配置： + +- `endpoint` + - 频道目录 JSON API 地址 +- `auth_type` + - `none` / `bearer` / `api_key` / `basic` +- `headers` + - 额外请求头 +- `config` + - 采集器请求与解析行为 + +### 支持的 `config` 字段 + +```json +{ + "timeout": 30, + "method": "GET", + "params": { + "region": "global" + }, + "body_type": "json", + "body": { + "include_disabled": false + }, + "response_path": "payload.channels" +} +``` + +- `timeout` + - 请求超时秒数 +- `method` + - `GET` 或 `POST` +- `params` + - 查询参数对象 +- `body_type` + - `json` 或 `form` +- `body` + - 配合 `POST` 使用的请求体 +- `json_body` + - 显式 JSON 请求体，优先级高于 `body` +- `form_body` + - 显式表单请求体，优先级高于 `body` +- `response_path` + - 返回 JSON 中频道数组所在路径，支持点路径，例如： + - `payload.channels` + - `data.items` + - `result.streams` + +### 认证补充 + +- `bearer` + - 使用 `Authorization: Bearer ` +- `api_key` + - 默认作为请求头发送 + - 如果 `auth_config.in = "query"`，则作为 query param 发送 +- `basic` + - 使用 HTTP Basic Authorization + +## 兼容的响应结构 + +采集器会优先读取： + +- 顶层数组 +- 或这些常见字段下的数组： + - `sources` + - `streams` + - `channels` + - `items` + - `results` + - `data` + +同时会兼容这些字段别名： + +- `id` / `source_id` / `slug` / `channel_id` / `code` +- `name` / `title` / `channel` / `display_name` +- `provider` / `publisher` / `network` +- `stream_url` / `stream` / `playback_url` / `hls_url` / `m3u8_url` +- `embed_url` / `embed` / `page_url` +- `homepage_url` / `source_url` / `website` +- `language` / `lang` / `locale` +- `youtube_video_id` / `video_id` +- `youtube_channel` / `channel_handle` diff --git a/docs/version-history.md b/docs/version-history.md index 1916da29..33b21828 100644 --- a/docs/version-history.md +++ b/docs/version-history.md @@ -16,12 +16,13 @@ ## Current Version - `main` 当前主线历史推导到：`0.16.5` -- `dev` 当前开发分支历史推导到：`0.32.0` +- `dev` 当前开发分支历史推导到：`0.33.0` ## Timeline | Version | Type | Branch | Commit | Summary | | --- | --- | --- | --- | --- | +| `0.33.0` | feature | `dev` | `pending` | `news_live_streams` 默认接入 iptv-org 频道目录，内置数据源支持直接编辑 override，并修复 TV 合并采集源后默认频道消失的问题 | | `0.32.0` | feature | `dev` | `pending` | Earth 设置新增默认地球大小真源，并继续收口卫星焦点层次、toolbar/scrollbar 性能与 HUD 设置面板细节 | | `0.31.3` | bugfix | `dev` | `pending` | 收口 Earth 图层注册表与启动任务框架，修复旋转/巡航切换、卫星地形遮挡与日夜关闭照明回归 | | `0.31.2` | bugfix | `dev` | `pending` | 将 Earth 巡航模式拆成通用 sequencer、通用连线和 BGP 巡航适配层，并修复空白点击推进与连线动画回归 | diff --git a/frontend/package.json b/frontend/package.json index 67ad99ce..0d2d4392 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "planet-frontend", - "version": "0.32.0", + "version": "0.33.0", "private": true, "packageManager": "bun@1", "dependencies": { diff --git a/frontend/public/earth/js/tv.js b/frontend/public/earth/js/tv.js index d19b68a0..53b75767 100644 --- a/frontend/public/earth/js/tv.js +++ b/frontend/public/earth/js/tv.js @@ -877,11 +877,12 @@ function renderSourceOptions() { const fragment = document.createDocumentFragment(); sources.forEach((source) => { + const sourceOriginLabel = source.collector_source ? "[采集]" : "[内置]"; const defaultMark = source.id === tvPayload?.default_source_id ? " · 默认" : ""; const failMark = failedSourceIds.has(source.id) ? " ⚠" : ""; const option = document.createElement("option"); option.value = source.id; - option.textContent = `${source.name}${defaultMark}${failMark}`; + option.textContent = `${sourceOriginLabel} ${source.name}${defaultMark}${failMark}`; fragment.appendChild(option); }); @@ -913,8 +914,12 @@ function renderSource(source) { const latestLabel = latestUpdatedAt ? `最近同步 ${new Date(latestUpdatedAt).toLocaleString("zh-CN", { hour12: false })}` : "尚未同步"; - const collectorLabel = source?.collector_source ? ` · 采集器 ${source.collector_source}` : ""; - catalog.textContent = `共 ${sourceCount} 个频道 · ${latestLabel}${collectorLabel}`; + const sourceOriginLabel = source?.collector_source + ? `采集源 ${source.collector_source}` + : source + ? "内置源" + : ""; + catalog.textContent = `共 ${sourceCount} 个频道 · ${latestLabel}${sourceOriginLabel ? ` · ${sourceOriginLabel}` : ""}`; } if (notes) { notes.textContent = source?.notes || "支持后台配置默认源与采集器补充源。"; diff --git a/frontend/src/index.css b/frontend/src/index.css index f4083b44..63994175 100644 --- a/frontend/src/index.css +++ b/frontend/src/index.css @@ -1562,6 +1562,14 @@ body { border-radius: 12px; } +.data-source-drawer-collapse { + margin-bottom: 12px; +} + +.data-source-drawer-collapse:last-of-type { + margin-bottom: 0; +} + .stat-card { background: white; padding: 24px; diff --git a/frontend/src/pages/DataSources/DataSources.tsx b/frontend/src/pages/DataSources/DataSources.tsx index 154c2a45..4b1812c1 100644 --- a/frontend/src/pages/DataSources/DataSources.tsx +++ b/frontend/src/pages/DataSources/DataSources.tsx @@ -19,6 +19,7 @@ import { useWebSocket } from '../../hooks/useWebSocket' interface BuiltInDataSource { id: number + source: string name: string module: string priority: string @@ -180,6 +181,19 @@ interface CustomDataSource { updated_at: string | null } +interface EditableDataSourceConfig { + id: number + name: string + description: string | null + source_type: string + endpoint: string + auth_type: string + auth_config: Record + headers: Record + config: Record + is_active?: boolean +} + interface ViewDataSource { id: number name: string @@ -205,6 +219,7 @@ function DataSources() { const [drawerVisible, setDrawerVisible] = useState(false) const [viewDrawerVisible, setViewDrawerVisible] = useState(false) const [editingConfig, setEditingConfig] = useState(null) + const [builtinEditingSource, setBuiltinEditingSource] = useState(null) const [viewingSource, setViewingSource] = useState(null) const [recordCount, setRecordCount] = useState(0) const [testing, setTesting] = useState(false) @@ -219,6 +234,81 @@ function DataSources() { const [customActionsCollapsed, customContainerRef] = useCollapsedActions() const [form] = Form.useForm() + const headersMapToList = useCallback((headers?: Record | null) => { + return Object.entries(headers || {}) + .filter(([key, value]) => key && value !== undefined && value !== null && String(value).trim() !== '') + .map(([key, value]) => ({ key, value })) + }, []) + + const headersListToMap = useCallback((headers?: Array<{ key?: string; value?: string }> | Record) => { + if (!headers) return {} + if (!Array.isArray(headers)) return headers + + return headers.reduce>((acc, item) => { + const key = item?.key?.trim() + const value = item?.value?.trim() + if (!key || value === undefined) return acc + acc[key] = value + return acc + }, {}) + }, []) + + const applyConfigToForm = useCallback((config?: Partial | null) => { + form.setFieldsValue({ + name: config?.name || '', + description: config?.description || '', + source_type: config?.source_type || 'http', + endpoint: config?.endpoint || '', + auth_type: config?.auth_type || 'none', + auth_config: config?.auth_config || {}, + headers: headersMapToList(config?.headers || {}), + config: config?.config || { timeout: 30, retry: 3 }, + }) + }, [form, headersMapToList]) + + const loadConfigDetail = useCallback(async (configId: number) => { + const res = await axios.get(`/api/v1/datasources/configs/${configId}`) + return res.data + }, []) + + const createDefaultConfigDraft = useCallback((overrides?: Partial) => ({ + source_type: 'http', + auth_type: 'none', + headers: {}, + config: { timeout: 30, retry: 3 }, + ...overrides, + }), []) + + const getBuiltinOverrideDescription = useCallback( + (source?: Pick | null) => + source ? `Built-in datasource override for ${source.name}` : undefined, + [], + ) + + const createFormPayload = useCallback((values: any) => ({ + ...values, + name: builtinEditingSource ? builtinEditingSource.source : values.name, + description: + values.description || + getBuiltinOverrideDescription(builtinEditingSource), + source_type: builtinEditingSource ? 'http' : values.source_type, + headers: headersListToMap(values.headers), + }), [builtinEditingSource, getBuiltinOverrideDescription, headersListToMap]) + + const closeDrawerAfterLoadError = useCallback(( + errorMessage: string, + options?: { clearBuiltin?: boolean; clearEditingConfig?: boolean }, + ) => { + messageApi.error(errorMessage) + setDrawerVisible(false) + if (options?.clearBuiltin) { + setBuiltinEditingSource(null) + } + if (options?.clearEditingConfig) { + setEditingConfig(null) + } + }, [messageApi]) + const fetchData = useCallback(async () => { setLoading(true) try { @@ -711,9 +801,11 @@ function DataSources() { const handleViewSource = async (source: BuiltInDataSource) => { try { - const [res, statsRes] = await Promise.all([ + const existingOverride = customSources.find((item) => item.name === source.source) + const [res, statsRes, overrideDetail] = await Promise.all([ axios.get(`/api/v1/datasources/${source.id}`), - axios.get(`/api/v1/datasources/${source.id}/stats`) + axios.get(`/api/v1/datasources/${source.id}/stats`), + existingOverride ? loadConfigDetail(existingOverride.id) : Promise.resolve(null), ]) const data = res.data setViewingSource({ @@ -721,10 +813,10 @@ function DataSources() { name: data.name, description: null, source_type: data.collector_class, - endpoint: data.endpoint || '', - auth_type: 'none', - headers: {}, - config: {}, + endpoint: overrideDetail?.endpoint || data.endpoint || '', + auth_type: overrideDetail?.auth_type || 'none', + headers: overrideDetail?.headers || {}, + config: overrideDetail?.config || {}, collector_class: data.collector_class, module: data.module, priority: data.priority, @@ -753,7 +845,8 @@ function DataSources() { const values = await form.validateFields() setTesting(true) setTestResult(null) - const res = await axios.post('/api/v1/datasources/configs/test', values) + const payload = createFormPayload(values) + const res = await axios.post('/api/v1/datasources/configs/test', payload) setTestResult(res.data) if (res.data.success) { messageApi.success('连接测试成功') @@ -771,16 +864,18 @@ function DataSources() { const handleSave = async () => { try { const values = await form.validateFields() + const payload = createFormPayload(values) if (editingConfig) { - await axios.put(`/api/v1/datasources/configs/${editingConfig.id}`, values) + await axios.put(`/api/v1/datasources/configs/${editingConfig.id}`, payload) messageApi.success('配置已更新') } else { - await axios.post('/api/v1/datasources/configs', values) + await axios.post('/api/v1/datasources/configs', payload) messageApi.success('配置已创建') } setDrawerVisible(false) form.resetFields() setEditingConfig(null) + setBuiltinEditingSource(null) setTestResult(null) fetchData() } catch (error: unknown) { @@ -800,6 +895,23 @@ function DataSources() { } } + const handleResetBuiltinOverride = async () => { + if (!builtinEditingSource || !editingConfig) return + try { + await axios.delete(`/api/v1/datasources/configs/${editingConfig.id}`) + messageApi.success(`已恢复 ${builtinEditingSource.name} 的默认配置`) + setDrawerVisible(false) + form.resetFields() + setEditingConfig(null) + setBuiltinEditingSource(null) + setTestResult(null) + fetchData() + } catch (error: unknown) { + const err = error as { response?: { data?: { detail?: string } } } + messageApi.error(err.response?.data?.detail || '恢复默认失败') + } + } + const handleToggleCustom = async (id: number, current: boolean) => { try { await axios.put(`/api/v1/datasources/configs/${id}`, { is_active: !current }) @@ -811,24 +923,53 @@ function DataSources() { } } - const openDrawer = (config?: CustomDataSource) => { + const openDrawer = async (config?: CustomDataSource) => { + setBuiltinEditingSource(null) setEditingConfig(config || null) - if (config) { - form.setFieldsValue({ - ...config, - auth_config: {}, - }) - } else { - form.resetFields() - form.setFieldsValue({ - source_type: 'http', - auth_type: 'none', - config: { timeout: 30, retry: 3 }, - headers: {}, - }) - } - setDrawerVisible(true) setTestResult(null) + setDrawerVisible(true) + + if (config) { + try { + const detail = await loadConfigDetail(config.id) + applyConfigToForm(detail) + } catch { + closeDrawerAfterLoadError('获取配置详情失败', { clearEditingConfig: true }) + } + return + } + + form.resetFields() + applyConfigToForm(createDefaultConfigDraft()) + } + + const openBuiltinConfigDrawer = async (source: BuiltInDataSource) => { + setBuiltinEditingSource(source) + setTestResult(null) + setDrawerVisible(true) + + const existingOverride = customSources.find((item) => item.name === source.source) + setEditingConfig(existingOverride || null) + + if (existingOverride) { + try { + const detail = await loadConfigDetail(existingOverride.id) + applyConfigToForm(detail) + } catch { + closeDrawerAfterLoadError('获取内置数据源配置失败', { + clearBuiltin: true, + clearEditingConfig: true, + }) + } + return + } + + form.resetFields() + applyConfigToForm(createDefaultConfigDraft({ + name: source.source, + description: getBuiltinOverrideDescription(source), + endpoint: source.endpoint || '', + })) } const handleCopyLink = async (value: string, successText: string) => { @@ -945,12 +1086,18 @@ function DataSources() { title: '操作', key: 'action', fixed: 'right' as const, - width: builtinActionsCollapsed ? 40 : 164, + width: builtinActionsCollapsed ? 40 : 228, onCell: () => actionCellProps, render: (_: unknown, record: BuiltInDataSource) => ( , + onClick: () => { void openBuiltinConfigDrawer(record) }, + }, { key: 'trigger', label: '触发', @@ -967,6 +1114,14 @@ function DataSources() { }, ]} > + + @@ -1215,24 +1370,40 @@ function DataSources() { { setDrawerVisible(false) form.resetFields() setEditingConfig(null) + setBuiltinEditingSource(null) setTestResult(null) }} footer={
- + + {builtinEditingSource && editingConfig ? ( + + + + ) : null} + +