"""'Your digest is ready' — a push with enough of the digest to judge it by. The digest is generated four times a day and shown only when somebody asks for it — a spoken "play my digest", or the button in Home Assistant. Nothing displays it because a person walked past a screen. That makes the notification the only thing that reaches you unprompted, and its job is exactly one decision: is this run worth going and asking for, or does it keep until tonight? So it pushes a short summary to ntfy after every run, readable on a phone or a wrist (Android relays notifications to a Pebble with nothing else installed), without opening the canvas at all. WHAT IT SENDS, AND WHY IT COSTS NO EXTRA LLM CALL ------------------------------------------------- Every section already produces a `narration` at `compact` detail — two to four sentences written to be read aloud, which is exactly the register a notification wants. The summary is those narrations, trimmed, plus the few things that are worth knowing before you open anything: how many political to-dos this run produced, and whether any section was withheld or came back unverified. Nothing here is generated: it is assembled from the documents the run already wrote, so a notification can never claim something the digest itself does not say. WHO GETS WHAT ------------- The per-person section settings (see preferences.py) decide the content of the push, not just the canvas. Somebody who has switched the political section off does not get political content pushed to their phone — that would make the setting a lie in the one place it is most visible. People are grouped by their own ntfy topic (identity's `notify_topic`, the same one arrival notifications use). Each topic gets the union of the sections wanted by the people who share it: a topic IS its audience, so two people sharing one have already agreed to share what arrives on it. Anyone without a topic of their own falls back to `NTFY_TOPIC`, the household topic — and with no preferences known at all (identity down, nobody registered), one message about everything generated goes there. NETWORK PLACEMENT ----------------- This POSTs to the self-hosted ntfy already running in this compose stack for `chores` and `identity` — container to container, never near the firewall. Getting it onto a phone away from home is the WireGuard split tunnel described in docs/network-integration.md §2.2, not a port forward. Nothing new is exposed. Best-effort throughout: a failed push logs a warning. The digest is already written by the time this runs, and no notification is worth failing a run over. """ import logging import os from urllib.parse import urlencode import requests LOG = logging.getLogger(__name__) HTTP_TIMEOUT = 10 # How much of a section's narration survives into the push. ntfy shows a few lines # before truncating, and a Pebble shows fewer still — the point is to be enough to # judge by, not to be the digest. MAX_SECTION_CHARS = 220 MAX_BODY_CHARS = 1400 SECTION_LABELS = { "personal": "Social", "political": "Political", "household": "Household", "network": "Network", } # ntfy renders these as icons next to the title. SECTION_TAGS = { "personal": "speech_balloon", "political": "newspaper", "household": "house", "network": "shield", } def _trim(text, limit): text = " ".join(str(text or "").split()) if len(text) <= limit: return text # Cut at a sentence end when there is one in reach, so the push doesn't stop # mid-clause; otherwise at a word boundary. window = text[:limit] for stop in (". ", "! ", "? "): cut = window.rfind(stop) if cut > limit * 0.5: return window[: cut + 1].strip() cut = window.rfind(" ") return (window[:cut] if cut > 0 else window).rstrip() + "…" def _todo_count(document): """How many political to-dos this run produced, or None if there is no todo window. Counted from the window's own content rather than inferred from prose — a number in a notification has to be a number the digest actually contains.""" for window in document.get("windows") or []: if str(window.get("id")) != "political-todo": continue content = window.get("content") if isinstance(content, list): return len(content) # A todo window that came back as prose still means "there are some". return 1 if str(content or "").strip() else 0 return None def _section_lines(documents, sections): """One line per section, in the order the canvas lays them out.""" by_section = { document.get("section"): document for document in documents.get("compact") or [] if isinstance(document, dict) } lines = {} for section in sections: document = by_section.get(section) if not document: continue label = SECTION_LABELS.get(section, section.title()) if document.get("withheld"): lines[section] = f"{label}: withheld — nothing could be verified against its sources." continue text = _trim(document.get("narration"), MAX_SECTION_CHARS) if not text: # A section with no narration still ran; saying so is more useful than # leaving a gap the reader has to interpret. text = "generated, no narration." if document.get("degraded"): text = f"unavailable — {text}" elif document.get("unverified"): text = f"[unverified] {text}" if section == "political": todos = _todo_count(document) if todos: text = f"{todos} todo{'s' if todos != 1 else ''}. {text}" lines[section] = f"{label}: {text}" return lines def _click_url(person_name): """Where the notification takes you when tapped: this person's own digest.""" base = os.environ.get("DIGEST_WEB_URL", "").strip().rstrip("/") if not base: return None if not person_name: return f"{base}/full.html" return f"{base}/full.html?{urlencode({'person': person_name})}" def _audiences(prefs, sections, default_topic): """topic -> (sections that topic may see, the names behind it). Grouping rather than one message per person: without it a household where nobody has set a personal topic would get one identical push per registered person, four times a day, which is how a useful notification becomes one people turn off. """ if prefs is None: return {default_topic: (list(sections), [])} if default_topic else {} grouped = {} for person in prefs["people"]: topic = (person.get("notify_topic") or default_topic or "").strip() wanted = [section for section in sections if section in person["digest_sections"]] if not topic: LOG.info("notify: %s has no ntfy topic and no household default, skipping", person["name"]) continue if not wanted: continue visible, names = grouped.setdefault(topic, (set(), [])) visible.update(wanted) names.append(person["name"]) return { topic: ([section for section in sections if section in visible], names) for topic, (visible, names) in grouped.items() } def _post(base_url, topic, title, body, tags, click): """Published as JSON to ntfy's root rather than as text to /. The header-based form (`Title:`, `Tags:`) is what `chores` uses and is fine for its plain-English nudges, but HTTP headers are latin-1: a title or a name with an em dash, an umlaut or a euro sign in it raises UnicodeEncodeError before the request is even sent. This digest quotes news headlines and household member names, so that would have failed on real content and worked on every test that used ASCII. The JSON body is UTF-8 throughout. """ payload = {"topic": topic, "title": title, "message": body} if tags: payload["tags"] = tags if click: payload["click"] = click response = requests.post(base_url, json=payload, timeout=HTTP_TIMEOUT) response.raise_for_status() def send(context, documents, sections, prefs): """Pushes one summary per audience. Never raises; returns how many went out.""" if os.environ.get("ENABLE_DIGEST_NOTIFY", "true").strip().lower() != "true": return 0 base_url = os.environ.get("NTFY_URL", "").strip().rstrip("/") if not base_url: LOG.info("notify: NTFY_URL is not set, not pushing a digest notification") return 0 default_topic = os.environ.get("NTFY_TOPIC", "").strip() audiences = _audiences(prefs, sections, default_topic) if not audiences: LOG.info("notify: nobody to notify about this run") return 0 lines = _section_lines(documents, sections) slot = str(context.get("slot_hour", "")).zfill(2) sent = 0 for topic, (visible, names) in audiences.items(): body = "\n".join(lines[section] for section in visible if section in lines) if not body: continue if len(body) > MAX_BODY_CHARS: body = body[:MAX_BODY_CHARS].rstrip() + "…" # One name in the click-through: a shared topic has no single owner, so it # gets the unfiltered page rather than an arbitrary person's. click = _click_url(names[0] if len(names) == 1 else None) labels = [SECTION_LABELS.get(section) or str(section) for section in visible] title = f"Digest {slot}:00 — {', '.join(labels)}" tags = [SECTION_TAGS[section] for section in visible if section in SECTION_TAGS] try: _post(base_url, topic, title, body, tags, click) sent += 1 LOG.info("notify: pushed the digest summary to %r (%s)", topic, ", ".join(visible)) except Exception: LOG.warning("notify: could not push to %r", topic, exc_info=True) return sent