import requests
import os

def download_m3u8_video(url, file_path):
    r = requests.get(url)
    if r.status_code != 200:
        print('m3u8視頻下載鏈接無效')
        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)
    print('m3u8視頻下載完成')
    return True

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

if __name__ == '__main__':
    url = '輸入m3u8視頻下載鏈接'
    ts_file_path = '輸入ts文件保存路徑'
    mp4_file_path = '輸入mp4文件保存路徑'

    download_m3u8_video(url, ts_file_path)
    convert_ts_to_mp4(ts_file_path, mp4_file_path)

import requests
import os
import threading

class Downloader(threading.Thread):
    def __init__(self, url, ts_url, file_path):
        threading.Thread.__init__(self)
        self.url = url
        self.ts_url = ts_url
        self.file_path = file_path

    def run(self):
        r = requests.get(self.ts_url, stream=True)
        if r.status_code == 200:
            with open(self.file_path, 'wb') as f:
                for chunk in r.iter_content(chunk_size=1024):
                    if chunk:
                        f.write(chunk)

def download_m3u8_video(url, file_path):
    r = requests.get(url)
    if r.status_code != 200:
        print('m3u8視頻下載鏈接無效')
        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)

    threads = []
    for i, ts_url in enumerate(ts_list):
        ts_file_path = file_path.rsplit('.', 1)[0] + f'_{i}.ts'
        thread = Downloader(url, ts_url, ts_file_path)
        thread.start()
        threads.append(thread)

    for thread in threads:
        thread.join()

    print('m3u8視頻下載完成')
    return True

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

if __name__ == '__main__':
    url = '輸入m3u8視頻下載鏈接'
    ts_file_path = '輸入ts文件保存路徑'
    mp4_file_path = '輸入mp4文件保存路徑'

    download_m3u8_video(url, ts_file_path)
    convert_ts_to_mp4(ts_file_path, mp4_file_path)

import aiohttp
import asyncio
import os

async def download_ts_file(ts_url, ts_file_path):
    # 防止ssl報(bào)錯(cuò)：
    # aiohttp.client_exceptions.ClientConnectorCertificateError: Cannot connect to host ***.****.com:443 ssl:True
    # [SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local
    # issuer certificate (_ssl.c:1123)')]
    conn = aiohttp.TCPConnector(ssl=False)
    async with aiohttp.ClientSession(connector=conn) as session:
        async with session.get(ts_url) as response:
            if response.status != 200:
                print(f'{ts_url} 下載失敗')
                return False
            with open(ts_file_path, 'wb') as f:
                while True:
                    chunk = await response.content.read(1024)
                    if not chunk:
                        break
                    f.write(chunk)
    print(f'{ts_url} 下載完成')
    return True

async def download_m3u8_video(url, file_path):
    # 防止ssl報(bào)錯(cuò)
    conn = aiohttp.TCPConnector(ssl=False)
    async with aiohttp.ClientSession(connector=conn) as session:
        async with session.get(url) as response:
            if response.status != 200:
                print('m3u8視頻下載鏈接無效')
                return False

            m3u8_text = await response.text()
            m3u8_list = m3u8_text.split('\n')
            m3u8_list = [i for i in m3u8_list if i and i[0] != '#']

            tasks = []
            for i, ts_url in enumerate(m3u8_list):
                ts_url = url.rsplit('/', 1)[0] + '/' + ts_url
                ts_file_path = file_path.rsplit('.', 1)[0] + f'_{i}.ts'
                task = asyncio.ensure_future(
                    download_ts_file(ts_url, ts_file_path))
                tasks.append(task)

            await asyncio.gather(*tasks)

    print('m3u8視頻下載完成')
    return True

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

if __name__ == '__main__':
    url = '輸入m3u8視頻下載鏈接'
    ts_file_path = '輸入ts文件保存路徑'
    mp4_file_path = '輸入mp4文件保存路徑'

    loop = asyncio.get_event_loop()
    loop.run_until_complete(download_m3u8_video(url, ts_file_path))
    convert_ts_to_mp4(ts_file_path, mp4_file_path)