import urllib.request import requests, os, threading from Crypto.Cipher import AES from src.Pacho.moviePa.tsdownload import aes_decodeclass m3u8down(object):def __init__(self, url, listheaders, dicheaders):self.url = url # 這里的url是index.m3u8地址self.headers = listheadersself.header = dicheadersself.ts_parts = []self.down_path = 'D:/workspace/download/Mp4'self.tsthreads = []self.key = Nonedef aes_decode(self, data, key):"""AES解密:param key: 密鑰（16.32）一般16的倍數:param data: 要解密的數據:return: 處理好的數據"""cryptor = AES.new(key, AES.MODE_CBC, key)plain_text = cryptor.decrypt(data)return plain_text.rstrip(b'\0') # .decode("utf-8")def to_ts(self):requests.packages.urllib3.disable_warnings()content_all = requests.get(self.url, verify=False, timeout=200).textif "#EXTM3U" not in content_all:raise BaseException("非M3U8的鏈接")if "EXT-X-VERSION" in content_all:file_line = content_all.split("\n")# print(file_line)self.get_tsurls(self.url, file_line)def get_tsurls(self, m3u8url, lines):for index, line in enumerate(lines): # m3u8文件中有ts，獲取ts地址并添加索引if "EXTINF" in line: # 找ts地址if "/" not in lines[index + 1]: # 判斷.ts是否是路徑 'DjbgADY7468014.ts' or '/20181221/.../VRYKBY4319009.ts'ts_url = m3u8url.rsplit("/", 1)[0] + "/" + lines[index + 1] # 拼出ts片段的URLelse:ts_url = m3u8url.rsplit("/", 1)[0] + "/" + lines[index + 1].rsplit("/", 1)[-1] # 拼出ts片段的URLself.ts_parts.append(ts_url)if "#EXT-X-KEY" in line:# #EXT-X-KEY:METHOD=AES-128,URI="encryption.key"key_url = m3u8url.rsplit("/", 1)[0] + "/" + line.split('"')[1]self.key = requests.get(url=key_url, timeout=120, headers=self.header).content # 獲取秘鑰def load_ts(self, ts_url, files, count):if self.key:self.auto_keydown(ts_url, files, self.header, self.key)else:self.auto_down(ts_url, files, self.headers)print('第 %d/%d 個文件下載完成, 下載地址是%s' % (count, len(self.ts_parts), ts_url))count += 1def auto_down(self, url, filename, headers): # 下載失敗后，自調用從新下載try:opener = urllib.request.build_opener() # 創建opener對象opener.addheaders = self.headersurllib.request.install_opener(opener) # 將opener對象裝入urllib.requesturllib.request.urlretrieve(url, filename)except Exception as ex:# print(ex.args, url)return self.auto_down(url, filename, headers)def auto_keydown(self, url, filename, headers, key): # 下載失敗后，自調用從新下載try:response = requests.get(url=url, timeout=120, headers=headers)with open(filename, 'ab+') as f:data = aes_decode(response.content, key)f.write(data)f.close()except Exception as ex:# print(ex.args, url)return self.auto_down(url, filename, headers, key)def threads(self):for i in range(len(self.ts_parts)):files = self.down_path + '/' + 'tsm{:0>5}.ts'.format(i)if os.path.exists(files): # 判斷文件是已下載，且文件大小變為空。是則結束本次循環，繼續循環sz = os.path.getsize(files)if not sz:os.remove(files) # 刪除空文件print("刪除空字節視頻文件", files.rsplit("/", 1)[-1])else:continuet = threading.Thread(target=self.load_ts, args=(self.ts_parts[i], files, i))self.tsthreads.append(t)def main(): # This is m3u8 urlurl = 'https://www.XXXXX.com/20200612/jDCLCWyb/1500kb/hls/index.m3u8'hd = [('User-Agent', 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1941.0 Safari/537.36')]headers = {"User-Agent": "Mozilla/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36","Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive","Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2"} m3u8 = m3u8down(url, hd, headers)m3u8.to_ts()m3u8.threads()for th in m3u8.tsthreads:th.start()for th in m3u8.tsthreads:th.join()print("{:-^20}".format("下載結束"))if __name__ == '__main__':main()

總結

以上是生活随笔為你收集整理的ts 视频下载的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。