Oleksandr Kozachuk
400 lines
16 KiB
Python
400 lines
16 KiB
Python
"""News digest fetcher (SPEC-013, FDB-012 news rewrite).
|
|
|
|
Replaces the broken pre-1.0-openai `news_feed.py`. Fetches configured
|
|
RSS/Atom feeds (stdlib, no feedparser dep), builds a compact sanitized
|
|
headline digest, and writes it to the `{news}` file the responder
|
|
injects (AIResponder.message). Feeds are external input: titles are
|
|
sanitized (SAF-03) and each feed URL is SSRF-guarded before fetching.
|
|
|
|
CLI: python -m fjerkroa_bot.news --config kroa.toml
|
|
"""
|
|
|
|
import argparse
|
|
import logging
|
|
import re
|
|
import sys
|
|
import time
|
|
from html import unescape
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
import defusedxml.ElementTree as ElementTree # hardened XML: feeds are untrusted (XXE/billion-laughs)
|
|
|
|
from .ai_responder import sanitize_external_text
|
|
|
|
DEFAULT_PER_FEED = 3
|
|
DEFAULT_MAX_ITEMS = 15
|
|
DEFAULT_SUMMARY_CHARS = 200
|
|
DEFAULT_NEWS_KEEP = 400
|
|
FETCH_TIMEOUT_S = 15
|
|
_ATOM = "{http://www.w3.org/2005/Atom}"
|
|
_RSS1 = "{http://purl.org/rss/1.0/}" # RSS 1.0 / RDF (e.g. 4gamer.net) namespaces <item>/<title>/<link>
|
|
_TAG_RE = re.compile(r"<[^>]+>")
|
|
|
|
|
|
def _clean_summary(raw: str, max_len: int = 300) -> str:
|
|
"""Strip HTML, unescape entities, collapse whitespace (feed descriptions are often HTML)."""
|
|
text = unescape(_TAG_RE.sub(" ", raw or ""))
|
|
return re.sub(r"\s+", " ", text).strip()[:max_len]
|
|
|
|
|
|
def _rss_items(root: Any, ns: str, source: str) -> List[Dict[str, str]]:
|
|
"""RSS 2.0 (ns='') and RSS 1.0/RDF (ns=_RSS1) both use <item><title><link><description>."""
|
|
out: List[Dict[str, str]] = []
|
|
for item in root.iter(f"{ns}item"):
|
|
title = (item.findtext(f"{ns}title") or "").strip()
|
|
link = (item.findtext(f"{ns}link") or "").strip()
|
|
summary = _clean_summary(item.findtext(f"{ns}description") or "")
|
|
if title:
|
|
out.append({"title": title, "link": link, "source": source, "summary": summary})
|
|
return out
|
|
|
|
|
|
def parse_feed(data: bytes, source: str = "") -> List[Dict[str, str]]:
|
|
"""Parse RSS 2.0, RSS 1.0/RDF, or Atom bytes into [{title, link, source, summary}] (tolerant)."""
|
|
try:
|
|
root = ElementTree.fromstring(data)
|
|
except Exception as err:
|
|
# malformed XML or a blocked entity/DTD attack — tolerate, never raise (NEWS-01)
|
|
logging.warning(f"news: unparseable/unsafe feed {source!r}: {err!r}")
|
|
return []
|
|
# RSS 2.0 (unqualified) + RSS 1.0/RDF (namespaced, e.g. 4gamer) share <item><title><link><description>
|
|
items: List[Dict[str, str]] = _rss_items(root, "", source) + _rss_items(root, _RSS1, source)
|
|
# Atom: <feed><entry><title/><link href=/><summary|content/>
|
|
for entry in root.iter(f"{_ATOM}entry"):
|
|
title = (entry.findtext(f"{_ATOM}title") or "").strip()
|
|
link_el = entry.find(f"{_ATOM}link")
|
|
link = link_el.get("href", "") if link_el is not None else ""
|
|
summary = _clean_summary(entry.findtext(f"{_ATOM}summary") or entry.findtext(f"{_ATOM}content") or "")
|
|
if title:
|
|
items.append({"title": title, "link": link, "source": source, "summary": summary})
|
|
return items
|
|
|
|
|
|
def render_digest(items: List[Dict[str, str]], max_items: int = DEFAULT_MAX_ITEMS, summary_chars: int = DEFAULT_SUMMARY_CHARS) -> str:
|
|
"""Compact sanitized digest for the {news} prompt slot (title + short summary + link)."""
|
|
lines = []
|
|
for item in items[:max_items]:
|
|
title = sanitize_external_text(item["title"], 200)
|
|
source = item.get("source", "")
|
|
link = item.get("link", "")
|
|
summary = sanitize_external_text(item.get("summary", ""), summary_chars) if summary_chars else ""
|
|
prefix = f"[{source}] " if source else ""
|
|
line = f"- {prefix}{title}"
|
|
if summary:
|
|
line += f" — {summary}"
|
|
if link:
|
|
line += f" ({link})"
|
|
lines.append(line)
|
|
return "\n".join(lines)
|
|
|
|
|
|
class NewsFetcher:
|
|
def __init__(self, guard, fetch_bytes) -> None:
|
|
# injected so tests need no network; production wires aiohttp + guard_url
|
|
self._guard = guard
|
|
self._fetch_bytes = fetch_bytes
|
|
|
|
async def collect(self, feeds: List[Tuple[str, str]], per_feed: int) -> List[Dict[str, str]]:
|
|
"""feeds = [(url, label)]; returns deduped items, order preserved."""
|
|
seen = set()
|
|
out: List[Dict[str, str]] = []
|
|
for url, label in feeds:
|
|
reason = self._guard(url)
|
|
if reason:
|
|
logging.warning(f"news: skipping feed {label} — {reason}")
|
|
continue
|
|
try:
|
|
data = await self._fetch_bytes(url)
|
|
except Exception as err:
|
|
logging.warning(f"news: fetch failed for {label}: {repr(err)}")
|
|
continue
|
|
for item in parse_feed(data, label)[:per_feed]:
|
|
key = item["title"]
|
|
if key not in seen:
|
|
seen.add(key)
|
|
out.append(item)
|
|
return out
|
|
|
|
|
|
DEFAULT_SEEN_CAP = 5000
|
|
DEFAULT_POST_PER_FEED = 5
|
|
DEFAULT_POST_MAX_PER_RUN = 8
|
|
|
|
|
|
def item_key(item: Dict[str, str]) -> str:
|
|
return item.get("link") or item.get("title") or ""
|
|
|
|
|
|
class NewsPoster:
|
|
"""Post NEW feed items to Discord channel webhooks (ggg model, SPEC-013 NEWS-04..06)."""
|
|
|
|
def __init__(self, guard, fetch_bytes, post_webhook, store: Any = None) -> None:
|
|
self._guard = guard
|
|
self._fetch_bytes = fetch_bytes
|
|
self._post_webhook = post_webhook
|
|
self._store = store
|
|
|
|
async def run_post(
|
|
self,
|
|
feeds: List[Tuple[str, str, str]],
|
|
webhooks: Dict[str, str],
|
|
seen: set,
|
|
per_feed: int,
|
|
max_per_run: int,
|
|
seed_only: bool,
|
|
keep: int = DEFAULT_NEWS_KEEP,
|
|
) -> Tuple[int, set]:
|
|
"""Returns (posted_count, updated_seen). seed_only marks new items seen without posting."""
|
|
posted = 0
|
|
harvested: List[Dict[str, str]] = []
|
|
for url, label, channel in feeds:
|
|
reason = self._guard(url)
|
|
if reason:
|
|
logging.warning(f"news-post: skipping feed {label} — {reason}")
|
|
continue
|
|
try:
|
|
data = await self._fetch_bytes(url)
|
|
except Exception as err:
|
|
logging.warning(f"news-post: fetch failed for {label}: {repr(err)}")
|
|
continue
|
|
for item in parse_feed(data, label)[:per_feed]:
|
|
harvested.append(item) # NEWS-09: everything parsed feeds the searchable store
|
|
key = item_key(item)
|
|
if not key or key in seen:
|
|
continue
|
|
seen.add(key)
|
|
may_post = not seed_only and posted < max_per_run
|
|
if may_post and await self._deliver(item, label, channel, webhooks):
|
|
posted += 1
|
|
if self._store is not None and harvested:
|
|
self._store.add_news_items(harvested)
|
|
self._store.prune_news(keep)
|
|
return posted, seen
|
|
|
|
async def _deliver(self, item: Dict[str, str], label: str, channel: str, webhooks: Dict[str, str]) -> bool:
|
|
hook = webhooks.get(channel)
|
|
if not hook:
|
|
logging.warning(f"news-post: no webhook for channel {channel!r} ({label})")
|
|
return False
|
|
title = sanitize_external_text(item["title"], 300)
|
|
link = item.get("link", "")
|
|
content = f"**[{label}]** {title}" + (f"\n{link}" if link else "")
|
|
try:
|
|
await self._post_webhook(hook, content)
|
|
return True
|
|
except Exception as err:
|
|
logging.warning(f"news-post: webhook post failed ({label}): {repr(err)}")
|
|
return False
|
|
|
|
|
|
# --- news memory + on-demand retrieval tool (SPEC-013 NEWS-09..12) ---
|
|
|
|
GET_NEWS_TOOL = {
|
|
"name": "get_news",
|
|
"description": "Fetch news the bot has collected from its RSS feeds — this is the SAME news that gets posted in the "
|
|
"server's news channels (e.g. #news, #newsjp / ニュース). Use this FIRST, before web_search, for anything about "
|
|
"current news or about something someone saw in a news channel; filter by topic (a keyword, also matches the source "
|
|
"label) or by source. Returns headlines with a short summary and a link; follow up with fetch_url for the full text.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"topic": {"type": "string", "description": "Optional keywords to filter by, e.g. 'Nordland', 'football', 'weather'."},
|
|
"source": {"type": "string", "description": "Optional source label, e.g. 'NRK', 'Aftenposten', 'Verden', 'Sport'."},
|
|
"limit": {"type": "integer", "description": "How many items to return (default 10, max 30)."},
|
|
},
|
|
"required": [],
|
|
},
|
|
}
|
|
|
|
|
|
def _news_terms(topic: Optional[str]) -> List[str]:
|
|
return [t for t in re.split(r"\W+", (topic or "").lower()) if len(t) > 1][:5]
|
|
|
|
|
|
def query_news(
|
|
store: Any, topic: Optional[str] = None, source: Optional[str] = None, limit: int = 10, summary_chars: int = DEFAULT_SUMMARY_CHARS
|
|
) -> Dict[str, Any]:
|
|
"""Retrieve stored news for the get_news tool: term/source-filtered, sanitized (NEWS-11)."""
|
|
if store is None:
|
|
return {"error": "news store unavailable"}
|
|
limit = max(1, min(int(limit or 10), 30))
|
|
src = (str(source).strip() or None) if source else None
|
|
terms = _news_terms(topic)
|
|
try:
|
|
rows = store.search_news(terms, limit, src) if terms else store.recent_news(limit, src)
|
|
except Exception as err:
|
|
logging.warning(f"news: query failed: {err!r}")
|
|
return {"error": "news lookup failed"}
|
|
results = [
|
|
{
|
|
"source": row.get("source", ""),
|
|
"title": sanitize_external_text(row.get("title", ""), 200),
|
|
"summary": sanitize_external_text(row.get("summary", ""), summary_chars),
|
|
"link": row.get("link", ""),
|
|
}
|
|
for row in rows
|
|
]
|
|
return {"topic": topic or "", "source": src or "", "results": results}
|
|
|
|
|
|
def _open_store(config: Dict[str, Any]) -> Any:
|
|
directory = config.get("history-directory")
|
|
if not directory:
|
|
return None
|
|
from pathlib import Path
|
|
|
|
from .persistence import PersistentStore
|
|
|
|
return PersistentStore(Path(str(directory)).expanduser() / "bot.db")
|
|
|
|
|
|
def persist_news(config: Dict[str, Any], items: List[Dict[str, str]]) -> int:
|
|
"""Upsert fetched items into the news store, prune to the rolling window (NEWS-09)."""
|
|
store = _open_store(config)
|
|
if store is None or not items:
|
|
return 0
|
|
added = store.add_news_items(items)
|
|
store.prune_news(int(config.get("news-keep", DEFAULT_NEWS_KEEP)))
|
|
return added
|
|
|
|
|
|
def load_seen(path: str) -> Tuple[set, bool]:
|
|
"""(seen-set, existed). Missing/broken state -> empty set, existed=False (seed run)."""
|
|
import json
|
|
import os
|
|
|
|
if not os.path.exists(path):
|
|
return set(), False
|
|
try:
|
|
with open(path, encoding="utf-8") as fd:
|
|
return set(json.load(fd)), True
|
|
except Exception as err:
|
|
logging.warning(f"news-post: unreadable state {path}: {err!r} — reseeding")
|
|
return set(), False
|
|
|
|
|
|
def save_seen(path: str, seen: set, cap: int = DEFAULT_SEEN_CAP) -> None:
|
|
import json
|
|
|
|
# keep the newest `cap` keys (insertion order preserved by Python sets? no — use a bounded slice)
|
|
keys = list(seen)[-cap:]
|
|
with open(path, "w", encoding="utf-8") as fd:
|
|
json.dump(keys, fd)
|
|
|
|
|
|
def _post_feeds_from_config(config: Dict[str, Any]) -> List[Tuple[str, str, str]]:
|
|
feeds = []
|
|
for entry in config.get("news-post-feeds", []):
|
|
if isinstance(entry, (list, tuple)) and len(entry) >= 3:
|
|
feeds.append((str(entry[0]), str(entry[1]), str(entry[2])))
|
|
return feeds
|
|
|
|
|
|
def _feeds_from_config(config: Dict[str, Any]) -> List[Tuple[str, str]]:
|
|
"""news-feeds = [["url", "label"], ...] or ["url", ...]."""
|
|
feeds = []
|
|
for entry in config.get("news-feeds", []):
|
|
if isinstance(entry, (list, tuple)):
|
|
feeds.append((str(entry[0]), str(entry[1]) if len(entry) > 1 else ""))
|
|
else:
|
|
feeds.append((str(entry), ""))
|
|
return feeds
|
|
|
|
|
|
async def _aiohttp_fetch(url: str) -> bytes:
|
|
import aiohttp
|
|
|
|
from .httpread import read_capped
|
|
|
|
timeout = aiohttp.ClientTimeout(total=FETCH_TIMEOUT_S)
|
|
async with aiohttp.ClientSession(timeout=timeout, headers={"User-Agent": "Mozilla/5.0 (compatible; FjerkroaBot-news/1.0)"}) as session:
|
|
async with session.get(url) as response:
|
|
response.raise_for_status()
|
|
return await read_capped(response, 4 * 1024 * 1024)
|
|
|
|
|
|
async def _aiohttp_post(hook: str, content: str) -> None:
|
|
import aiohttp
|
|
|
|
timeout = aiohttp.ClientTimeout(total=FETCH_TIMEOUT_S)
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
# allowed_mentions none: a headline can never ping the channel (SAF-02 spirit)
|
|
payload = {"content": content[:2000], "allowed_mentions": {"parse": []}}
|
|
async with session.post(hook, json=payload) as response:
|
|
response.raise_for_status()
|
|
|
|
|
|
async def run_post(config: Dict[str, Any]) -> int:
|
|
"""Webhook-posting mode (ggg): post new items to channels. Returns posted count."""
|
|
from .url_reader import guard_url
|
|
|
|
webhooks = dict(config.get("news-post-webhooks", {}))
|
|
feeds = _post_feeds_from_config(config)
|
|
state_path = config.get("news-post-state", "news_state.json")
|
|
if not webhooks or not feeds:
|
|
logging.error("news-post: need news-post-webhooks and news-post-feeds")
|
|
return 0
|
|
seen, existed = load_seen(state_path)
|
|
poster = NewsPoster(guard_url, _aiohttp_fetch, _aiohttp_post, store=_open_store(config))
|
|
posted, seen = await poster.run_post(
|
|
feeds,
|
|
webhooks,
|
|
seen,
|
|
int(config.get("news-post-per-feed", DEFAULT_POST_PER_FEED)),
|
|
int(config.get("news-post-max-per-run", DEFAULT_POST_MAX_PER_RUN)),
|
|
seed_only=not existed, # first run seeds without flooding the channels
|
|
keep=int(config.get("news-keep", DEFAULT_NEWS_KEEP)),
|
|
)
|
|
save_seen(state_path, seen, int(config.get("news-post-seen-cap", DEFAULT_SEEN_CAP)))
|
|
logging.info(f"news-post: posted {posted} item(s)" + (" (seed run — nothing posted)" if not existed else ""))
|
|
return posted
|
|
|
|
|
|
async def run(config: Dict[str, Any]) -> Optional[str]:
|
|
from .url_reader import guard_url
|
|
|
|
out_path = config.get("news")
|
|
if not out_path:
|
|
logging.error("news: no `news` output path in config")
|
|
return None
|
|
feeds = _feeds_from_config(config)
|
|
if not feeds:
|
|
logging.error("news: no `news-feeds` configured")
|
|
return None
|
|
fetcher = NewsFetcher(guard_url, _aiohttp_fetch)
|
|
items = await fetcher.collect(feeds, int(config.get("news-per-feed", DEFAULT_PER_FEED)))
|
|
persist_news(config, items) # NEWS-09: feed the searchable rolling store for get_news
|
|
digest = render_digest(
|
|
items, int(config.get("news-max-items", DEFAULT_MAX_ITEMS)), int(config.get("news-summary-chars", DEFAULT_SUMMARY_CHARS))
|
|
)
|
|
header = f"News as of {time.strftime('%Y-%m-%d %H:%M UTC', time.gmtime())}:\n"
|
|
with open(out_path, "w", encoding="utf-8") as fd:
|
|
fd.write(header + digest + "\n")
|
|
logging.info(f"news: wrote {len(items)} items to {out_path}")
|
|
return out_path
|
|
|
|
|
|
def main() -> int:
|
|
import asyncio
|
|
|
|
import tomlkit
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
|
parser = argparse.ArgumentParser(
|
|
description="Fetch RSS/Atom feeds: --post to channel webhooks (ggg) or default {news} digest file (kroa)"
|
|
)
|
|
parser.add_argument("--config", required=True)
|
|
parser.add_argument("--post", action="store_true", help="webhook-posting mode (post new items to Discord channels)")
|
|
args = parser.parse_args()
|
|
with open(args.config, encoding="utf-8") as fd:
|
|
config = tomlkit.load(fd)
|
|
if args.post:
|
|
asyncio.run(run_post(config))
|
|
return 0
|
|
result = asyncio.run(run(config))
|
|
return 0 if result else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|