Allow dashboard import to pull complete file from github (#3982)

esphome/components/dashboard_import/__init__.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+import requests
 
 import esphome.codegen as cg
 import esphome.config_validation as cv
@@ -6,6 +7,7 @@ from esphome.components.packages import validate_source_shorthand
 from esphome.const import CONF_WIFI
 from esphome.wizard import wizard_file
 from esphome.yaml_util import dump
+from esphome import git
 
 
 dashboard_import_ns = cg.esphome_ns.namespace("dashboard_import")
@@ -25,9 +27,12 @@ def validate_import_url(value):
 
 
 CONF_PACKAGE_IMPORT_URL = "package_import_url"
+CONF_IMPORT_FULL_CONFIG = "import_full_config"
+
 CONFIG_SCHEMA = cv.Schema(
     {
         cv.Required(CONF_PACKAGE_IMPORT_URL): validate_import_url,
+        cv.Optional(CONF_IMPORT_FULL_CONFIG, default=False): cv.boolean,
     }
 )
 
@@ -41,6 +46,9 @@ wifi:
 
 async def to_code(config):
     cg.add_define("USE_DASHBOARD_IMPORT")
+    url = config[CONF_PACKAGE_IMPORT_URL]
+    if config[CONF_IMPORT_FULL_CONFIG]:
+        url += "?full_config"
     cg.add(dashboard_import_ns.set_package_import_url(config[CONF_PACKAGE_IMPORT_URL]))
 
 
@@ -63,6 +71,19 @@ def import_config(
             ),
             encoding="utf8",
         )
+    else:
+        git_file = git.GitFile.from_shorthand(import_url)
+
+        if git_file.query and "full_config" in git_file.query:
+            url = git_file.raw_url
+            try:
+                req = requests.get(url, timeout=30)
+                req.raise_for_status()
+            except requests.exceptions.RequestException as e:
+                raise ValueError(f"Error while fetching {url}: {e}") from e
+
+            p.write_text(req.text, encoding="utf8")
+
         else:
             config = {
                 "substitutions": {"name": name},

esphome/components/packages/__init__.py

@@ -1,23 +1,22 @@
-import re
 from pathlib import Path
-from esphome.core import EsphomeError
-from esphome.config_helpers import merge_config
 
+import esphome.config_validation as cv
 from esphome import git, yaml_util
+from esphome.config_helpers import merge_config
 from esphome.const import (
     CONF_ESPHOME,
     CONF_FILE,
     CONF_FILES,
     CONF_MIN_VERSION,
     CONF_PACKAGES,
+    CONF_PASSWORD,
     CONF_REF,
     CONF_REFRESH,
     CONF_URL,
     CONF_USERNAME,
-    CONF_PASSWORD,
-    __version__ as ESPHOME_VERSION,
 )
-import esphome.config_validation as cv
+from esphome.const import __version__ as ESPHOME_VERSION
+from esphome.core import EsphomeError
 
 DOMAIN = CONF_PACKAGES
 
@@ -55,23 +54,15 @@ def validate_source_shorthand(value):
     if not isinstance(value, str):
         raise cv.Invalid("Shorthand only for strings")
 
-    m = re.match(
-        r"github://([a-zA-Z0-9\-]+)/([a-zA-Z0-9\-\._]+)/([a-zA-Z0-9\-_.\./]+)(?:@([a-zA-Z0-9\-_.\./]+))?",
-        value,
-    )
-    if m is None:
-        raise cv.Invalid(
-            "Source is not a file system path or in expected github://username/name/[sub-folder/]file-path.yml[@branch-or-tag] format!"
-        )
+    git_file = git.GitFile.from_shorthand(value)
 
     conf = {
-        CONF_URL: f"https://github.com/{m.group(1)}/{m.group(2)}.git",
-        CONF_FILE: m.group(3),
+        CONF_URL: git_file.git_url,
+        CONF_FILE: git_file.filename,
     }
-    if m.group(4):
-        conf[CONF_REF] = m.group(4)
+    if git_file.ref:
+        conf[CONF_REF] = git_file.ref
 
-    # print(conf)
     return BASE_SCHEMA(conf)
 
 

esphome/dashboard/dashboard.py

@@ -416,6 +416,10 @@ class ImportRequestHandler(BaseHandler):
             self.set_status(500)
             self.write("File already exists")
             return
+        except ValueError:
+            self.set_status(422)
+            self.write("Invalid package url")
+            return
 
         self.set_status(200)
         self.finish()

esphome/git.py

@@ -1,14 +1,15 @@
-from pathlib import Path
-import subprocess
 import hashlib
 import logging
-from typing import Callable, Optional
+import re
+import subprocess
 import urllib.parse
-
+from dataclasses import dataclass
 from datetime import datetime
+from pathlib import Path
+from typing import Callable, Optional
 
-from esphome.core import CORE, TimePeriodSeconds
 import esphome.config_validation as cv
+from esphome.core import CORE, TimePeriodSeconds
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -103,3 +104,57 @@ def clone_or_update(
             return repo_dir, revert
 
     return repo_dir, None
+
+
+GIT_DOMAINS = {
+    "github": "github.com",
+    "gitlab": "gitlab.com",
+}
+
+
+@dataclass(frozen=True)
+class GitFile:
+    domain: str
+    owner: str
+    repo: str
+    filename: str
+    ref: str = None
+    query: str = None
+
+    @property
+    def git_url(self) -> str:
+        return f"https://{self.domain}/{self.owner}/{self.repo}.git"
+
+    @property
+    def raw_url(self) -> str:
+        if self.ref is None:
+            raise ValueError("URL has no ref")
+        if self.domain == "github":
+            return f"https://raw.githubusercontent.com/{self.owner}/{self.repo}/{self.ref}/{self.filename}"
+        if self.domain == "gitlab":
+            return f"https://gitlab.com/{self.owner}/{self.repo}/-/raw/{self.ref}/{self.filename}"
+        raise NotImplementedError(f"Git domain {self.domain} not supported")
+
+    @classmethod
+    def from_shorthand(cls, shorthand):
+        """Parse a git shorthand URL into its components."""
+        if not isinstance(shorthand, str):
+            raise ValueError("Git shorthand must be a string")
+        m = re.match(
+            r"(?P<domain>[a-zA-Z0-9\-]+)://(?P<owner>[a-zA-Z0-9\-]+)/(?P<repo>[a-zA-Z0-9\-\._]+)/(?P<filename>[a-zA-Z0-9\-_.\./]+)(?:@(?P<ref>[a-zA-Z0-9\-_.\./]+))?(?:\?(?P<query>[a-zA-Z0-9\-_.\./]+))?",
+            shorthand,
+        )
+        if m is None:
+            raise ValueError(
+                "URL is not in expected github://username/name/[sub-folder/]file-path.yml[@branch-or-tag] format!"
+            )
+        if m.group("domain") not in GIT_DOMAINS:
+            raise ValueError(f"Unknown git domain {m.group('domain')}")
+        return cls(
+            domain=GIT_DOMAINS[m.group("domain")],
+            owner=m.group("owner"),
+            repo=m.group("repo"),
+            filename=m.group("filename"),
+            ref=m.group("ref"),
+            query=m.group("query"),
+        )