from __future__ import annotations import email.utils import sqlite3 import urllib.error import urllib.request import xml.etree.ElementTree as ET from dataclasses import dataclass from datetime import UTC, datetime from .scoring import score_article from .text import canonicalize_url, clean_text, sha256_text USER_AGENT = "goodNews/0.1 (+local constructive news prototype)" @dataclass class FeedItem: title: str url: str description: str | None = None author: str | None = None published_at: str | None = None image_url: str | None = None language: str | None = None raw_guid: str | None = None def poll_all_sources(conn: sqlite3.Connection, limit: int | None = None) -> dict: return _poll_rows(conn, conn.execute( "SELECT * FROM sources WHERE active = 1 ORDER BY id" ).fetchall(), limit) def poll_due_sources(conn: sqlite3.Connection, limit: int | None = None) -> dict: """Poll only active sources whose last successful poll is older than their poll_interval_minutes (or that have never been polled successfully). This is what makes poll_interval_minutes meaningful and lets a scheduler run frequently without re-hitting feeds that are not yet due. """ rows = conn.execute( """ SELECT s.* FROM sources s WHERE s.active = 1 AND ( NOT EXISTS ( SELECT 1 FROM ingest_runs r WHERE r.source_id = s.id AND r.status = 'ok' ) OR ( SELECT MAX(r.finished_at) FROM ingest_runs r WHERE r.source_id = s.id AND r.status = 'ok' ) <= datetime('now', '-' || s.poll_interval_minutes || ' minutes') ) ORDER BY s.id """ ).fetchall() return _poll_rows(conn, rows, limit) def _poll_rows(conn: sqlite3.Connection, rows: list[sqlite3.Row], limit: int | None) -> dict: if limit is not None: rows = rows[:limit] totals = {"sources": 0, "seen": 0, "inserted": 0, "duplicate": 0, "failed": 0} for source in rows: result = poll_source(conn, source) totals["sources"] += 1 totals["seen"] += result["seen"] totals["inserted"] += result["inserted"] totals["duplicate"] += result["duplicate"] totals["failed"] += 1 if result["status"] == "failed" else 0 return totals def poll_source(conn: sqlite3.Connection, source: sqlite3.Row) -> dict: run_id = conn.execute( "INSERT INTO ingest_runs (source_id) VALUES (?)", (source["id"],), ).lastrowid conn.commit() seen = inserted = duplicate = 0 try: xml = fetch_feed(source["feed_url"]) items = parse_feed(xml) seen = len(items) for item in items: inserted_now = insert_article(conn, source, item) if inserted_now: inserted += 1 else: duplicate += 1 conn.execute( """ UPDATE ingest_runs SET finished_at = CURRENT_TIMESTAMP, status = 'ok', items_seen = ?, items_inserted = ?, items_duplicate = ? WHERE id = ? """, (seen, inserted, duplicate, run_id), ) conn.commit() return {"status": "ok", "seen": seen, "inserted": inserted, "duplicate": duplicate} except Exception as exc: conn.execute( """ UPDATE ingest_runs SET finished_at = CURRENT_TIMESTAMP, status = 'failed', items_seen = ?, items_inserted = ?, items_duplicate = ?, error = ? WHERE id = ? """, (seen, inserted, duplicate, str(exc), run_id), ) conn.commit() return { "status": "failed", "seen": seen, "inserted": inserted, "duplicate": duplicate, "error": str(exc), } def fetch_feed(url: str, timeout: int = 20) -> bytes: request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) try: with urllib.request.urlopen(request, timeout=timeout) as response: return response.read() except urllib.error.HTTPError as exc: raise RuntimeError(f"HTTP {exc.code} fetching {url}") from exc except urllib.error.URLError as exc: raise RuntimeError(f"failed fetching {url}: {exc.reason}") from exc def parse_feed(xml: bytes) -> list[FeedItem]: root = ET.fromstring(xml) root_name = _local_name(root.tag) if root_name == "feed": return _parse_atom(root) return _parse_rss(root) def insert_article(conn: sqlite3.Connection, source: sqlite3.Row, item: FeedItem) -> bool: canonical_url = canonicalize_url(item.url) if not canonical_url or not item.title: return False title = clean_text(item.title, max_len=500) description = clean_text(item.description, max_len=1000) if not title: return False url_hash = sha256_text(canonical_url) title_hash = sha256_text(title) try: cursor = conn.execute( """ INSERT INTO articles ( source_id, canonical_url, title, description, author, published_at, image_url, language, raw_guid, url_hash, title_hash ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( source["id"], canonical_url, title, description, clean_text(item.author, max_len=250), item.published_at, canonicalize_url(item.image_url), item.language, item.raw_guid, url_hash, title_hash, ), ) except sqlite3.IntegrityError: return False scores = score_article(title, description, int(source["pr_risk_score"])) conn.execute( """ INSERT INTO article_scores ( article_id, constructive_score, cortisol_score, ragebait_score, agency_score, human_benefit_score, novelty_score, pr_risk_score, accepted, reason_code, reason_text, model_name ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( cursor.lastrowid, scores["constructive_score"], scores["cortisol_score"], scores["ragebait_score"], scores["agency_score"], scores["human_benefit_score"], scores["novelty_score"], scores["pr_risk_score"], scores["accepted"], scores["reason_code"], scores["reason_text"], scores["model_name"], ), ) conn.commit() return True def _parse_rss(root: ET.Element) -> list[FeedItem]: channel = _first_child(root, "channel") or root language = _first_text(channel, "language") items = [element for element in root.iter() if _local_name(element.tag) == "item"] parsed = [] for item in items: title = _first_text(item, "title") link = _first_text(item, "link") guid = _first_text(item, "guid") url = link or guid if not title or not url: continue parsed.append( FeedItem( title=title, url=url, description=_first_text(item, "description", "summary", "encoded"), author=_first_text(item, "author", "creator"), published_at=_parse_date(_first_text(item, "pubDate", "published", "updated", "date")), image_url=_find_image_url(item), language=language, raw_guid=guid, ) ) return parsed def _parse_atom(root: ET.Element) -> list[FeedItem]: language = root.attrib.get("{http://www.w3.org/XML/1998/namespace}lang") entries = [element for element in root if _local_name(element.tag) == "entry"] parsed = [] for entry in entries: title = _first_text(entry, "title") url = _atom_link(entry) if not title or not url: continue author = None author_el = _first_child(entry, "author") if author_el is not None: author = _first_text(author_el, "name") or _text(author_el) parsed.append( FeedItem( title=title, url=url, description=_first_text(entry, "summary", "content"), author=author, published_at=_parse_date(_first_text(entry, "published", "updated")), image_url=_find_image_url(entry), language=language, raw_guid=_first_text(entry, "id"), ) ) return parsed def _atom_link(entry: ET.Element) -> str | None: fallback = None for child in entry: if _local_name(child.tag) != "link": continue href = child.attrib.get("href") if not href: continue if child.attrib.get("rel", "alternate") == "alternate": return href fallback = fallback or href return fallback def _find_image_url(element: ET.Element) -> str | None: for child in element.iter(): name = _local_name(child.tag) if name in {"thumbnail", "content"} and child.attrib.get("url"): if child.attrib.get("medium") in {None, "image"}: return child.attrib["url"] if name == "enclosure" and child.attrib.get("url"): mime = child.attrib.get("type", "") if mime.startswith("image/"): return child.attrib["url"] return None def _parse_date(value: str | None) -> str | None: if not value: return None value = value.strip() try: parsed = email.utils.parsedate_to_datetime(value) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=UTC) return parsed.astimezone(UTC).isoformat() except (TypeError, ValueError): pass try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=UTC) return parsed.astimezone(UTC).isoformat() except ValueError: return None def _first_child(element: ET.Element, name: str) -> ET.Element | None: for child in element: if _local_name(child.tag) == name: return child return None def _first_text(element: ET.Element, *names: str) -> str | None: for child in element: if _local_name(child.tag) in names: value = _text(child) if value: return value return None def _text(element: ET.Element) -> str | None: if element.text: return element.text.strip() return None def _local_name(tag: str) -> str: if "}" in tag: return tag.rsplit("}", 1)[1] return tag