37 lines
949 B
Python
37 lines
949 B
Python
"""抓取远程直播源内容。"""
|
|
|
|
import logging
|
|
|
|
import requests
|
|
|
|
from .sources import REQUEST_TIMEOUT, USER_AGENT
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def fetch(url: str) -> str | None:
|
|
"""下载单个源的原始文本，失败返回 None。"""
|
|
try:
|
|
resp = requests.get(
|
|
url,
|
|
timeout=REQUEST_TIMEOUT,
|
|
headers={"User-Agent": USER_AGENT},
|
|
)
|
|
resp.raise_for_status()
|
|
resp.encoding = resp.apparent_encoding or "utf-8"
|
|
log.info("fetched %s (%d bytes)", url, len(resp.text))
|
|
return resp.text
|
|
except Exception as exc: # noqa: BLE001
|
|
log.warning("fetch failed %s: %s", url, exc)
|
|
return None
|
|
|
|
|
|
def fetch_all(urls: list[str]) -> list[str]:
|
|
"""批量抓取，返回成功的内容列表。"""
|
|
results: list[str] = []
|
|
for url in urls:
|
|
text = fetch(url)
|
|
if text:
|
|
results.append(text)
|
|
return results
|