importer.py

Go to the documentation of this file.

"""Import logic for blueprint."""
 
from __future__ import annotations
 
from contextlib import suppress
from dataclasses import dataclass
import html
import re
 
import voluptuous as vol
import yarl
 
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import aiohttp_client, config_validation as cv
from homeassistant.util import yaml
 
from .models import Blueprint
from .schemas import BLUEPRINT_SCHEMA, is_blueprint_config
 
COMMUNITY_TOPIC_PATTERN = re.compile(
    r"^https://community.home-assistant.io/t/[a-z0-9-]+/(?P<topic>\d+)(?:/(?P<post>\d+)|)$"
)
 
COMMUNITY_CODE_BLOCK = re.compile(
    r'<code class="lang-(?P<syntax>[a-z]+)">(?P<content>(?:.|\n)*)</code>', re.MULTILINE
)
 
GITHUB_FILE_PATTERN = re.compile(
    r"^https://github.com/(?P<repository>.+)/blob/(?P<path>.+)$"
)
 
WEBSITE_PATTERN = re.compile(
    r"^https://(?P<subdomain>[a-z0-9-]+)\.home-assistant\.io/(?P<path>.+).yaml$"
)
 
COMMUNITY_TOPIC_SCHEMA = vol.Schema(
    {
        "slug": str,
        "title": str,
        "post_stream": {"posts": [{"updated_at": cv.datetime, "cooked": str}]},
    },
    extra=vol.ALLOW_EXTRA,
)
 
 
class UnsupportedUrl(HomeAssistantError):
    """When the function doesn't support the url."""
 
 
@dataclass(frozen=True)
class ImportedBlueprint:
    """Imported blueprint."""
 
    suggested_filename: str
    raw_data: str
    blueprint: Blueprint
 
 
def _get_github_import_url(url: str) -> str:
    """Convert a GitHub url to the raw content.
 
    Async friendly.
    """
    if url.startswith("https://raw.githubusercontent.com/"):
        return url
 
    if (match := GITHUB_FILE_PATTERN.match(url)) is None:
        raise UnsupportedUrl("Not a GitHub file url")
 
    repo, path = match.groups()
 
    return f"https://raw.githubusercontent.com/{repo}/{path}"
 
 
def _get_community_post_import_url(url: str) -> str:
    """Convert a forum post url to an import url.
 
    Async friendly.
    """
    if (match := COMMUNITY_TOPIC_PATTERN.match(url)) is None:
        raise UnsupportedUrl("Not a topic url")
 
    _topic, post = match.groups()
 
    json_url = url
 
    if post is not None:
        # Chop off post part, ie /2
        json_url = json_url[: -len(post) - 1]
 
    json_url += ".json"
 
    return json_url
 
 
def _extract_blueprint_from_community_topic(
    url: str,
    topic: dict,
) -> ImportedBlueprint:
    """Extract a blueprint from a community post JSON.
 
    Async friendly.
    """
    block_content: str
    blueprint = None
    post = topic["post_stream"]["posts"][0]
 
    for match in COMMUNITY_CODE_BLOCK.finditer(post["cooked"]):
        block_syntax, block_content = match.groups()
 
        if block_syntax not in ("auto", "yaml"):
            continue
 
        block_content = html.unescape(block_content.strip())
 
        try:
            data = yaml.parse_yaml(block_content)
        except HomeAssistantError:
            if block_syntax == "yaml":
                raise
 
            continue
 
        if not is_blueprint_config(data):
            continue
        assert isinstance(data, dict)
 
        blueprint = Blueprint(data, schema=BLUEPRINT_SCHEMA)
        break
 
    if blueprint is None:
        raise HomeAssistantError(
            "No valid blueprint found in the topic. Blueprint syntax blocks need to be"
            " marked as YAML or no syntax."
        )
 
    return ImportedBlueprint(
        f'{post["username"]}/{topic["slug"]}', block_content, blueprint
    )
 
 
async def fetch_blueprint_from_community_post(
    hass: HomeAssistant, url: str
) -> ImportedBlueprint:
    """Get blueprints from a community post url.
 
    Method can raise aiohttp client exceptions, vol.Invalid.
 
    Caller needs to implement own timeout.
    """
    import_url = _get_community_post_import_url(url)
    session = aiohttp_client.async_get_clientsession(hass)
 
    resp = await session.get(import_url, raise_for_status=True)
    json_resp = await resp.json()
    json_resp = COMMUNITY_TOPIC_SCHEMA(json_resp)
    return _extract_blueprint_from_community_topic(url, json_resp)
 
 
