esh-pfi-infrastructure/stacks/news-digest/digest.py

"""news-digest — twice-daily LLM-curated briefing across subreddits + Miniflux.

Runs from cron at 0800 / 2000 local. Each invocation:

  1. Pulls the subreddit list from Miniflux (any feed whose URL starts
     with https://www.reddit.com/r/) — single source of truth, no
     duplicated config.
  2. Hits Reddit's public JSON API per subreddit for top-of-day,
     filters by score + upvote ratio.
  3. Pulls non-Reddit recent items from Miniflux (Tech aggregators
     category — HN, Lobste.rs).
  4. Batches each source through llama-swap on ana-ml2 with a
     terse summarization prompt (one call per source).
  5. Renders the Jinja2 template + CSS to /output/index.html
     (atomic write via .tmp + rename).
  6. Also writes /output/edition-YYYY-MM-DD-<am|pm>.html as an archive.

All tunables are environment-driven; see .env.example for the full
list. Designed to be a one-shot invocation — it does not loop or daemon.
"""
from __future__ import annotations

import hashlib
import json
import os
import re
import sys
import time
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Iterable, Optional

import requests
from jinja2 import Environment, FileSystemLoader, select_autoescape


def _stable_id(*parts: str) -> str:
    """12-char sha1 prefix used as the per-item id for the X-button-to-hide
    feature. Stable across editions (built from the source's native id),
    cross-source-unique (prefixed with the source kind), and short enough
    to live in JSON without bloat."""
    h = hashlib.sha1()
    for p in parts:
        h.update(p.encode("utf-8", errors="replace"))
        h.update(b"\x00")
    return h.hexdigest()[:12]

# ── env config ───────────────────────────────────────────────────────

LLAMA_SWAP_URL    = os.environ.get("LLAMA_SWAP_URL", "http://10.250.50.54:9292")
LLAMA_SWAP_MODEL  = os.environ.get("LLAMA_SWAP_MODEL", "qwen3.5-35-a3b")
LLAMA_SWAP_TIMEOUT = int(os.environ.get("LLAMA_SWAP_TIMEOUT", "180"))

MINIFLUX_URL      = os.environ.get("MINIFLUX_URL", "http://miniflux:8080")
MINIFLUX_USER     = os.environ.get("MINIFLUX_USER", "lkraven")
MINIFLUX_PASS     = os.environ.get("MINIFLUX_PASSWORD", "")

OUTPUT_DIR        = Path(os.environ.get("DIGEST_OUTPUT_DIR", "/output"))
TEMPLATE_DIR      = Path(os.environ.get("DIGEST_TEMPLATE_DIR", "/app/templates"))

REDDIT_HOURS      = int(os.environ.get("DIGEST_REDDIT_HOURS", "12"))
REDDIT_MIN_SCORE  = int(os.environ.get("DIGEST_MIN_SCORE", "50"))
REDDIT_MIN_RATIO  = float(os.environ.get("DIGEST_MIN_RATIO", "0.85"))
REDDIT_MAX_PER_SUB = int(os.environ.get("DIGEST_MAX_PER_SUB", "8"))
REDDIT_USER_AGENT = os.environ.get(
    "DIGEST_REDDIT_USER_AGENT",
    "news-digest:phasefinal:0.1.0 (anonymous)",
)

MINIFLUX_TECH_CATEGORY = os.environ.get(
    "DIGEST_MINIFLUX_TECH_CATEGORY", "Tech aggregators"
)
MINIFLUX_HOURS    = int(os.environ.get("DIGEST_MINIFLUX_HOURS", "12"))
MINIFLUX_MAX_PER_SOURCE = int(os.environ.get("DIGEST_MINIFLUX_MAX", "8"))

TZ_NAME           = os.environ.get("TZ", "America/Los_Angeles")

# ── data shapes ──────────────────────────────────────────────────────

@dataclass
class Item:
    """A single curated post — Reddit or Miniflux origin."""
    id: str
    title: str
    url: str            # external link or HTML permalink
    permalink: str      # discussion / source URL (Reddit thread, HN comments)
    body: str           # selftext / description (may be empty)
    author: str
    score: Optional[int]    # Reddit upvotes if known
    comments: Optional[int]
    upvote_ratio: Optional[float]
    posted_at: datetime
    # Filled by summarize step:
    tldr: str = ""
    tag: str = ""

@dataclass
class Source:
    """A logical grouping of items shown as one section in the digest."""
    name: str            # display name ("r/selfhosted", "Hacker News")
    kind: str            # "reddit" | "miniflux"
    href: str            # link to the source's homepage / sub
    items: list[Item] = field(default_factory=list)

