Files
daily-robots/daily/news/fetch.py

747 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""抓取并整理国际 / 国内 AI 时讯 RSS。"""
from __future__ import annotations
import logging
import re
import time
import html
import xml.etree.ElementTree as ET
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, time as dt_time, timezone, timedelta
from email.utils import parsedate_to_datetime
from typing import Any
from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
from zoneinfo import ZoneInfo
import certifi
import httpx
from daily.config import env, env_int, news_summary_limit, wecom_news_desc_limit
from daily.news.feeds import NEWS_CATEGORIES, NewsCategory, NewsFeed
from daily.news.feeds_cn import CN_AI_TITLE_KEYWORDS, CN_NEWS_CATEGORIES
from daily.text_utils import trim_brief
logger = logging.getLogger(__name__)
USER_AGENT = "Mozilla/5.0 (compatible; skills-hot-daily/1.0; +https://skills.sh)"
BROWSER_USER_AGENT = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/131.0.0.0 Safari/537.36"
)
STRIP_HTML = re.compile(r"<[^>]+>")
WS = re.compile(r"\s+")
ATOM_NS = {"a": "http://www.w3.org/2005/Atom"}
RSS_NS = {"r": "http://purl.org/rss/1.0/modules/content/"}
def _enabled() -> bool:
raw = (env("DAILY_AI_NEWS") or "1").strip().lower()
return raw not in {"0", "false", "no", "off"}
def _cn_enabled() -> bool:
raw = (env("DAILY_CN_AI_NEWS") or "1").strip().lower()
return raw not in {"0", "false", "no", "off"}
def _hours_window() -> int:
return max(1, env_int("DAILY_AI_NEWS_HOURS", 24))
def _news_tz_name() -> str:
return (env("DAILY_AI_NEWS_TZ") or env("DAILY_SCHEDULE_TZ") or "Asia/Shanghai").strip()
def _floor_today_enabled() -> bool:
raw = env("DAILY_AI_NEWS_FLOOR_TODAY")
if raw is None:
return True
return raw.strip().lower() not in {"0", "false", "no", "off"}
def _cutoff_datetime(*, floor_today: bool) -> datetime:
"""滚动 N 小时窗口;国际新闻可叠加「不早于今日 0 点(本地时区)」。"""
now = _now_utc()
rolling = now - timedelta(hours=_hours_window())
if not floor_today:
return rolling
tz = ZoneInfo(_news_tz_name())
local = now.astimezone(tz)
start_today = local.replace(hour=0, minute=0, second=0, microsecond=0).astimezone(timezone.utc)
return max(rolling, start_today)
def _per_feed_limit() -> int:
want = max(_wecom_limit(), _wecom_cn_limit())
return max(want // 2, env_int("DAILY_AI_NEWS_PER_FEED", 5))
def _per_category_limit() -> int:
want = max(_wecom_limit(), _wecom_cn_limit())
return max(want, env_int("DAILY_AI_NEWS_PER_CATEGORY", 10))
def _wecom_limit() -> int:
return max(1, env_int("DAILY_WECOM_AI_NEWS", 10))
def _wecom_cn_limit() -> int:
return max(1, env_int("DAILY_WECOM_CN_AI_NEWS", 10))
def _matches_cn_ai_title(title: str) -> bool:
text = title.strip()
if not text:
return False
lower = text.lower()
for keyword in CN_AI_TITLE_KEYWORDS:
if keyword.lower() in lower:
return True
return False
def _filter_ai_entries(entries: list[dict[str, Any]], *, ai_filter: bool) -> list[dict[str, Any]]:
if not ai_filter:
return entries
return [item for item in entries if _matches_cn_ai_title(item.get("title", ""))]
def _now_utc() -> datetime:
return datetime.now(timezone.utc)
def _parse_datetime(value: str | None) -> datetime | None:
if not value:
return None
text = value.strip()
if not text:
return None
try:
dt = parsedate_to_datetime(text)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)
except (TypeError, ValueError, OverflowError):
pass
for fmt in (
"%Y-%m-%dT%H:%M:%SZ",
"%Y-%m-%dT%H:%M:%S%z",
):
try:
dt = datetime.strptime(text[: len(fmt.replace("%z", "+0000"))], fmt.replace("%z", ""))
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)
except ValueError:
continue
if re.match(r"^\d{4}-\d{2}-\d{2}$", text):
try:
tz = ZoneInfo(_news_tz_name())
day = datetime.strptime(text, "%Y-%m-%d").date()
# 仅日期时按本地中午估算,避免 UTC 0 点误判为「前一天」
dt = datetime.combine(day, dt_time(12, 0), tzinfo=tz)
return dt.astimezone(timezone.utc)
except ValueError:
pass
return None
def _clean_text(text: str | None, limit: int = 200) -> str:
if not text:
return ""
plain = _strip_summary_plain(text)
if limit <= 0 or len(plain) <= limit:
return plain
return trim_brief(plain, limit)
def _strip_summary_plain(text: str | None) -> str:
if not text:
return ""
plain = STRIP_HTML.sub(" ", html.unescape(text))
return WS.sub(" ", plain).strip()
_JUNK_SUMMARY_RE = re.compile(
r"^(点击查看原文|article url:|comments url:|discussion on hn|read more)",
re.IGNORECASE,
)
def _is_junk_news_summary(text: str) -> bool:
if not text:
return True
if _JUNK_SUMMARY_RE.match(text.strip()):
return True
if text.strip().endswith(">") and "点击" in text:
return True
return False
def brief_news_summary(text: str | None, limit: int | None = None) -> str:
"""企微新闻一句摘要:去 HTML、过滤占位文案、句读处截断。"""
plain = _strip_summary_plain(text)
if _is_junk_news_summary(plain):
return ""
lim = wecom_news_desc_limit() if limit is None else limit
return trim_brief(plain, lim)
def sync_wecom_news_rows(items: list[dict[str, Any]], flat: list[dict[str, Any]]) -> None:
"""中文化后,用 flat 最新 summary 刷新企微 desc_short。"""
by_link = {_normalize_link(str(i.get("link") or "")): i for i in flat if i.get("link")}
for row in items:
link = _normalize_link(str(row.get("link") or ""))
src = by_link.get(link)
if src:
row["desc_short"] = brief_news_summary(src.get("summary"))
def finalize_wecom_news_items(
items: list[dict[str, Any]],
*,
force_chinese: bool = False,
) -> None:
"""企微新闻摘要:确保 desc_short 为中文(国际源 force_chinese=True"""
from daily.localize import LocalizeJob, localize_brief_descriptions, needs_chinese
from daily.news.sanitize import strip_relax_window_prefix
limit = wecom_news_desc_limit()
jobs: list[LocalizeJob] = []
keyed: list[tuple[str, dict[str, Any]]] = []
for idx, item in enumerate(items):
text = strip_relax_window_prefix(
(item.get("desc_short") or item.get("summary_plain") or "").strip()
)
if text:
item["desc_short"] = text
if not text or _is_junk_news_summary(text):
item["desc_short"] = ""
continue
if force_chinese or needs_chinese(text):
key = f"wecom-news:{item.get('link') or idx}"
jobs.append(LocalizeJob(key, text, limit))
keyed.append((key, item))
elif not item.get("desc_short"):
item["desc_short"] = brief_news_summary(text, limit)
if not jobs:
return
zh_map = localize_brief_descriptions(jobs, archive=True)
for key, item in keyed:
if key in zh_map:
item["desc_short"] = strip_relax_window_prefix(zh_map[key])
def _normalize_link(link: str) -> str:
parsed = urlparse(link.strip())
query = parse_qs(parsed.query, keep_blank_values=False)
for key in list(query.keys()):
if key.lower().startswith("utm_") or key.lower() in {"ref", "source"}:
query.pop(key, None)
clean_query = urlencode({k: v[0] for k, v in query.items() if v}, doseq=False)
return urlunparse((parsed.scheme, parsed.netloc, parsed.path.rstrip("/"), "", clean_query, ""))
def _normalize_title(title: str) -> str:
return WS.sub(" ", title.strip().lower())
def _entry_datetime(entry: dict[str, Any]) -> datetime | None:
for key in ("published", "updated"):
dt = _parse_datetime(entry.get(key))
if dt:
return dt
return None
def _parse_atom(content: str, feed_name: str, category: NewsCategory) -> list[dict[str, Any]]:
items: list[dict[str, Any]] = []
try:
root = ET.fromstring(content)
except ET.ParseError:
return items
for entry in root.findall("a:entry", ATOM_NS):
title_el = entry.find("a:title", ATOM_NS)
link_el = entry.find("a:link", ATOM_NS)
summary_el = entry.find("a:summary", ATOM_NS) or entry.find("a:content", ATOM_NS)
updated_el = entry.find("a:updated", ATOM_NS) or entry.find("a:published", ATOM_NS)
title = title_el.text.strip() if title_el is not None and title_el.text else ""
link = ""
if link_el is not None:
link = link_el.get("href") or (link_el.text or "").strip()
if not title or not link:
continue
items.append(
{
"title": title,
"link": link,
"summary": _clean_text(summary_el.text if summary_el is not None else ""),
"published": updated_el.text.strip() if updated_el is not None and updated_el.text else "",
"source_name": feed_name,
"category_id": category.id,
"category_name": category.name,
"category_icon": category.icon,
}
)
return items
def _parse_rss(content: str, feed_name: str, category: NewsCategory) -> list[dict[str, Any]]:
items: list[dict[str, Any]] = []
try:
root = ET.fromstring(content)
except ET.ParseError:
return items
channel = root.find("channel")
if channel is None:
return items
for item in channel.findall("item"):
title_el = item.find("title")
link_el = item.find("link")
desc_el = item.find("description")
pub_el = item.find("pubDate")
title = title_el.text.strip() if title_el is not None and title_el.text else ""
link = link_el.text.strip() if link_el is not None and link_el.text else ""
if not title or not link:
continue
items.append(
{
"title": title,
"link": link,
"summary": _clean_text(desc_el.text if desc_el is not None else ""),
"published": pub_el.text.strip() if pub_el is not None and pub_el.text else "",
"source_name": feed_name,
"category_id": category.id,
"category_name": category.name,
"category_icon": category.icon,
}
)
return items
def _parse_feed(content: str, feed_name: str, category: NewsCategory) -> list[dict[str, Any]]:
text = content.lstrip("\ufeff").strip()
if not text:
return []
if text.startswith("<rss") or "<channel>" in text[:500]:
return _parse_rss(text, feed_name, category)
if text.startswith("<feed") or "<entry>" in text[:500]:
return _parse_atom(text, feed_name, category)
if "<item>" in text[:2000]:
return _parse_rss(text, feed_name, category)
return _parse_atom(text, feed_name, category)
def _is_reddit_url(url: str) -> bool:
host = urlparse(url).netloc.lower()
return host.endswith("reddit.com")
def _reddit_auth_params() -> dict[str, str]:
user = (env("REDDIT_RSS_USER") or "").strip()
feed = (env("REDDIT_RSS_FEED") or "").strip()
if user and feed:
return {"user": user, "feed": feed}
return {}
def _with_query_params(url: str, extra: dict[str, str]) -> str:
if not extra:
return url
parsed = urlparse(url)
query = parse_qs(parsed.query, keep_blank_values=True)
for key, value in extra.items():
if value and key not in query:
query[key] = [value]
clean_query = urlencode({k: v[0] for k, v in query.items() if v and v[0]}, doseq=False)
return urlunparse((parsed.scheme, parsed.netloc, parsed.path, "", clean_query, ""))
def _reddit_old_url(url: str) -> str:
parsed = urlparse(url)
host = parsed.netloc.lower()
if host.startswith("old."):
return url
if host in {"www.reddit.com", "reddit.com"}:
return urlunparse((parsed.scheme, "old.reddit.com", parsed.path, "", parsed.query, ""))
return url
def _request_headers(base: dict[str, str], url: str) -> dict[str, str]:
if not _is_reddit_url(url):
return base
return {
**base,
"User-Agent": BROWSER_USER_AGENT,
"Accept-Language": "en-US,en;q=0.9",
}
def _reddit_fetch_urls(feed_url: str) -> list[str]:
primary = _with_query_params(feed_url, _reddit_auth_params())
if not _is_reddit_url(primary):
return [primary]
fallback = _reddit_old_url(primary)
if fallback == primary:
return [primary]
return [primary, fallback]
def _fetch_one(
client: httpx.Client,
category: NewsCategory,
feed: NewsFeed,
) -> list[dict[str, Any]]:
last_exc: Exception | None = None
for url in _reddit_fetch_urls(feed.url):
try:
headers = _request_headers(dict(client.headers), url)
resp = client.get(url, headers=headers)
resp.raise_for_status()
entries = _parse_feed(resp.text, feed.name, category)
return _filter_ai_entries(entries, ai_filter=feed.ai_filter)
except Exception as exc:
last_exc = exc
continue
logger.warning("RSS fetch failed [%s] %s: %s", feed.name, feed.url, last_exc)
return []
def _dedupe_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
seen_links: set[str] = set()
seen_titles: set[str] = set()
result: list[dict[str, Any]] = []
for item in items:
link_key = _normalize_link(item["link"])
title_key = _normalize_title(item["title"])
if link_key in seen_links or title_key in seen_titles:
continue
seen_links.add(link_key)
seen_titles.add(title_key)
result.append(item)
return result
def _filter_flat_in_window(
flat: list[dict[str, Any]],
*,
floor_today: bool,
) -> list[dict[str, Any]]:
cutoff = _cutoff_datetime(floor_today=floor_today)
items = [i for i in _dedupe_items(flat) if _within_window(i, cutoff)]
items.sort(key=_sort_key, reverse=True)
return items
def _pick_news_items(
flat: list[dict[str, Any]],
limit: int,
preferred: tuple[str, ...],
*,
one_per_source: bool = False,
) -> list[dict[str, Any]]:
picked: list[dict[str, Any]] = []
seen_links: set[str] = set()
seen_sources: set[str] = set()
def _try_take(item: dict[str, Any]) -> bool:
link = _normalize_link(item.get("link", ""))
if not link or link in seen_links:
return False
if one_per_source:
source = item.get("source_name", "?")
if source in seen_sources:
return False
seen_sources.add(source)
seen_links.add(link)
picked.append(item)
return True
for cat in preferred:
for item in flat:
if item.get("category_id") != cat:
continue
if _try_take(item) and len(picked) >= limit:
return picked[:limit]
for item in flat:
if _try_take(item) and len(picked) >= limit:
break
return picked[:limit]
def _fill_picked_to_limit(
picked: list[dict[str, Any]],
pools: list[list[dict[str, Any]]],
limit: int,
) -> list[dict[str, Any]]:
seen_links = {_normalize_link(i.get("link", "")) for i in picked}
for pool in pools:
for item in pool:
if len(picked) >= limit:
return picked[:limit]
link = _normalize_link(item.get("link", ""))
if not link or link in seen_links:
continue
picked.append(item)
seen_links.add(link)
return picked[:limit]
def _to_wecom_news_row(item: dict[str, Any]) -> dict[str, Any]:
plain = _strip_summary_plain(item.get("summary", ""))
return {
"title": item.get("title", "?"),
"link": item.get("link", ""),
"source_name": item.get("source_name", "?"),
"published_fmt": item.get("published_fmt", ""),
"desc_short": brief_news_summary(plain),
"summary_plain": plain,
}
def _apply_pushed_dedup_with_backfill(
items: list[dict[str, Any]],
picked: list[dict[str, Any]],
*,
date_str: str | None,
limit: int,
) -> list[dict[str, Any]]:
if not date_str:
return items[:limit]
from daily.config import news_backfill_enabled
from daily.news.pushed_links import filter_unpushed_items
fresh = filter_unpushed_items(items, date_str=date_str)
if len(fresh) >= limit:
return fresh[:limit]
if not news_backfill_enabled():
if len(fresh) < limit:
logger.info("news_short:%s", len(fresh))
return fresh[:limit]
seen = {_normalize_link(i.get("link", "")) for i in fresh if i.get("link")}
for item in picked:
if len(fresh) >= limit:
break
link = _normalize_link(item.get("link", ""))
if not link or link in seen:
continue
fresh.append(_to_wecom_news_row(item))
seen.add(link)
return fresh[:limit]
def _within_window(item: dict[str, Any], cutoff: datetime) -> bool:
dt = _entry_datetime(item)
if dt is None:
return False
return dt >= cutoff
def _sort_key(item: dict[str, Any]) -> tuple[int, datetime]:
dt = _entry_datetime(item)
if dt is None:
return (1, datetime.min.replace(tzinfo=timezone.utc))
return (0, dt)
def _fetch_news(categories: tuple[NewsCategory, ...], *, floor_today: bool = False) -> dict[str, Any]:
hours = _hours_window()
per_feed = _per_feed_limit()
per_category = _per_category_limit()
cutoff = _cutoff_datetime(floor_today=floor_today)
headers = {"User-Agent": USER_AGENT, "Accept": "application/rss+xml, application/atom+xml, application/xml, text/xml, */*"}
tasks: list[tuple[NewsCategory, NewsFeed]] = []
for category in categories:
for feed in category.feeds:
tasks.append((category, feed))
raw_by_category: dict[str, list[dict[str, Any]]] = {c.id: [] for c in categories}
stats = {"feeds_total": len(tasks), "feeds_ok": 0, "items_raw": 0}
with httpx.Client(timeout=15.0, verify=certifi.where(), follow_redirects=True, headers=headers) as client:
fast_tasks = [t for t in tasks if not t[1].slow]
slow_tasks = [t for t in tasks if t[1].slow]
with ThreadPoolExecutor(max_workers=8) as pool:
futures = {
pool.submit(_fetch_one, client, cat, feed): (cat.id, feed.name)
for cat, feed in fast_tasks
}
for future in as_completed(futures):
cat_id, feed_name = futures[future]
try:
entries = future.result()
except Exception as exc:
logger.warning("RSS 任务异常 [%s]: %s", feed_name, exc)
continue
if entries:
stats["feeds_ok"] += 1
stats["items_raw"] += len(entries)
raw_by_category[cat_id].extend(entries[:per_feed])
for cat, feed in slow_tasks:
entries = _fetch_one(client, cat, feed)
if entries:
stats["feeds_ok"] += 1
stats["items_raw"] += len(entries)
raw_by_category[cat.id].extend(entries[:per_feed])
if _is_reddit_url(feed.url):
time.sleep(2.0)
else:
time.sleep(1.0)
categories_out: list[dict[str, Any]] = []
flat: list[dict[str, Any]] = []
for category in categories:
items = raw_by_category[category.id]
items = [i for i in items if _within_window(i, cutoff)]
items.sort(key=_sort_key, reverse=True)
items = _dedupe_items(items)[:per_category]
for item in items:
dt = _entry_datetime(item)
item["published_fmt"] = dt.astimezone(timezone(timedelta(hours=8))).strftime("%m-%d %H:%M") if dt else ""
if items:
categories_out.append(
{
"id": category.id,
"name": category.name,
"icon": category.icon,
"items": items,
}
)
flat.extend(items)
flat.sort(key=_sort_key, reverse=True)
flat = _dedupe_items(flat)
return {
"enabled": True,
"hours": hours,
"floor_today": floor_today,
"categories": categories_out,
"flat": flat,
"stats": stats,
}
def fetch_ai_news() -> dict[str, Any]:
"""按类别抓取国际 AI 时讯,返回 {enabled, hours, categories, flat, stats}。"""
if not _enabled():
return {"enabled": False, "categories": [], "flat": [], "stats": {}}
return _fetch_news(NEWS_CATEGORIES, floor_today=_floor_today_enabled())
def fetch_cn_ai_news() -> dict[str, Any]:
"""按类别抓取国内 AI 时讯,返回 {enabled, hours, categories, flat, stats}。"""
if not _cn_enabled():
return {"enabled": False, "categories": [], "flat": [], "stats": {}}
return _fetch_news(CN_NEWS_CATEGORIES)
def format_news_section(news: dict[str, Any], *, section_no: int, wecom_limit: int | None = None) -> list[str]:
return _format_news_section(
news,
section_no=section_no,
title="国际 AI 时讯",
disabled_hint="AI 时讯已关闭(`DAILY_AI_NEWS=0`)。",
wecom_limit=wecom_limit,
)
def format_cn_news_section(news: dict[str, Any], *, section_no: int, wecom_limit: int | None = None) -> list[str]:
return _format_news_section(
news,
section_no=section_no,
title="国内 AI 时讯",
disabled_hint="国内 AI 时讯已关闭(`DAILY_CN_AI_NEWS=0`)。",
wecom_limit=wecom_limit,
)
def _format_news_section(
news: dict[str, Any],
*,
section_no: int,
title: str,
disabled_hint: str,
wecom_limit: int | None = None,
) -> list[str]:
if not news.get("enabled"):
return ["---", "", f"## {section_no}{title}", "", f"*{disabled_hint}*", ""]
categories = news.get("categories") or []
hours = news.get("hours", 72)
floor_note = " · 仅今日" if news.get("floor_today") else ""
lines = [
"---",
"",
f"## {section_no}{title}",
"",
f"> 近 **{hours}h**{floor_note} · {news.get('stats', {}).get('feeds_ok', 0)}/{news.get('stats', {}).get('feeds_total', 0)} 源可用",
"",
]
if not categories:
lines.append("*暂无可用条目(网络/RSS 源异常或时间窗口内无更新)。*")
lines.append("")
return lines
if wecom_limit is not None:
flat = (news.get("flat") or [])[:wecom_limit]
for i, item in enumerate(flat, 1):
pub = f" · {item['published_fmt']}" if item.get("published_fmt") else ""
lines.append(
f"{i}. [{item['title']}]({item['link']}) · `{item['source_name']}`{pub}"
)
lines.append("")
return lines
for cat in categories:
lines.append(f"### {cat['icon']} {cat['name']}")
lines.append("")
for i, item in enumerate(cat["items"], 1):
pub = f" · {item['published_fmt']}" if item.get("published_fmt") else ""
lines.append(f"{i}. **[{item['title']}]({item['link']})** · `{item['source_name']}`{pub}")
summary = item.get("summary", "")
if summary:
lines.append(f" - {_clean_text(summary, news_summary_limit())}")
lines.append("")
return lines
def prepare_wecom_news_items(news: dict[str, Any], *, date_str: str | None = None) -> list[dict[str, Any]]:
if not news.get("enabled"):
return []
limit = _wecom_limit()
floor = bool(news.get("floor_today", _floor_today_enabled()))
flat_strict = _filter_flat_in_window(news.get("flat") or [], floor_today=floor)
flat_relaxed = _filter_flat_in_window(news.get("flat") or [], floor_today=False)
preferred = ("media", "newsletter", "official", "community", "research", "developer")
picked = _pick_news_items(flat_strict, limit, preferred)
picked = _fill_picked_to_limit(picked, [flat_relaxed, news.get("flat") or []], limit)
items = [_to_wecom_news_row(item) for item in picked[:limit]]
return _apply_pushed_dedup_with_backfill(items, picked, date_str=date_str, limit=limit)
def prepare_wecom_cn_news_items(news: dict[str, Any], *, date_str: str | None = None) -> list[dict[str, Any]]:
if not news.get("enabled"):
return []
limit = _wecom_cn_limit()
flat = _filter_flat_in_window(news.get("flat") or [], floor_today=False)
preferred = ("media", "tech")
picked = _pick_news_items(flat, limit, preferred, one_per_source=True)
picked = _fill_picked_to_limit(picked, [news.get("flat") or []], limit)
items = [_to_wecom_news_row(item) for item in picked[:limit]]
return _apply_pushed_dedup_with_backfill(items, picked, date_str=date_str, limit=limit)