330 lines
13 KiB
Python
330 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""digest-engine entrypoint — one digest run, then exit.
|
|
|
|
Invoked as `docker compose run --rm digest-engine` by the smart-home-digest.timer
|
|
systemd timer, 4x/day. It is deliberately a oneshot, not a daemon: scheduling
|
|
lives in systemd, exactly like the restic backup job in
|
|
hosts/container-host/scripts/setup-container-host.sh.
|
|
|
|
Every ingestion source is independently toggleable and independently fallible. A
|
|
source that is disabled, misconfigured, or simply broken (a stale Telegram
|
|
session, a signal-cli container that is down) logs a warning and contributes an
|
|
empty list — it must never take down the rest of the run, because a digest with
|
|
two of three sections is worth far more than no digest at all.
|
|
|
|
Everything here is read-only. See docs/project-plan.md Phase 12 step 8.
|
|
|
|
WHICH RUN IS THIS?
|
|
------------------
|
|
One feature — the evening recipe suggestion in synth/prompts/household.md — only
|
|
applies to one of the four daily runs. The slot is derived here from the local
|
|
wall clock rather than passed in by the caller: the systemd unit installed by
|
|
hosts/container-host/scripts/setup-container-host.sh runs a bare
|
|
`docker compose run --rm digest-engine` with no arguments, and a manual run uses
|
|
exactly the same command, so anything argument- or unit-based would have to be
|
|
threaded through both and would silently do the wrong thing on a hand-run digest.
|
|
The container already has the host's timezone (`TZ` plus a bind-mounted
|
|
/etc/localtime), which is the same clock systemd's `OnCalendar` fires against.
|
|
|
|
The run is attributed to the most recent DIGEST_SCHEDULE slot at or before now,
|
|
not to an exact hour match, because the timer is `Persistent=true`: a host that
|
|
was asleep at 18:00 fires the run late, and an exact match would drop the evening
|
|
feature precisely on the days the digest is read late.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from ingest import (
|
|
caldav as caldav_ingest,
|
|
discord_ingest,
|
|
email_imap,
|
|
financial,
|
|
flight_traffic,
|
|
grocy,
|
|
naval_traffic,
|
|
news_rss,
|
|
opnsense_ids,
|
|
signal_ingest,
|
|
telegram_ingest,
|
|
whatsapp_ingest,
|
|
)
|
|
from synth import llm_client
|
|
import viewed_tracker
|
|
|
|
LOG = logging.getLogger("digest")
|
|
|
|
# (env toggle, context key, module) — order is the order they run in.
|
|
SOURCES = (
|
|
("ENABLE_EMAIL_INGEST", "email", email_imap),
|
|
("ENABLE_SIGNAL_INGEST", "signal", signal_ingest),
|
|
("ENABLE_TELEGRAM_INGEST", "telegram", telegram_ingest),
|
|
("ENABLE_DISCORD_INGEST", "discord", discord_ingest),
|
|
("ENABLE_WHATSAPP_INGEST", "whatsapp", whatsapp_ingest),
|
|
("ENABLE_NEWS_INGEST", "news", news_rss),
|
|
("ENABLE_FINANCIAL_INGEST", "financial", financial),
|
|
("ENABLE_FLIGHT_TRAFFIC_INGEST", "flight_traffic", flight_traffic),
|
|
("ENABLE_NAVAL_TRAFFIC_INGEST", "naval_traffic", naval_traffic),
|
|
("ENABLE_OPNSENSE_IDS_INGEST", "opnsense_ids", opnsense_ids),
|
|
("ENABLE_CALDAV_INGEST", "calendar", caldav_ingest),
|
|
("ENABLE_GROCY_INGEST", "grocy", grocy),
|
|
)
|
|
|
|
PERSONAL_SOURCES = ("email", "signal", "telegram", "discord", "whatsapp")
|
|
POLITICAL_SOURCES = ("news", "financial", "email", "flight_traffic", "naval_traffic")
|
|
|
|
DEFAULT_SCHEDULE = "00,06,12,18"
|
|
DEFAULT_EVENING_HOUR = 18
|
|
|
|
|
|
def env_flag(name, default="false"):
|
|
return os.environ.get(name, default).strip().lower() == "true"
|
|
|
|
|
|
def schedule_hours():
|
|
hours = sorted(
|
|
{int(part.strip()) for part in os.environ.get("DIGEST_SCHEDULE", DEFAULT_SCHEDULE).split(",")
|
|
if part.strip().isdigit() and 0 <= int(part.strip()) <= 23}
|
|
)
|
|
if not hours:
|
|
LOG.warning("DIGEST_SCHEDULE is unusable, falling back to %s", DEFAULT_SCHEDULE)
|
|
return [int(part) for part in DEFAULT_SCHEDULE.split(",")]
|
|
return hours
|
|
|
|
|
|
def slot_hour(now_local, hours):
|
|
earlier = [hour for hour in hours if hour <= now_local.hour]
|
|
# Before the first slot of the day the run still belongs to yesterday's last one.
|
|
return earlier[-1] if earlier else hours[-1]
|
|
|
|
|
|
def resolve_slot():
|
|
now_local = datetime.now().astimezone()
|
|
hours = schedule_hours()
|
|
try:
|
|
evening_hour = int(os.environ.get("DIGEST_EVENING_HOUR", DEFAULT_EVENING_HOUR))
|
|
except ValueError:
|
|
# Unlike an ingestion module, this runs outside collect()'s backstop, and a
|
|
# typo in one env var must not take down a whole digest.
|
|
LOG.warning("DIGEST_EVENING_HOUR is not a number, falling back to %s", DEFAULT_EVENING_HOUR)
|
|
evening_hour = DEFAULT_EVENING_HOUR
|
|
|
|
if evening_hour not in hours:
|
|
LOG.warning(
|
|
"DIGEST_EVENING_HOUR=%s is not one of the DIGEST_SCHEDULE slots %s, "
|
|
"so no run will ever be the evening one",
|
|
evening_hour,
|
|
hours,
|
|
)
|
|
|
|
current = slot_hour(now_local, hours)
|
|
is_evening = current == evening_hour or env_flag("DIGEST_FORCE_EVENING")
|
|
LOG.info("run attributed to the %02d:00 slot (evening run: %s)", current, is_evening)
|
|
return current, evening_hour, is_evening
|
|
|
|
|
|
def collect(lookback_hours):
|
|
collected = {}
|
|
for toggle, key, module in SOURCES:
|
|
if not env_flag(toggle):
|
|
LOG.info("%s is not enabled, skipping %s ingestion", toggle, key)
|
|
collected[key] = []
|
|
continue
|
|
try:
|
|
collected[key] = module.fetch(lookback_hours) or []
|
|
except Exception:
|
|
# Modules already swallow their own failures; this is the backstop for
|
|
# anything they miss (an import-time error, an unexpected exception type).
|
|
LOG.warning("%s ingestion raised, continuing without it", key, exc_info=True)
|
|
collected[key] = []
|
|
return collected
|
|
|
|
|
|
def load_previous_run(output_dir):
|
|
"""Best-effort read of the run currently pointed at by output/latest — the
|
|
candidate for merging if it turns out not to have been viewed. Any failure (first
|
|
run ever, corrupt JSON, missing file) is silently "no previous run", which just
|
|
means no merge is attempted, never a crash."""
|
|
try:
|
|
return json.loads((output_dir / "latest" / "digest.json").read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return None
|
|
|
|
|
|
def should_merge(previous_run, previous_viewed_at):
|
|
if previous_run is None:
|
|
return False
|
|
# None (unknown) is handled identically to "viewed" — see viewed_tracker's
|
|
# last_viewed_at() docstring for why treating unknown as unviewed is the riskier
|
|
# default, not the safer one.
|
|
if previous_viewed_at is None:
|
|
return False
|
|
return previous_viewed_at < previous_run.get("generated_at", "")
|
|
|
|
|
|
def previous_section_document(previous_run, section):
|
|
"""The previous run's own rendered document for one section, at `full` detail —
|
|
the richest version, since what gets folded in is content, not layout, and the
|
|
merge instruction in synth/llm_client.py asks the model to write fresh prose at
|
|
whatever detail_level this pass actually is, not to reuse this verbatim."""
|
|
if not previous_run:
|
|
return None
|
|
for doc in (previous_run.get("sections") or {}).get("full") or []:
|
|
if isinstance(doc, dict) and doc.get("section") == section:
|
|
return doc
|
|
return None
|
|
|
|
|
|
def build_section_contexts(collected, lookback_hours, is_evening_run, previous_run=None):
|
|
# The full pantry is only ever needed to work out what the evening recipe still
|
|
# requires. On the other three runs it is a few hundred lines of prompt that
|
|
# buys nothing, so it is dropped rather than sent and then ignored.
|
|
grocy_entries = [
|
|
entry
|
|
for entry in collected.get("grocy", [])
|
|
if is_evening_run or entry.get("category") != "in_stock"
|
|
]
|
|
|
|
contexts = {
|
|
"personal": {
|
|
"lookback_hours": lookback_hours,
|
|
"messages": {key: collected.get(key, []) for key in PERSONAL_SOURCES},
|
|
},
|
|
"political": {
|
|
"lookback_hours": lookback_hours,
|
|
"news": collected.get("news", []),
|
|
"financial": collected.get("financial", []),
|
|
"mail": collected.get("email", []),
|
|
# Extra evidence for the existing political synthesis, not a fourth
|
|
# window type — see the traffic-data section of synth/prompts/political.md.
|
|
"flight_traffic": collected.get("flight_traffic", []),
|
|
"naval_traffic": collected.get("naval_traffic", []),
|
|
},
|
|
# Both keys stay present even when their source is off or broken, so the
|
|
# prompt sees the shape it is promised and says "nothing scheduled" instead
|
|
# of hallucinating an event.
|
|
"household": {
|
|
"lookback_hours": lookback_hours,
|
|
"calendar": collected.get("calendar", []),
|
|
"grocy": grocy_entries,
|
|
# Drives the evening-only recipe/shopping-list section of
|
|
# synth/prompts/household.md. Nothing is ever written back to Grocy.
|
|
"is_evening_run": is_evening_run,
|
|
# Home network status sits with the household, not the political
|
|
# section — it is a "something in this house needs your attention"
|
|
# item. See the network-security section of synth/prompts/household.md.
|
|
"network_security": collected.get("opnsense_ids", []),
|
|
},
|
|
}
|
|
|
|
if previous_run is not None:
|
|
for section, context in contexts.items():
|
|
doc = previous_section_document(previous_run, section)
|
|
if doc:
|
|
context["previous_unviewed_digest"] = {
|
|
"generated_at": doc.get("generated_at"),
|
|
"windows": doc.get("windows", []),
|
|
"narration": doc.get("narration", ""),
|
|
}
|
|
|
|
return contexts
|
|
|
|
|
|
def write_output(output_dir, run_id, context, documents):
|
|
run_dir = output_dir / run_id
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Persisted for the Phase 12 step 7 follow-up voice Q&A: a spoken follow-up
|
|
# re-queries Ollama against this cached context instead of re-ingesting.
|
|
(run_dir / "context.json").write_text(
|
|
json.dumps(context, indent=2, ensure_ascii=False, default=str), encoding="utf-8"
|
|
)
|
|
|
|
digest = {
|
|
"run_id": run_id,
|
|
"generated_at": context["generated_at"],
|
|
"sections": documents,
|
|
}
|
|
payload = json.dumps(digest, indent=2, ensure_ascii=False, default=str)
|
|
(run_dir / "digest.json").write_text(payload, encoding="utf-8")
|
|
|
|
# digest-web serves output/ read-only and has no idea which run is newest, so
|
|
# the pointer has to live in the served tree itself. latest.json holds the whole
|
|
# digest so a template needs exactly one fetch; the `latest` symlink is for
|
|
# humans poking around the volume.
|
|
(output_dir / "latest.json").write_text(payload, encoding="utf-8")
|
|
|
|
symlink = output_dir / "latest"
|
|
try:
|
|
if symlink.is_symlink() or symlink.exists():
|
|
symlink.unlink()
|
|
symlink.symlink_to(run_id, target_is_directory=True)
|
|
except OSError:
|
|
LOG.warning("could not update the output/latest symlink", exc_info=True)
|
|
|
|
return run_dir
|
|
|
|
|
|
def main():
|
|
logging.basicConfig(
|
|
level=os.environ.get("LOG_LEVEL", "INFO").upper(),
|
|
format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
|
|
)
|
|
|
|
lookback_hours = float(os.environ.get("DIGEST_LOOKBACK_HOURS", "6"))
|
|
output_dir = Path(os.environ.get("DIGEST_OUTPUT_DIR", "/output"))
|
|
started_at = datetime.now(timezone.utc)
|
|
run_id = started_at.strftime("%Y%m%dT%H%M%SZ")
|
|
|
|
LOG.info("digest run %s starting (lookback %sh)", run_id, lookback_hours)
|
|
|
|
current_slot, evening_hour, is_evening_run = resolve_slot()
|
|
collected = collect(lookback_hours)
|
|
|
|
previous_run = load_previous_run(output_dir)
|
|
previous_viewed_at = viewed_tracker.last_viewed_at()
|
|
merge_previous = should_merge(previous_run, previous_viewed_at)
|
|
LOG.info(
|
|
"previous run viewed_at=%s, previous run generated_at=%s -> merging: %s",
|
|
previous_viewed_at,
|
|
previous_run.get("generated_at") if previous_run else None,
|
|
merge_previous,
|
|
)
|
|
|
|
context = {
|
|
"run_id": run_id,
|
|
"generated_at": started_at.replace(microsecond=0).isoformat().replace("+00:00", "Z"),
|
|
"lookback_hours": lookback_hours,
|
|
"slot_hour": current_slot,
|
|
"evening_hour": evening_hour,
|
|
"is_evening_run": is_evening_run,
|
|
"enabled_sources": [key for toggle, key, _ in SOURCES if env_flag(toggle)],
|
|
"item_counts": {key: len(items) for key, items in collected.items()},
|
|
"sources": collected,
|
|
"merged_unviewed_previous_run": merge_previous,
|
|
}
|
|
|
|
section_contexts = build_section_contexts(
|
|
collected, lookback_hours, is_evening_run,
|
|
previous_run=previous_run if merge_previous else None,
|
|
)
|
|
documents = llm_client.generate_all(section_contexts)
|
|
|
|
run_dir = write_output(output_dir, run_id, context, documents)
|
|
|
|
LOG.info(
|
|
"digest run %s complete: %s -> %s",
|
|
run_id,
|
|
context["item_counts"],
|
|
run_dir,
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|