# ── http session shared across calls ─────────────────────────────────

S = requests.Session()
S.headers["User-Agent"] = REDDIT_USER_AGENT

def log(msg: str) -> None:
    print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)

# ── miniflux: discover subreddits + pull tech-aggregator items ───────

def miniflux_get(path: str, **params) -> Any:
    url = f"{MINIFLUX_URL.rstrip('/')}{path}"
    r = S.get(url, params=params, auth=(MINIFLUX_USER, MINIFLUX_PASS), timeout=20)
    r.raise_for_status()
    return r.json()

REDDIT_FEED_RE = re.compile(r"^https?://(?:www\.)?reddit\.com/r/([^/]+)/", re.I)

def discover_subreddits_from_miniflux() -> list[str]:
    """Return list of subreddit names extracted from Miniflux's feed URLs."""
    feeds = miniflux_get("/v1/feeds")
    subs: list[str] = []
    for f in feeds:
        m = REDDIT_FEED_RE.match(f.get("feed_url", ""))
        if m:
            subs.append(m.group(1))
    seen, deduped = set(), []
    for s in subs:
        k = s.lower()
        if k not in seen:
            deduped.append(s)
            seen.add(k)
    return deduped

def fetch_miniflux_tech_items() -> list[Source]:
    """Return one Source per non-Reddit feed in the configured category."""
    cats = miniflux_get("/v1/categories")
    tech_cat = next(
        (c for c in cats if c["title"].lower() == MINIFLUX_TECH_CATEGORY.lower()),
        None,
    )
    if not tech_cat:
        log(f"miniflux: category {MINIFLUX_TECH_CATEGORY!r} not found, skipping")
        return []

    cutoff = int((datetime.now(timezone.utc) - timedelta(hours=MINIFLUX_HOURS)).timestamp())
    entries = miniflux_get(
        "/v1/entries",
        category_id=tech_cat["id"],
        published_after=cutoff,
        order="published_at",
        direction="desc",
        limit=200,
    )

    by_feed: dict[int, Source] = {}
    for e in entries.get("entries", []):
        feed = e.get("feed") or {}
        if REDDIT_FEED_RE.match(feed.get("feed_url", "")):
            continue                                  # handled in Reddit pass
        fid = feed.get("id")
        if fid is None:
            continue
        src = by_feed.get(fid)
        if src is None:
            src = Source(
                name=feed.get("title", "?"),
                kind="miniflux",
                href=feed.get("site_url") or feed.get("feed_url") or "",
            )
            by_feed[fid] = src
        if len(src.items) >= MINIFLUX_MAX_PER_SOURCE:
            continue
        src.items.append(Item(
            id=_stable_id("miniflux", str(e["id"])),
            title=e.get("title", "(untitled)"),
            url=e.get("url", ""),
            permalink=e.get("url", ""),
            body=(e.get("content") or "")[:1500],
            author=e.get("author", ""),
            score=None,
            comments=None,
            upvote_ratio=None,
            posted_at=_parse_dt(e.get("published_at")),
        ))
    return [s for s in by_feed.values() if s.items]

def _parse_dt(s: Optional[str]) -> datetime:
    if not s:
        return datetime.now(timezone.utc)
    try:
        return datetime.fromisoformat(s.replace("Z", "+00:00"))
    except Exception:
        return datetime.now(timezone.utc)

# ── reddit JSON: top-of-day per subreddit ────────────────────────────

def fetch_reddit_top(sub: str) -> Source:
    log(f"reddit: r/{sub}")
    url = f"https://www.reddit.com/r/{sub}/top/.json"
    params = {"t": "day", "limit": 25}
    r = S.get(url, params=params, timeout=20)
    if not r.ok:
        log(f"  HTTP {r.status_code}: skipping")
        return Source(name=f"r/{sub}", kind="reddit", href=f"https://reddit.com/r/{sub}")
    posts = r.json().get("data", {}).get("children", [])
    cutoff_ts = (datetime.now(timezone.utc) - timedelta(hours=REDDIT_HOURS)).timestamp()

    items: list[Item] = []
    for p in posts:
        d = p.get("data", {})
        score = d.get("score", 0)
        ratio = d.get("upvote_ratio", 0.0)
        created = d.get("created_utc", 0)
        if score < REDDIT_MIN_SCORE: continue
        if ratio < REDDIT_MIN_RATIO: continue
        if created < cutoff_ts: continue
        items.append(Item(
            id=_stable_id("reddit", d.get("id", "")),
            title=d.get("title", "(untitled)"),
            url=d.get("url", ""),
            permalink=f"https://reddit.com{d.get('permalink', '')}",
            body=(d.get("selftext") or "")[:1500],
            author=d.get("author", "[deleted]"),
            score=score,
            comments=d.get("num_comments"),
            upvote_ratio=ratio,
            posted_at=datetime.fromtimestamp(created, tz=timezone.utc),
        ))
    items.sort(key=lambda x: (x.score or 0), reverse=True)
    items = items[:REDDIT_MAX_PER_SUB]
    log(f"  kept {len(items)} (score>={REDDIT_MIN_SCORE}, ratio>={REDDIT_MIN_RATIO})")
    return Source(name=f"r/{sub}", kind="reddit",
                  href=f"https://reddit.com/r/{sub}", items=items)

