Python學(xué)習(xí)筆記之爬蟲(五) 進程、線程、協(xié)程 實戰(zhàn) 丨【生長吧！Python】（Python爬蟲筆記）

''' 異步爬蟲實戰(zhàn)：爬取小說?'''

# http://dushu.baidu.com/api/pc/getCatalog?data={"book_id":"4306063500"} => 所有章節(jié)的內(nèi)容(名稱，cid)

# 章節(jié)內(nèi)部的內(nèi)容

# http://dushu.baidu.com/api/pc/getChapterContent?data={"book_id":"4306063500","cid":"4306063500|11349571","need_bookinfo":1}

import requests

import asyncio

import aiohttp

import aiofiles

import json

'''

1、同步操作：訪問getCatalog 拿到所有章節(jié)的書名和cid

2、異步操作：訪問getChapterContent 下載所有的文章內(nèi)容

'''

async def aiodownload(b_id, cid, title):

headers = {

'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.77 Safari/537.36 Edg/91.0.864.41'

}

data = {

"book_id": b_id,

"cid": f"{b_id}|{cid}",

"need_bookinfo": 1

}

# 將json對象變成json格式的字符串

data = json.dumps(data)

url = f'http://dushu.baidu.com/api/pc/getChapterContent?data={data}'

# 異步發(fā)送請求

async with aiohttp.ClientSession() as session:

async with session.get(url, headers=headers) as resp:

dct = await resp.json()

# 將內(nèi)容保存到文件中

async with aiofiles.open('novel/' + title, 'w', encoding='utf-8') as f:

await f.write(dct['data']['novel']['content']) ?# 把小說內(nèi)容寫出來

async def getCatalog(url):

resp = requests.get(url) ???# 同步操作，該操作沒完成，后面的就沒法進行

dct = resp.json()

tasks = []

for item in dct['data']['novel']['items']: ?# item就是對應(yīng)的每一個章節(jié)的名稱和cid

title = item['title']

cid = item['cid']

# 準備異步任務(wù)

tasks.append(aiodownload(b_id, cid, title))

await asyncio.wait(tasks)

if __name__ == '__main__':

b_id = '4306063500'

# 這里的鏈接如果用f'' 則大括號{}會發(fā)生轉(zhuǎn)義，所以用拼接的方式

url = 'http://dushu.baidu.com/api/pc/getCatalog?data={"book_id":"' + b_id + '"}'

asyncio.run(getCatalog(url))

''' 綜合訓(xùn)練：視頻網(wǎng)站的工作原理?'''

# 需要一個文件記錄：1、視頻播放順序?2、視頻存放的路徑

# m3u8 ?txt ?json ?==> 文本

# 想要抓取一個視頻：

# 1、找到m3u8(各種手段)

# 2、通過m3u8下載到ts文件

# 3、通過各種手段(不僅是編程手段)把ts文件合并為一個mp4文件

''' 抓取91看劇 簡單版-熟悉m3u8結(jié)構(gòu)?'''

