carry_video_rewriting/common/download_video.py

import html
import os
import requests


class DownLoad:
    @classmethod
    def download_video(cls, video_url, video_path_url):
        video = video_path_url + 'video.mp4'
        url_type = html.unescape(video_url).split('?')[0]
        if ".m3u8" in url_type:
            ts_video = video_path_url + 'video.ts'
            status_code = cls.download_m3u8_video(video_url, ts_video)
            if status_code == False:
                return video
            cls.convert_ts_to_mp4(ts_video, video)
            return video
        else:
            try:
                for i in range(3):
                    payload = {}
                    headers = {}
                    response = requests.request("GET", video_url, headers=headers, data=payload, timeout=240)
                    if response.status_code == 200:
                        # 以二进制写入模式打开文件

                        with open(f"{video}", "wb") as file:
                            # 将响应内容写入文件
                            file.write(response.content)
                        return video
                return video
            except Exception:
                return video

    @classmethod
    def download_m3u8_video(cls ,url, file_path):
        r = requests.get(url)
        if r.status_code != 200:
            return False
        m3u8_list = r.text.split('\n')
        m3u8_list = [i for i in m3u8_list if i and i[0] != '#']

        ts_list = []
        for ts_url in m3u8_list:
            ts_url = url.rsplit('/', 1)[0] + '/' + ts_url
            ts_list.append(ts_url)
        with open(file_path, 'wb') as f:
            for ts_url in ts_list:
                r = requests.get(ts_url)
                if r.status_code == 200:
                    f.write(r.content)
        return True

    @classmethod
    def convert_ts_to_mp4(cls, ts_file_path, mp4_file_path):
        os.system(f'ffmpeg -i {ts_file_path} -c copy {mp4_file_path}')