catalog/pub/utils/fileutil.py

   1 # Copyright 2017 ZTE Corporation.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 import os
  15 import shutil
  16 import logging
  17 import tempfile
  18 import traceback
  19 import urllib
  20 import zipfile
  21
  22 logger = logging.getLogger(__name__)
  23
  24
  25 def make_dirs(path):
  26     """
  27     Make directories
  28     :param path:
  29     :return:
  30     """
  31     if not os.path.exists(path):
  32         os.makedirs(path, 0o777)
  33
  34
  35 def delete_dirs(path):
  36     """
  37     Delete directories
  38     :param path:
  39     :return:
  40     """
  41     try:
  42         if os.path.exists(path):
  43             shutil.rmtree(path)
  44     except Exception as e:
  45         logger.error(traceback.format_exc())
  46         logger.error("Failed to delete %s:%s", path, e.args[0])
  47
  48
  49 def download_file_from_http(url, local_dir, file_name):
  50     """
  51     Download file from http and save to local dir
  52     :param url:
  53     :param local_dir:
  54     :param file_name:
  55     :return:
  56     """
  57     local_file_name = os.path.join(local_dir, file_name)
  58     is_download_ok = False
  59     try:
  60         make_dirs(local_dir)
  61         req = urllib.request.urlopen(url)
  62         save_file = open(local_file_name, 'w')
  63         save_file.write(req.read())
  64         save_file.close()
  65         req.close()
  66         is_download_ok = True
  67     except:
  68         logger.error(traceback.format_exc())
  69         logger.error("Failed to download %s to %s.", url, local_file_name)
  70     return is_download_ok, local_file_name
  71
  72
  73 def unzip_file(zip_src, dst_dir, csar_path):
  74     """
  75     Unzip the zip file to given path
  76     :param zip_src:
  77     :param dst_dir:
  78     :param csar_path:
  79     :return:
  80     """
  81     logger.debug("unzip_file %s to %s.", zip_src, dst_dir)
  82     if os.path.exists(zip_src):
  83         logger.debug("unzip_file %s.", zip_src)
  84         fz = zipfile.ZipFile(zip_src, 'r')
  85         for file in fz.namelist():
  86             fz.extract(file, dst_dir)
  87         return os.path.join(dst_dir, csar_path)
  88     else:
  89         return ""
  90
  91
  92 def unzip_csar(zip_src, dst_dir):
  93     """
  94     Unzip csar package
  95     :param zip_src:
  96     :param dst_dir:
  97     :return:
  98     """
  99     if os.path.exists(zip_src):
 100         fz = zipfile.ZipFile(zip_src, 'r')
 101         for file in fz.namelist():
 102             fz.extract(file, dst_dir)
 103         return dst_dir
 104     else:
 105         logger.error("%s doesn't exist", zip_src)
 106         return ""
 107
 108
 109 def unzip_csar_to_tmp(zip_src):
 110     """
 111     Unzip csar package to temp path
 112     :param zip_src:
 113     :return:
 114     """
 115     dirpath = tempfile.mkdtemp()
 116     zip_ref = zipfile.ZipFile(zip_src, 'r')
 117     zip_ref.extractall(dirpath)
 118     return dirpath
 119
 120
 121 def get_artifact_path(vnf_path, artifact_file):
 122     """
 123     Get the path of artifact file
 124     :param vnf_path:
 125     :param artifact_file:
 126     :return:
 127     """
 128     for root, dirs, files in os.walk(vnf_path):
 129         if artifact_file in files:
 130             return os.path.join(root, artifact_file)
 131     return None
 132
 133
 134 def end_with(_s_in, *suffix):
 135     """
 136     Check if end with given suffix
 137     :param _s_in:
 138     :param suffix:
 139     :return:
 140     """
 141     array = map(_s_in.endswith, suffix)
 142     if True in array:
 143         return True
 144     return False
 145
 146
 147 def filter_files(search_path, suffix):
 148     """
 149     Filter file by given suffix
 150     :param search_path:
 151     :param suffix:
 152     :return:
 153     """
 154     f_find = []
 155     file_list = os.listdir(search_path)
 156     for file_item in file_list:
 157         if end_with(file_item, suffix):
 158             f_find.append(file_item)
 159     return f_find
 160
 161
 162 def recreate_dir(path):
 163     """
 164     Recreate directory
 165     :param path:
 166     :return:
 167     """
 168     if os.path.exists(path):
 169         shutil.rmtree(path)
 170     os.makedirs(path, mode=0o777)
 171
 172
 173 def copy(src_file, dest_dir, new_file_name=None):
 174     """
 175     Copy file to given dest dir
 176     :param src_file:
 177     :param dest_dir:
 178     :param new_file_name:
 179     :return:
 180     """
 181     if not os.path.exists(dest_dir):
 182         os.makedirs(dest_dir)
 183     if new_file_name is None:
 184         dst = os.path.join(dest_dir, os.path.basename(src_file))
 185     else:
 186         dst = os.path.join(dest_dir, new_file_name)
 187     shutil.copyfile(src_file, dst)
 188     shutil.copymode(src_file, dst)