'''

案例鏈接：

1、Python通過m3u8文件下載合并ts視頻

https://blog.csdn.net/weixin_38819889/article/details/103434122

2、利用python爬蟲通過m3u8文件下載ts視頻

https://zhuanlan.zhihu.com/p/70290764

什么是m3u8文件：

m3u8是蘋果公司推出一種視頻播放標準，是一種文件檢索格式，將視頻切割成一小段一小段的ts格式的視頻文件，

然后存在服務(wù)器中（現(xiàn)在為了減少I/o訪問次數(shù)，一般存在服務(wù)器的內(nèi)存中），通過m3u8解析出來路徑，然后去請求，

是現(xiàn)在比較流行的一種加載方式。目前，很多新聞視頻網(wǎng)站都是采用這種模式去加載視頻。

M3U8文件是指UTF-8編碼格式的M3U文件。M3U文件是記錄了一個索引純文本文件，打開它時播放軟件并不是播放它，

而是根據(jù)它的索引找到對應(yīng)的音視頻文件的網(wǎng)絡(luò)地址進行在線播放。原視頻數(shù)據(jù)分割為很多個TS流，每個TS流的地址記錄在m3u8文件列表中。

m3u8文件中的?m3u8標簽與屬性說明：

#EXTM3U

每個M3U文件第一行必須是這個tag，請標示作用

#EXT-X-VERSION:3

該屬性可以沒有

#EXT-X-MEDIA-SEQUENCE:140651513

每一個media URI在PlayList中只有唯一的序號，相鄰之間序號+1,

一個media URI并不是必須要包含的，如果沒有，默認為0

#EXT-X-ALLOW-CACHE:NO

是否允許客戶端對下載的視頻分段緩存用于以后播放？

#EXT-X-TARGETDURATION

指定最大的媒體段時間長（秒）。所以#EXTINF中指定的時間長度必須小于或是等于這

個最大值。這個tag在整個PlayList文件中只能出現(xiàn)一 次（在嵌套的情況下，一般有

真正ts url的m3u8才會出現(xiàn)該tag）

#EXT-X-PLAYLIST-TYPE

提供關(guān)于PlayList的可變性的信息，這個對整個PlayList文件有效，是可選的，格式

如下：#EXT-X-PLAYLIST-TYPE:：如果是VOD，則服務(wù)器不能改變PlayList 文件；

如果是EVENT，則服務(wù)器不能改變或是刪除PlayList文件中的任何部分，但是可以向該

文件中增加新的一行內(nèi)容。

#EXTINF

duration指定每個媒體段(ts)的持續(xù)時間（秒），僅對其后面的URI有效，title是下載資源的url

#EXT-X-KEY

表示怎么對media segments進行解碼。其作用范圍是下次該tag出現(xiàn)前的所有media

URI，屬性為NONE 或者?AES-128。NONE表示?URI以及IV（Initialization

Vector）屬性必須不存在，?AES-128(Advanced EncryptionStandard)表示URI

必須存在，IV可以不存在。

#EXT-X-PROGRAM-DATE-TIME

將一個絕對時間或是日期和一個媒體段中的第一個sample相關(guān)聯(lián)，只對下一個meida

URI有效，格式如#EXT-X-PROGRAM-DATE-TIME:

For example: #EXT-X-PROGRAM-DATETIME:2010-02-19T14:54:23.031+08:00

#EXT-X-ALLOW-CACHE

是否允許做cache，這個可以在PlayList文件中任意地方出現(xiàn)，并且最多出現(xiàn)一次，作

用效果是所有的媒體段。格式如下：#EXT-X-ALLOW-CACHE:

#EXT-X-ENDLIST

表示PlayList的末尾了，它可以在PlayList中任意位置出現(xiàn)，但是只能出現(xiàn)一個，格

式如下：#EXT-X-ENDLIST

流程：

1、拿到54812-1-1.html的頁面源代碼

2、從源代碼中提取到m3u8的url

3、下載m3u8

4、讀取m3u8文件，下載視頻

5、合并視頻

'''

import requests

import re

import os

url = 'https://91kanju.com/vod-play/54812-1-1.html'

headers = {

'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.77 Safari/537.36 Edg/91.0.864.41'

}

resp = requests.get(url, headers=headers)

pattern = re.compile(r"url: '(?P.*?)',", re.S)

m3u8_url = pattern.search(resp.text).group('url') ??# 拿到m3u8的地址

# print(m3u8_url)

resp.close()

# 下載m3u8文件

resp2 = requests.get(m3u8_url, headers=headers)

if not os.path.exists('video'):

os.mkdir('video', 0o755)

with open('TSVideo/哲仁王后' + '哲仁王后.m3u8', 'wb') as f:

f.write(resp2.content)

resp2.close()

print('下載完畢')

# 下載完畢以后，注釋以上全部代碼

# 解析m3u8文件

