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
from typing import TYPE_CHECKING

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 as yaml_util

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_util.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_util.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]}"
    suggested_filename = suggested_filename.removesuffix(".yaml")

    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: Blueprint | None = None
    filename: str | None = None
    content: str

    for filename, info in gist["files"].items():
        if not filename.endswith(".yaml"):
            continue

        content = info["content"]
        data = yaml_util.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'"
        )
    if TYPE_CHECKING:
        assert isinstance(filename, str)

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