SmartestHome/trash-calendar/sync.py

"""trash-calendar — mirrors the household's trash/recycling collection dates into
the same Nextcloud calendar everything else already uses, from
docs/project-plan.md Phase 19.

THE PROBLEM: Kennelbach's own Abfallkalender (kennelbach.at) and Vorarlberg's
Umweltverband (umweltv.at) both publish collection dates as a **personal ICS feed**
keyed to your street/house number — real, live infrastructure, not something this
repo re-implements. What's missing is getting those dates into the SAME calendar the
household already looks at (Nextcloud, via `digest-engine/ingest/caldav.py`'s
existing read path) instead of a second place nobody checks.

Oneshot script + systemd timer (daily), not an always-on service — same shape as
`digest-engine/run.py`, and deliberately not folded into digest-engine itself: this
writes to the calendar, digest-engine's own `ingest/caldav.py` is READ-ONLY by
explicit invariant (see that file's docstring), and mixing a write path into a
component whose one documented guarantee is "never mutates" is exactly the kind of
scope creep worth a separate, smaller tool instead.

CREDENTIALS: reuses digest-engine's own CALDAV_URL/CALDAV_USERNAME/CALDAV_PASSWORD/
CALDAV_VERIFY_TLS names and the same "Nextcloud app password, not the account
password" reasoning (see `digest-engine/ingest/caldav.py`'s docstring) — one
Nextcloud app password, shared by both the read and write paths, not two separate
credentials to manage. `CALDAV_TARGET_CALENDAR` is new: which calendar (by display
name) this writes into — unlike the read path, which can read several, this writes
to exactly one, on purpose (never guess which of several calendars a light
household chore belongs in).

OWNERSHIP INVARIANT: every event this script creates gets a UID prefixed
`smartesthome-trash-`, deterministic from the source feed's own event content. It
only ever creates events under that prefix and only ever checks for their existence
before creating — it never reads, modifies, or deletes anything else in the target
calendar. A no-op re-run (nothing new in the source feed) touches nothing.
"""

from __future__ import annotations

import hashlib
import logging
import os
import sys
import urllib.request
from datetime import date, datetime, timedelta, timezone

LOG = logging.getLogger("trash-calendar")

UID_PREFIX = "smartesthome-trash-"
DEFAULT_LOOKAHEAD_DAYS = 60


def _stable_uid(summary: str, start: date) -> str:
    digest = hashlib.sha1(f"{summary}|{start.isoformat()}".encode("utf-8")).hexdigest()[:16]
    return f"{UID_PREFIX}{digest}@smartesthome"


def fetch_source_events(ics_url: str, lookahead_days: int) -> list[dict]:
    """Downloads and parses the household's personal collection-date ICS feed.
    Degrades to an empty list (never raises past this point) if the feed is
    unreachable or malformed — a missed sync run just means tomorrow's run tries
    again, same "degrade, don't blank" rule as every renderer in this project.
    """
    from icalendar import Calendar as ICalendar  # see module docstring: caldav.py's identical import-inside-function reasoning

    try:
        req = urllib.request.Request(ics_url, headers={"User-Agent": "smartesthome-trash-calendar/1"})
        with urllib.request.urlopen(req, timeout=30) as resp:
            raw = resp.read()
    except Exception:
        LOG.warning("trash-calendar: could not fetch WASTE_ICS_URL", exc_info=True)
        return []

    today = date.today()
    horizon = today + timedelta(days=lookahead_days)

    events = []
    try:
        for component in ICalendar.from_ical(raw).walk("VEVENT"):
            dtstart = component.get("dtstart")
            if dtstart is None:
                continue
            start = dtstart.dt
            start_date = start.date() if isinstance(start, datetime) else start
            if not (today <= start_date <= horizon):
                continue
            summary = str(component.get("summary") or "Müllabfuhr").strip()
            events.append({"summary": summary, "start": start_date})
    except Exception:
        LOG.warning("trash-calendar: could not parse WASTE_ICS_URL as iCalendar", exc_info=True)
        return []

    LOG.info("trash-calendar: %d collection date(s) in the next %d days", len(events), lookahead_days)
    return events


def sync_to_caldav(events: list[dict]) -> None:
    url = os.environ.get("CALDAV_URL", "").strip()
    username = os.environ.get("CALDAV_USERNAME", "").strip()
    password = os.environ.get("CALDAV_PASSWORD", "")
    target_name = os.environ.get("CALDAV_TARGET_CALENDAR", "").strip()

    if not (url and username and password and target_name):
        LOG.error(
            "trash-calendar: CALDAV_URL/CALDAV_USERNAME/CALDAV_PASSWORD/CALDAV_TARGET_CALENDAR "
            "must all be set — see trash-calendar.env.example"
        )
        return

    verify = os.environ.get("CALDAV_VERIFY_TLS", "true").strip().lower() == "true"
    if not verify:
        LOG.warning("trash-calendar: CALDAV_VERIFY_TLS is false, the app password is sent over an unverified session")

    import caldav as caldav_lib

    with caldav_lib.DAVClient(url=url, username=username, password=password, ssl_verify_cert=verify) as client:
        target = None
        for calendar in client.principal().calendars():
            try:
                name = str(calendar.name or "")
            except Exception:
                continue
            if name == target_name:
                target = calendar
                break

        if target is None:
            LOG.error("trash-calendar: no calendar named %r found on this Nextcloud account", target_name)
            return

        created = 0
        for event in events:
            uid = _stable_uid(event["summary"], event["start"])
            try:
                target.event_by_uid(uid)
                continue  # already exists — this script never updates, only creates (see module docstring)
            except Exception:
                pass  # not found — fall through and create it

            ics = (
                "BEGIN:VCALENDAR\r\n"
                "VERSION:2.0\r\n"
                "PRODID:-//SmartestHome//trash-calendar//EN\r\n"
                "BEGIN:VEVENT\r\n"
                f"UID:{uid}\r\n"
                f"DTSTAMP:{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}\r\n"
                f"DTSTART;VALUE=DATE:{event['start'].strftime('%Y%m%d')}\r\n"
                f"SUMMARY:{event['summary']}\r\n"
                "END:VEVENT\r\n"
                "END:VCALENDAR\r\n"
            )
            try:
                target.save_event(ics)
                created += 1
            except Exception:
                LOG.warning("trash-calendar: could not create event for %s on %s", event["summary"], event["start"], exc_info=True)

        LOG.info("trash-calendar: created %d new event(s) in %r (existing ones left untouched)", created, target_name)


def main() -> int:
    logging.basicConfig(
        level=os.environ.get("LOG_LEVEL", "INFO").upper(),
        format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
    )

    ics_url = os.environ.get("WASTE_ICS_URL", "").strip()
    if not ics_url:
        LOG.error("trash-calendar: WASTE_ICS_URL is not set — see trash-calendar.env.example and README.md")
        return 1

    lookahead_days = int(os.environ.get("WASTE_LOOKAHEAD_DAYS") or DEFAULT_LOOKAHEAD_DAYS)

    events = fetch_source_events(ics_url, lookahead_days)
    if not events:
        LOG.info("trash-calendar: nothing to sync this run")
        return 0

    sync_to_caldav(events)
    return 0


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