i = 1

with open('TSVideo/哲仁王后/哲仁王后.m3u8', 'r', encoding='utf-8')as f:

# print(f) ???# <_io.TextIOWrapper name='video/哲仁王后.m3u8' mode='r' encoding='utf-8'>

for line in f:

line = line.strip() ????????# 先去掉空格，空白，換行符

if line.startswith('#'): ???# 過濾掉以#開頭的內(nèi)容

continue

# print(line)

# 下載視頻片段

with requests.get(line, headers=headers) as resp3:

with open(f'video/{i}.ts', 'wb') as f2:

f2.write(resp3.content)

print(f'{line} {i} 完成')

i += 1

# 下載完成后，視頻中使用的是QuickTime Player播放器播放的

# Windows默認播放器也可以播放

''' 91看劇 復(fù)雜版?'''

'''

思路：

1、拿到主頁面的源代碼，找到iframe對應(yīng)的url

2、從iframe的頁面源代碼中拿到m3u8文件的地址

3、下載第一層m3u8文件?-> 下載第二層m3u8文件(視頻存放路徑)

4、下載視頻

5、下載密鑰，進行解密操作

6、合并所有ts文件為一個mp4文件

'''

import requests

from bs4 import BeautifulSoup

import re

import asyncio

import aiohttp

import aiofiles

import os

from Cryptodome.Cipher import AES

'''

官方文檔：

首頁：https://www.pycryptodome.org/en/latest/

AES：https://www.pycryptodome.org/en/latest/src/examples.html#encrypt-data-with-aes

python安裝AES庫(Advanced Encryption Standard)及使用

https://blog.csdn.net/m0_52693073/article/details/110943828

python3.8上Crypto用不了，改用pycryptodomex

安裝pycryptodomex：

pip install pycryptodomex

用法：

from Cryptodome.Cipher import AES

aes = AES.new(key, AES.MODE_CBC, IV=)

content = aes.decrypt(content)

IV格式如果是下面這種會報錯：

'0x674B2D91C06186399C33DC5901307461'

報錯如下：

ValueError: Incorrect IV length (it must be 16 bytes long)

解決方法：需要把hex轉(zhuǎn)換成bytes:

from binascii import unhexlify

iv = unhexlify(IV_str.replace('0x', ''))

結(jié)果：

unhexlify('672482D91C06186399C33DC5901307461')

b'gK-\x91\xc0a\x869\x9c3\xdcY\x010ta'

'''

headers = {

'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.77 Safari/537.36 Edg/91.0.864.41'

}

def get_iframe_src(url):

resp = requests.get(url, headers=headers)

# 獲取iframe標簽的src，整個頁面只有一個iframe標簽，使用bs最合適

main_page = BeautifulSoup(resp.text, 'html.parser')

src = main_page.find('iframe').get('src')

# print(src) ???# https://boba.52kuyun.com/share/xfPs9NPHvYGhNzFp

return src

def get_first_meu8_url(url):

resp = requests.get(url, headers=headers)

# 想從script中獲取內(nèi)容，最好的方式是re

obj = re.compile(r'var main = "(.*?)"', re.S)

m3u8_url = obj.search(resp.text).group('m3u8_url')

# print(m3u8_url) ??# /20170907/Moh2l9zV/index.m3u8?sign=548ae366a075f0fie7c76af215aa18e1

return m3u8_url

def download_m3u8_file(url, file_name):

resp = requests.get(url, headers=headers)

with open(file_name, 'wb') as f:

f.write(resp.content)

# 視頻https://www.bilibili.com/video/BV1Mf4y1s7ds?p=76

async def download_ts(ts_url, name, session):

async with session.get(ts_url, headers=headers) as resp:

async with aiofiles.open(f'TSVideo/{name}', 'wb') as f:

# 把下載到的內(nèi)容寫入到文件中

# 不論是發(fā)送請求也好，創(chuàng)建文件也好，都是異步操作

