Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def process_notifications(login_obj, raw_notifications=None):
"""Process raw notifications json."""
from alexapy import AlexaAPI
if not raw_notifications:
raw_notifications = await AlexaAPI.get_notifications(login_obj)
email: Text = login_obj.email
notifications = {}
for notification in raw_notifications:
n_dev_id = notification['deviceSerialNumber']
n_type = notification['type']
if n_type == "MusicAlarm":
n_type = "Alarm"
n_id = notification['notificationIndex']
n_date = notification['originalDate']
n_time = notification['originalTime']
notification['date_time'] = f"{n_date} {n_time}"
if n_dev_id not in notifications:
notifications[n_dev_id] = {}
if n_type not in notifications[n_dev_id]:
notifications[n_dev_id][n_type] = {}
notifications[n_dev_id][n_type][n_id] = notification
async def process_notifications(login_obj, raw_notifications=None):
"""Process raw notifications json."""
from alexapy import AlexaAPI
if not raw_notifications:
raw_notifications = await AlexaAPI.get_notifications(login_obj)
email: Text = login_obj.email
notifications = {"process_timestamp": datetime.utcnow()}
for notification in raw_notifications:
n_dev_id = notification.get("deviceSerialNumber")
if n_dev_id is None:
# skip notifications untied to a device for now
# https://github.com/custom-components/alexa_media_player/issues/633#issuecomment-610705651
continue
n_type = notification.get("type")
if n_type is None:
continue
if n_type == "MusicAlarm":
n_type = "Alarm"
n_id = notification["notificationIndex"]
if n_type == "Alarm":
n_date = notification.get("originalDate")
async def process_notifications(login_obj, raw_notifications=None):
"""Process raw notifications json."""
from alexapy import AlexaAPI
if not raw_notifications:
raw_notifications = await AlexaAPI.get_notifications(login_obj)
email: Text = login_obj.email
notifications = {"process_timestamp": datetime.utcnow()}
for notification in raw_notifications:
n_dev_id = notification.get("deviceSerialNumber")
if n_dev_id is None:
# skip notifications untied to a device for now
# https://github.com/custom-components/alexa_media_player/issues/633#issuecomment-610705651
continue
n_type = notification.get("type")
if n_type is None:
continue
if n_type == "MusicAlarm":
n_type = "Alarm"
n_id = notification["notificationIndex"]
if n_type == "Alarm":
n_date = notification.get("originalDate")
async def process_notifications(login_obj, raw_notifications=None):
"""Process raw notifications json."""
from alexapy import AlexaAPI
if not raw_notifications:
raw_notifications = await AlexaAPI.get_notifications(login_obj)
email: Text = login_obj.email
notifications = {"process_timestamp": datetime.utcnow()}
for notification in raw_notifications:
n_dev_id = notification.get("deviceSerialNumber")
if n_dev_id is None:
# skip notifications untied to a device for now
# https://github.com/custom-components/alexa_media_player/issues/633#issuecomment-610705651
continue
n_type = notification.get("type")
if n_type is None:
continue
if n_type == "MusicAlarm":
n_type = "Alarm"
n_id = notification["notificationIndex"]
if n_type == "Alarm":
n_date = notification.get("originalDate")
existing_entities = hass.data[DATA_ALEXAMEDIA]["accounts"][email]["entities"][
"media_player"
].values()
auth_info = hass.data[DATA_ALEXAMEDIA]["accounts"][email].get("auth_info")
new_devices = hass.data[DATA_ALEXAMEDIA]["accounts"][email]["new_devices"]
devices = {}
bluetooth = {}
preferences = {}
dnd = {}
raw_notifications = {}
tasks = [
AlexaAPI.get_devices(login_obj),
AlexaAPI.get_bluetooth(login_obj),
AlexaAPI.get_device_preferences(login_obj),
AlexaAPI.get_dnd_state(login_obj),
AlexaAPI.get_notifications(login_obj),
]
if new_devices:
tasks.append(AlexaAPI.get_authentication(login_obj))
try:
# Note: asyncio.TimeoutError and aiohttp.ClientError are already
# handled by the data update coordinator.
async with async_timeout.timeout(30):
if new_devices:
(
devices,
bluetooth,
preferences,
dnd,
raw_notifications,
auth_info,
existing_entities = hass.data[DATA_ALEXAMEDIA]["accounts"][email]["entities"][
"media_player"
].values()
auth_info = hass.data[DATA_ALEXAMEDIA]["accounts"][email].get("auth_info")
new_devices = hass.data[DATA_ALEXAMEDIA]["accounts"][email]["new_devices"]
devices = {}
bluetooth = {}
preferences = {}
dnd = {}
raw_notifications = {}
tasks = [
AlexaAPI.get_devices(login_obj),
AlexaAPI.get_bluetooth(login_obj),
AlexaAPI.get_device_preferences(login_obj),
AlexaAPI.get_dnd_state(login_obj),
AlexaAPI.get_notifications(login_obj),
]
if new_devices:
tasks.append(AlexaAPI.get_authentication(login_obj))
try:
# Note: asyncio.TimeoutError and aiohttp.ClientError are already
# handled by the data update coordinator.
async with async_timeout.timeout(30):
if new_devices:
(
devices,
bluetooth,
preferences,
dnd,
raw_notifications,
auth_info,
existing_entities = hass.data[DATA_ALEXAMEDIA]["accounts"][email]["entities"][
"media_player"
].values()
auth_info = hass.data[DATA_ALEXAMEDIA]["accounts"][email].get("auth_info")
new_devices = hass.data[DATA_ALEXAMEDIA]["accounts"][email]["new_devices"]
devices = {}
bluetooth = {}
preferences = {}
dnd = {}
raw_notifications = {}
tasks = [
AlexaAPI.get_devices(login_obj),
AlexaAPI.get_bluetooth(login_obj),
AlexaAPI.get_device_preferences(login_obj),
AlexaAPI.get_dnd_state(login_obj),
AlexaAPI.get_notifications(login_obj),
]
if new_devices:
tasks.append(AlexaAPI.get_authentication(login_obj))
try:
# Note: asyncio.TimeoutError and aiohttp.ClientError are already
# handled by the data update coordinator.
async with async_timeout.timeout(10):
if new_devices:
(
devices,
bluetooth,
preferences,
dnd,
raw_notifications,
auth_info,
[email]
['entities']
['media_player'].values())
if ('websocket' in hass.data[DATA_ALEXAMEDIA]['accounts'][email]
and hass.data[DATA_ALEXAMEDIA]['accounts'][email]['websocket']
and not (hass.data[DATA_ALEXAMEDIA]
['accounts'][email]['new_devices'])):
return
hass.data[DATA_ALEXAMEDIA]['accounts'][email]['new_devices'] = False
try:
auth_info = await AlexaAPI.get_authentication(login_obj)
devices = await AlexaAPI.get_devices(login_obj)
bluetooth = await AlexaAPI.get_bluetooth(login_obj)
preferences = await AlexaAPI.get_device_preferences(login_obj)
dnd = await AlexaAPI.get_dnd_state(login_obj)
raw_notifications = await AlexaAPI.get_notifications(login_obj)
_LOGGER.debug("%s: Found %s devices, %s bluetooth",
hide_email(email),
len(devices) if devices is not None else '',
len(bluetooth) if bluetooth is not None else '')
if ((devices is None or bluetooth is None)
and not (hass.data[DATA_ALEXAMEDIA]
['accounts'][email]['configurator'])):
raise AlexapyLoginError()
except (AlexapyLoginError, RuntimeError):
_LOGGER.debug("%s: Alexa API disconnected; attempting to relogin",
hide_email(email))
await login_obj.login_with_cookie()
await test_login_status(hass,
config_entry, login_obj,
setup_platform_callback)
return