mirror of
https://github.com/netbox-community/Device-Type-Library-Import.git
synced 2024-12-01 12:03:12 +01:00
0dbd34956d
Check if slug exists in data. Without this check, import fail when SLUGS are not none in .env
110 lines
4.1 KiB
Python
110 lines
4.1 KiB
Python
import os
|
|
from glob import glob
|
|
from re import sub as re_sub
|
|
from git import Repo, exc
|
|
import yaml
|
|
|
|
|
|
class DTLRepo:
|
|
def __new__(cls, *args, **kwargs):
|
|
return super().__new__(cls)
|
|
|
|
def __init__(self, args, repo_path, exception_handler):
|
|
self.handle = exception_handler
|
|
self.yaml_extensions = ['yaml', 'yml']
|
|
self.url = args.url
|
|
self.repo_path = repo_path
|
|
self.branch = args.branch
|
|
self.repo = None
|
|
self.cwd = os.getcwd()
|
|
|
|
if os.path.isdir(self.repo_path):
|
|
self.pull_repo()
|
|
else:
|
|
self.clone_repo()
|
|
|
|
def get_relative_path(self):
|
|
return self.repo_path
|
|
|
|
def get_absolute_path(self):
|
|
return os.path.join(self.cwd, self.repo_path)
|
|
|
|
def get_devices_path(self):
|
|
return os.path.join(self.get_absolute_path(), 'device-types')
|
|
|
|
def get_modules_path(self):
|
|
return os.path.join(self.get_absolute_path(), 'module-types')
|
|
|
|
def slug_format(self, name):
|
|
return re_sub('\W+', '-', name.lower())
|
|
|
|
def pull_repo(self):
|
|
try:
|
|
self.handle.log("Package devicetype-library is already installed, "
|
|
+ f"updating {self.get_absolute_path()}")
|
|
self.repo = Repo(self.repo_path)
|
|
if not self.repo.remotes.origin.url.endswith('.git'):
|
|
self.handle.exception("GitInvalidRepositoryError", self.repo.remotes.origin.url,
|
|
f"Origin URL {self.repo.remotes.origin.url} does not end with .git")
|
|
self.repo.remotes.origin.pull()
|
|
self.repo.git.checkout(self.branch)
|
|
self.handle.verbose_log(
|
|
f"Pulled Repo {self.repo.remotes.origin.url}")
|
|
except exc.GitCommandError as git_error:
|
|
self.handle.exception(
|
|
"GitCommandError", self.repo.remotes.origin.url, git_error)
|
|
except Exception as git_error:
|
|
self.handle.exception(
|
|
"Exception", 'Git Repository Error', git_error)
|
|
|
|
def clone_repo(self):
|
|
try:
|
|
self.repo = Repo.clone_from(
|
|
self.url, self.get_absolute_path(), branch=self.branch)
|
|
self.handle.log(
|
|
f"Package Installed {self.repo.remotes.origin.url}")
|
|
except exc.GitCommandError as git_error:
|
|
self.handle.exception("GitCommandError", self.url, git_error)
|
|
except Exception as git_error:
|
|
self.handle.exception(
|
|
"Exception", 'Git Repository Error', git_error)
|
|
|
|
def get_devices(self, base_path, vendors: list = None):
|
|
files = []
|
|
discovered_vendors = []
|
|
vendor_dirs = os.listdir(base_path)
|
|
|
|
for folder in [vendor for vendor in vendor_dirs if not vendors or vendor.casefold() in vendors]:
|
|
if folder.casefold() != "testing":
|
|
discovered_vendors.append({'name': folder,
|
|
'slug': self.slug_format(folder)})
|
|
for extension in self.yaml_extensions:
|
|
files.extend(glob(base_path + folder + f'/*.{extension}'))
|
|
return files, discovered_vendors
|
|
|
|
def parse_files(self, files: list, slugs: list = None):
|
|
deviceTypes = []
|
|
for file in files:
|
|
with open(file, 'r') as stream:
|
|
try:
|
|
data = yaml.safe_load(stream)
|
|
except yaml.YAMLError as excep:
|
|
self.handle.verbose_log(excep)
|
|
continue
|
|
manufacturer = data['manufacturer']
|
|
data['manufacturer'] = {
|
|
'name': manufacturer, 'slug': self.slug_format(manufacturer)}
|
|
|
|
# Save file location to resolve any relative paths for images
|
|
data['src'] = file
|
|
|
|
match_any_given_slug_wildcard = any(
|
|
[s.lower() in data["slug"].lower() for s in slugs if "slug" in data]
|
|
)
|
|
if not match_any_given_slug_wildcard:
|
|
self.handle.verbose_log(f"Skipping {data['model']}")
|
|
continue
|
|
|
|
deviceTypes.append(data)
|
|
return deviceTypes
|