release: bump version to 0.33.0

This commit is contained in:
rayd1o
2026-04-22 05:28:54 +08:00
parent 3ae4acdff8
commit 0082cf3fbd
18 changed files with 1069 additions and 91 deletions

View File

@@ -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(),
}
)