背景:
客户端上报md5校验失败,想复现,但是又不能一条一条的手工验证cdn资源
目标:
给定m3u8文件地址,自动下载与校验全部分片的md5
代码:
# coding=utf-8
import sys
import requests
import hashlib
def checkCdnFile(url):
"""
检测一个segment
"""
# url = 'http://valipl.cp31.ott.cibntv.net/67756D6080932713CFC02204E/03000500005DBC8F1334A405C99AC17FE460FC-F6E0-4DC0-8AE4-579B006E3EC6-00143.ts?ccode=01010201&duration=2647&expire=18000&psid=56bfdf60419bdb7f42e668114df58a1341cb3&ups_client_netip=ddc2f33f&ups_ts=1593570410&ups_userid=&utid=WkJ1nK5mJecDAE9Wggk%2FsUOv&vid=XMjQyODU1MDMy&sm=1&operate_type=1&dre=u29&si=79&eo=1&dst=1&iv=0&s=cbffd1c2962411de83b1&type=flvhdv3&bc=2&vkey=Bc2a976f747039b61fc2b55a717af1867'
requestHeaders = {'User-Agent': 'Android'}
response = requests.get(url)
headers = response.headers
md5 = headers.get('ETag')
md5 = md5[1: (len(md5) - 1)]
md5 = md5.lower()
print("headerMd5:" + md5)
with open('temp.ts', 'wb') as tempTs:
tempTs.write(response.content)
with open('temp.ts', 'rb') as tempTs:
data = tempTs.read()
realMd5 = hashlib.md5(data).hexdigest()
print("realMd5:" + realMd5)
if (md5 == realMd5):
print("correct_segment")
else:
print("error_segment:" + url)
print("error_header_md5" + md5)
print("error_real_md5" + realMd5)
sys.exit(1)
# checkCdnFile('http://m-vali.cp31.ott.cibntv.net/6581031CC4F03E719C05A73654/03000C14125CF5672A664BC457BC38537F0F2F-887E-447C-9791-5CA361FBA9C8.mp4?ccode=01010201&duration=131&expire=18000&psid=37ccbe5449a577cf9e11a940f90cdd1f41cb3&ups_client_netip=758850ca&ups_ts=1593577163&ups_userid=1865339496&utid=WNdzjS7%2Bx8oDAJ2fy%2F0l6haR&vid=XNDIxNjUyNDQ4OA&vkey=Babcac71d12d0f156b6435e8ad9b1eb49&iv=1&eo=0&bc=2&dre=u13&si=43&dst=1')
# 根据url下载m3u8文件.当然也可以注释掉这个下载
m3u8_url = 'http://valipl.cp31.ott.cibntv.net/657227C075630713D75A441E8/05000A00005EBBBA588BB780000000690827C6-432D-4F9E-855E-A710EB3F0518.m3u8?ccode=01010201&duration=2744&expire=18000&psid=4b9f64c3c529489de9dbf10de7cc77bb41cb3&ups_client_netip=df659a7b&ups_ts=1593576619&ups_userid=1434082693&utid=XjtxFE23ve8DAF7F2Cm1G2Yf&vid=XNDU0MTc1ODE0NA&vkey=B9b36304de1830bfadec05c7525f1cded&sm=1&operate_type=1&dre=u29&si=79&eo=1&dst=1&iv=0&s=e585299e47d911e6abda&type=mp5hd2v3&bc=2'
response = requests.get(m3u8_url)
with open('youku.m3u8', 'wb') as tempTs:
tempTs.write(response.content)
# 读取m3u8每行文本
with open('youku.m3u8', 'r') as m3u8:
segmentCount = 0
for line in m3u8:
line = line.strip('\n')
if line.startswith("http"):
segmentCount += 1
print('no:%d'%(segmentCount))
checkCdnFile(line)