| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243 |
- # -*- coding: utf-8 -*-
- """
- Satellite-Calendar Mapper.
- Ordnet Satellites den konfigurierten Kalendern zu.
- Unterstuetzt Matching per Alias und Room-ID.
- """
- from trixy_core.utils.debug import pdebug
- from plugins.calendar.models import CalendarInfo
- class SatelliteCalendarMapper:
- """
- Verwaltet die Zuordnung zwischen Satellites und Kalendern.
- Matching-Logik:
- 1. Satellite-Alias oder Room-ID gegen konfigurierte Eintraege pruefen
- 2. Globale Kalender werden immer hinzugefuegt
- 3. Standalone-Modus: Fallback auf ersten Eintrag
- """
- def __init__(self, config: dict, application: object) -> None:
- """
- Args:
- config: Plugin-Konfiguration (komplett)
- application: Application-Instanz
- """
- self._config = config
- self._application = application
- self._satellite_mapping = config.get("satellite_mapping", {})
- self._calendars_config = config.get("calendars", {})
- self._global_calendars = self._satellite_mapping.get("global_calendars", [])
- self._entries = self._satellite_mapping.get("satellites", [])
- def get_calendars_for_satellite(self, satellite_id: str) -> list[CalendarInfo]:
- """
- Gibt alle Kalender fuer einen Satellite zurueck.
- Args:
- satellite_id: Satellite-ID
- Returns:
- Liste von CalendarInfo (persoenliche + globale Kalender)
- """
- entry = self._find_entry_for_satellite(satellite_id)
- calendar_keys: list[str] = []
- # Persoenliche Kalender
- if entry:
- calendar_keys.extend(entry.get("calendars", []))
- # Globale Kalender
- for key in self._global_calendars:
- if key not in calendar_keys:
- calendar_keys.append(key)
- return self._keys_to_infos(calendar_keys)
- def get_write_calendar(self, satellite_id: str) -> CalendarInfo | None:
- """
- Gibt den Standard-Schreib-Kalender fuer einen Satellite zurueck.
- Args:
- satellite_id: Satellite-ID
- Returns:
- CalendarInfo oder None
- """
- entry = self._find_entry_for_satellite(satellite_id)
- if not entry:
- return None
- write_key = entry.get("default_write_calendar", "")
- if not write_key:
- return None
- cal_config = self._calendars_config.get(write_key, {})
- if not cal_config:
- return None
- if cal_config.get("readonly", True):
- return None
- return CalendarInfo(
- calendar_id=cal_config.get("calendar_id", ""),
- provider_name=cal_config.get("provider", ""),
- display_name=cal_config.get("display_name", write_key),
- readonly=False,
- )
- def get_person_name(self, satellite_id: str) -> str:
- """
- Gibt den Personennamen fuer einen Satellite zurueck.
- Args:
- satellite_id: Satellite-ID
- Returns:
- Personenname oder leerer String
- """
- entry = self._find_entry_for_satellite(satellite_id)
- if entry:
- return entry.get("person_name", "")
- return ""
- def get_satellites_for_calendar(self, calendar_key: str) -> list[str]:
- """
- Gibt alle Satellite-IDs zurueck, die einen bestimmten Kalender sehen.
- Args:
- calendar_key: Kalender-Schluessel
- Returns:
- Liste von Satellite-IDs
- """
- satellite_ids: list[str] = []
- # Globaler Kalender: Alle Satellites
- if calendar_key in self._global_calendars:
- return self._get_all_satellite_ids()
- # Per Entry pruefen
- for entry in self._entries:
- if calendar_key in entry.get("calendars", []):
- ids = self._get_satellite_ids_for_entry(entry)
- satellite_ids.extend(ids)
- return satellite_ids
- def find_person_calendars(self, person_name: str) -> list[str]:
- """
- Findet Kalender-Keys fuer eine Person (z.B. "Wann hat André Zeit?").
- Args:
- person_name: Personenname
- Returns:
- Liste von Kalender-Keys
- """
- person_lower = person_name.lower()
- for entry in self._entries:
- entry_person = entry.get("person_name", "").lower()
- if entry_person == person_lower:
- return entry.get("calendars", [])
- return []
- def get_calendar_config(self, calendar_key: str) -> dict:
- """
- Gibt die Konfiguration eines Kalenders zurueck.
- Args:
- calendar_key: Kalender-Schluessel
- Returns:
- Kalender-Konfiguration
- """
- return self._calendars_config.get(calendar_key, {})
- # =========================================================================
- # Private Methoden
- # =========================================================================
- def _find_entry_for_satellite(self, satellite_id: str) -> dict | None:
- """Findet den Mapping-Eintrag fuer einen Satellite."""
- # Satellite-Daten holen
- satellites = getattr(self._application, "satellites", None)
- if not satellites:
- # Standalone-Modus: Ersten Eintrag als Fallback verwenden
- if self._entries:
- return self._entries[0]
- return None
- satellite = satellites.get(satellite_id)
- if not satellite:
- return None
- sat_alias = getattr(satellite, "alias", "").lower()
- sat_room = getattr(satellite, "room_id", "").lower()
- for entry in self._entries:
- # Alias-Matching
- alias_list = [a.strip().lower() for a in entry.get("alias", "").split(",")]
- if sat_alias and sat_alias in alias_list:
- return entry
- # Room-Matching
- room_list = [r.strip().lower() for r in entry.get("room", "").split(",")]
- if sat_room and sat_room in room_list:
- return entry
- return None
- def _get_satellite_ids_for_entry(self, entry: dict) -> list[str]:
- """Gibt alle Satellite-IDs zurueck, die zu einem Eintrag passen."""
- satellites = getattr(self._application, "satellites", None)
- if not satellites:
- return []
- ids: list[str] = []
- alias_list = [a.strip().lower() for a in entry.get("alias", "").split(",")]
- room_list = [r.strip().lower() for r in entry.get("room", "").split(",")]
- for sat in satellites:
- sat_alias = getattr(sat, "alias", "").lower()
- sat_room = getattr(sat, "room_id", "").lower()
- if (sat_alias and sat_alias in alias_list) or (sat_room and sat_room in room_list):
- ids.append(sat.id)
- return ids
- def _get_all_satellite_ids(self) -> list[str]:
- """Gibt alle verbundenen Satellite-IDs zurueck."""
- satellites = getattr(self._application, "satellites", None)
- if not satellites:
- return []
- return [sat.id for sat in satellites if sat.is_connected]
- def _keys_to_infos(self, calendar_keys: list[str]) -> list[CalendarInfo]:
- """Konvertiert Kalender-Keys in CalendarInfo-Objekte."""
- infos: list[CalendarInfo] = []
- for key in calendar_keys:
- cal_config = self._calendars_config.get(key, {})
- if not cal_config:
- continue
- infos.append(CalendarInfo(
- calendar_id=cal_config.get("calendar_id", ""),
- provider_name=cal_config.get("provider", ""),
- display_name=cal_config.get("display_name", key),
- readonly=cal_config.get("readonly", True),
- is_global=key in self._global_calendars,
- ))
- return infos
|