git/orchard/main
updater/news.py
171 lines6 KB last change raw · history
"""Arch news, which occasionally tells you an update needs a human.

The front page of archlinux.org is the one place upstream says "this upgrade
requires manual intervention", and it is exactly the kind of thing a tool that
applies updates for you should not let you miss. The feed is RSS 2.0, so this
parses it with the standard library rather than adding a dependency.

Never raises: a feed that cannot be fetched is a normal state, and the caller
decides what to say about it.
"""

from __future__ import annotations

import re
import time
import xml.etree.ElementTree as ET
from dataclasses import dataclass, replace
from email.utils import parsedate_to_datetime

from . import fetch

FEED = "https://archlinux.org/feeds/news/"
#: the feed is a handful of items and changes rarely
TTL = 6 * 3600
#: how far back to look when nothing has been read yet
DEFAULT_WINDOW = 90 * 24 * 3600

#: the phrase upstream uses when an upgrade will not complete on its own
_ACTION_PHRASES = (
    "manual intervention",
    "requires manual",
    "manual action",
    "before upgrading",
    "you will need to",
)
_TAGS = re.compile(r"<[^>]+>")
_SPACE = re.compile(r"\s+")


def plain_text(html: str) -> str:
    """The description is HTML; the terminal wants prose."""
    return _SPACE.sub(" ", _TAGS.sub(" ", html or "")).strip()


@dataclass(frozen=True)
class NewsItem:
    title: str
    link: str
    published: float
    author: str = ""
    summary: str = ""

    @property
    def needs_action(self) -> bool:
        """Whether this reads like it will not apply cleanly on its own."""
        haystack = f"{self.title} {self.summary}".lower()
        return any(phrase in haystack for phrase in _ACTION_PHRASES)

    @property
    def age(self) -> float:
        return max(0.0, time.time() - self.published)


def _parse(text: str) -> list[NewsItem]:
    try:
        root = ET.fromstring(text)
    except ET.ParseError:
        return []

    out: list[NewsItem] = []
    for item in root.iter("item"):
        title = (item.findtext("title") or "").strip()
        if not title:
            continue
        link = (item.findtext("link") or "").strip()
        stamp = 0.0
        raw_date = (item.findtext("pubDate") or "").strip()
        if raw_date:
            try:
                stamp = parsedate_to_datetime(raw_date).timestamp()
            except (TypeError, ValueError):
                stamp = 0.0
        creator = ""
        for child in item:
            if child.tag.split("}")[-1] == "creator":
                creator = (child.text or "").strip()
        out.append(
            NewsItem(
                title=title,
                link=link,
                published=stamp,
                author=creator,
                summary=plain_text(item.findtext("description") or ""),
            )
        )
    # An item whose pubDate is missing or unparseable would keep a stamp of
    # zero, and `recent()` filters on a positive cutoff -- so it could never be
    # shown, however urgent it was. Falling back to the newest stamp in the
    # feed keeps it visible; being ordered slightly wrong is a smaller problem
    # than an Arch news item nobody ever sees.
    newest = max((n.published for n in out), default=0.0)
    fallback = newest if newest > 0 else time.time()
    out = [n if n.published > 0 else replace(n, published=fallback) for n in out]
    out.sort(key=lambda n: n.published, reverse=True)
    return out


def fetch_news(*, timeout: float = 20) -> tuple[list[NewsItem], str]:
    """Recent news, newest first. Returns `(items, error)`; error is empty on success."""
    path = fetch.cache_dir() / "news.xml"
    body, error = fetch.get(FEED, timeout=timeout, cache=path, ttl=TTL)
    if error:
        # a stale copy beats nothing, and the feed rarely changes
        if path.is_file():
            try:
                body = path.read_text()
            except OSError:
                return [], error
        else:
            return [], error
    items = _parse(body)
    if not items:
        return [], "the news feed could not be read"
    return items, ""


def recent(
    items: list[NewsItem], *, since: float = 0.0, window: float = DEFAULT_WINDOW
) -> list[NewsItem]:
    """Items published after `since`, or within `window` if nothing is recorded."""
    cutoff = since if since > 0 else time.time() - window
    return [item for item in items if item.published > cutoff]


def summary_line(items: list[NewsItem]) -> str:
    if not items:
        return "nothing new"
    urgent = [i for i in items if i.needs_action]
    if urgent:
        return f"{len(items)} recent item(s), {len(urgent)} needing a decision before updating"
    return f"{len(items)} recent item(s)"


def explain(items: list[NewsItem], *, limit: int = 12) -> list[str]:
    """The news list, as lines, with anything needing action called out first."""
    if not items:
        return ["nothing new since you last looked"]

    urgent = [i for i in items if i.needs_action]
    routine = [i for i in items if not i.needs_action]
    out = [
        f"{len(items)} item(s) since you last looked, {len(urgent)} of which ask for a decision",
        "",
    ]
    for group, label in ((urgent, "READ BEFORE UPDATING"), (routine, "ALSO RECENT")):
        if not group:
            continue
        out.append(label)
        for item in group[:limit]:
            when = time.strftime("%Y-%m-%d", time.localtime(item.published))
            out.append(f"  {when}  {item.title}")
            if item.summary:
                out.append(f"            {item.summary[:150]}")
            if item.link:
                out.append(f"            {item.link}")
        if len(group) > limit:
            out.append(f"  + {len(group) - limit} more")
        out.append("")
    out.append("mark these as read with n")
    return out