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