"""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())