# ── llama-swap: batched summarization per source ─────────────────────

SUMMARIZE_SYSTEM = (
    "You are a curator producing a tight intelligence briefing for an "
    "engineer who reads many feeds. You are concise, neutral, and never "
    "editorialize. You skip pure shitposts and screenshots-without-context."
)

SUMMARIZE_USER_TEMPLATE = """Given the {n} posts from {source} below, return a JSON ARRAY where each element has:

- "id":  the post id from the input
- "tldr": ONE sentence, 25 words max, capturing the substantive point. Lead with a verb. No "this post discusses". No "a user shares".
- "tag": ONE word from {{news, tutorial, release, discussion, question, showcase, drama, meme, other}}

If a post is a pure shitpost / screenshot-without-context / duplicate of an item already in this batch, set "tldr" to "" and "tag" to "skip".

Output ONLY the JSON array. No prose, no markdown fence.

POSTS:
{posts_json}
"""

def summarize_source(src: Source) -> None:
    if not src.items:
        return
    posts_json = json.dumps([
        {"id": it.id, "title": it.title, "body": it.body[:600], "url": it.url}
        for it in src.items
    ], ensure_ascii=False)
    user = SUMMARIZE_USER_TEMPLATE.format(
        n=len(src.items),
        source=src.name,
        posts_json=posts_json,
    )
    log(f"  llm: summarizing {len(src.items)} items from {src.name}")
    try:
        r = S.post(
            f"{LLAMA_SWAP_URL.rstrip('/')}/v1/chat/completions",
            json={
                "model": LLAMA_SWAP_MODEL,
                "messages": [
                    {"role": "system", "content": SUMMARIZE_SYSTEM},
                    {"role": "user", "content": user},
                ],
                "temperature": 0.2,
                "max_tokens": 2000,
            },
            timeout=LLAMA_SWAP_TIMEOUT,
        )
        r.raise_for_status()
        msg = r.json()["choices"][0]["message"]
        # Models in extended-thinking mode (e.g. Qwen3.x defaults) put
        # output in reasoning_content and leave content empty until they
        # exit thinking — fall back so we get *something* to parse.
        content = (msg.get("content") or msg.get("reasoning_content") or "").strip()
        # Some models wrap JSON in ```...``` even when told not to.
        content = re.sub(r"^```(?:json)?\s*|\s*```$", "", content, flags=re.M).strip()
        mapped = {x.get("id"): x for x in json.loads(content)}
    except Exception as e:
        log(f"  ! llm failed for {src.name}: {e!r} — keeping raw titles")
        return

    for it in src.items:
        m = mapped.get(it.id, {})
        it.tldr = (m.get("tldr") or "").strip()
        it.tag = (m.get("tag") or "").strip().lower()

    # Drop skipped entries from the source.
    src.items = [it for it in src.items if it.tag != "skip" and (it.tldr or it.score is None)]
    log(f"  -> {len(src.items)} kept after llm filter")

# ── render ───────────────────────────────────────────────────────────

def render(reddit_sources: list[Source], tech_sources: list[Source],
           generated_at: datetime) -> str:
    env = Environment(
        loader=FileSystemLoader(str(TEMPLATE_DIR)),
        autoescape=select_autoescape(["html"]),
        trim_blocks=True,
        lstrip_blocks=True,
    )
    env.filters["humanago"] = _humanago
    env.filters["domain"] = _domain
    template = env.get_template("digest.html.j2")
    edition = "morning" if generated_at.hour < 14 else "evening"
    reddit_kept = [s for s in reddit_sources if s.items]
    tech_kept   = [s for s in tech_sources if s.items]
    return template.render(
        reddit_sources=reddit_kept,
        tech_sources=tech_kept,
        reddit_total=sum(len(s.items) for s in reddit_kept),
        tech_total=sum(len(s.items) for s in tech_kept),
        generated_at=generated_at,
        edition=edition,
        edition_short="AM" if edition == "morning" else "PM",
        model=LLAMA_SWAP_MODEL,
        date_long=generated_at.strftime("%A %B %-d, %Y"),
        time_short=generated_at.strftime("%-I:%M %p"),
        tz=generated_at.tzname() or TZ_NAME,
        next_edition=("evening" if edition == "morning" else "morning"),
    )

