from __future__ import annotations from datetime import UTC, datetime from typing import Any from urllib.parse import urlparse from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.time import to_iso8601_utc from app.models.collected_data import CollectedData from app.models.system_setting import SystemSetting DEFAULT_TV_SOURCE_ID = "cgtn-en" TV_SETTINGS_CATEGORY = "tv" 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, "auto_fallback": True, "sources": [ { "id": DEFAULT_TV_SOURCE_ID, "name": "CCTV-4 中文国际", "provider": "CCTV", "region": "China", "language": "zh-CN", "source_type": "hls", "embed_url": "https://tv.cctv.com/live/cctv4/", "stream_url": "https://ldocctvwbcdtxy.liveplay.myqcloud.com/ldocctvwbcd/cdrmldcctv4_1_td.m3u8", "homepage_url": "https://tv.cctv.com/live/cctv4/", "poster_url": "", "is_enabled": True, "is_fallback": True, "sort_order": 10, "collector_source": None, "notes": "默认兜底新闻直播源。优先尝试 CCTV-4 官方 HLS 播放流，若直播放失败则回退到央视官网直播页。", }, { "id": "reuters-tv", "name": "Reuters TV", "provider": "Reuters", "region": "Global", "language": "en", "source_type": "hls", "embed_url": "", "stream_url": "https://reuters-reutersnow-1-eu.rakuten.wurl.tv/playlist.m3u8", "homepage_url": "https://www.reuters.com/video/live/", "poster_url": "", "is_enabled": True, "is_fallback": False, "sort_order": 20, "collector_source": None, "notes": "参考 worldmonitor 的默认新闻频道清单，优先作为全球英文新闻直播放源。", }, { "id": "cgtn-en", "name": "CGTN English", "provider": "CGTN", "region": "Global", "language": "en", "source_type": "youtube", "embed_url": "https://www.youtube.com/watch?v=BOy2xDU1LC8", "stream_url": "https://news.cgtn.com/resource/live/english/cgtn-news.m3u8", "youtube_video_id": "BOy2xDU1LC8", "homepage_url": "https://news.cgtn.com/", "poster_url": "", "is_enabled": True, "is_fallback": False, "sort_order": 30, "collector_source": None, "notes": "优先使用官方 YouTube 直播源，保留 HLS 直播放流作为候选信息。", }, { "id": "cgtn-es", "name": "CGTN Espanol", "provider": "CGTN", "region": "Latin America", "language": "es", "source_type": "hls", "embed_url": "", "stream_url": "https://news.cgtn.com/resource/live/espanol/cgtn-e.m3u8", "homepage_url": "https://news.cgtn.com/", "poster_url": "", "is_enabled": True, "is_fallback": False, "sort_order": 40, "collector_source": None, "notes": "西语国际新闻频道，覆盖拉美方向态势。", }, { "id": "dw-espanol", "name": "DW Espanol", "provider": "Deutsche Welle", "region": "Europe", "language": "es", "source_type": "hls", "embed_url": "", "stream_url": "https://dwamdstream104.akamaized.net/hls/live/2015530/dwstream104/stream04/streamPlaylist.m3u8", "homepage_url": "https://www.dw.com/es/", "poster_url": "", "is_enabled": True, "is_fallback": False, "sort_order": 50, "collector_source": None, "notes": "来自 worldmonitor 可选频道清单的直播放源。", }, { "id": "dw-arabic", "name": "DW Arabic", "provider": "Deutsche Welle", "region": "Middle East", "language": "ar", "source_type": "hls", "embed_url": "", "stream_url": "https://dwamdstream103.akamaized.net/hls/live/2015526/dwstream103/index.m3u8", "homepage_url": "https://www.dw.com/ar/", "poster_url": "", "is_enabled": True, "is_fallback": False, "sort_order": 60, "collector_source": None, "notes": "阿拉伯语新闻流，适合作为中东方向新闻补充源。", }, { "id": "aljazeera-mubasher", "name": "Al Jazeera Mubasher", "provider": "Al Jazeera", "region": "Middle East", "language": "ar", "source_type": "hls", "embed_url": "", "stream_url": "https://live-hls-web-ajm.getaj.net/AJM/index.m3u8", "homepage_url": "https://www.aljazeera.net/live", "poster_url": "", "is_enabled": True, "is_fallback": False, "sort_order": 70, "collector_source": None, "notes": "中东实时新闻流，来自 worldmonitor HLS 频道目录。", }, { "id": "arirang-news", "name": "Arirang News", "provider": "Arirang", "region": "Korea", "language": "en", "source_type": "hls", "embed_url": "", "stream_url": "https://amdlive-ch01-ctnd-com.akamaized.net/arirang_1ch/smil:arirang_1ch.smil/playlist.m3u8", "homepage_url": "https://www.arirang.com/", "poster_url": "", "is_enabled": True, "is_fallback": False, "sort_order": 80, "collector_source": None, "notes": "东北亚英语新闻源，适合补充韩半岛与东亚视角。", }, { "id": "abp-news", "name": "ABP News", "provider": "ABP", "region": "India", "language": "hi", "source_type": "hls", "embed_url": "", "stream_url": "https://abplivetv.pc.cdn.bitgravity.com/httppush/abp_livetv/abp_abpnews/master.m3u8", "homepage_url": "https://news.abplive.com/live-tv", "poster_url": "", "is_enabled": True, "is_fallback": False, "sort_order": 90, "collector_source": None, "notes": "印度新闻直播放源，补充南亚区域视角。", }, { "id": "sabc-news", "name": "SABC News", "provider": "SABC", "region": "Africa", "language": "en", "source_type": "hls", "embed_url": "", "stream_url": "https://sabconetanw.cdn.mangomolo.com/news/smil:news.stream.smil/playlist.m3u8", "homepage_url": "https://www.sabcnews.com/sabcnews/", "poster_url": "", "is_enabled": True, "is_fallback": False, "sort_order": 100, "collector_source": None, "notes": "非洲英语新闻源，补充非洲区域新闻覆盖。", }, ], } def _clean_text(value: Any) -> str: if value is None: return "" return str(value).strip() def _clean_url(value: Any) -> str: text = _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 def _clean_bool(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"}: return True if lowered in {"0", "false", "no", "off"}: return False return bool(value) def _clean_int(value: Any, *, default: int) -> int: try: return int(value) except (TypeError, ValueError): return default def normalize_tv_source(source: dict[str, Any] | None, *, index: int = 0) -> dict[str, Any]: payload = dict(source or {}) source_id = _clean_text(payload.get("id")) or f"tv-source-{index + 1}" source_type = _clean_text(payload.get("source_type")).lower() youtube_video_id = _clean_text(payload.get("youtube_video_id")) youtube_channel = _clean_text(payload.get("youtube_channel")) if source_type not in {"iframe", "hls", "video", "external", "youtube"}: if youtube_video_id or youtube_channel: source_type = "youtube" else: source_type = "iframe" if _clean_text(payload.get("embed_url")) else "external" if source_type == "youtube" and not youtube_video_id and not youtube_channel: source_type = "iframe" if _clean_text(payload.get("embed_url")) else "external" return { "id": source_id, "name": _clean_text(payload.get("name")) or f"新闻直播源 {index + 1}", "provider": _clean_text(payload.get("provider")) or "Unknown", "region": _clean_text(payload.get("region")) or "Global", "language": _clean_text(payload.get("language")) or "und", "source_type": source_type, "embed_url": _clean_url(payload.get("embed_url")), "stream_url": _clean_url(payload.get("stream_url")), "homepage_url": _clean_url(payload.get("homepage_url")), "poster_url": _clean_url(payload.get("poster_url")), "youtube_video_id": youtube_video_id, "youtube_channel": youtube_channel, "is_enabled": _clean_bool(payload.get("is_enabled"), default=True), "is_fallback": _clean_bool(payload.get("is_fallback"), default=False), "sort_order": _clean_int(payload.get("sort_order"), default=(index + 1) * 10), "collector_source": payload.get("collector_source"), "notes": _clean_text(payload.get("notes")), "updated_at": _clean_text(payload.get("updated_at")), } def normalize_tv_settings(payload: dict[str, Any] | None) -> dict[str, Any]: merged = { "default_source_id": DEFAULT_TV_SETTINGS["default_source_id"], "auto_fallback": DEFAULT_TV_SETTINGS["auto_fallback"], "sources": [], } raw_sources = [] if isinstance(payload, dict): merged["default_source_id"] = ( _clean_text(payload.get("default_source_id")) or merged["default_source_id"] ) merged["auto_fallback"] = _clean_bool( payload.get("auto_fallback"), default=DEFAULT_TV_SETTINGS["auto_fallback"], ) if isinstance(payload.get("sources"), list): raw_sources = payload["sources"] if not raw_sources: raw_sources = DEFAULT_TV_SETTINGS["sources"] normalized_sources = [ normalize_tv_source(source, index=index) for index, source in enumerate(raw_sources) ] if not any(source["id"] == DEFAULT_TV_SOURCE_ID for source in normalized_sources): normalized_sources.append( normalize_tv_source(DEFAULT_TV_SETTINGS["sources"][0], index=len(normalized_sources)) ) default_source_exists = any( source["id"] == merged["default_source_id"] and source["is_enabled"] for source in normalized_sources ) if not default_source_exists: fallback_source = next( (source for source in normalized_sources if source["is_fallback"] and source["is_enabled"]), None, ) first_enabled_source = next( (source for source in normalized_sources if source["is_enabled"]), None, ) merged["default_source_id"] = ( fallback_source["id"] if fallback_source else first_enabled_source["id"] if first_enabled_source else DEFAULT_TV_SOURCE_ID ) merged["sources"] = sorted( normalized_sources, key=lambda item: (item["sort_order"], item["name"], item["id"]), ) return merged async def get_tv_settings_payload(db: AsyncSession) -> dict[str, Any]: result = await db.execute( select(SystemSetting).where(SystemSetting.category == TV_SETTINGS_CATEGORY) ) record = result.scalar_one_or_none() payload = record.payload if record else None return normalize_tv_settings(payload) def _build_collected_tv_source(record: CollectedData, index: int) -> dict[str, Any]: metadata = dict(record.extra_data or {}) return normalize_tv_source( { "id": metadata.get("id") or record.source_id or record.entity_key, "name": record.name or record.title or metadata.get("name") or f"采集直播源 {index + 1}", "provider": metadata.get("provider") or metadata.get("publisher") or "Collector", "region": metadata.get("region") or metadata.get("country") or "Global", "language": metadata.get("language") or "und", "source_type": metadata.get("source_type") or "iframe", "embed_url": metadata.get("embed_url") or metadata.get("url") or "", "stream_url": metadata.get("stream_url") or "", "homepage_url": metadata.get("homepage_url") or metadata.get("source_url") or "", "poster_url": metadata.get("poster_url") or "", "youtube_video_id": metadata.get("youtube_video_id") or metadata.get("video_id") or "", "youtube_channel": metadata.get("youtube_channel") or metadata.get("channel_handle") or "", "is_enabled": metadata.get("is_enabled", True), "is_fallback": False, "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)), }, index=index, ) async def get_collected_tv_sources(db: AsyncSession) -> list[dict[str, Any]]: result = await db.execute( select(CollectedData) .where(CollectedData.source == TV_LIVE_SOURCE_COLLECTOR) .where(CollectedData.data_type == TV_LIVE_SOURCE_DATA_TYPE) .where(CollectedData.is_current.is_(True)) .where(CollectedData.is_valid == 1) .order_by(CollectedData.reference_date.desc().nullslast(), CollectedData.id.desc()) ) rows = result.scalars().all() return [_build_collected_tv_source(record, index) for index, record in enumerate(rows)] def build_public_tv_payload( settings_payload: dict[str, Any], collected_sources: list[dict[str, Any]], ) -> dict[str, Any]: configured_sources = [ source for source in settings_payload["sources"] if source["is_enabled"] ] merged_by_id = {source["id"]: source for source in configured_sources} for source in collected_sources: if source["id"] in merged_by_id or not source["is_enabled"]: continue merged_by_id[source["id"]] = source available_sources = sorted( merged_by_id.values(), key=lambda item: (item["sort_order"], item["name"], item["id"]), ) default_source = next( ( source for source in available_sources if source["id"] == settings_payload["default_source_id"] ), None, ) fallback_source = next( (source for source in available_sources if source["is_fallback"]), None, ) resolved_source = default_source or fallback_source or (available_sources[0] if available_sources else None) latest_updated_at = max( (source.get("updated_at") or "" for source in available_sources), default="", ) return { "default_source_id": settings_payload["default_source_id"], "auto_fallback": settings_payload["auto_fallback"], "selected_source": resolved_source, "fallback_source": fallback_source, "sources": available_sources, "source_count": len(available_sources), "latest_updated_at": latest_updated_at or to_iso8601_utc(datetime.now(UTC)), "generated_at": to_iso8601_utc(datetime.now(UTC)), } async def get_public_tv_payload(db: AsyncSession) -> dict[str, Any]: settings_payload = await get_tv_settings_payload(db) collected_sources = await get_collected_tv_sources(db) return build_public_tv_payload(settings_payload, collected_sources) def _extract_allowed_tv_hosts(sources: list[dict[str, Any]]) -> set[str]: hosts: set[str] = set() for source in sources: for field in ("stream_url", "embed_url", "homepage_url", "youtube_channel"): value = _clean_url(source.get(field)) if not value: continue parsed = urlparse(value) if parsed.hostname: hosts.add(parsed.hostname.lower()) return hosts def is_allowed_tv_proxy_url(url: str, sources: list[dict[str, Any]]) -> bool: cleaned = _clean_url(url) if not cleaned: return False parsed = urlparse(cleaned) hostname = (parsed.hostname or "").lower() if not hostname: return False allowed_hosts = _extract_allowed_tv_hosts(sources) if hostname in allowed_hosts: return True return any(hostname.endswith(f".{allowed_host}") for allowed_host in allowed_hosts)