waste_api/app/collector.py

"""
Collector — wraps waste_collection_schedule library calls.

Key insight: icalevents has a timezone bug (offset-naive vs offset-aware datetime
comparison). We fix it by monkey-patching the library's ICS.convert() to use icalendar.
"""
import datetime
import logging
import sys
from pathlib import Path
from typing import Any, List

import requests

log = logging.getLogger(__name__)

# ── Library path setup ────────────────────────────────────────────────────────
WASTE_LIB_BASE = Path(__file__).resolve().parent.parent.parent
_WASTE_OUTER = WASTE_LIB_BASE / "waste_lib" / "custom_components" / "waste_collection_schedule"
_WASTE_INNER = _WASTE_OUTER / "waste_collection_schedule"
_PATCHED = False


def _ensure_path():
    global _PATCHED
    if str(_WASTE_INNER) not in sys.path:
        sys.path.insert(0, str(_WASTE_INNER))
        sys.path.insert(1, str(_WASTE_OUTER))
    if not _PATCHED:
        _patch_icalevents_bug()
        _PATCHED = True


def _patch_icalevents_bug():
    """
    Monkey-patch waste_collection_schedule.service.ICS.ICS.convert()
    to use icalendar instead of icalevents (which has a timezone bug).
    Returns List[Tuple[datetime.date, str]] as the original does.
    """
    import icalendar
    import jinja2

    from waste_collection_schedule.service.ICS import ICS as OriginalICS

    _original_convert = OriginalICS.convert

    def patched_convert(self, text: str) -> List[Any]:
        try:
            return _original_convert(self, text)
        except TypeError as ex:
            if "offset-naive" not in str(ex):
                raise
        log.debug("ICS.convert: icalevents bug detected, falling back to icalendar")

        cal = icalendar.Calendar.from_ical(text)
        now = datetime.datetime.now().date()
        end = now + datetime.timedelta(days=365)
        entries: List[Any] = []

        for component in cal.walk():
            if component.name != "VEVENT":
                continue
            dtstart = component.get("DTSTART")
            if dtstart is None:
                continue
            dt = dtstart.dt
            if isinstance(dt, datetime.datetime):
                dt = dt.date()
            if not isinstance(dt, datetime.date):
                continue
            if dt < now or dt > end:
                continue
            if self._offset:
                dt = dt + datetime.timedelta(days=self._offset)
            # Try SUMMARY first, then DESCRIPTION
            summary = str(component.get("SUMMARY", "") or component.get("DESCRIPTION", "") or "")
            if not summary and self._title_template:
                env = jinja2.Environment()
                title_tpl = env.from_string(self._title_template)
                summary = title_tpl.render(date=dt)
            entries.append((dt, summary))
        # Sort by date (icalevents returns sorted results)
        entries.sort(key=lambda x: x[0])
        return entries

    OriginalICS.convert = patched_convert
    log.info("Patched ICS.convert to use icalendar fallback")


# ── Collection mapping ────────────────────────────────────────────────────────
def _map_collection(wc) -> dict:
    return {
        "date": wc.date.isoformat(),
        "type": wc.get("type", "Unknown"),
        "daysUntil": (wc.date - datetime.datetime.now().date()).days,
        "icon": wc.icon,
    }


# ── ICS via icalendar (bypasses icalevents timezone bug) ─────────────────────
def _fetch_ics_via_icalendar(url: str, count: int) -> list[dict]:
    import icalendar

    resp = requests.get(url, timeout=15)
    resp.raise_for_status()
    cal = icalendar.Calendar.from_ical(resp.text)
    now = datetime.datetime.now().date()
    entries = []
    for component in cal.walk():
        if component.name != "VEVENT":
            continue
        dtstart = component.get("DTSTART")
        if dtstart is None:
            continue
        dt = dtstart.dt
        if isinstance(dt, datetime.datetime):
            dt = dt.date()
        if not isinstance(dt, datetime.date):
            continue
        if dt < now:
            continue
        summary = str(component.get("SUMMARY", "Unknown"))
        entries.append({
            "date": dt.isoformat(),
            "type": summary,
            "daysUntil": (dt - now).days,
            "icon": None,
        })
    entries.sort(key=lambda x: x["date"])
    return entries[:count]


# ── Public API ───────────────────────────────────────────────────────────────
def fetch_ics(url: str, count: int = 20) -> list[dict]:
    _ensure_path()
    from waste_collection_schedule.source.ics import Source as ICSSource

    source = ICSSource(url=url)
    entries = source.fetch()
    return [_map_collection(e) for e in entries[:count]]


def fetch_abfall_io(
    key: str, f_id_kommune: int, f_id_strasse: int, count: int
) -> list[dict]:
    _ensure_path()
    from waste_collection_schedule.source.abfall_io import Source as AbfallIOSource

    source = AbfallIOSource(
        key=key,
        f_id_kommune=f_id_kommune,
        f_id_strasse=f_id_strasse,
    )
    entries = source.fetch()
    return [_map_collection(e) for e in entries[:count]]


def fetch_awm_muenchen(street: str, house_number: str, count: int) -> list[dict]:
    _ensure_path()
    from waste_collection_schedule.source.awm_muenchen_de import Source as AWMSource

    source = AWMSource(street=street, house_number=house_number)
    entries = source.fetch()
    return [_map_collection(e) for e in entries[:count]]


def fetch_stuttgart(street: str, streetnr: int, count: int) -> list[dict]:
    _ensure_path()
    from waste_collection_schedule.source.stuttgart_de import Source

    source = Source(street=street, streetnr=streetnr)
    entries = source.fetch()
    return [_map_collection(e) for e in entries[:count]]