core/homeassistant/components/blueprint/importer.py

"""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")