446 lines
16 KiB
Python
446 lines
16 KiB
Python
"""Diagnostics support for Alexa Media Player."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from dataclasses import fields, is_dataclass
|
|
from datetime import datetime
|
|
from itertools import islice
|
|
import re
|
|
from typing import Any
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers import device_registry as dr
|
|
from homeassistant.helpers.redact import async_redact_data
|
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
|
|
|
from .const import (
|
|
COMMON_BUCKET_COUNTS,
|
|
COMMON_DIAGNOSTIC_BUCKETS,
|
|
COMMON_DIAGNOSTIC_NAMES,
|
|
DEVICE_PLAYER_BUCKETS,
|
|
DOMAIN,
|
|
TO_REDACT,
|
|
)
|
|
|
|
|
|
# --------------------
|
|
# Local Functions
|
|
# --------------------
|
|
def _safe_dt(val: Any) -> str | None:
|
|
"""Serialize datetimes safely for JSON diagnostics."""
|
|
if isinstance(val, datetime):
|
|
return val.isoformat()
|
|
return None
|
|
|
|
|
|
def _maybe_len(val: Any) -> int | None:
|
|
"""Return the length of common container types or None if not applicable."""
|
|
if isinstance(val, (list, tuple, dict, set)):
|
|
return len(val)
|
|
return None
|
|
|
|
|
|
def _maybe_keys(val: Any, limit: int = 50) -> list[str] | None:
|
|
"""Return a sanitized sample of mapping keys for diagnostics.
|
|
|
|
If ``val`` is a mapping, return up to ``limit`` obfuscated keys to provide
|
|
structural insight without exposing sensitive data. Email-like keys are
|
|
redacted when possible; otherwise keys are shortened to a non-identifying
|
|
form. Returns ``None`` if ``val`` is not a mapping or keys cannot be read.
|
|
"""
|
|
|
|
if isinstance(val, Mapping):
|
|
try:
|
|
# Sample up to `limit` keys to keep diagnostics small.
|
|
def _safe_key(k: Any) -> str:
|
|
s = str(k)
|
|
# Emails/titles/tokens sometimes appear as keys in AMP structures.
|
|
if re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", s):
|
|
try:
|
|
from alexapy import ( # pylint: disable=import-outside-toplevel
|
|
hide_email,
|
|
)
|
|
|
|
return hide_email(s)
|
|
except (ImportError, AttributeError, TypeError, ValueError):
|
|
pass
|
|
return _obfuscate_identifier(s)
|
|
|
|
return sorted(_safe_key(k) for k in islice(val.keys(), limit))
|
|
except (TypeError, AttributeError):
|
|
return None
|
|
return None
|
|
|
|
|
|
def _sample_names(val: Any, *, limit: int = 5) -> list[str] | None:
|
|
"""Try to sample human-friendly names from a list/dict of device-like objects."""
|
|
names: list[str] = []
|
|
|
|
def add_name(x: Any) -> None:
|
|
if isinstance(x, Mapping):
|
|
for key in COMMON_DIAGNOSTIC_NAMES:
|
|
v = x.get(key)
|
|
if isinstance(v, str) and v:
|
|
names.append(v)
|
|
return
|
|
v = getattr(x, "name", None)
|
|
if isinstance(v, str) and v:
|
|
names.append(v)
|
|
|
|
if isinstance(val, Mapping):
|
|
for v in islice(val.values(), limit * 2):
|
|
add_name(v)
|
|
if len(names) >= limit:
|
|
break
|
|
return names[:limit] if names else None
|
|
|
|
if isinstance(val, (list, tuple)):
|
|
for v in val[: limit * 2]:
|
|
add_name(v)
|
|
if len(names) >= limit:
|
|
break
|
|
return names[:limit] if names else None
|
|
|
|
return None
|
|
|
|
|
|
# --------------------
|
|
# Coordinator discovery + summary
|
|
# --------------------
|
|
def _find_coordinators(obj: Any) -> list[DataUpdateCoordinator]:
|
|
"""Recursively find DataUpdateCoordinator instances in an object tree."""
|
|
found: list[DataUpdateCoordinator] = []
|
|
visited: set[int] = set()
|
|
|
|
def walk(x: Any) -> None:
|
|
obj_id = id(x)
|
|
if obj_id in visited:
|
|
return
|
|
visited.add(obj_id)
|
|
|
|
if isinstance(x, DataUpdateCoordinator):
|
|
found.append(x)
|
|
return
|
|
if is_dataclass(x):
|
|
try:
|
|
# Walk dataclass attributes directly; asdict() can lose/mangle objects.
|
|
for f in fields(x):
|
|
try:
|
|
walk(getattr(x, f.name))
|
|
except (AttributeError, TypeError, ValueError):
|
|
# Skip fields that can't be read safely
|
|
pass
|
|
except (TypeError, ValueError):
|
|
# Fallback: vars() can work for some dataclass/slots variations
|
|
try:
|
|
for v in vars(x).values():
|
|
walk(v)
|
|
except (AttributeError, TypeError, ValueError):
|
|
# Ignore attributes that cannot be introspected via vars()
|
|
pass
|
|
return
|
|
if isinstance(x, Mapping):
|
|
for v in x.values():
|
|
walk(v)
|
|
return
|
|
if isinstance(x, (list, tuple, set)):
|
|
for v in x:
|
|
walk(v)
|
|
return
|
|
# Ignore everything else.
|
|
|
|
walk(obj)
|
|
return found
|
|
|
|
|
|
def _summarize_coordinator_data(cdata: Any) -> dict:
|
|
"""
|
|
Allowlisted summary of coordinator.data.
|
|
|
|
Never dump raw coordinator data. Only return counts + small samples.
|
|
Optimized for AMP: coordinator.data is often a mapping keyed by UUIDs.
|
|
"""
|
|
out: dict[str, Any] = {}
|
|
|
|
if isinstance(cdata, Mapping):
|
|
out["data_key_count"] = len(cdata)
|
|
|
|
key_sample = list(islice(cdata.keys(), 10))
|
|
|
|
out["data_key_types_sample"] = [type(k).__name__ for k in key_sample]
|
|
|
|
sample_vals = [type(cdata.get(k)).__name__ for k in key_sample[:3]]
|
|
if sample_vals:
|
|
out["data_value_types_sample"] = sample_vals
|
|
|
|
# If coordinator.data sometimes contains named buckets (future-proof),
|
|
# include just counts (but only if those keys actually exist).
|
|
for key in COMMON_DIAGNOSTIC_BUCKETS:
|
|
if key in cdata:
|
|
out[f"{key}_count"] = _maybe_len(cdata.get(key))
|
|
|
|
# If AMP ever exposes last_called through coordinator.data, include only safe fields.
|
|
last_called = cdata.get("last_called")
|
|
if isinstance(last_called, Mapping):
|
|
ts = last_called.get("timestamp")
|
|
out["last_called"] = {
|
|
"timestamp": _safe_dt(ts) or ts,
|
|
"summary": last_called.get("summary"),
|
|
}
|
|
|
|
# If there are device/player buckets, sample friendly names (no IDs).
|
|
for key in DEVICE_PLAYER_BUCKETS:
|
|
if key in cdata:
|
|
sample = _sample_names(cdata.get(key))
|
|
if sample:
|
|
out[f"{key}_sample_names"] = sample
|
|
break
|
|
|
|
return out
|
|
|
|
if isinstance(cdata, (list, tuple)):
|
|
out["data_len"] = len(cdata)
|
|
sample = _sample_names(cdata)
|
|
if sample:
|
|
out["sample_names"] = sample
|
|
return out
|
|
|
|
if cdata is not None:
|
|
out["data_type"] = type(cdata).__name__
|
|
return out
|
|
|
|
|
|
def _summarize_coordinator(coordinator: DataUpdateCoordinator) -> dict:
|
|
"""Return a safe, compact view of a coordinator."""
|
|
exc = getattr(coordinator, "last_exception", None)
|
|
|
|
data = {
|
|
"name": getattr(coordinator, "name", None),
|
|
"last_update_success": getattr(coordinator, "last_update_success", None),
|
|
"has_exception": exc is not None,
|
|
"last_exception_type": type(exc).__name__ if exc else None,
|
|
"update_interval": (
|
|
str(getattr(coordinator, "update_interval", None))
|
|
if getattr(coordinator, "update_interval", None) is not None
|
|
else None
|
|
),
|
|
"last_update": _safe_dt(getattr(coordinator, "last_update", None)),
|
|
}
|
|
|
|
try:
|
|
data["data_summary"] = _summarize_coordinator_data(
|
|
getattr(coordinator, "data", None)
|
|
)
|
|
except (
|
|
Exception
|
|
) as exc: # noqa: BLE001 - intentionally broad; diagnostics must not crash
|
|
data["data_summary_error"] = type(exc).__name__
|
|
data["data_summary_error_present"] = True
|
|
|
|
return data
|
|
|
|
|
|
# --------------------
|
|
# AMP-specific (non-coordinator) runtime summaries
|
|
# --------------------
|
|
def _summarize_amp_entry_runtime(entry_runtime: Any) -> dict:
|
|
"""
|
|
Best-effort summary of hass.data[DOMAIN][entry_id] runtime.
|
|
|
|
AMP may not store anything here; keep robust.
|
|
"""
|
|
out: dict[str, Any] = {"present": entry_runtime is not None}
|
|
|
|
if isinstance(entry_runtime, Mapping):
|
|
out["runtime_type"] = "mapping"
|
|
out["runtime_keys"] = _maybe_keys(entry_runtime)
|
|
# Common “bucket” counts if they happen to exist.
|
|
for key in COMMON_BUCKET_COUNTS:
|
|
if key in entry_runtime:
|
|
out[f"{key}_count"] = _maybe_len(entry_runtime.get(key))
|
|
# Small sample of names
|
|
for key in DEVICE_PLAYER_BUCKETS:
|
|
if key in entry_runtime:
|
|
sample = _sample_names(entry_runtime.get(key))
|
|
if sample:
|
|
out[f"{key}_sample_names"] = sample
|
|
break
|
|
else:
|
|
if entry_runtime is not None:
|
|
out["runtime_type"] = type(entry_runtime).__name__
|
|
|
|
return out
|
|
|
|
|
|
def _obfuscate_identifier(val: Any) -> str:
|
|
"""Return a shortened, non-identifying representation of a value.
|
|
|
|
Non-string, empty, or very short values are fully masked. Longer strings
|
|
are reduced to a minimal prefix and suffix to aid debugging without
|
|
exposing the original identifier.
|
|
"""
|
|
if not isinstance(val, str) or not val or len(val) <= 4:
|
|
return "****"
|
|
return f"{val[:2]}...{val[-2:]}"
|
|
|
|
|
|
def _obfuscate_title_with_email(title: str | None, email: str | None) -> str | None:
|
|
"""Obfuscate email in config entry title using the same mechanism as AMP logs."""
|
|
if not title or not email:
|
|
return title
|
|
|
|
try:
|
|
# Lazy import to keep diagnostics import cheap
|
|
from alexapy import hide_email # pylint: disable=import-outside-toplevel
|
|
|
|
redacted = hide_email(email)
|
|
except (ImportError, AttributeError, TypeError, ValueError):
|
|
redacted = _obfuscate_identifier(email)
|
|
|
|
return title.replace(email, redacted)
|
|
|
|
|
|
def _get_safe_config_entry_title(config_entry: ConfigEntry) -> str | None:
|
|
"""Get obfuscated config entry title."""
|
|
email = config_entry.data.get("email")
|
|
return _obfuscate_title_with_email(config_entry.title, email)
|
|
|
|
|
|
def _summarize_amp_domain(domain_data: Any, config_entry: ConfigEntry) -> dict:
|
|
"""
|
|
Best-effort summary of hass.data[DOMAIN] for AMP.
|
|
|
|
AMP historically stores account/login state in custom structures, not always
|
|
keyed by entry_id, and often not using DataUpdateCoordinator.
|
|
"""
|
|
out: dict[str, Any] = {}
|
|
out["domain_data_present"] = domain_data is not None
|
|
out["domain_data_type"] = (
|
|
type(domain_data).__name__ if domain_data is not None else None
|
|
)
|
|
|
|
if not isinstance(domain_data, Mapping):
|
|
return out
|
|
|
|
out["domain_keys"] = _maybe_keys(domain_data)
|
|
|
|
# Try a few common/likely buckets without dumping contents.
|
|
# NOTE: We deliberately avoid copying values; only report counts/types/samples.
|
|
for key in COMMON_DIAGNOSTIC_BUCKETS:
|
|
if key in domain_data:
|
|
val = domain_data.get(key)
|
|
out[f"{key}_type"] = type(val).__name__
|
|
out[f"{key}_len"] = _maybe_len(val)
|
|
sample = _sample_names(val)
|
|
if sample:
|
|
out[f"{key}_sample_names"] = sample
|
|
|
|
# Try to locate the specific account blob by email/title if present.
|
|
# The config entry title often contains "email - url". We'll only use it to
|
|
# match keys; we won't add the email to diagnostics (redaction will remove it).
|
|
raw_title = config_entry.title or ""
|
|
email = config_entry.data.get("email")
|
|
out["entry_title_hint"] = _obfuscate_title_with_email(raw_title, email)
|
|
# Some integrations store per-entry runtime keyed by entry_id *or* by title/email.
|
|
# Report whether those keys exist.
|
|
out["has_entry_id_key"] = config_entry.entry_id in domain_data
|
|
out["has_title_key"] = raw_title in domain_data if raw_title else False
|
|
|
|
return out
|
|
|
|
|
|
# --------------------
|
|
# Diagnostics entry points
|
|
# --------------------
|
|
async def async_get_config_entry_diagnostics(
|
|
hass: HomeAssistant, config_entry: ConfigEntry
|
|
) -> dict:
|
|
"""Return diagnostics for a config entry."""
|
|
domain_data = hass.data.get(DOMAIN)
|
|
safe_title = _get_safe_config_entry_title(config_entry)
|
|
|
|
# AMP currently doesn't store runtime under entry_id.
|
|
# This adds future-proofing for if and when it does.
|
|
entry_runtime = None
|
|
if isinstance(domain_data, Mapping):
|
|
entry_runtime = domain_data.get(config_entry.entry_id)
|
|
|
|
# Coordinator discovery:
|
|
# 1) Try under entry_runtime (best practice)
|
|
# 2) If none found and domain_data is a mapping, try domain_data as a whole
|
|
coordinators: list[DataUpdateCoordinator] = []
|
|
searched: list[str] = []
|
|
|
|
if entry_runtime is not None:
|
|
searched.append("hass.data[DOMAIN][entry_id]")
|
|
coordinators = _find_coordinators(entry_runtime)
|
|
|
|
if not coordinators and isinstance(domain_data, Mapping):
|
|
searched.append("hass.data[DOMAIN]")
|
|
coordinators = _find_coordinators(domain_data)
|
|
|
|
coordinator_summaries = [_summarize_coordinator(c) for c in coordinators]
|
|
|
|
data: dict = {
|
|
"entry": {
|
|
"entry_id": config_entry.entry_id,
|
|
"title": safe_title,
|
|
"domain": config_entry.domain,
|
|
"version": config_entry.version,
|
|
"minor_version": config_entry.minor_version,
|
|
},
|
|
# Include config + options; sensitive values are redacted below.
|
|
"data": dict(config_entry.data),
|
|
"options": dict(config_entry.options),
|
|
"account": {
|
|
"searched_for_coordinators_in": searched,
|
|
"coordinator_count": len(coordinator_summaries),
|
|
"coordinators": coordinator_summaries,
|
|
# AMP-specific summaries (useful when coordinator_count == 0)
|
|
"amp_entry_runtime_summary": _summarize_amp_entry_runtime(entry_runtime),
|
|
"amp_domain_summary": _summarize_amp_domain(domain_data, config_entry),
|
|
},
|
|
}
|
|
|
|
return async_redact_data(data, TO_REDACT)
|
|
|
|
|
|
async def async_get_device_diagnostics(
|
|
_hass: HomeAssistant, config_entry: ConfigEntry, device: dr.DeviceEntry
|
|
) -> dict:
|
|
"""Return diagnostics for a specific device."""
|
|
safe_title = _get_safe_config_entry_title(config_entry)
|
|
|
|
try:
|
|
# Lazy import to keep diagnostics import cheap
|
|
from alexapy import hide_serial # pylint: disable=import-outside-toplevel
|
|
|
|
safe_serial = hide_serial(device.serial_number)
|
|
except (ImportError, AttributeError, TypeError, ValueError):
|
|
safe_serial = _obfuscate_identifier(device.serial_number)
|
|
|
|
data: dict = {
|
|
"device": {
|
|
"id": _obfuscate_identifier(device.id),
|
|
"name": device.name,
|
|
"name_by_user": device.name_by_user,
|
|
"manufacturer": device.manufacturer,
|
|
"model": device.model,
|
|
"sw_version": device.sw_version,
|
|
"serial_number": safe_serial,
|
|
"identifiers": sorted(
|
|
(domain, _obfuscate_identifier(value))
|
|
for domain, value in device.identifiers
|
|
),
|
|
"via_device_id": _obfuscate_identifier(device.via_device_id),
|
|
},
|
|
"config_entry": {
|
|
"entry_id": config_entry.entry_id,
|
|
"title": safe_title,
|
|
},
|
|
}
|
|
|
|
return async_redact_data(data, TO_REDACT)
|