home_control/lib/_cloud_logger.py

"""Send controller events to the cloud event logger."""

import logging
from datetime import datetime, timezone
from pprint import pprint
import requests


LOG = logging.getLogger("home_control.cloud_logger")


def _filter_values(config: dict, mode: str, field: str) -> set[str]:
    value = config.get("filter", {}).get(mode, {}).get(field, [])
    if not isinstance(value, list) or any(not isinstance(item, str) for item in value):
        raise ValueError(
            f"cloud_logger.filter.{mode}.{field} must be a list of strings"
        )
    return set(value)


def accepts_event(config: dict, message: dict) -> bool:
    """Return whether an event passes configured event and ID filters."""
    accepted_events = _filter_values(config, "accept", "events")
    accepted_ids = _filter_values(config, "accept", "ids")
    ignored_events = _filter_values(config, "ignore", "events")
    ignored_ids = _filter_values(config, "ignore", "ids")

    if message["event"] in ignored_events or message["id"] in ignored_ids:
        return False
    if accepted_events and message["event"] not in accepted_events:
        return False
    if accepted_ids and message["id"] not in accepted_ids:
        return False
    return True


def send_event(message: dict, event_url: str, timeout: float = 10) -> None:
    """POST an event dictionary as JSON to the configured logger endpoint."""
    if not isinstance(message, dict):
        raise TypeError("message must be a dictionary")
    if not event_url:
        raise ValueError("event_url is required")

    response = requests.post(
        event_url,
        json=message,
        timeout=timeout,
        allow_redirects=False,
    )
    if response.is_redirect:
        raise requests.HTTPError(
            f"event endpoint redirected to {response.headers.get('Location', '')}",
            response=response,
        )
    response.raise_for_status()


def on_event(config: dict, message: dict) -> None:
    """Controller action that forwards event data to the cloud logger."""
    if not config.get("enabled", True):
        return

    try:
        if not accepts_event(config, message):
            LOG.debug(
                "cloud event ignored by filter: event=%s id=%s",
                message["event"],
                message["id"],
            )
            return

        timestamp = datetime.now(timezone.utc).isoformat()
        values = {**message, "datetime": timestamp}
        messages = config.get("messages", {})
        template = config.get(
            message["event"],
            config.get("message", "{datetime} {sender}: {event} ({id})"),
        )
        payload = {
            **message,
            "datetime": timestamp,
            "text": message.get("text") or str(template).format(**values),
        }
        send_event(
            payload,
            event_url=str(config.get("event_url", "")),
            timeout=float(config.get("timeout", 10)),
        )
    except (TypeError, ValueError, requests.RequestException) as error:
        LOG.error("cloud event logging failed: %s", error)