aigc-server/utils.py

import requests
import os
import json
import traceback
from log import Log
from config import set_config
log_ = Log()
config_ = set_config()


def request_post(request_url, headers, request_data):
    """
    post 请求 HTTP接口
    :param request_url: 接口URL
    :param headers: 请求头
    :param request_data: 请求参数
    :return: res_data json格式
    """
    try:
        response = requests.post(url=request_url, json=request_data, headers=headers)
        # print(response)
        if response.status_code == 200:
            res_data = json.loads(response.text)
            return res_data
        else:
            return None
    except Exception as e:
        log_.error('url: {}, exception: {}, traceback: {}'.format(request_url, e, traceback.format_exc()))
        return None


def request_get(request_url, headers, params=None):
    """
    get 请求 HTTP接口
    :param request_url: 接口URL
    :param headers: 请求头
    :param params: 请求参数
    :return: res_data json格式
    """
    try:
        response = requests.get(url=request_url, headers=headers, params=params)
        if response.status_code == 200:
            res_data = json.loads(response.text)
            return res_data
        else:
            return None
    except Exception as e:
        log_.error('url: {}, exception: {}, traceback: {}'.format(request_url, e, traceback.format_exc()))
        return None


def download_video(video_path, video_id, download_folder, ftype='mp4'):
    """下载视频"""
    if not os.path.exists(download_folder):
        os.makedirs(download_folder)
    filename = f"{download_folder}/{video_id}.{ftype}"
    # 视频已存在，则不重复下载
    if os.path.exists(filename):
        return filename
    response = requests.get(video_path, stream=True)
    if response.status_code == 200:
        with open(filename, "wb") as video_file:
            for chunk in response.iter_content(chunk_size=8192):
                video_file.write(chunk)
        return filename