经常用到的函数

创建文件夹和删除文件夹的函数

def make_dirs(*dirs):for new_dir in dirs:if not os.path.exists(new_dir):try:os.makedirs(new_dir)except RuntimeError:return Falsereturn Truedef remove_files(file_path_list):""" 删除列表中指定路径文件Args:file_path_list: 待删除文件路径 list"""try:if not isinstance(file_path_list, list):file_path_list = [file_path_list]for file_path in file_path_list:if file_path and os.path.isfile(file_path):os.remove(file_path)except Exception as ex:return exdef remove_dirs(dir_list):""" 删除列表中指定路径文件Args:dir_list: 待删除文件目录 list"""try:if not isinstance(dir_list, list):dir_list = [dir_list]for file_dir in dir_list:if file_dir and os.path.exists(file_dir):shutil.rmtree(file_dir)except Exception as ex:return ex

下载的函数

图像下载

def load_image(image: Union[str, PIL.Image.Image]) -> PIL.Image.Image:"""Loads `image` to a PIL Image.Args:image (`str` or `PIL.Image.Image`):The image to convert to the PIL Image format.Returns:`PIL.Image.Image`:A PIL Image."""if isinstance(image, str):if image.startswith("http://") or image.startswith("https://"):image = PIL.Image.open(requests.get(image, stream=True).raw)elif os.path.isfile(image):image = PIL.Image.open(image)else:raise ValueError(f"Incorrect path or url, URLs must start with `http://` or `https://`, and {image} is not a valid path")elif isinstance(image, PIL.Image.Image):image = imageelse:raise ValueError("Incorrect format used for image. Should be an url linking to an image, a local path, or a PIL image.")image = PIL.ImageOps.exif_transpose(image)image = image.convert("RGB")return image

单个文件的下载

下面的两个函数都可以进行文件的下载

import urllib
from urllib import request, errordef download_file(url, save_path, retry=5):"""采用 urlretrieve 下载单个文件, 并根据保存目录及指定文件名, 保存下载文件Args:url: 待下载文件 urlsave_path: 下载文件保存路径retry: 最多重复下载次数Returns:exitstatus: 是否下载成功url: 待下载文件 url (若下载失败, 可查看相应的 url)"""headers = {  # 用户代理，伪装浏览器用户访问网址'User-Agent': 'Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3941.4 ''Safari/537.36'}count = 1while count <= retry:try:r = request.Request(url, headers=headers)r1 = request.urlopen(r, timeout=2)try:urllib.request.urlretrieve(url, save_path)return Trueexcept socket.timeout:count += 1except Exception as e:count += 1except error.HTTPError as e:count += 1except error.URLError as e:count += 1except Exception as e:count += 1if count > retry:return False

上下文处理器函数Profile

import contextlibclass Profile(contextlib.ContextDecorator):# Usage: @Profile() decorator or 'with Profile():' context managerdef __init__(self, t=0.0):self.t = tself.cuda = torch.cuda.is_available()def __enter__(self):self.start = self.time()return selfdef __exit__(self, type, value, traceback):self.dt = self.time() - self.start  # delta-timeself.t += self.dt  # accumulate dtdef time(self):if self.cuda:torch.cuda.synchronize()return time.time()# 用法入下：
upload_oss_elapsed = tools.Profile()
with upload_oss_elapsed:oss_dir = os.path.join(server_conf.path.oss_root_dir, day, key)stat_up = tools.upload_dir2oss(bucket, result_data_dir, oss_dir, upload=server_conf.oss.OSS_UPLOAD)if not stat_up:_info = "[{}] upload to oss failed!!".format(key)self.logger.error(_info)response_dict = make_response(message=_info, requestId=requestId, typeId=typeId,userId=userId, labLogId=labLogId,imgsourceUrl=ori_image_url, videosourceUrl=ori_video_url,logosourceUrl=ori_logo_url, service_status_code=406)tools.remove_dirs([ori_data_dir, result_data_dir])return self.post_process(response_dict), 406self.logger.info(f"[{key}] byte adx result upload, elapsed time: {process_video_elapsed.t}")