async def fetch_blueprint_from_github_url(
    hass: HomeAssistant, url: str
) -> ImportedBlueprint:
    """Get a blueprint from a github url."""
    import_url = _get_github_import_url(url)
    session = aiohttp_client.async_get_clientsession(hass)
 
    resp = await session.get(import_url, raise_for_status=True)
    raw_yaml = await resp.text()
    data = yaml.parse_yaml(raw_yaml)
    assert isinstance(data, dict)
    blueprint = Blueprint(data, schema=BLUEPRINT_SCHEMA)
 
    parsed_import_url = yarl.URL(import_url)
    suggested_filename = f"{parsed_import_url.parts[1]}/{parsed_import_url.parts[-1]}"
    if suggested_filename.endswith(".yaml"):
        suggested_filename = suggested_filename[:-5]
 
    return ImportedBlueprint(suggested_filename, raw_yaml, blueprint)
 
 
async def fetch_blueprint_from_github_gist_url(
    hass: HomeAssistant, url: str
) -> ImportedBlueprint:
    """Get a blueprint from a Github Gist."""
    if not url.startswith("https://gist.github.com/"):
        raise UnsupportedUrl("Not a GitHub gist url")
 
    parsed_url = yarl.URL(url)
    session = aiohttp_client.async_get_clientsession(hass)
 
    resp = await session.get(
        f"https://api.github.com/gists/{parsed_url.parts[2]}",
        headers={"Accept": "application/vnd.github.v3+json"},
        raise_for_status=True,
    )
    gist = await resp.json()
 
    blueprint = None
    filename = None
    content: str
 
    for filename, info in gist["files"].items():
        if not filename.endswith(".yaml"):
            continue
 
        content = info["content"]
        data = yaml.parse_yaml(content)
 
        if not is_blueprint_config(data):
            continue
        assert isinstance(data, dict)
 
        blueprint = Blueprint(data, schema=BLUEPRINT_SCHEMA)
        break
 
    if blueprint is None:
        raise HomeAssistantError(
            "No valid blueprint found in the gist. The blueprint file needs to end with"
            " '.yaml'"
        )
 
    return ImportedBlueprint(
        f"{gist['owner']['login']}/{filename[:-5]}", content, blueprint
    )
 
 
async def fetch_blueprint_from_website_url(
    hass: HomeAssistant, url: str
) -> ImportedBlueprint:
    """Get a blueprint from our website."""
    if (WEBSITE_PATTERN.match(url)) is None:
        raise UnsupportedUrl("Not a Home Assistant website URL")
 
    session = aiohttp_client.async_get_clientsession(hass)
 
    resp = await session.get(url, raise_for_status=True)
    raw_yaml = await resp.text()
    data = yaml.parse_yaml(raw_yaml)
    assert isinstance(data, dict)
    blueprint = Blueprint(data, schema=BLUEPRINT_SCHEMA)
 
    parsed_import_url = yarl.URL(url)
    suggested_filename = f"homeassistant/{parsed_import_url.parts[-1][:-5]}"
    return ImportedBlueprint(suggested_filename, raw_yaml, blueprint)
 
 
async def fetch_blueprint_from_generic_url(
    hass: HomeAssistant, url: str
) -> ImportedBlueprint:
    """Get a blueprint from a generic website."""
    session = aiohttp_client.async_get_clientsession(hass)
 
    resp = await session.get(url, raise_for_status=True)
    raw_yaml = await resp.text()
    data = yaml.parse_yaml(raw_yaml)
 
    assert isinstance(data, dict)
    blueprint = Blueprint(data, schema=BLUEPRINT_SCHEMA)
 
    parsed_import_url = yarl.URL(url)
    suggested_filename = f"{parsed_import_url.host}/{parsed_import_url.parts[-1][:-5]}"
    return ImportedBlueprint(suggested_filename, raw_yaml, blueprint)
 
 
FETCH_FUNCTIONS = (
    fetch_blueprint_from_community_post,
    fetch_blueprint_from_github_url,
    fetch_blueprint_from_github_gist_url,
    fetch_blueprint_from_website_url,
    fetch_blueprint_from_generic_url,
)
 
 
async def fetch_blueprint_from_url(hass: HomeAssistant, url: str) -> ImportedBlueprint:
    """Get a blueprint from a url.
 
    The returned blueprint will only be validated with BLUEPRINT_SCHEMA, not the domain
    specific schema.
    """
    for func in FETCH_FUNCTIONS:
        with suppress(UnsupportedUrl):
            imported_bp = await func(hass, url)
            imported_bp.blueprint.update_metadata(source_url=url)
            return imported_bp
 
    raise HomeAssistantError("Unsupported URL")