# 不加await報錯Coroutine 'write' is not awaited

await f.write(await resp.content.read())

# 最后的下載提示信息

print(f'{name} 下載完畢！')

async def aio_download(prefix_url): # https://boba.52kuyun.com//20170907/Moh2l9zV/hls

tasks = []

# 提前準備好session，傳遞給download_ts函數(shù)使用

async with aiohttp.ClientSession() as session:

async with ?aiofiles.open('越獄第一季第一集_second_m3u8_url.txt', 'r', encoding='utf-8') as f:

# for前不加async則f報錯：Expected type 'collections.Iterable', got 'AsyncTextIOWrapper' instead

async for line in f:

if line.startswith('#'):

continue

# line就是xxxxxx.ts

line = line.strip() # 去點空白和換行

# 拼接真正的ts路徑

ts_url = prefix_url + line

# 準備任務(wù)列表tsks之后，創(chuàng)建任務(wù)

task = asyncio.create_task(download_ts(ts_url, line, session)) ??# 創(chuàng)建任務(wù)，asyncio.create_task放入?yún)f(xié)程對象

# 把每個任務(wù)放入tasks列表

tasks.append(task)

# 必須在for循環(huán)結(jié)束之后

await asyncio.wait(tasks) ??# 等待任務(wù)結(jié)束

'''

async def wait(fs, *, loop=None, timeout=None, return_when=ALL_COMPLETED):

Wait for the Futures and coroutines given by fs to complete.

等待fs給出的Futures和協(xié)程完成

The sequence futures must not be empty.

序列futures不能為空。

Coroutines will be wrapped in Tasks.

協(xié)程將封裝在任務(wù)中

Returns two sets of Future: (done, pending).

返回兩組Future:(done, pending) ???done:已完成的，pending:掛起的

Usage:

done, pending = await asyncio.wait(fs)

Note: This does not raise TimeoutError! Futures that aren't done

when the timeout occurs are returned in the second set.

'''

def get_key(url):

resp = requests.get(url, headers=headers)

return resp.text

'''

AES加密用法：

from Cryptodome.Cipher import AES

aes = AES.new(key, AES.MODE_CBC, IV=)

content = aes.decrypt(content)

'''

async def dec_ts(file_name, key):

# AES.new(key, mode, iv, IV, nonce, segment_size, mac_len, assoc_len, initial_value, counter, use_aesni)

aes = AES.new(key, AES.MODE_CBC, IV=b'0000000000000000')# IV為16位

async with aiofiles.open(f'TSVideo/{file_name}', 'rb') as f1,\

aiofiles.open(f'TSVideo/temp_{file_name}', 'wb') as f2: # \是換行

read = await f1.read() ?# 從源文件讀取內(nèi)容、

await f2.write(aes.decrypt(read)) ??# 把解密后的內(nèi)容寫入文件

print(f'{file_name} 處理完畢！')

# 解密，少了前8分鐘 視頻https://www.bilibili.com/video/BV1Mf4y1s7ds?p=77

async def aio_dec(key):

tasks = []

async with aiofiles.open('越獄第一季第一集_second_m3u8_url.txt', 'r', encoding='utf-8') as f:

async for line in f:

if line.startswith('#'):

continue

line = line.strip() # 去掉空白、換行符

# 開始創(chuàng)建異步任務(wù)

task = asyncio.create_task(dec_ts(line, key))

tasks.append(task)

await asyncio.wait(tasks)

# 合并成mp4文件：https://www.bilibili.com/video/BV1Mf4y1s7ds?p=78

def merge_ts():

# 合并ts文件命令：

# mac: cat 1.ts 2.ts 3.ts > xxx.mp4

# windows: copy /b 1.ts+2.ts+3.ts xxx.mp4

# 準備一個空列表，存放所有的ts文件，用來執(zhí)行合并操作

lst = []

with open('越獄第一季第一集_second_m3u8_url.txt', 'r', encoding='utf-8') as f:

