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         for key,value in self.publicUrlFormat.items():
  30             print(f'fmt={key}->{value}')
  31         self.branch = branch
  32         self.baseFolder = dstFolder
  33         self.subfolder = self.createSubFolder(url, branch)
  34
  35     def createPublicUrlFormat(self, url:str, branch:str)->dict:
  36         fmt: dict = {}
  37         if url.endswith('.git'):
  38             url = url[:-4]
  39         if url.startswith('git@'):
  40             url = 'https://'+url[4:]
  41         fmt["raw"]=url+'/raw/'+branch+'/{}'
  42         fmt["blob"]=url+'/blob/'+branch+'/{}'
  43         return fmt
  44     def createSubFolder(self, gitUrl:str, branch:str) -> str:
  45         regex = r"^[^\/]+\/\/(.*)$"
  46         matches = re.finditer(regex, gitUrl)
  47         match = next(matches)
  48         name = match.group(1)
  49         if name.endswith('.git'):
  50             name=name[:-4]
  51         tmp:List[str]=[]
  52         hlp1 = name.split('/')
  53         for h in hlp1:
  54             if '.' in h:
  55                 hlp2=h.split('.')
  56                 for h2 in hlp2:
  57                     tmp.append(h2)
  58             else:
  59                 tmp.append(h)
  60
  61         return '/'.join(tmp)+'/'+branch
  62
  63     def getDstFolder(self)->str:
  64         return f'{self.baseFolder}/{self.subfolder}'
  65
  66     def exec(self, cmd:str):
  67         output = subprocess.Popen(
  68             cmd, shell=True, stdout=subprocess.PIPE).stdout.read()
  69         return output
  70
  71     def download(self) -> bool:
  72         print(f'try to download repo {self.downloadUrl} to {self.getDstFolder()}')
  73         self.exec(f'git clone --single-branch --branch {self.branch} {self.downloadUrl} {self.getDstFolder()}')
  74
  75     def getFilesFiltered(self, lst:List[str]=None, path=None, root=None, filter=['yaml','yml'])->List[str]:
  76         if lst is None:
  77             lst=[]
  78         if root is None:
  79             root=str(self.getDstFolder())
  80         if path is None:
  81             path=self.getDstFolder()
  82         if os.path.exists(path) and os.path.isdir(path):
  83             # Iterate over all files and directories in the given path
  84             for filename in os.listdir(path):
  85                 if filename.startswith("."):
  86                     continue
  87                 fmatch=False
  88                 # Get the absolute path of the file/directory
  89                 abs_path = os.path.join(path, filename)
  90                 # If it is a directory, recursively call this function on it
  91                 if os.path.isdir(abs_path):
  92                     self.getFilesFiltered(lst=lst, path=abs_path, root=root, filter=filter )
  93                 # If it is a file, print its absolute path
  94                 elif os.path.isfile(abs_path):
  95                     for fi in filter:
  96                         if abs_path.endswith(fi):
  97                             fmatch=True
  98                             break
  99                     if not fmatch:
 100                         continue
 101                     relpath=abs_path[len(root)+1:]
 102                     lst.append(relpath)
 103         return lst
 104
 105     def urlAlreadyInData(self, data:List[dict], pubUrl:str, key='publicURL'):
 106         for item in data:
 107             if key in item and item[key]==pubUrl:
 108                 return True
 109         return False
 110
 111     def createSchemaMap(self):
 112         schemaMapFile = f'{self.baseFolder}/schema-map.json'
 113         if os.path.isfile(schemaMapFile):
 114             with open(schemaMapFile) as fp:
 115                 data = json.load(fp)
 116         else:
 117             data:List[dict] = []
 118         files = self.getFilesFiltered()
 119         for file in files:
 120             print(file)
 121             for key,value in self.publicUrlFormat.items():
 122                 pubUrl = value.format(file)
 123                 if self.urlAlreadyInData(data,pubUrl):
 124                     print(f'entry with url {pubUrl} already exists. ignoring')
 125                     continue
 126                 data.append({
 127                     'publicURL': pubUrl,
 128                     'localURL': f'{self.subfolder}/{file}'
 129                 })
 130         with open(schemaMapFile,'w') as fp:
 131             json.dump(data,fp)
 132
 133 def printHelp(msg:str = None):
 134     if msg is not None:
 135         print('ERR: {msg}')
 136     print('Installation script for VES additional formats')
 137     print(' usage: ')
 138     print('    install.py [OPTIONS]')
 139     print('       -c  CONFIG_FILE')
 140     print('       -d  DESTINATION_PATH')
 141
 142
 143 args = sys.argv
 144 args.pop(0)
 145 configFilename = None
 146 dstPath = None
 147 while True:
 148     arg = args.pop(0)
 149     if arg == '-c':
 150         configFilename = args.pop(0)
 151     elif arg == '-d':
 152         dstPath = args.pop(0)
 153     else:
 154         printHelp(f'bad parameter {arg}')
 155         exit(1)
 156     if len(args)<=0:
 157         break
 158
 159 if configFilename is None or dstPath is None:
 160     printHelp('missing parameter')
 161     exit(1)
 162
 163
 164 config = json.load(open(configFilename))
 165 if not isinstance(config, list):
 166     printHelp('invalid config json. has to be a array')
 167     exit(1)
 168
 169 for item in config:
 170     dlRepo = item['repository']
 171     dlBranch = item['branch']
 172     installer = Installer(dlRepo, dlBranch, dstPath)
 173     installer.download()
 174     installer.createSchemaMap()