aigc-server/gpt_process.py

import os
import time
import requests
import json
import traceback
from utils import download_video
from audio_process import get_wav
from xunfei_asr import call_asr
from config import set_config
from log import Log

config_ = set_config()
log_ = Log()


def request_gpt(prompt):
    """
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {config_.GPT_OPENAI_API_KEY}',
    }
    proxies = config_.PROXIES

    json_data = {
        'model': 'gpt-3.5-turbo',
        'messages': [
            {
                'role': 'user',
                'content': prompt,
            },
        ],
    }
    response = requests.post(url=config_.GPT_HOST, headers=headers, json=json_data, proxies=proxies)
    """
    retry_count = 0
    result_content = None
    while retry_count < config_.RETRY_MAX_COUNT:
        retry_count += 1
        try:
            response = requests.post(url=config_.GPT_URL, json={'content': prompt, 'auth': config_.GPT_OPENAI_API_KEY})
            # print(response.json())
            # print(response.json()['choices'][0]['message']['content'])
            # print('\n')
            # result_content = response.json()['choices'][0]['message']['content']
            # log_.info(f"response.text: {response.text}")
            res_data = json.loads(response.text)
            if res_data['code'] != 0:
                time.sleep(10)
                continue
            result_content = res_data['data']['choices'][0]['message']['content']
        except Exception:
            time.sleep(10)
            continue
    return result_content


def title_generate(video_id, video_path):
    """
    视频生成标题
    :param video_id: videoId
    :param video_path: videoPath
    :return:
    """
    generate_filepath = dict()
    # 1. 下载视频
    # log_.info(f"debug: title_generate 1")
    video_file_path = download_video(video_path=video_path, video_id=video_id, download_folder='videos')
    generate_filepath['video_file_path'] = video_file_path
    # log_.info({'videoId': video_id, 'video_file_path': video_file_path})
    # 2. 获取视频中的音频
    # log_.info(f"debug: title_generate 2")
    audio_path = get_wav(video_path=video_file_path)
    generate_filepath['audio_path'] = audio_path
    # log_.info({'videoId': video_id, 'audio_path': audio_path})
    # 3. asr
    # log_.info(f"debug: title_generate 3")
    dialogue_path, asr_res = call_asr(audio_path=audio_path)
    generate_filepath['dialogue_path'] = dialogue_path
    log_.info({
        'asrResult': {'videoId': video_id, 'asrRes': asr_res}
    })
    # 4. gpt产出结果
    # log_.info(f"debug: title_generate 4")
    prompt = f"{config_.GPT_PROMPT['title']['prompt2']}{asr_res.strip()}"
    gpt_res = request_gpt(prompt=prompt)

    return gpt_res, generate_filepath
    # except ConnectionResetError:
    #     log_.info(video_id)
    # except Exception as e:
    #     log_.info(traceback.format_exc())
    # else:
    #     print(gpt_res)


if __name__ == '__main__':
    title_generate(video_id='001', video_path='')