solution/smo/oam/ves-collector/install.py

   1 #!/usr/bin/python3
   2 ################################################################################
   3 # Copyright 2023 highstreet technologies GmbH
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the 'License');
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an 'AS IS' BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16
  17 import sys
  18 import os
  19 import subprocess
  20 import json
  21 import re
  22 from typing import List
  23
  24 class Installer:
  25
  26     def __init__(self, url, branch, dstFolder) -> None:
  27         self.downloadUrl = url
  28         self.publicUrlFormat = self.createPublicUrlFormat(url, branch)
  29         print(f'fmt={self.publicUrlFormat}')
  30         self.branch = branch
  31         self.baseFolder = dstFolder
  32         self.subfolder = self.createSubFolder(url, branch)
  33
  34     def createPublicUrlFormat(self, url:str, branch:str)->str:
  35         if url.endswith('.git'):
  36             url = url[:-4]
  37         if url.startswith('git@'):
  38             url = 'https://'+url[4:]
  39         fmt=url+'/raw/'+branch+'/{}'
  40         return fmt
  41     def createSubFolder(self, gitUrl:str, branch:str) -> str:
  42         regex = r"^[^\/]+\/\/(.*)$"
  43         matches = re.finditer(regex, gitUrl)
  44         match = next(matches)
  45         name = match.group(1)
  46         if name.endswith('.git'):
  47             name=name[:-4]
  48         tmp:List[str]=[]
  49         hlp1 = name.split('/')
  50         for h in hlp1:
  51             if '.' in h:
  52                 hlp2=h.split('.')
  53                 for h2 in hlp2:
  54                     tmp.append(h2)
  55             else:
  56                 tmp.append(h)
  57
  58         return '/'.join(tmp)+'/'+branch
  59
  60     def getDstFolder(self)->str:
  61         return f'{self.baseFolder}/{self.subfolder}'
  62
  63     def exec(self, cmd:str):
  64         output = subprocess.Popen(
  65             cmd, shell=True, stdout=subprocess.PIPE).stdout.read()
  66         return output
  67
  68     def download(self) -> bool:
  69         print(f'try to download repo {self.downloadUrl} to {self.getDstFolder()}')
  70         self.exec(f'git clone --single-branch --branch {self.branch} {self.downloadUrl} {self.getDstFolder()}')
  71
  72     def getFilesFiltered(self, lst:List[str]=None, path=None, root=None, filter=['yaml','yml'])->List[str]:
  73         if lst is None:
  74             lst=[]
  75         if root is None:
  76             root=str(self.getDstFolder())
  77         if path is None:
  78             path=self.getDstFolder()
  79         if os.path.exists(path) and os.path.isdir(path):
  80             # Iterate over all files and directories in the given path
  81             for filename in os.listdir(path):
  82                 if filename.startswith("."):
  83                     continue
  84                 fmatch=False
  85                 # Get the absolute path of the file/directory
  86                 abs_path = os.path.join(path, filename)
  87                 # If it is a directory, recursively call this function on it
  88                 if os.path.isdir(abs_path):
  89                     self.getFilesFiltered(lst=lst, path=abs_path, root=root, filter=filter )
  90                 # If it is a file, print its absolute path
  91                 elif os.path.isfile(abs_path):
  92                     for fi in filter:
  93                         if abs_path.endswith(fi):
  94                             fmatch=True
  95                             break
  96                     if not fmatch:
  97                         continue
  98                     relpath=abs_path[len(root)+1:]
  99                     lst.append(relpath)
 100         return lst
 101
 102     def urlAlreadyInData(self, data:List[dict], pubUrl:str, key='publicURL'):
 103         for item in data:
 104             if key in item and item[key]==pubUrl:
 105                 return True
 106         return False
 107
 108     def createSchemaMap(self):
 109         schemaMapFile = f'{self.baseFolder}/schema-map.json'
 110         if os.path.isfile(schemaMapFile):
 111             with open(schemaMapFile) as fp:
 112                 data = json.load(fp)
 113         else:
 114             data:List[dict] = []
 115         files = self.getFilesFiltered()
 116         for file in files:
 117             print(file)
 118             pubUrl = self.publicUrlFormat.format(file)
 119             if self.urlAlreadyInData(data,pubUrl):
 120                 print(f'entry with url {pubUrl} already exists. ignoring')
 121                 continue
 122             data.append({
 123                 'publicURL': pubUrl,
 124                 'localURL': f'{self.subfolder}/{file}'
 125             })
 126         with open(schemaMapFile,'w') as fp:
 127             json.dump(data,fp)
 128
 129 def printHelp(msg:str = None):
 130     if msg is not None:
 131         print('ERR: {msg}')
 132     print('Installation script for VES additional formats')
 133     print(' usage: ')
 134     print('    install.py [OPTIONS]')
 135     print('       -c  CONFIG_FILE')
 136     print('       -d  DESTINATION_PATH')
 137
 138
 139 args = sys.argv
 140 args.pop(0)
 141 configFilename = None
 142 dstPath = None
 143 while True:
 144     arg = args.pop(0)
 145     if arg == '-c':
 146         configFilename = args.pop(0)
 147     elif arg == '-d':
 148         dstPath = args.pop(0)
 149     else:
 150         printHelp(f'bad parameter {arg}')
 151         exit(1)
 152     if len(args)<=0:
 153         break
 154
 155 if configFilename is None or dstPath is None:
 156     printHelp('missing parameter')
 157     exit(1)
 158
 159
 160 config = json.load(open(configFilename))
 161 if not isinstance(config, list):
 162     printHelp('invalid config json. has to be a array')
 163     exit(1)
 164
 165 for item in config:
 166     dlRepo = item['repository']
 167     dlBranch = item['branch']
 168     installer = Installer(dlRepo, dlBranch, dstPath)
 169     installer.download()
 170     installer.createSchemaMap()