Coverage for custom_components/supernotify/target.py: 98%
337 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 14:29 +0000
1from __future__ import annotations
3import logging
4import re
5from dataclasses import dataclass
6from typing import TYPE_CHECKING, Any, ClassVar
8import voluptuous as vol
9from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
10from homeassistant.const import (
11 ATTR_AREA_ID,
12 ATTR_DEVICE_ID,
13 ATTR_ENTITY_ID,
14 ATTR_FLOOR_ID,
15 ATTR_LABEL_ID,
16)
17from homeassistant.core import valid_entity_id
18from homeassistant.helpers.redact import partial_redact
20from .common import ensure_list
21from .const import (
22 ATTR_EMAIL,
23 ATTR_MOBILE_APP_ID,
24 ATTR_PERSON_ID,
25 ATTR_PHONE,
26 RE_DEVICE_ID,
27 TARGET_CATEGORY_VALUES,
28)
29from .schema import phone
31if TYPE_CHECKING:
32 from collections.abc import Collection, Sequence
34 from .hass_api import HomeAssistantAPI
35 from .model import SelectionRule
38_LOGGER = logging.getLogger(__name__)
40# Avoid importing this from HA component to dodge heavy imports until lazyimport available
41MOBILE_APP_DOMAIN = "mobile_app"
44@dataclass(frozen=True)
45class TargetEntityCategory:
46 """Declares which entities a transport accepts for the `entity_id` target category.
48 Used in a `Transport.target_categories` list in place of a plain category name, since
49 "any entity_id" is too broad - most entity-based transports only want entities of a
50 particular domain (or domains), or ones registered by a particular platform/integration
51 (or both). `domain`/`platform` may each be a single value or a list; `None` means
52 "don't filter on this".
53 """
55 domain: str | list[str] | None = None
56 platform: str | list[str] | None = None
58 def matches(self, entity_id: str, hass_api: HomeAssistantAPI) -> bool:
59 if self.domain is not None:
60 domains = [self.domain] if isinstance(self.domain, str) else self.domain
61 if entity_id.split(".", 1)[0] not in domains:
62 return False
63 # only look the platform up, in the entity registry, when there's a platform to check against
64 if self.platform is not None:
65 platforms = [self.platform] if isinstance(self.platform, str) else self.platform
66 if hass_api.platform_for_entity(entity_id) not in platforms:
67 return False
68 return True
71class Target:
72 # actual targets, that can positively identified with a validator
73 DIRECT_CATEGORIES: ClassVar[list[str]] = [ATTR_ENTITY_ID, ATTR_DEVICE_ID, ATTR_EMAIL, ATTR_PHONE, ATTR_MOBILE_APP_ID]
74 # references that lead to targets, that can positively identified with a validator
75 AUTO_INDIRECT_CATEGORIES: ClassVar[list[str]] = [ATTR_PERSON_ID]
76 # references that lead to targets, that can't be positively identified with a validator
77 EXPLICIT_INDIRECT_CATEGORIES: ClassVar[list[str]] = [ATTR_AREA_ID, ATTR_FLOOR_ID, ATTR_LABEL_ID]
78 INDIRECT_CATEGORIES = EXPLICIT_INDIRECT_CATEGORIES + AUTO_INDIRECT_CATEGORIES
79 AUTO_CATEGORIES = DIRECT_CATEGORIES + AUTO_INDIRECT_CATEGORIES
81 CATEGORIES = DIRECT_CATEGORIES + INDIRECT_CATEGORIES
83 UNKNOWN_CUSTOM_CATEGORY = "_UNKNOWN_"
85 def __init__(
86 self,
87 target: str
88 | list[str]
89 | dict[str, str]
90 | dict[str, Sequence[str]]
91 | dict[str, list[str]]
92 | dict[str, str | list[str]]
93 | None = None,
94 target_data: dict[str, Any] | None = None,
95 target_specific_data: bool = False,
96 ) -> None:
97 self.target_data: dict[str, Any] | None = None
98 self.target_specific_data: dict[tuple[str, str], dict[str, Any]] | None = None
99 self.targets: dict[str, list[str]] = {}
101 matched: list[str]
103 if isinstance(target, str):
104 target = [target]
106 if target is None:
107 pass # empty constructor is valid case for target building
108 elif isinstance(target, list):
109 # simplified and legacy way of assuming list of entities that can be discriminated by validator
110 targets = list(target)
111 # "<category>:<value>" entries scope a custom target without needing the full
112 # mapping form - only recognised here, in the flat list/scalar form, so the mapping
113 # form (`target: {category: value}`) stays available verbatim for any value that
114 # collides with this syntax (e.g. one already containing a colon). A prefix is always
115 # a target *category* (e.g. `topic:`, `email:`), never a transport or delivery name
116 # directly - those are dynamic and only addressable via the mapping form.
117 unprefixed: list[str] = []
118 for t in targets:
119 prefix, sep, rest = t.partition(":") if isinstance(t, str) else ("", "", "")
120 if sep and rest and prefix in TARGET_CATEGORY_VALUES:
121 self.targets.setdefault(prefix, [])
122 if rest not in self.targets[prefix]:
123 self.targets[prefix].append(rest)
124 else:
125 unprefixed.append(t)
126 targets = unprefixed
127 for category in self.AUTO_CATEGORIES:
128 matched = self._filter_by_category(category, targets)
129 if matched:
130 self.targets.setdefault(category, [])
131 self.targets[category].extend([t for t in matched if t not in self.targets[category]])
132 targets = [t for t in targets if t not in matched]
133 if not targets:
134 break
135 if targets:
136 self.targets[self.UNKNOWN_CUSTOM_CATEGORY] = targets
138 elif isinstance(target, dict):
139 for category in target:
140 targets = ensure_list(target[category])
141 if not targets:
142 continue
143 if category in self.AUTO_CATEGORIES:
144 matched = self._filter_by_category(category, targets)
145 if matched:
146 self.targets.setdefault(category, [])
147 self.targets[category].extend([t for t in matched if t not in self.targets[category]])
149 elif category in self.CATEGORIES:
150 # categories that can't be automatically detected, like label_id
151 self.targets[category] = targets
152 else:
153 # custom categories
154 self.targets[category] = targets
155 else:
156 _LOGGER.warning("SUPERNOTIFY Target created with no valid targets: %s", target)
158 if target_data and target_specific_data:
159 self.target_specific_data = {}
160 for category, targets in self.targets.items():
161 for t in targets:
162 self.target_specific_data[category, t] = target_data
163 if target_data and not target_specific_data:
164 self.target_data = target_data
166 def _filter_by_category(self, category: str, candidates: list[str]) -> list[str]:
167 matched: list[str] = []
168 validator = getattr(self, f"is_{category}", None)
169 if validator is not None:
170 for t in candidates:
171 if t not in matched and validator(t):
172 matched.append(t)
173 else:
174 _LOGGER.debug("SUPERNOTIFY Missing validator for selective target category %s", category)
175 return matched
177 # Targets by category
179 @property
180 def email(self) -> list[str]:
181 return self.targets.get(ATTR_EMAIL, [])
183 @property
184 def entity_ids(self) -> list[str]:
185 return self.targets.get(ATTR_ENTITY_ID, [])
187 @property
188 def person_ids(self) -> list[str]:
189 return self.targets.get(ATTR_PERSON_ID, [])
191 @property
192 def device_ids(self) -> list[str]:
193 return self.targets.get(ATTR_DEVICE_ID, [])
195 @property
196 def phone(self) -> list[str]:
197 return self.targets.get(ATTR_PHONE, [])
199 @property
200 def mobile_app_ids(self) -> list[str]:
201 return self.targets.get(ATTR_MOBILE_APP_ID, [])
203 def domain_entity_ids(self, domain: str | None) -> list[str]:
204 return [t for t in self.targets.get(ATTR_ENTITY_ID, []) if domain is not None and t and t.startswith(f"{domain}.")]
206 def custom_ids(self, category: str) -> list[str]:
207 return self.targets.get(category, []) if category not in self.CATEGORIES else []
209 @property
210 def area_ids(self) -> list[str]:
211 return self.targets.get(ATTR_AREA_ID, [])
213 @property
214 def floor_ids(self) -> list[str]:
215 return self.targets.get(ATTR_FLOOR_ID, [])
217 @property
218 def label_ids(self) -> list[str]:
219 return self.targets.get(ATTR_LABEL_ID, [])
221 # Selectors / validators
223 @classmethod
224 def is_device_id(cls, target: str) -> bool:
225 return re.fullmatch(RE_DEVICE_ID, target) is not None
227 @classmethod
228 def is_entity_id(cls, target: str) -> bool:
229 return valid_entity_id(target) and not target.startswith(("person.", "user."))
231 @classmethod
232 def is_person_id(cls, target: str) -> bool:
233 """True for a real Person entity_id, or a Recipient's synthetic `user.<name>` id - used
234 to identify a recipient with no Person record (see people.Recipient.entity_id). Both are
235 `person_id`-category target values, resolved the same way downstream."""
236 return target.startswith(("person.", "user.")) and valid_entity_id(target)
238 @classmethod
239 def is_phone(cls, target: str) -> bool:
240 try:
241 return phone(target) is not None
242 except vol.Invalid:
243 return False
245 @classmethod
246 def is_mobile_app_id(cls, target: str) -> bool:
247 return not valid_entity_id(target) and target.startswith(f"{MOBILE_APP_DOMAIN}_")
249 @classmethod
250 def is_notify_entity(cls, target: str) -> bool:
251 return valid_entity_id(target) and target.startswith(f"{NOTIFY_DOMAIN}.")
253 @classmethod
254 def is_email(cls, target: str) -> bool:
255 try:
256 return vol.Email()(target) is not None # type: ignore[call-arg] # ty: ignore[missing-argument]
257 except vol.Invalid:
258 return False
260 def has_targets(self) -> bool:
261 return any(targets for targets in self.targets.values())
263 def has_resolved_target(self) -> bool:
264 return any(targets for category, targets in self.targets.items() if category not in self.INDIRECT_CATEGORIES)
266 def has_unknown_targets(self) -> bool:
267 return len(self.targets.get(self.UNKNOWN_CUSTOM_CATEGORY, [])) > 0
269 def for_category(self, category: str) -> list[str]:
270 return self.targets.get(category, [])
272 def resolved_targets(self) -> list[str]:
273 result: list[str] = []
274 for category, targets in self.targets.items():
275 if category not in self.INDIRECT_CATEGORIES:
276 result.extend(targets)
277 return result
279 def hash_resolved(self) -> int:
280 targets = []
281 for category in self.targets:
282 if category not in self.INDIRECT_CATEGORIES:
283 targets.extend(self.targets[category])
284 return hash(tuple(targets))
286 @property
287 def direct_categories(self) -> list[str]:
288 return self.DIRECT_CATEGORIES + [cat for cat in self.targets if cat not in self.CATEGORIES]
290 def direct(self) -> Target:
291 t = Target(
292 {cat: targets for cat, targets in self.targets.items() if cat in self.direct_categories},
293 target_data=self.target_data,
294 )
295 if self.target_specific_data:
296 t.target_specific_data = {k: v for k, v in self.target_specific_data.items() if k[0] in self.direct_categories}
297 return t
299 def extend(self, category: str, targets: list[str] | str) -> None:
300 targets = ensure_list(targets)
301 self.targets.setdefault(category, [])
302 self.targets[category].extend(t for t in targets if t not in self.targets[category])
304 def remove(self, category: str, targets: list[str] | str) -> None:
305 targets = ensure_list(targets)
306 if category in self.targets:
307 self.targets[category] = [t for t in self.targets[category] if t not in targets]
309 def safe_copy(self) -> Target:
310 t = Target(dict(self.targets), target_data=dict(self.target_data) if self.target_data else None)
311 t.target_specific_data = dict(self.target_specific_data) if self.target_specific_data else None
312 return t
314 def resolve_selectors(self, hass_api: HomeAssistantAPI) -> Target:
315 """Replace `area_id`/`floor_id`/`label_id` targets with the entities they reference
317 Resolution goes through the same core helper as a Home Assistant entity action, so
318 groups are expanded and an entity inherits the area of its device. An entity in more
319 than one of them - the kitchen, the first floor and the `voice` label - is kept once,
320 and data attached to a selector is inherited by each of its entities, the same way as
321 for a group member. Transports therefore never see a selector, and only the exception
322 of an action that genuinely knows about areas needs `extra_data`.
324 Returns this target untouched when there is no selector to resolve.
325 """
326 if not any(self.targets.get(category) for category in self.EXPLICIT_INDIRECT_CATEGORIES):
327 return self
328 kwarg: dict[str, str] = {ATTR_AREA_ID: "area_ids", ATTR_FLOOR_ID: "floor_ids", ATTR_LABEL_ID: "label_ids"}
329 targets: dict[str, list[str]] = {
330 category: list(values)
331 for category, values in self.targets.items()
332 if category not in self.EXPLICIT_INDIRECT_CATEGORIES
333 }
334 entity_ids: list[str] = targets.setdefault(ATTR_ENTITY_ID, [])
335 inherited: dict[tuple[str, str], dict[str, Any]] = {}
336 unknown: list[str] = []
337 for category in self.EXPLICIT_INDIRECT_CATEGORIES:
338 for value in self.targets.get(category, []):
339 # one selector at a time, so data attached to it follows its own entities
340 resolution = hass_api.resolve_target_selectors(**{kwarg[category]: [value]})
341 if resolution.has_missing():
342 unknown.append(f"{category}:{value}")
343 data: dict[str, Any] | None = (self.target_specific_data or {}).get((category, value))
344 for entity_id in resolution.entity_ids:
345 if entity_id not in entity_ids:
346 entity_ids.append(entity_id)
347 if data and (ATTR_ENTITY_ID, entity_id) not in inherited:
348 inherited[(ATTR_ENTITY_ID, entity_id)] = data
349 if unknown:
350 # a typo in an area or label would otherwise silently resolve to nothing
351 _LOGGER.warning("SUPERNOTIFY Unknown target selectors %s", ", ".join(unknown))
352 resolved = Target(targets, target_data=self.target_data)
353 if inherited or self.target_specific_data:
354 # data attached to the entity itself wins over data inherited from a selector
355 resolved.target_specific_data = inherited | {
356 key: data
357 for key, data in (self.target_specific_data or {}).items()
358 if key[0] not in self.EXPLICIT_INDIRECT_CATEGORIES
359 }
360 return resolved
362 def select(
363 self,
364 categories: Sequence[str | TargetEntityCategory],
365 own_names: Collection[str],
366 hass_api: HomeAssistantAPI,
367 target_selector: SelectionRule | None = None,
368 ) -> Target:
369 """Narrow this target to what a delivery can use, leaving this one untouched
371 `categories` are the delivery's declared target categories, and `own_names` its own
372 name and its transport's. A target category named after either is always destined
373 for that delivery. The two serve different purposes and both stay available:
374 - the TRANSPORT name (`sms:value`) reaches every delivery of that transport, so
375 scenario/time/occupancy selection logic can still decide which one actually
376 fires - the same as it would for a plain, auto-matched value
377 - a specific DELIVERY name (`shortcode_sms:value`) pins the target to just that
378 one delivery, for when two deliveries of the same transport must stay distinct
379 (e.g. `email` vs `html_email`)
381 `person_id`s are always kept, whatever the delivery declares, since they aren't
382 delivered to but are the link back to the recipients a delivery reaches (see
383 `Notification.generate_targets()`, which narrows them to those actually in each
384 envelope). They are kept out of the `target_selector` too, as it's for choosing
385 between values a transport can address.
386 """
387 if any(self.targets.get(category) for category in self.EXPLICIT_INDIRECT_CATEGORIES):
388 # area/floor/label always become entities first, so everything downstream - the
389 # category checks, `target_select`, the transports - only ever sees entity_ids
390 return self.resolve_selectors(hass_api).select(categories, own_names, hass_api, target_selector)
392 plain_categories = {c for c in categories if isinstance(c, str)}
393 entity_selectors = [c for c in categories if isinstance(c, TargetEntityCategory)]
394 # HA groups (`group.*` helpers and platform groups such as media player groups) are
395 # expanded into their members before the category and target_select checks, so a
396 # transport that can't address a group itself still reaches its members
397 expansions: dict[tuple[str, str], list[str]] = {}
398 groups: set[tuple[str, str]] = set()
400 def accepted(category: str, t: str, restricted: bool) -> bool:
401 if (
402 restricted
403 and entity_selectors
404 and category == ATTR_ENTITY_ID
405 and not any(sel.matches(t, hass_api) for sel in entity_selectors)
406 ):
407 return False
408 return target_selector is None or target_selector.match(t)
410 def selected(category: str, targets: list[str]) -> list[str]:
411 if category == ATTR_PERSON_ID:
412 return targets
413 restricted = category not in own_names
414 if (
415 restricted
416 and not (entity_selectors and category == ATTR_ENTITY_ID)
417 and plain_categories
418 and category not in plain_categories
419 ):
420 # this delivery declares fixed categories (from its transport, its own
421 # config, or both) - anything outside that set is rejected
422 return []
423 # else: this delivery declares no categories at all (e.g. `generic` with no
424 # config) - nothing to restrict against
425 chosen: list[str] = []
426 for t in targets:
427 members = hass_api.group_members(t) if category == ATTR_ENTITY_ID else None
428 if members is None:
429 matched = [t] if accepted(category, t, restricted) else []
430 else:
431 groups.add((category, t))
432 matched = [m for m in members if accepted(category, m, restricted)]
433 if not matched and accepted(category, t, restricted):
434 # group members unusable but the group id itself accepted, e.g. alexa_devices
435 matched = [t]
436 expansions[(category, t)] = matched
437 chosen.extend(m for m in matched if m not in chosen)
438 return chosen
440 filtered_target = Target({k: selected(k, v) for k, v in self.targets.items()}, target_data=self.target_data)
441 if self.target_specific_data:
442 # data inherited from a group first, so data explicitly attached to a member always wins
443 specific_data: dict[tuple[str, str], dict[str, Any]] = {}
444 for (c, t), data in self.target_specific_data.items():
445 if (c, t) in groups:
446 for m in expansions.get((c, t), []):
447 specific_data[(c, m)] = data
448 for (c, t), data in self.target_specific_data.items():
449 if (c, t) not in groups and c in filtered_target.targets and t in filtered_target.targets[c]:
450 specific_data[(c, t)] = data
451 filtered_target.target_specific_data = specific_data
452 return filtered_target
454 def split_by_target_data(self) -> list[Target]:
455 if not self.target_specific_data:
456 result = self.safe_copy()
457 result.target_specific_data = None
458 return [result]
459 results: list[Target] = []
460 default: Target = self.safe_copy()
461 default.target_specific_data = None
462 last_found: dict[str, Any] | None = None
463 collected: dict[str, list[str]] = {}
464 for (category, target), data in self.target_specific_data.items():
465 if last_found is None:
466 last_found = data
467 collected = {category: [target]}
468 elif data != last_found and last_found is not None:
469 new_target: Target = Target(collected, target_data=last_found)
470 results.append(new_target)
471 default -= new_target
472 last_found = data
473 collected = {category: [target]}
474 else:
475 collected.setdefault(category, [])
476 collected[category].append(target)
477 new_target = Target(collected, target_data=last_found)
478 results.append(new_target)
479 default -= new_target
480 if default.has_resolved_target():
481 results.append(default)
482 return results
484 def __len__(self) -> int:
485 """How many targets, whether direct or indirect"""
486 return sum(len(targets) for targets in self.targets.values())
488 def __add__(self, other: Target) -> Target:
489 """Create a new target by adding another to this one"""
490 new = Target()
491 categories = set(list(self.targets.keys()) + list(other.targets.keys()))
492 for category in categories:
493 new.targets[category] = list(self.targets.get(category, []))
494 new.targets[category].extend(t for t in other.targets.get(category, []) if t not in new.targets[category])
496 new.target_data = dict(self.target_data) if self.target_data else None
497 if other.target_data:
498 if new.target_data is None:
499 new.target_data = dict(other.target_data)
500 else:
501 new.target_data.update(other.target_data)
502 new.target_specific_data = dict(self.target_specific_data) if self.target_specific_data else None
503 if other.target_specific_data:
504 if new.target_specific_data is None:
505 new.target_specific_data = dict(other.target_specific_data)
506 else:
507 new.target_specific_data.update(other.target_specific_data)
508 return new
510 def __sub__(self, other: Target) -> Target:
511 """Create a new target by removing another from this one, ignoring target_data"""
512 new = Target()
513 new.target_data = self.target_data
514 if self.target_specific_data:
515 new.target_specific_data = {
516 k: v for k, v in self.target_specific_data.items() if k[1] not in other.targets.get(k[0], ())
517 }
518 categories = set(list(self.targets.keys()) + list(other.targets.keys()))
519 for category in categories:
520 new.targets[category] = []
521 new.targets[category].extend(t for t in self.targets.get(category, []) if t not in other.targets.get(category, []))
523 return new
525 def __eq__(self, other: object) -> bool:
526 """Compare two targets"""
527 if other is self:
528 return True
529 if other is None:
530 return False
531 if not isinstance(other, Target):
532 return NotImplemented
533 if self.target_data != other.target_data:
534 return False
535 if self.target_specific_data != other.target_specific_data:
536 return False
537 return all(self.targets.get(category, []) == other.targets.get(category, []) for category in self.CATEGORIES)
539 def as_dict(self, *, redact: bool = False, **_kwargs: Any) -> dict[str, list[str]]:
540 result = {k: v for k, v in self.targets.items() if v}
541 if redact:
542 if ATTR_EMAIL in result:
543 result[ATTR_EMAIL] = [partial_redact(v, unmasked_prefix=2, unmasked_suffix=1) for v in result[ATTR_EMAIL]]
544 if ATTR_PHONE in result:
545 result[ATTR_PHONE] = [partial_redact(v, unmasked_prefix=2, unmasked_suffix=1) for v in result[ATTR_PHONE]]
546 return result