277 lines
8.7 KiB
Python
277 lines
8.7 KiB
Python
|
#!/usr/bin/python
|
||
|
|
# coding: utf-8 -*-
|
||
|
|
|
||
|
|
"""
|
||
|
|
CVP Uploader content
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
from typing import List, Optional, Any
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from loguru import logger
|
||
|
|
from cvprac.cvp_client import CvpClient
|
||
|
|
from cvprac.cvp_client_errors import CvpLoginError
|
||
|
|
|
||
|
|
# from eos_downloader.tools import exc_to_str
|
||
|
|
|
||
|
|
# logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class CvpAuthenticationItem:
|
||
|
|
"""
|
||
|
|
Data structure to represent Cloudvision Authentication
|
||
|
|
"""
|
||
|
|
server: str
|
||
|
|
port: int = 443
|
||
|
|
token: Optional[str] = None
|
||
|
|
timeout: int = 1200
|
||
|
|
validate_cert: bool = False
|
||
|
|
|
||
|
|
|
||
|
|
class Filer():
|
||
|
|
# pylint: disable=too-few-public-methods
|
||
|
|
"""
|
||
|
|
Filer Helper for file management
|
||
|
|
"""
|
||
|
|
def __init__(self, path: str) -> None:
|
||
|
|
self.file_exist = False
|
||
|
|
self.filename = ''
|
||
|
|
self.absolute_path = ''
|
||
|
|
self.relative_path = path
|
||
|
|
if os.path.exists(path):
|
||
|
|
self.file_exist = True
|
||
|
|
self.filename = os.path.basename(path)
|
||
|
|
self.absolute_path = os.path.realpath(path)
|
||
|
|
|
||
|
|
def __repr__(self) -> str:
|
||
|
|
return self.absolute_path if self.file_exist else ''
|
||
|
|
|
||
|
|
|
||
|
|
class CvFeatureManager():
|
||
|
|
"""
|
||
|
|
CvFeatureManager Object to interect with Cloudvision
|
||
|
|
"""
|
||
|
|
def __init__(self, authentication: CvpAuthenticationItem) -> None:
|
||
|
|
"""
|
||
|
|
__init__ Class Creator
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
authentication : CvpAuthenticationItem
|
||
|
|
Authentication information to use to connect to Cloudvision
|
||
|
|
"""
|
||
|
|
self._authentication = authentication
|
||
|
|
# self._cv_instance = CvpClient()
|
||
|
|
self._cv_instance = self._connect(authentication=authentication)
|
||
|
|
self._cv_images = self.__get_images()
|
||
|
|
# self._cv_bundles = self.__get_bundles()
|
||
|
|
|
||
|
|
def _connect(self, authentication: CvpAuthenticationItem) -> CvpClient:
|
||
|
|
"""
|
||
|
|
_connect Connection management
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
authentication : CvpAuthenticationItem
|
||
|
|
Authentication information to use to connect to Cloudvision
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
CvpClient
|
||
|
|
cvprac session to cloudvision
|
||
|
|
"""
|
||
|
|
client = CvpClient()
|
||
|
|
if authentication.token is not None:
|
||
|
|
try:
|
||
|
|
client.connect(
|
||
|
|
nodes=[authentication.server],
|
||
|
|
username='',
|
||
|
|
password='',
|
||
|
|
api_token=authentication.token,
|
||
|
|
is_cvaas=True,
|
||
|
|
port=authentication.port,
|
||
|
|
cert=authentication.validate_cert,
|
||
|
|
request_timeout=authentication.timeout
|
||
|
|
)
|
||
|
|
except CvpLoginError as error_data:
|
||
|
|
logger.error(f'Cannot connect to Cloudvision server {authentication.server}')
|
||
|
|
logger.debug(f'Error message: {error_data}')
|
||
|
|
logger.info('connected to Cloudvision server')
|
||
|
|
logger.debug(f'Connection info: {authentication}')
|
||
|
|
return client
|
||
|
|
|
||
|
|
def __get_images(self) -> List[Any]:
|
||
|
|
"""
|
||
|
|
__get_images Collect information about images on Cloudvision
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
dict
|
||
|
|
Fact returned by Cloudvision
|
||
|
|
"""
|
||
|
|
images = []
|
||
|
|
logger.debug(' -> Collecting images')
|
||
|
|
images = self._cv_instance.api.get_images()['data']
|
||
|
|
return images if self.__check_api_result(images) else []
|
||
|
|
|
||
|
|
# def __get_bundles(self):
|
||
|
|
# """
|
||
|
|
# __get_bundles [Not In use] Collect information about bundles on Cloudvision
|
||
|
|
|
||
|
|
# Returns
|
||
|
|
# -------
|
||
|
|
# dict
|
||
|
|
# Fact returned by Cloudvision
|
||
|
|
# """
|
||
|
|
# bundles = []
|
||
|
|
# logger.debug(' -> Collecting images bundles')
|
||
|
|
# bundles = self._cv_instance.api.get_image_bundles()['data']
|
||
|
|
# # bundles = self._cv_instance.post(url='/cvpservice/image/getImageBundles.do?queryparam=&startIndex=0&endIndex=0')['data']
|
||
|
|
# return bundles if self.__check_api_result(bundles) else None
|
||
|
|
|
||
|
|
def __check_api_result(self, arg0: Any) -> bool:
|
||
|
|
"""
|
||
|
|
__check_api_result Check API calls return content
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
arg0 : any
|
||
|
|
Element to test
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
bool
|
||
|
|
True if data are correct False in other cases
|
||
|
|
"""
|
||
|
|
logger.debug(arg0)
|
||
|
|
return len(arg0) > 0
|
||
|
|
|
||
|
|
def _does_image_exist(self, image_name: str) -> bool:
|
||
|
|
"""
|
||
|
|
_does_image_exist Check if an image is referenced in Cloudvision facts
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
image_name : str
|
||
|
|
Name of the image to search for
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
bool
|
||
|
|
True if present
|
||
|
|
"""
|
||
|
|
return any(image_name == image['name'] for image in self._cv_images) if isinstance(self._cv_images, list) else False
|
||
|
|
|
||
|
|
def _does_bundle_exist(self, bundle_name: str) -> bool:
|
||
|
|
# pylint: disable=unused-argument
|
||
|
|
"""
|
||
|
|
_does_bundle_exist Check if an image is referenced in Cloudvision facts
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
bool
|
||
|
|
True if present
|
||
|
|
"""
|
||
|
|
# return any(bundle_name == bundle['name'] for bundle in self._cv_bundles)
|
||
|
|
return False
|
||
|
|
|
||
|
|
def upload_image(self, image_path: str) -> bool:
|
||
|
|
"""
|
||
|
|
upload_image Upload an image to Cloudvision server
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
image_path : str
|
||
|
|
Path to the local file to upload
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
bool
|
||
|
|
True if succeeds
|
||
|
|
"""
|
||
|
|
image_item = Filer(path=image_path)
|
||
|
|
if image_item.file_exist is False:
|
||
|
|
logger.error(f'File not found: {image_item.relative_path}')
|
||
|
|
return False
|
||
|
|
logger.info(f'File path for image: {image_item}')
|
||
|
|
if self._does_image_exist(image_name=image_item.filename):
|
||
|
|
logger.error("Image found in Cloudvision , Please delete it before running this script")
|
||
|
|
return False
|
||
|
|
try:
|
||
|
|
upload_result = self._cv_instance.api.add_image(filepath=image_item.absolute_path)
|
||
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
||
|
|
logger.error('An error occurred during upload, check CV connection')
|
||
|
|
logger.error(f'Exception message is: {e}')
|
||
|
|
return False
|
||
|
|
logger.debug(f'Upload Result is : {upload_result}')
|
||
|
|
return True
|
||
|
|
|
||
|
|
def build_image_list(self, image_list: List[str]) -> List[Any]:
|
||
|
|
"""
|
||
|
|
Builds a list of the image data structures, for a given list of image names.
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
image_list : list
|
||
|
|
List of software image names
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
List:
|
||
|
|
Returns a list of images, with complete data or None in the event of failure
|
||
|
|
"""
|
||
|
|
internal_image_list = []
|
||
|
|
image_data = None
|
||
|
|
success = True
|
||
|
|
|
||
|
|
for entry in image_list:
|
||
|
|
for image in self._cv_images:
|
||
|
|
if image["imageFileName"] == entry:
|
||
|
|
image_data = image
|
||
|
|
|
||
|
|
if image_data is not None:
|
||
|
|
internal_image_list.append(image_data)
|
||
|
|
image_data = None
|
||
|
|
else:
|
||
|
|
success = False
|
||
|
|
|
||
|
|
return internal_image_list if success else []
|
||
|
|
|
||
|
|
def create_bundle(self, name: str, images_name: List[str]) -> bool:
|
||
|
|
"""
|
||
|
|
create_bundle Create a bundle with a list of images.
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
name : str
|
||
|
|
Name of the bundle
|
||
|
|
images_name : List[str]
|
||
|
|
List of images available on Cloudvision
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
bool
|
||
|
|
True if succeeds
|
||
|
|
"""
|
||
|
|
logger.debug(f'Init creation of an image bundle {name} with following images {images_name}')
|
||
|
|
all_images_present: List[bool] = []
|
||
|
|
self._cv_images = self.__get_images()
|
||
|
|
all_images_present.extend(
|
||
|
|
self._does_image_exist(image_name=image_name)
|
||
|
|
for image_name in images_name
|
||
|
|
)
|
||
|
|
# Bundle Create
|
||
|
|
if self._does_bundle_exist(bundle_name=name) is False:
|
||
|
|
logger.debug(f'Creating image bundle {name} with following images {images_name}')
|
||
|
|
images_data = self.build_image_list(image_list=images_name)
|
||
|
|
if images_data is not None:
|
||
|
|
logger.debug('Images information: {images_data}')
|
||
|
|
try:
|
||
|
|
data = self._cv_instance.api.save_image_bundle(name=name, images=images_data)
|
||
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
||
|
|
logger.critical(f'{e}')
|
||
|
|
else:
|
||
|
|
logger.debug(data)
|
||
|
|
return True
|
||
|
|
logger.critical('No data found for images')
|
||
|
|
return False
|