python爬取基于m3u8協(xié)議的ts文件并合并
前言
簡單學(xué)習(xí)過網(wǎng)絡(luò)爬蟲,只是之前都是照著書上做并發(fā),大概能理解,卻還是無法自己用到自己項目中,這里自己研究實現(xiàn)一個網(wǎng)頁嗅探HTML5播放控件中基于m3u8協(xié)議ts格式視頻資源的項目,并未考慮過復(fù)雜情況,畢竟只是練練手.
源碼
# coding=utf-8
import asyncio
import multiprocessing
import os
import re
import time
from math import floor
from multiprocessing import Manager
import aiohttp
import requests
from lxml import html
import threading
from src.my_lib import retry
from src.my_lib import time_statistics
class M3U8Download:
_path = "./resource\\" # 本地文件路徑
_url_seed = None # 資源所在鏈接前綴
_target_url = {} # 資源任務(wù)目標字典
_mode = ""
_headers = {"User-agent": "Mozilla/5.0"} # 瀏覽器代理
_target_num = 100
def __init__(self):
self._ml = Manager().list() # 進程通信列表
if not os.path.exists(self._path): # 檢測本地目錄存在否
os.makedirs(self._path)
exec_str = r'chcp 65001'
os.system(exec_str) # 先切換utf-8輸出,防止控制臺亂碼
def sniffing(self, url):
self._url = url
print("開始嗅探...")
try:
r = requests.get(self._url) # 訪問嗅探網(wǎng)址,獲取網(wǎng)頁信息
except:
print("嗅探失敗,網(wǎng)址不正確")
os.system("pause")
else:
tree = html.fromstring(r.content)
try:
source_url = tree.xpath('//video//source/@src')[0] # 嗅探資源控制文件鏈接,這里只針對一個資源控制文件
# self._url_seed = re.split("/\w+\.m3u8", source_url)[0] # 從資源控制文件鏈接解析域名
except:
print("嗅探失敗,未發(fā)現(xiàn)資源")
os.system("pause")
else:
self.analysis(source_url)
def analysis(self, source_url):
try:
self._url_seed = re.split("/\w+\.m3u8", source_url)[0] # 從資源控制文件鏈接解析域名
with requests.get(source_url) as r: # 訪問資源控制文件,獲得資源信息
src = re.split("\n*#.+\n", r.text) # 解析資源信息
for sub_src in src: # 將資源地址儲存到任務(wù)字典
if sub_src:
self._target_url[sub_src] = self._url_seed + "/" + sub_src
except Exception as e:
print("資源無法成功解析", e)
os.system("pause")
else:
self._target_num = len(self._target_url)
print("sniffing success!!!,found", self._target_num, "url.")
self._mode = input(
"1:-> 單進程(Low B)\n2:-> 多進程+多線程(網(wǎng)速開始biubiu飛起!)\n3:-> 多進程+協(xié)程(最先進的并發(fā)!!!)\n")
if self._mode == "1":
for path, url in self._target_url.items():
self._download(path, url)
elif self._mode == "2" or self._mode == "3":
self._multiprocessing()
def _multiprocessing(self, processing_num=4): # 多進程,多線程
target_list = {} # 進程任務(wù)字典,儲存每個進程分配的任務(wù)
pool = multiprocessing.Pool(processes=processing_num) # 開啟進程池
i = 0 # 任務(wù)分配標識
for path, url in self._target_url.items(): # 分配進程任務(wù)
target_list[path] = url
i += 1
if i % 10 == 0 or i == len(self._target_url): # 每個進程分配十個任務(wù)
if self._mode == "2":
pool.apply_async(self._sub_multithreading, kwds=target_list) # 使用多線程驅(qū)動方法
else:
pool.apply_async(self._sub_coroutine, kwds=target_list) # 使用協(xié)程驅(qū)動方法
target_list = {}
pool.close() # join函數(shù)等待所有子進程結(jié)束
pool.join() # 調(diào)用join之前,先調(diào)用close函數(shù),否則會出錯。執(zhí)行完close后不會有新的進程加入到pool
while True:
if self._judge_over():
self._combine()
break
def _sub_multithreading(self, **kwargs):
for path, url in kwargs.items(): # 根據(jù)進程任務(wù)開啟線程
t = threading.Thread(target=self._download, args=(path, url,))
t.start()
@retry()
def _download(self, path, url): # 同步下載方法
with requests.get(url, headers=self._headers) as r:
if r.status_code == 200:
with open(self._path + path, "wb")as file:
file.write(r.content)
self._ml.append(0) # 每成功一個就往進程通信列表增加一個值
percent = '%.2f' % (len(self._ml) / self._target_num * 100)
print(len(self._ml), ": ", path, "->OK", "\tcomplete:", percent, "%") # 顯示下載進度
else:
print(path, r.status_code, r.reason)
def _sub_coroutine(self, **kwargs):
tasks = []
for path, url in kwargs.items(): # 根據(jù)進程任務(wù)創(chuàng)建協(xié)程任務(wù)列表
tasks.append(asyncio.ensure_future(self._async_download(path, url)))
loop = asyncio.get_event_loop() # 創(chuàng)建異步事件循環(huán)
loop.run_until_complete(asyncio.wait(tasks)) # 注冊任務(wù)列表
async def _async_download(self, path, url): # 異步下載方法
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=self._headers) as resp:
try:
assert resp.status == 200, "E" # 斷言狀態(tài)碼為200,否則拋異常,觸發(fā)重試裝飾器
with open(self._path + path, "wb")as file:
file.write(await resp.read())
except Exception as e:
print(e)
else:
self._ml.append(0) # 每成功一個就往進程通信列表增加一個值
percent = '%.2f' % (len(self._ml) / self._target_num * 100)
print(len(self._ml), ": ", path, "->OK", "\tcomplete:", percent, "%") # 顯示下載進度
def _combine(self): # 組合資源方法
try:
print("開始組合資源...")
identification = str(floor(time.time()))
exec_str = r'copy /b "' + self._path + r'*.ts" "' + self._path + 'video' + identification + '.mp4"'
os.system(exec_str) # 使用cmd命令將資源整合
exec_str = r'del "' + self._path + r'*.ts"'
os.system(exec_str) # 刪除原來的文件
except:
print("資源組合失敗")
else:
print("資源組合成功!")
def _judge_over(self): # 判斷是否全部下載完成
if len(self._ml) == len(self._target_url):
return True
return False
@time_statistics
def app():
multiprocessing.freeze_support()
url = input("輸入嗅探網(wǎng)址:\n")
m3u8 = M3U8Download()
m3u8.sniffing(url)
# m3u8.analysis(url)
if __name__ == "__main__":
app()
這里是兩個裝飾器的實現(xiàn):
import time
def time_statistics(fun):
def function_timer(*args, **kwargs):
t0 = time.time()
result = fun(*args, **kwargs)
t1 = time.time()
print("Total time running %s: %s seconds" % (fun.__name__, str(t1 - t0)))
return result
return function_timer
def retry(retries=3):
def _retry(fun):
def wrapper(*args, **kwargs):
for _ in range(retries):
try:
return fun(*args, **kwargs)
except Exception as e:
print("@", fun.__name__, "->", e)
return wrapper
return _retry
打包成exe文件
使用PyInstaller -F download.py將程序打包成單個可執(zhí)行文件.
這里需要注意一下,因為程序含有多進程,需要在執(zhí)行前加一句multiprocessing.freeze_support(),不然程序會反復(fù)執(zhí)行多進程前的功能.
關(guān)于協(xié)程
協(xié)程在Python3.5進化到了async await版本,用 async 標記異步方法,在異步方法里對耗時操作使用await標記.這里使用了一個進程驅(qū)動協(xié)程的方法,在進程池創(chuàng)建多個協(xié)程任務(wù),使用asyncio.get_event_loop()創(chuàng)建協(xié)程事件循環(huán),使用run_until_complete()注冊協(xié)程任務(wù),asyncio.wait()方法接收一個任務(wù)列表進行協(xié)程注冊.
關(guān)于裝飾器
裝飾器源于閉包原理,這里使用了兩種裝飾器.
- @time_statistics:統(tǒng)計耗時,裝飾器自己無參型
- @retry():設(shè)置重試次數(shù),裝飾器自己有參型
- 按我理解是有參型是將無參型裝飾器包含在內(nèi)部,而調(diào)用是加()的,關(guān)于():
- 不帶括號時,調(diào)用的是這個函數(shù)本身
- 帶括號(此時必須傳入需要的參數(shù)),調(diào)用的是函數(shù)的return結(jié)果
關(guān)于CMD控制臺
程序會使用CMD命令來將下載的ts文件合并.
因為CMD默認使用GB2312編碼,調(diào)用os.system()需要先切換成通用的UTF-8輸出,否則系統(tǒng)信息會亂碼.
而且使用cmd命令時參數(shù)最好加雙引號,以避免特殊符號報錯.
以上就是本文的全部內(nèi)容,希望對大家的學(xué)習(xí)有所幫助,也希望大家多多支持腳本之家。
相關(guān)文章
Python尾遞歸優(yōu)化實現(xiàn)代碼及原理詳解
這篇文章主要介紹了Python尾遞歸優(yōu)化實現(xiàn)代碼及原理詳解,文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-10-10
python 實現(xiàn)GUI(圖形用戶界面)編程詳解
今天小編就為大家分享一篇python 實現(xiàn)GUI(圖形用戶界面)編程詳解,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2019-07-07
python用selenium打開chrome瀏覽器保持登錄方式
大家好,本篇文章主要講的是python用selenium打開chrome瀏覽器保持登錄方式,感興趣的同學(xué)趕快來看一看吧,對你有幫助的話記得收藏一下2022-02-02

