Code quality scrape (#65441)

This commit is contained in:
G Johansson 2022-02-12 15:28:54 +01:00 committed by GitHub
parent 62d49dcf98
commit 6d20e68e6d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 334 additions and 49 deletions

View file

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from bs4 import BeautifulSoup
import httpx
@ -11,7 +12,7 @@ from homeassistant.components.rest.data import RestData
from homeassistant.components.sensor import (
CONF_STATE_CLASS,
DEVICE_CLASSES_SCHEMA,
PLATFORM_SCHEMA,
PLATFORM_SCHEMA as PARENT_PLATFORM_SCHEMA,
STATE_CLASSES_SCHEMA,
SensorEntity,
)
@ -33,6 +34,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
_LOGGER = logging.getLogger(__name__)
@ -44,7 +46,7 @@ CONF_INDEX = "index"
DEFAULT_NAME = "Web scrape"
DEFAULT_VERIFY_SSL = True
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
PLATFORM_SCHEMA = PARENT_PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_RESOURCE): cv.string,
vol.Required(CONF_SELECT): cv.string,
@ -73,32 +75,32 @@ async def async_setup_platform(
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Web scrape sensor."""
name = config.get(CONF_NAME)
resource = config.get(CONF_RESOURCE)
method = "GET"
payload = None
headers = config.get(CONF_HEADERS)
verify_ssl = config.get(CONF_VERIFY_SSL)
select = config.get(CONF_SELECT)
attr = config.get(CONF_ATTR)
index = config.get(CONF_INDEX)
unit = config.get(CONF_UNIT_OF_MEASUREMENT)
device_class = config.get(CONF_DEVICE_CLASS)
state_class = config.get(CONF_STATE_CLASS)
username = config.get(CONF_USERNAME)
password = config.get(CONF_PASSWORD)
name: str = config[CONF_NAME]
resource: str = config[CONF_RESOURCE]
method: str = "GET"
payload: str | None = None
headers: str | None = config.get(CONF_HEADERS)
verify_ssl: bool = config[CONF_VERIFY_SSL]
select: str | None = config.get(CONF_SELECT)
attr: str | None = config.get(CONF_ATTR)
index: int = config[CONF_INDEX]
unit: str | None = config.get(CONF_UNIT_OF_MEASUREMENT)
device_class: str | None = config.get(CONF_DEVICE_CLASS)
state_class: str | None = config.get(CONF_STATE_CLASS)
username: str | None = config.get(CONF_USERNAME)
password: str | None = config.get(CONF_PASSWORD)
value_template: Template | None = config.get(CONF_VALUE_TEMPLATE)
if (value_template := config.get(CONF_VALUE_TEMPLATE)) is not None:
if value_template is not None:
value_template.hass = hass
auth: httpx.DigestAuth | tuple[str, str] | None
auth: httpx.DigestAuth | tuple[str, str] | None = None
if username and password:
if config.get(CONF_AUTHENTICATION) == HTTP_DIGEST_AUTHENTICATION:
auth = httpx.DigestAuth(username, password)
else:
auth = (username, password)
else:
auth = None
rest = RestData(hass, method, resource, auth, headers, None, payload, verify_ssl)
await rest.async_update()
@ -128,19 +130,19 @@ class ScrapeSensor(SensorEntity):
def __init__(
self,
rest,
name,
select,
attr,
index,
value_template,
unit,
device_class,
state_class,
):
rest: RestData,
name: str,
select: str | None,
attr: str | None,
index: int,
value_template: Template | None,
unit: str | None,
device_class: str | None,
state_class: str | None,
) -> None:
"""Initialize a web scrape sensor."""
self.rest = rest
self._state = None
self._attr_native_value = None
self._select = select
self._attr = attr
self._index = index
@ -150,12 +152,7 @@ class ScrapeSensor(SensorEntity):
self._attr_device_class = device_class
self._attr_state_class = state_class
@property
def native_value(self):
"""Return the state of the device."""
return self._state
def _extract_value(self):
def _extract_value(self) -> Any:
"""Parse the html extraction in the executor."""
raw_data = BeautifulSoup(self.rest.data, "html.parser")
_LOGGER.debug(raw_data)
@ -180,30 +177,26 @@ class ScrapeSensor(SensorEntity):
_LOGGER.debug(value)
return value
async def async_update(self):
async def async_update(self) -> None:
"""Get the latest data from the source and updates the state."""
await self.rest.async_update()
await self._async_update_from_rest_data()
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Ensure the data from the initial update is reflected in the state."""
await self._async_update_from_rest_data()
async def _async_update_from_rest_data(self):
async def _async_update_from_rest_data(self) -> None:
"""Update state from the rest data."""
if self.rest.data is None:
_LOGGER.error("Unable to retrieve data for %s", self.name)
return
try:
value = await self.hass.async_add_executor_job(self._extract_value)
except IndexError:
_LOGGER.error("Unable to extract data from HTML for %s", self.name)
return
value = await self.hass.async_add_executor_job(self._extract_value)
if self._value_template is not None:
self._state = self._value_template.async_render_with_possible_json_value(
value, None
self._attr_native_value = (
self._value_template.async_render_with_possible_json_value(value, None)
)
else:
self._state = value
self._attr_native_value = value