DC學(xué)院(https://www.dcxueyuan.com/index.html)是數(shù)據(jù)城堡（https://www.pkbigdata.com/）下面的一個(gè)教學(xué)平臺(tái)，有許多大數(shù)據(jù)和人工智能領(lǐng)域的優(yōu)質(zhì)課程萄金。
image.png
下面介紹如何使用python來(lái)爬取DC學(xué)院上已經(jīng)購(gòu)買的課程业崖。
環(huán)境
開(kāi)發(fā)環(huán)境
- window 7 x64
- python3.7 (Anacona 3)
- vscode 編輯器
使用的python包有 - requests 用于模擬http請(qǐng)求
- bs4 用來(lái)解析html文檔
- re 正則表達(dá)式
- pycryptodome 用于AES解密
代碼實(shí)現(xiàn)
這個(gè)代碼主要實(shí)現(xiàn)了使用python爬取m3u8格式的加密文件，代碼很簡(jiǎn)單
爬取m3u8文件也可以使用m3u8包,感興趣的話可以試試，本代碼使用的requests+re 來(lái)實(shí)現(xiàn)m3u8列表文件的解析。
直接看代碼比較好。
#-*- coding:utf-8 -*-
import requests
from requests.packages import urllib3
import os,re
from Crypto.Cipher import AES
import hashlib
from bs4 import BeautifulSoup
urllib3.disable_warnings()
#全局變量胶惰，requests庫(kù)請(qǐng)求使用headers參數(shù)
headers = {
'User-Agent':'Mozilla/5.0 (Windows NT 6.1; Win64; x64; rv:71.0) Gecko/20100101 Firefox/71.0',
'Referer':'https://www.dcxueyuan.com/'
}
#用于保持會(huì)話
s = requests.Session()
#下載視頻
#m3u8_url為視頻的m3u8地址,filename為視頻保存的文件名
def download_video(m3u8_url,filename):
global headers,s
res = s.get(m3u8_url,headers=headers,verify=False)
if res.status_code == 200:
data = res.text.strip()
#print data
#使用正則表達(dá)式提取出AES加密方法，密鑰的url和iv
aes_method,key_url,iv_str = re.findall(r'#EXT-X-KEY:METHOD=(.*?),URI="(.*?)",IV=0x(.*?)\n',data)[0]
#提取所有的ts文件的uri
ts_uri_list = re.findall(r'(.*?.ts)\n',data)
#獲取密鑰
key_str = get_key(key_url)
print(key_str)
print(iv_str)
#print(ts_uri_list)
#下載視頻內(nèi)容
content=b''
for ts_url in ts_uri_list:
url_base = m3u8_url[:m3u8_url.rfind('/')+1]
res1 = s.get(url_base+ts_url,headers=headers,verify=False)
if res1.status_code == 200:
#對(duì)ts片斷進(jìn)行解密霞溪，拼接
content += decrypt_single_ts(res1.content,iv_str,key_str)
#保存為文件
open('%s' % filename,'wb').write(content)
print(filename,'下載完畢')
#根據(jù)key和iv對(duì)ts進(jìn)行解密
#key_str為AES密鑰孵滞，為十六進(jìn)制字符串
#iv_str為AES的初始向量，為十六進(jìn)制字符串
def decrypt_single_ts(ts,iv_str,key_str):
#將key和iv轉(zhuǎn)化成bytes類型
iv = bytes.fromhex(iv_str)
key = bytes.fromhex(key_str)
#填充最后一個(gè)塊
pad_len = AES.block_size - len(ts) % AES.block_size
if pad_len != AES.block_size:
ts = ts[:-pad_len] + bytes([0] * pad_len)
#解密
cipher = AES.new(key, AES.MODE_CBC, iv=iv)
out_data = cipher.decrypt(ts)
#提出解密后的真實(shí)數(shù)據(jù)
if pad_len != AES.block_size:
out_data = out_data[:-pad_len]
return out_data
#根據(jù)密鑰的url獲取密鑰
def get_key(key_url):
res = s.get(key_url,headers=headers,verify=False)
if res.status_code == 200:
return res.content.hex()
return ''
#計(jì)算x的md5值
def md5(x):
m = hashlib.md5()
m.update(x.encode('utf-8'))
return m.hexdigest()
#下載課程資料
#url為課程資料的url鸯匹，filename為文件名
def download_info(url,filename):
#其文件為pdf
if filename.endswith('.pdf'):
res = s.get(
url,
headers=headers,
verify=False
)
if res.status_code == 200:
open(filename,'wb').write(res.content)
print(filename,'下載完畢')
#若文件為html
if filename.endswith('.html'):
res = s.get(
url,
headers=headers,
verify=False
)
if res.status_code == 200:
content = res.text
#爬取頁(yè)面中的css
for x in BeautifulSoup(res.text,'lxml').find_all('link'):
url = x['href']
target_filename = 'data\\static\\%s' % md5(url)
open(target_filename,'w',encoding='utf-8').write(s.get(url,headers=headers,verify=False).text)
content = content.replace(url,'../../static/%s' % md5(url))
print(res.encoding)
open(filename,'w',encoding=res.encoding).write(content)
print(filename,'下載完畢')
def main():
#登陸
res = s.post(
'https://www.dcxueyuan.com/api/user/common/login.json',
headers = headers,
verify=False,
data={
'username':'xxxxxxx',#用戶名
'password':'xxxxxxx'#密碼
}
)
if res.status_code == 200:
if res.json()['login']:
print('登陸成功')
userid = res.json()['data']['loginUserIds']['id']
print('userid',userid)
#請(qǐng)求該用戶購(gòu)買的所有課程
res = s.get(
'https://www.dcxueyuan.com/api/common/center/courseCertificateList.json',
headers=headers,
verify=False,
params={"pageNo":"1","pageSize":"100","userId":userid}
)
if res.status_code == 200:
print(res.json())
for x in res.json()['data']['certificateList']:
#課程名稱
course_name = x['name']
#課程id
course_id = x['course_id']
print(course_name)
res = s.get(
'https://www.dcxueyuan.com/api/common/course/getCourseCatalogue.json',
headers = headers,
verify = False,
params={"courseId":course_id}
)
if res.status_code == 200:
for i,x in enumerate(res.json()['data']['courseCatalogue']):
#章的名字
catalogue = x['name']
catalogue = '第%d章 %s' % (i+1,catalogue)
print(catalogue)
for j,y in enumerate(x['trainClassMapList']):
#節(jié)的名稱
class_name = y['name']
#課時(shí)id
class_id = y['id']
#請(qǐng)求課程的下載連接
res = s.get(
'https://www.dcxueyuan.com/api/user/getVideoUrl.json',
headers=headers,
verify=False,
params = {"classId":class_id,"pixel":"1080P","videoFrom":"2"}
)
if res.status_code == 200:
#print(res.json())
#print(res.json()['data']['class']['name'])
#構(gòu)造課時(shí)的完整名稱
class_name = '%d-%d %s'%(i,j+1,class_name)
print(class_name)
#課時(shí)的視頻m3u8地址
m3u8_url = res.json()['data']['url']
print(m3u8_url)
#存儲(chǔ)的路徑
path = 'data\\%s\\%s' % (course_name,catalogue)
#若不存在該路徑坊饶，創(chuàng)建這個(gè)路徑
if not os.path.isdir(path):
os.makedirs(path)
#視頻文件名
filename = '%s\\%s.mp4' % (path,class_name)
#若該視頻文件不存在，下載它
if not os.path.isfile(filename):
download_video(m3u8_url,filename)
#請(qǐng)求課程資料的url
res = s.get(
'https://www.dcxueyuan.com/api/user/course/downDataOuter.json',
headers=headers,
verify=False,
params={"dataId":class_id}
)
if res.status_code == 200:
#print(res.json())
#課程資料的title
x = res.json()['data'].get('file','')
if x:
#課程資料的文件名
filename = '%s\\%s' % (path,x['name'])
try:
#下載課程資料
download_info(x['path'],filename)
except Exception as e:
#報(bào)錯(cuò)的話不處理
print(e)
pass
main()