def _humanago(d: datetime) -> str:
    delta = datetime.now(timezone.utc) - d
    s = int(delta.total_seconds())
    if s < 60: return f"{s}s"
    if s < 3600: return f"{s // 60}m"
    if s < 86400: return f"{s // 3600}h"
    return f"{s // 86400}d"

def _domain(url: str) -> str:
    m = re.match(r"^https?://(?:www\.)?([^/]+)", url or "")
    return m.group(1) if m else ""

def write_output(html: str, generated_at: datetime) -> None:
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    edition = "am" if generated_at.hour < 14 else "pm"
    archive = OUTPUT_DIR / f"edition-{generated_at:%Y-%m-%d}-{edition}.html"
    index = OUTPUT_DIR / "index.html"

    archive_tmp = archive.with_suffix(".html.tmp")
    archive_tmp.write_text(html, encoding="utf-8")
    archive_tmp.rename(archive)

    index_tmp = index.with_suffix(".html.tmp")
    index_tmp.write_text(html, encoding="utf-8")
    index_tmp.rename(index)
    log(f"wrote {index} (and archive {archive.name})")
    write_archive_index(generated_at)

ARCHIVE_FNAME_RE = re.compile(r"^edition-(\d{4}-\d{2}-\d{2})-(am|pm)\.html$")

def write_archive_index(generated_at: datetime) -> None:
    """Render /output/archive.html — list every edition-*.html in
    OUTPUT_DIR, newest-first. Cheap (re-runs every digest fire);
    template loads from the same TEMPLATE_DIR."""
    editions = []
    for p in OUTPUT_DIR.glob("edition-*.html"):
        m = ARCHIVE_FNAME_RE.match(p.name)
        if not m:
            continue
        date_str, ed = m.group(1), m.group(2)
        try:
            date = datetime.strptime(date_str, "%Y-%m-%d").date()
        except ValueError:
            continue
        editions.append({
            "filename": p.name,
            "date": date,
            "edition": "morning" if ed == "am" else "evening",
            "edition_short": ed.upper(),
            "date_long": date.strftime("%A %B %-d, %Y"),
            # Sort key: date descending, then PM before AM (within a day,
            # PM is the most recent edition).
            "_sort": (date, 1 if ed == "pm" else 0),
        })
    editions.sort(key=lambda e: e["_sort"], reverse=True)

    env = Environment(
        loader=FileSystemLoader(str(TEMPLATE_DIR)),
        autoescape=select_autoescape(["html"]),
        trim_blocks=True,
        lstrip_blocks=True,
    )
    tpl = env.get_template("archive.html.j2")
    html = tpl.render(
        editions=editions,
        generated_at=generated_at,
        total=len(editions),
    )

    out = OUTPUT_DIR / "archive.html"
    out_tmp = out.with_suffix(".html.tmp")
    out_tmp.write_text(html, encoding="utf-8")
    out_tmp.rename(out)
    log(f"wrote {out} ({len(editions)} editions indexed)")

# ── main ─────────────────────────────────────────────────────────────

def main() -> int:
    if not MINIFLUX_PASS:
        print("MINIFLUX_PASSWORD not set — bailing", file=sys.stderr)
        return 2

    try:
        from zoneinfo import ZoneInfo
        now_local = datetime.now(ZoneInfo(TZ_NAME))
    except Exception:
        now_local = datetime.now()
    log(f"starting digest run at {now_local.isoformat()} ({'AM' if now_local.hour < 14 else 'PM'})")

    log("phase 1: discovering subreddits from miniflux")
    subs = discover_subreddits_from_miniflux()
    log(f"  found {len(subs)} subreddits: {', '.join(subs) or '(none)'}")

    log("phase 2: fetching reddit top-of-day per subreddit")
    reddit_sources: list[Source] = []
    for sub in subs:
        reddit_sources.append(fetch_reddit_top(sub))
        time.sleep(1.5)                              # gentle to anonymous Reddit

    log("phase 3: fetching tech-aggregator items from miniflux")
    tech_sources = fetch_miniflux_tech_items()
    log(f"  found {len(tech_sources)} non-reddit feeds with recent items")

    log("phase 4: summarizing each source via llama-swap")
    for src in reddit_sources + tech_sources:
        summarize_source(src)

    log("phase 5: rendering")
    html = render(reddit_sources, tech_sources, now_local)
    write_output(html, now_local)
    log("done")
    return 0

if __name__ == "__main__":
    sys.exit(main())