for line in f:

if line.startswith('#'):

continue

line = line.strip()

lst.append(f'TSVideo/temp_{line}')

# 拼接操作

s = '+'.join(lst) ??# 得到1.ts+2.ts+3.ts

# 執(zhí)行操作系統(tǒng)命令，合并視頻

os.system(f'copy /b {s}?movie.mp4')

print('合并視頻成功！')

def main(url):

# 1、拿到主頁面的源代碼，找到iframe對應(yīng)的url

iframe_src = get_iframe_src(url)

# 2、拿到第一層meu8文件的地址

first_m3u8_url = get_first_meu8_url(iframe_src)

# 拿到iframe的域名??https://boba.52kuyun.com/share/xfPs9NpHvYGhNzFp

iframe_domain = iframe_src.split('/share')[0]

# 拼接出真正的m3u8的-

first_m3u8_url = iframe_domain + first_m3u8_url

# print(first_m3u8_url)

# https://boba.52kuyun.com//20170907/Moh2l9zV/index.m3u8?sign=548ae366a075f0fie7c76af215aa18e1

# 3.1、下載第一層m3u8文件

download_m3u8_file(first_m3u8_url, '越獄第一季第一集_first_m3u8_url.txt')

print('第一層m3u8下載完畢！')

# 3.2、下載第二層m3u8文件

with open('越獄第一季第一集_first_m3u8_url.txt', 'r', encoding='utf-8') as f:

for line in f:

if line.startswith('#'):

continue

else:

line = line.strip() # 去點空白、換行符

# 準備拼接第二層m3u8的下載路徑

# https://boba.52kuyun.com//20170907/Moh2l9zV/ + hls/index.m3u8

second_m3u8_url = first_m3u8_url.split('index.m3u8')[0] + line

# print(second_m3u8_url)

# https://boba.52kuyun.com//20170907/Moh2l9zV/hls/index.m3u8

# ts文件下載路徑

# https://boba.52kuyun.com//20170907/Moh2l9zV/hls/cFN803436000.ts

# 下載操作

download_m3u8_file(second_m3u8_url, '越獄第一季第一集_second_m3u8_url.txt')

print('第二層m3u8下載完畢！')

# 4、下載視頻。 ?視頻https://www.bilibili.com/video/BV1Mf4y1s7ds?p=75

second_m3u8_url_prefix = second_m3u8_url.replace('index.m3u8', '')

# 異步協(xié)程，IO操作多，任務(wù)量大

asyncio.run(aio_download(second_m3u8_url_prefix))

# 5.1 拿到密鑰

key_url = second_m3u8_url_prefix + 'key.key' ???# 偷懶寫法，正常應(yīng)該去m3u8文件里找

key = get_key(key_url)

# 5.2 解密

asyncio.run(aio_dec(key))

# 6、合并ts文件為mp4文件

merge_ts()

if __name__ == '__main__':

url = 'https://www.91kanju.com/vod-play/541-2-1.html'

main(url)

'''

https://www.bilibili.com/video/BV1Mf4y1s7ds?p=79 ?02:31

簡單的問題復(fù)雜化，復(fù)雜的問題簡單化

復(fù)雜的問題簡單化：

在這么復(fù)雜的需求下我們怎么一步一步地搞定呢？

首先要縷清思路，把要做的事先羅列出來，一個一個拿出來分解。

在編寫過程中，你可能會有新的思路，把它加入到原先的結(jié)構(gòu)中，按照結(jié)構(gòu)逐個解決。

簡單的問題復(fù)雜化：

你的編程能力為什么一直停滯不前？干了好幾年還在做CRUD。

把簡單的問題做到極致，比如拿 秒殺 來舉例，

'''

Python 爬蟲

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實的內(nèi)容，請聯(lián)系我們jiasou666@gmail.com 處理，核實后本網(wǎng)站將在24小時內(nèi)刪除侵權(quán)內(nèi)容。