Violet
/
ttt
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

merge into: Violet:old
Branches Tags
Violet:main
Violet:old
...
pull from: Violet:main
Branches Tags
Violet:main
Violet:old

6 Commits
old ... main

Author SHA1 Message Date
Violet 1f617717a6 更新 'yk.py'
7 months ago
Violet 903d188823 添加 'README.md'
7 months ago
Violet 17f75d1baf 更新 'main.py'
7 months ago
Violet 2f1ca99981 更新 'iqy.py'
7 months ago
Violet 038118e3f0 删除 'test.pdf'
8 months ago
baipzy123 94e0c1f08f
Add files via upload
12 months ago
4 changed files with 856 additions and 601 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats
  1. 75
      README.md
  2. 593
      iqy.py
  3. 110
      main.py
  4. 679
      yk.py

75
README.md
Unescape View File

@ -0,0 +1,75 @@
爱优腾视频下载器(web端）
## 基本说明
#### 支持内容
1.爱奇艺普通视频以及wvdrm内容
2.腾讯普通视频(不再支持)
3.优酷普通视频，wv加密内容以及自研加密内容
### 使用说明
自行编译，下载相关依赖
本软件支持解析，不支持破解
1.下载链接中全部文件，放到同一个文件夹
2.第一次运行需要输入三大网站cookie,必须输，**没有会员也要输，除非自己编译**
3.输入链接，然后不报错的话，会生成bat文件或者txt,
bat文件直接运行，
txt需要下载https://www.52pojie.cn/thread-1631141-1-1.html，直接拖进下载器就行
报错的话，根据源码自行修改
## 运行截图
![img](https://s3.ananas.chaoxing.com/sv-w8/doc/c9/08/85/36834fabf123a4ce08f07dfc736d9d50/thumb/1.png)
![img](https://s3.ananas.chaoxing.com/sv-w8/doc/c9/08/85/36834fabf123a4ce08f07dfc736d9d50/thumb/2.png)
![img](https://s3.ananas.chaoxing.com/sv-w8/doc/c9/08/85/36834fabf123a4ce08f07dfc736d9d50/thumb/3.png)
![img](https://s3.ananas.chaoxing.com/sv-w8/doc/c9/08/85/36834fabf123a4ce08f07dfc736d9d50/thumb/4.png)
## 郑重声明
**本软件仅限研究web视频下载技术，所以开源了，软件本身并不重要，其中的部分参数如何逆向才重要**
**比如爱奇艺的vf,虽然只是一个md5，但混淆的相当复杂，以及优酷解密key的jsvmp，还有腾讯的ckey:vmp+wasm以及响应的解密也是一个jsvmp**
## 2023-10-31更新
1.增加腾讯杜比
2.增加爱奇艺4k
3.下载器修复，优酷存在多个map导致下载只有6s，修复爱奇艺wv加密，读取不到init.mp4
## 2023-11-20更新
1.删除腾讯支持
2.增加优酷帧享
![image](https://github.com/Tontonnow/ttt/assets/122251963/199b9f19-399a-48b7-af0d-37cc44d4c4ed)
## 2023-11-25更新
1.修复优酷key出错
## 2023-12-8更新
1.增加优酷dts等音效
![QQ截图20231208185853](https://github.com/Tontonnow/ttt/assets/122251963/34cf85a4-18e4-4177-a07a-c57e9dc6303a)
下载选项选择0，自动生成m3u8大师列表，暂时支持单语种
![image](https://github.com/Tontonnow/ttt/assets/122251963/efb5746c-33df-48cc-9077-fb93f45e8975)
## 2024-1-25更新
1.增加音效和杜比
2.测试支持多音轨
![image](https://github.com/Tontonnow/ttt/assets/122251963/770479d9-ce45-4083-8766-801cb164c6f3)
![image](https://github.com/Tontonnow/ttt/assets/122251963/d8200b25-7f95-4eed-9df0-82a94d3bbaa4)

593
iqy.py
Unescape View File

@ -1,296 +1,297 @@
import base64
import json
import time
from urllib import parse
import requests
from tabulate import tabulate
from pywidevineb.L3.cdm import deviceconfig
from pywidevineb.L3.decrypt.wvdecryptcustom import WvDecrypt
from tools import dealck, md5, get_size, get_pssh
def get_key(pssh):
LicenseUrl = "https://drml.video.iqiyi.com/drm/widevine?ve=0"
wvdecrypt = WvDecrypt(init_data_b64=pssh, device=deviceconfig.device_android_generic)
widevine_license = requests.post(url=LicenseUrl, data=wvdecrypt.get_challenge())
license_b64 = base64.b64encode(widevine_license.content)
wvdecrypt.update_license(license_b64)
correct, keys = wvdecrypt.start_process()
for key in keys:
print('--key ' + key)
key_string = ' '.join([f"--key {key}" for key in keys])
return key_string
class iqy:
def __init__(self, aqy):
self.ck = aqy
ckjson = dealck(aqy)
self.P00003 = ckjson.get('P00003', "1008611")
self.pck = ckjson.get('P00001')
self.dfp = ckjson.get('__dfp', "").split("@")[0]
self.QC005 = ckjson.get('QC005', "")
self.requests = requests.Session()
self.requests.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Cookie": self.ck,
})
self.bop = f"{{\"version\":\"10.0\",\"dfp\":\"{self.dfp}\",\"b_ft1\":8}}"
@staticmethod
def parse(shareurl):
try:
url = "https://iface2.iqiyi.com/video/3.0/v_play"
params = {
"app_k": "20168006319fc9e201facfbd7c2278b7",
"app_v": "8.9.5",
"platform_id": "10",
"dev_os": "8.0.1",
"dev_ua": "Android",
"net_sts": "1",
"secure_p": "GPhone",
"secure_v": "1",
"dev_hw": "{\"cpu\":0,\"gpu\":\"\",\"mem\":\"\"}",
"app_t": "0",
"h5_url": shareurl
}
response = requests.get(url, params=params)
data = response.json()
pid = data['play_pid']
aid = data['play_aid']
tvid = data['play_tvid']
Album = data['album']
Title = Album['_t']
Cid = Album['_cid']
return pid, aid, tvid, Title, Cid
except Exception as e:
print(e)
return None, None, None, None, None
@staticmethod
def get_avlistinfo(title, albumId, cid, pid):
rets = []
page = 1
size = 200
def getlist6():
url = "https://pcw-api.iqiyi.com/album/source/svlistinfo"
params = {
"cid": "6",
"sourceid": pid,
"timelist": ",".join([str(i) for i in range(2000, 2026)]),
}
response = requests.get(url, params=params)
data = response.json()['data']
for a, b in data.items():
for i in b:
ret = {
"album": title,
"name": i['name'],
"tvId": i['tvId'],
}
rets.append(ret)
def getlist():
aid = albumId
url = "https://pcw-api.iqiyi.com/albums/album/avlistinfo"
params = {
"aid": aid,
"page": page,
"size": size
}
response = requests.get(url, params=params).json()
if response['code'] != 'A00000':
return None
data = response['data']
total = data['total']
if total > size:
for i in range(2, total // size + 2):
params['page'] = i
response = requests.get(url, params=params).json()
data['epsodelist'].extend(response['data']['epsodelist'])
for i in data['epsodelist']:
ret = {
"album": title,
"name": i['name'],
"tvId": i['tvId'],
}
rets.append(ret)
if cid == 1:
ret = {
"album": title,
"name": title,
"tvId": albumId,
}
rets.append(ret)
elif cid == 6:
getlist6()
else:
getlist()
return rets
def get_param(self, tvid="", vid=""):
tm = str(int(time.time() * 1000))
authKey = md5("d41d8cd98f00b204e9800998ecf8427e" + tm + str(tvid))
params = {
"tvid": tvid,
"bid": "600",
"vid": "",
"src": "01010031010000000000",
"vt": "0",
"rs": "1",
"uid": self.P00003,
"ori": "pcw",
"ps": "0",
"k_uid": "dc7c8156286e94182d2843ada4ef6050",
"pt": "0",
"d": "0",
"s": "",
"lid": "0",
"cf": "0",
"ct": "0",
"authKey": authKey,
"k_tag": "1",
"dfp": self.dfp,
"locale": "zh_cn",
"pck": self.pck,
"k_err_retries": "0",
"up": "",
"sr": "1",
"qd_v": "5",
"tm": tm,
"qdy": "u",
"qds": "0",
"ppt": "0",
"k_ft1": "706436220846084",
"k_ft4": "1162321298202628",
"k_ft2": "262335",
"k_ft5": "134217729",
"k_ft6": "128",
"k_ft7": "688390148",
"fr_300": "120_120_120_120_120_120",
"fr_500": "120_120_120_120_120_120",
"fr_600": "120_120_120_120_120_120",
"fr_800": "120_120_120_120_120_120",
"fr_1020": "120_120_120_120_120_120",
}
dash = f'/dash?'
for a, b in params.items():
dash += f"{a}={b}&"
dash = dash[:-1] + "&bop=" + parse.quote(self.bop) + "&ut=14"
vf = md5(dash + "tle8orw4vetejc62int3uewiniecr18i")
dash += f"&vf={vf}"
return dash
def get_dash(self, tvid="", vid=""):
params = self.get_param(tvid=tvid, vid=vid)
url = "https://cache.video.iqiyi.com" + params
res = self.requests.get(url)
return res.json()
def run(self, url=None):
url = input("请输入爱奇艺分享链接：") if url is None else url
pid, aid, tvid, title, cid = self.parse(url)
if pid is None:
print("解析失败")
return
avlist = self.get_avlistinfo(title, aid, cid, pid)
if avlist is None:
print("获取列表失败")
return
table = tabulate(avlist, headers="keys", tablefmt="grid", showindex=range(1, len(avlist) + 1))
print(table)
index = input("请输入序号：")
index = index.split(",")
for i in index:
if i.isdigit():
i = int(i)
if i > len(avlist):
print("序号错误")
continue
tvId = avlist[i - 1]['tvId']
name = avlist[i - 1]['name']
ctitle = avlist[i - 1]['album']
print(f"正在获取{ctitle} {name}的m3u8")
response = self.get_dash(tvid=tvId)
try:
if response['data']['boss_ts']['code'] != 'A00000':
print(f'获取m3u8失败\n')
print(response['data']['boss_ts']['msg'])
continue
except:
pass
data = response['data']
program = data['program']
if 'video' not in program:
print("无视频")
continue
video = program['video']
audio = program['audio']
stl = program.get("stl", [])
'''
list = []
for a in video:
scrsz = a.get('scrsz', '')
size = a['vsize']
vid = a['vid']
list.append((scrsz, vid, size))
list.sort(key=lambda x: x[-1], reverse=True)
tb = tabulate(list, headers=["分辨率", "vid", "大小"], tablefmt="grid",
showindex=range(1, len(list) + 1))
print(tb)
index = input("请输入序号：")
index = index.split(",")
for i in index:
vid = list[int(i) - 1][1]
response = self.get_dash(tvid=tvId, vid=vid)
try:
if response['data']['boss_ts']['code'] != 'A00000':
print(f'获取m3u8失败\n')
print(response['data']['boss_ts']['msg'])
continue
except:
pass
data = response['data']
program = data['program']
if 'video' not in program:
print("无视频")
continue
video = program['video']
'''
for a in video:
try:
scrsz = a.get('scrsz', '')
vsize = get_size(a['vsize'])
m3u8data = a['m3u8']
fr = str(a['fr'])
name = name + "_" + scrsz + "_" + vsize + "_" + fr + 'fps'
name = name.replace(' ', '_')
file = f"./chache/{name}.m3u8"
savepath = f"./download/iqy/{ctitle}"
with open(file, 'w') as f:
f.write(m3u8data)
if m3u8data.startswith('{"payload"'):
m3u8data = json.loads(m3u8data)
init = m3u8data['payload']['wm_a']['audio_track1']['codec_init']
pssh = get_pssh(init)
key_string = get_key(pssh)
cmd = f"N_m3u8DL-RE.exe \"{file} \" --tmp-dir ./cache --save-name \"{name}\" --save-dir \"{savepath}\" --thread-count 16 --download-retry-count 30 --auto-select --check-segments-count " + key_string + " --decryption-binary-path ./mp4decrypt.exe -M format=mp4"
else:
cmd = f"N_m3u8DL-RE.exe \"{file} \" --tmp-dir ./cache --save-name \"{name}\" --save-dir \"{savepath}\" --thread-count 16 --download-retry-count 30 --auto-select --check-segments-count "
with open(f"{ctitle}.bat", 'a', encoding='gbk') as f:
f.write(cmd)
f.write("\n")
print(f"获取{name}成功")
except:
continue
else:
continue
if __name__ == '__main__':
ck = ""
iq = iqy(ck)
iq.run()
import base64
import json
import time
from urllib import parse
import requests
from tabulate import tabulate
from pywidevine.L3.cdm import deviceconfig
from pywidevine.L3.decrypt.wvdecryptcustom import WvDecrypt
from tools import dealck, md5, get_size, get_pssh
def get_key(pssh):
LicenseUrl = "https://drml.video.iqiyi.com/drm/widevine?ve=0"
wvdecrypt = WvDecrypt(init_data_b64=pssh, cert_data_b64="",device=deviceconfig.device_android_generic)
widevine_license = requests.post(url=LicenseUrl, data=wvdecrypt.get_challenge())
license_b64 = base64.b64encode(widevine_license.content)
wvdecrypt.update_license(license_b64)
correct, keys = wvdecrypt.start_process()
for key in keys:
print('--key ' + key)
key_string = ' '.join([f"--key {key}" for key in keys])
return key_string
class iqy:
def __init__(self, aqy):
self.ck = aqy
ckjson = dealck(aqy)
self.P00003 = ckjson.get('P00003', "1008611")
self.pck = ckjson.get('P00001')
self.dfp = ckjson.get('__dfp', "").split("@")[0]
self.QC005 = ckjson.get('QC005', "")
self.requests = requests.Session()
self.requests.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
})
self.bop = f"{{\"version\":\"10.0\",\"dfp\":\"{self.dfp}\",\"b_ft1\":8}}"
@staticmethod
def parse(shareurl):
try:
url = "https://iface2.iqiyi.com/video/3.0/v_play"
params = {
"app_k": "20168006319fc9e201facfbd7c2278b7",
"app_v": "8.9.5",
"platform_id": "10",
"dev_os": "8.0.1",
"dev_ua": "Android",
"net_sts": "1",
"secure_p": "GPhone",
"secure_v": "1",
"dev_hw": "{\"cpu\":0,\"gpu\":\"\",\"mem\":\"\"}",
"app_t": "0",
"h5_url": shareurl
}
response = requests.get(url, params=params)
data = response.json()
pid = data['play_pid']
aid = data['play_aid']
tvid = data['play_tvid']
Album = data['album']
Title = Album['_t']
Cid = Album['_cid']
return pid, aid, tvid, Title, Cid
except Exception as e:
print(e)
return None, None, None, None, None
@staticmethod
def get_avlistinfo(title, albumId, cid, pid):
rets = []
page = 1
size = 200
def getlist6():
url = "https://pcw-api.iqiyi.com/album/source/svlistinfo"
params = {
"cid": "6",
"sourceid": pid,
"timelist": ",".join([str(i) for i in range(2000, 2026)]),
}
response = requests.get(url, params=params)
data = response.json()['data']
for a, b in data.items():
for i in b:
ret = {
"album": title,
"name": i['name'],
"tvId": i['tvId'],
}
rets.append(ret)
def getlist():
aid = albumId
url = "https://pcw-api.iqiyi.com/albums/album/avlistinfo"
params = {
"aid": aid,
"page": page,
"size": size
}
response = requests.get(url, params=params).json()
if response['code'] != 'A00000':
return None
data = response['data']
total = data['total']
if total > size:
for i in range(2, total // size + 2):
params['page'] = i
response = requests.get(url, params=params).json()
data['epsodelist'].extend(response['data']['epsodelist'])
for i in data['epsodelist']:
ret = {
"album": title,
"name": i['name'],
"tvId": i['tvId'],
}
rets.append(ret)
if cid == 1:
ret = {
"album": title,
"name": title,
"tvId": albumId,
}
rets.append(ret)
elif cid == 6:
getlist6()
else:
getlist()
return rets
def get_param(self, tvid="", vid=""):
tm = str(int(time.time() * 1000))
authKey = md5("d41d8cd98f00b204e9800998ecf8427e" + tm + str(tvid))
params = {
"tvid": tvid,
"bid": "800",
"src": "01010031010000000000",
"uid": self.P00003,
"k_uid": self.QC005,
"authKey": authKey,
"dfp": self.dfp,
"pck": self.pck,
"vid": "",
"tm": tm,
"vt": "0",
"rs": "1",
"ori": "pcw",
"ps": "1",
"pt": "0",
"d": "0",
"s": "",
"lid": "0",
"cf": "0",
"ct": "0",
"k_tag": "1",
"locale": "zh_cn",
"k_err_retries": "0",
"up": "",
"sr": "1",
"qd_v": "5",
"qdy": "u",
"qds": "0",
"k_ft1": "706436220846084",
"k_ft4": "1162321298202628",
"k_ft2": "262335",
"k_ft5": "134217729",
"k_ft6": "128",
"k_ft7": "688390148",
"fr_300": "120_120_120_120_120_120",
"fr_500": "120_120_120_120_120_120",
"fr_600": "120_120_120_120_120_120",
"fr_800": "120_120_120_120_120_120",
"fr_1020": "120_120_120_120_120_120",
}
dash = f'/dash?'
for a, b in params.items():
dash += f"{a}={b}&"
dash = dash[:-1] + "&bop=" + parse.quote(self.bop) + "&ut=13&ut=4&ut=5&ut=14&ut=57"
vf = md5(dash + "tle8orw4vetejc62int3uewiniecr18i")
dash += f"&vf={vf}"
return dash
def get_dash(self, tvid="", vid=""):
params = self.get_param(tvid=tvid, vid=vid)
url = "https://cache.video.iqiyi.com" + params
res = self.requests.get(url)
return res.json()
def run(self, url=None):
url = input("请输入爱奇艺分享链接：") if url is None else url
pid, aid, tvid, title, cid = self.parse(url)
if pid is None:
print("解析失败")
return
avlist = self.get_avlistinfo(title, aid, cid, pid)
if avlist is None:
print("获取列表失败")
return
table = tabulate(avlist, headers="keys", tablefmt="grid", showindex=range(1, len(avlist) + 1))
print(table)
index = input("请输入序号：")
index = index.split(",")
for i in index:
if i.isdigit():
i = int(i)
if i > len(avlist):
print("序号错误")
continue
tvId = avlist[i - 1]['tvId']
name = avlist[i - 1]['name']
ctitle = avlist[i - 1]['album']
print(f"正在获取{ctitle} {name}的m3u8")
response = self.get_dash(tvid=tvId)
try:
if response['data']['boss_ts']['code'] != 'A00000':
print(f'获取m3u8失败\n')
print(response['data']['boss_ts']['msg'])
continue
except:
pass
data = response['data']
program = data['program']
if 'video' not in program:
print("无视频")
continue
video = program['video']
audio = program['audio']
stl = program.get("stl", [])
'''
list = []
for a in video:
scrsz = a.get('scrsz', '')
size = a['vsize']
vid = a['vid']
list.append((scrsz, vid, size))
list.sort(key=lambda x: x[-1], reverse=True)
tb = tabulate(list, headers=["分辨率", "vid", "大小"], tablefmt="grid",
showindex=range(1, len(list) + 1))
print(tb)
index = input("请输入序号：")
index = index.split(",")
for i in index:
vid = list[int(i) - 1][1]
response = self.get_dash(tvid=tvId, vid=vid)
try:
if response['data']['boss_ts']['code'] != 'A00000':
print(f'获取m3u8失败\n')
print(response['data']['boss_ts']['msg'])
continue
except:
pass
data = response['data']
program = data['program']
if 'video' not in program:
print("无视频")
continue
video = program['video']
'''
for a in video:
try:
scrsz = a.get('scrsz', '')
vsize = get_size(a['vsize'])
m3u8data = a['m3u8']
fr = str(a['fr'])
name = name + "_" + scrsz + "_" + vsize + "_" + fr + 'fps'
name = name.replace(' ', '_')
file = f"./chache/{name}.m3u8"
savepath = f"./download/iqy/{ctitle}"
with open(file, 'w') as f:
f.write(m3u8data)
if m3u8data.startswith('{"payload"'):
m3u8data = json.loads(m3u8data)
init = m3u8data['payload']['wm_a']['audio_track1']['codec_init']
pssh = get_pssh(init)
key_string = get_key(pssh)
cmd = f"N_m3u8DL-RE.exe \"{file} \" --tmp-dir ./cache --save-name \"{name}\" --save-dir \"{savepath}\" --thread-count 16 --download-retry-count 30 --auto-select --check-segments-count " + key_string + " --decryption-binary-path ./mp4decrypt.exe -M format=mp4"
if m3u8data.startswith('<?xml'):
pssh = m3u8data.split('<cenc:pssh>')[1].split('</cenc:pssh>')[0]
key_string = get_key(pssh)
cmd = f"N_m3u8DL-RE.exe \"{file} \" --tmp-dir ./cache --save-name \"{name}\" --save-dir \"{savepath}\" --thread-count 16 --download-retry-count 30 --auto-select --check-segments-count " + key_string + " --decryption-binary-path ./mp4decrypt.exe -M format=mp4"
else:
cmd = f"N_m3u8DL-RE.exe \"{file} \" --tmp-dir ./cache --save-name \"{name}\" --save-dir \"{savepath}\" --thread-count 16 --download-retry-count 30 --auto-select --check-segments-count "
with open(f"{ctitle}.bat", 'a', encoding='gbk') as f:
f.write(cmd)
f.write("\n")
print(f"获取{name}成功")
except:
continue
else:
continue
if __name__ == '__main__':
ck = ""
iq = iqy(ck)
iq.run()

110
main.py
Unescape View File

@ -1,54 +1,56 @@
import base64
import hashlib
import json
import random
import time
from urllib import parse
from pathlib import Path
from urllib.parse import parse_qsl, urlsplit
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from Cryptodome.Random import get_random_bytes
from Cryptodome.Random import random
from Cryptodome.Cipher import PKCS1_OAEP, AES
from Cryptodome.Hash import CMAC, SHA256, HMAC, SHA1
from Cryptodome.PublicKey import RSA
from Cryptodome.Signature import pss
from Cryptodome.Util import Padding
from google.protobuf.message import DecodeError
from google.protobuf import text_format
import logging
import yaml
from tabulate import tabulate
from wasmer_compiler_cranelift import Compiler
from wasmer import Store, Type, Function, Memory, Module, ImportObject, engine, Instance, Table
from pywidevine.L3.cdm import deviceconfig
from pywidevine.L3.decrypt.wvdecryptcustom import WvDecrypt
import re, requests, time, json
from hashlib import md5
import base64
from tools import rsa_dec, aes_decrypt, djb2Hash, b64decode, sha1withrsa, check_file, get_config
from tx import TX
from iqy import iqy
from yk import YouKu
if __name__ == '__main__':
check_file()
config = get_config()
txck = config["txck"]
yk = config["yk"]
aqy = config["aqy"]
tx = TX(txck)
iq = iqy(aqy)
youku = YouKu(yk)
while True:
url = input("请输入视频链接：")
if "v.qq.com" in url:
tx.run(url)
elif "iqiyi.com" in url:
iq.run(url)
elif "youku.com" in url:
youku.start(url)
else:
print("暂不支持该链接")
import base64
import hashlib
import json
import random
import time
from urllib import parse
from pathlib import Path
from urllib.parse import parse_qsl, urlsplit
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from Cryptodome.Random import get_random_bytes
from Cryptodome.Random import random
from Cryptodome.Cipher import PKCS1_OAEP, AES
from Cryptodome.Hash import CMAC, SHA256, HMAC, SHA1
from Cryptodome.PublicKey import RSA
from Cryptodome.Signature import pss
from Cryptodome.Util import Padding
from google.protobuf.message import DecodeError
from google.protobuf import text_format
import logging
import yaml
from tabulate import tabulate
from wasmer_compiler_cranelift import Compiler
from wasmer import Store, Type, Function, Memory, Module, ImportObject, engine, Instance, Table
from pywidevine.L3.cdm import deviceconfig
from pywidevine.L3.decrypt.wvdecryptcustom import WvDecrypt
import re, requests, time, json
from hashlib import md5
import base64
from tools import rsa_dec, aes_decrypt, djb2Hash, b64decode, sha1withrsa, check_file, get_config
from iqy import iqy
from yk import YouKu
if __name__ == '__main__':
check_file()
config = get_config()
yk = config["yk"]
aqy = config["aqy"]
try:
iq = iqy(aqy)
youku = YouKu(yk)
except Exception as e:
print("配置文件有误，请检查")
print(e)
while True:
url = input("请输入视频链接：")
if "iqiyi.com" in url:
iq.run(url)
elif "youku.com" in url:
youku.start(url)
else:
print("暂不支持该链接")

679
yk.py
Unescape View File

@ -1,251 +1,428 @@
import os
import re, requests, time, json
from hashlib import md5
from urllib.parse import parse_qsl, urlsplit
import base64
from Crypto.Cipher import AES
from tabulate import tabulate
from pywidevineb.L3.cdm import deviceconfig
from pywidevineb.L3.decrypt.wvdecryptcustom import WvDecrypt
from tools import get_pssh, dealck
requests = requests.Session()
class YouKu:
def __init__(self, cookie):
self.cookie = dealck(cookie)
self.r = "xWrtQpP4Z4RsrRCY"
self.R = "aq1mVooivzaolmJY5NrQ3A=="
self.key = ""
self.headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36",
}
requests.headers.update(self.headers)
requests.cookies.update(self.cookie)
def youku_sign(self, t, data, token):
appKey = '24679788' # 固定值
'''token值在cookie中'''
sign = token + '&' + t + '&' + appKey + '&' + data
md = md5()
md.update(sign.encode('UTF-8'))
sign = md.hexdigest()
return sign
def utid(self):
json_cookie = requests.cookies.get_dict()
requests.cookies.clear()
requests.cookies.update(json_cookie)
utid = json_cookie.get("cna")
token = json_cookie.get("_m_h5_tk").split("_")[0]
return {"utid": utid, "token": token}
# 若直接在首页小窗口上复制的视频网址，是重定向的网址。
def redirect(self, url):
headers = {
"referer": "https://www.youku.com/",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36",
}
resp = requests.get(url=url)
return resp.url
def page_parser(self, url):
vid = re.findall(r"id_(.*?)\.html", url)[0]
url = "https://openapi.youku.com/v2/videos/show.json"
params = {
"client_id": "53e6cc67237fc59a",
"package": "com.huawei.hwvplayer.youku",
"ext": "show",
"video_id": vid
}
try:
response = requests.get(url, params=params).json()
showid = response["show"]["id"]
return {"current_showid": showid, "videoId": 0, "vid": vid}
except Exception as e:
print(f"获取showid失败:{e}")
print(f"[red]获取showid失败[/red]")
def get_emb(self, videoId):
emb = base64.b64encode(("%swww.youku.com/" % videoId).encode('utf-8')).decode('utf-8')
return emb
# 这个函数用来获取元素的第一个值
def takeOne(self, elem):
return float(elem[0])
def m3u8_url(self, t, params_data, sign):
url = "https://acs.youku.com/h5/mtop.youku.play.ups.appinfo.get/1.1/"
params = {
"jsv": "2.5.8",
"appKey": "24679788",
"t": t,
"sign": sign,
"api": "mtop.youku.play.ups.appinfo.get",
"v": "1.1",
"timeout": "20000",
"YKPid": "20160317PLF000211",
"YKLoginRequest": "true",
"AntiFlood": "true",
"AntiCreep": "true",
"type": "jsonp",
"dataType": "jsonp",
"callback": "mtopjsonp1",
"data": params_data,
}
resp = requests.get(url=url, params=params)
result = resp.text
# print(result)
data = json.loads(result[12:-1])
# print(data)
ret = data["ret"]
video_lists = []
if ret == ["SUCCESS::调用成功"]:
stream = data["data"]["data"]["stream"]
title = data["data"]["data"]["video"]["title"]
print("解析成功:")
for video in stream:
m3u8_url = video["m3u8_url"]
width = video["width"]
height = video["height"]
size = video.get("size", 0)
size = '{:.1f}'.format(float(size) / 1048576)
drm_type = video["drm_type"]
if video['drm_type'] == "default":
key = ""
elif drm_type == "cbcs":
license_url = video["stream_ext"]["uri"]
key = self.get_cbcs_key(license_url, m3u8_url)
if key[0]:
key = key[1][0]
else:
encryptR_server = video['encryptR_server']
copyright_key = video['stream_ext']['copyright_key']
key = self.copyrightDRM(self.r, encryptR_server, copyright_key)
video_lists.append([title, size + "M", f"{width}x{height}", drm_type, key, m3u8_url])
tb = tabulate([[*video_lists[i][:5]] for i in range(len(video_lists))],
headers=["标题", "分辨率", "视频大小", "drm_type", "base64key"], tablefmt="pretty",
showindex=range(1, len(video_lists) + 1))
ch = input(f"{tb}\n请输入要下载的视频序号：")
ch = ch.split(",")
for i in ch:
title, size, resolution, drm_type, key, m3u8_url = video_lists[int(i) - 1]
savename = f"{title}_{resolution}_{size}"
savepath = os.path.join(os.getcwd(), "/download/yk")
rm3u8_url = m3u8_url.replace("%", "%%")
common_args = f"N_m3u8DL-RE.exe \"{rm3u8_url}\" --tmp-dir ./cache --save-name \"{title}\" --save-dir \"{savepath}\" --thread-count 16 --download-retry-count 30 --auto-select --check-segments-count"
if drm_type == "default":
cmd = common_args
elif drm_type == "cbcs":
cmd = f"{common_args} --key {key} -M format=mp4"
else:
txt = f'''
#OUT,{savepath}
#DECMETHOD,ECB
#KEY,{key}
{title}_{resolution}_{size},{m3u8_url}
'''
with open("{}.txt".format(title), "a", encoding="gbk") as f:
f.write(txt)
print("下载链接已生成")
continue
with open("{}.bat".format(title), "a", encoding="gbk") as f:
f.write(cmd)
f.write("\n")
print("下载链接已生成")
elif ret == ["FAIL_SYS_ILLEGAL_ACCESS::非法请求"]:
print("请求参数错误")
elif ret == ["FAIL_SYS_TOKEN_EXOIRED::令牌过期"]:
print("Cookie过期")
return 10086
else:
print(ret[0])
return 0
def copyrightDRM(self, r, encryptR_server, copyright_key):
crypto_1 = AES.new(r.encode(), AES.MODE_ECB)
key_2 = crypto_1.decrypt(base64.b64decode(encryptR_server))
crypto_2 = AES.new(key_2, AES.MODE_ECB)
return base64.b64encode(base64.b64decode(crypto_2.decrypt(base64.b64decode(copyright_key)))).decode()
def get_cbcs_key(self, license_url, m3u8_url):
headers = {
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.82"
}
m3u8data = requests.get(m3u8_url, headers=headers).text
key_url = re.findall(r"URI=\"(.*?)\"", m3u8data)[0]
response = requests.get(key_url, headers=headers).text
pssh = response.split("data:text/plain;base64,").pop().split('",')[0]
wvdecrypt = WvDecrypt(init_data_b64=pssh, cert_data_b64="", device=deviceconfig.device_android_generic)
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.82",
}
dic = dict(parse_qsl(urlsplit(license_url).query))
url = license_url.split("?")[0]
dic["licenseRequest"] = base64.b64encode(wvdecrypt.get_challenge()).decode()
dic["drmType"] = "widevine"
response = requests.post(url, data=dic, headers=headers)
license_b64 = response.json()["data"]
wvdecrypt.update_license(license_b64)
Correct, keyswvdecrypt = wvdecrypt.start_process()
if Correct:
return Correct, keyswvdecrypt
def get(self, url):
t = str(int(time.time() * 1000))
user_info = self.utid()
userid = user_info["utid"]
page_info = self.page_parser(url)
emb = self.get_emb(page_info["videoId"])
steal_params = {
"ccode": "0502",
"utid": userid,
"version": "9.4.39",
"ckey": "DIl58SLFxFNndSV1GFNnMQVYkx1PP5tKe1siZu/86PR1u/Wh1Ptd+WOZsHHWxysSfAOhNJpdVWsdVJNsfJ8Sxd8WKVvNfAS8aS8fAOzYARzPyPc3JvtnPHjTdKfESTdnuTW6ZPvk2pNDh4uFzotgdMEFkzQ5wZVXl2Pf1/Y6hLK0OnCNxBj3+nb0v72gZ6b0td+WOZsHHWxysSo/0y9D2K42SaB8Y/+aD2K42SaB8Y/+ahU+WOZsHcrxysooUeND",
"client_ip": "192.168.1.1",
"client_ts": 1698373135
}
biz_params = {
"vid": page_info["vid"],
"h265": 1,
"preferClarity": 4,
"media_type": "standard,subtitle",
"app_ver": "9.4.39",
"extag": "EXT-X-PRIVINF",
"play_ability": 16782592,
"master_m3u8": 1,
"drm_type": 19,
"key_index": "web01",
"encryptR_client": self.R,
"skh": 1,
"last_clarity": 5,
"clarity_chg_ts": 1689341442
}
ad_params = {
"vip": 1,
}
params_data = {
"steal_params": json.dumps(steal_params),
"biz_params": json.dumps(biz_params),
"ad_params": json.dumps(ad_params),
}
params_data = json.dumps(params_data)
sign = self.youku_sign(t, params_data, user_info["token"])
return self.m3u8_url(t, params_data, sign)
def start(self, url=None):
url = input("请输入视频链接：") if url is None else url
url = self.redirect(url)
for i in range(3):
ret = self.get(url)
if ret:
continue
break
if __name__ == '__main__':
cookie =""
youku = YouKu(cookie)
youku.start()
import os
import re, requests, time, json
from hashlib import md5
from urllib.parse import parse_qsl, urlsplit
import base64
from Crypto.Cipher import AES
from tabulate import tabulate
from pywidevine.L3.cdm import deviceconfig
from pywidevine.L3.decrypt.wvdecryptcustom import WvDecrypt
from tools import get_pssh, dealck
requests = requests.Session()
class YouKu:
def __init__(self, cookie):
self.cookie = dealck(cookie)
self.r = "xWrtQpP4Z4RsrRCY"
self.R = "aq1mVooivzaolmJY5NrQ3A=="
self.key = ""
self.headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36",
}
requests.headers.update(self.headers)
requests.cookies.update(self.cookie)
self.ptoken = self.cookie.get("P_pck_rm")
self.utida = "ZIH81OVlRSMDAOQQiG52i4cO"
def youku_sign(self, t, data, token):
appKey = '24679788' # 固定值
'''token值在cookie中'''
sign = token + '&' + t + '&' + appKey + '&' + data
md = md5()
md.update(sign.encode('UTF-8'))
sign = md.hexdigest()
return sign
def utid(self):
json_cookie = requests.cookies.get_dict()
requests.cookies.clear()
requests.cookies.update(json_cookie)
utid = json_cookie.get("cna")
token = json_cookie.get("_m_h5_tk").split("_")[0]
return {"utid": utid, "token": token}
# 若直接在首页小窗口上复制的视频网址，是重定向的网址。
def redirect(self, url):
headers = {
"referer": "https://www.youku.com/",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36",
}
resp = requests.get(url=url)
return resp.url
def page_parser(self, url):
vid = re.findall(r"id_(.*?)\.html", url)[0]
url = "https://openapi.youku.com/v2/videos/show.json"
params = {
"client_id": "53e6cc67237fc59a",
"package": "com.huawei.hwvplayer.youku",
"ext": "show",
"video_id": vid
}
try:
response = requests.get(url, params=params).json()
showid = response["show"]["id"]
return {"current_showid": showid, "videoId": 0, "vid": vid}
except Exception as e:
print(f"获取showid失败:{e}")
print(f"[red]获取showid失败[/red]")
def get_emb(self, videoId):
emb = base64.b64encode(("%swww.youku.com/" % videoId).encode('utf-8')).decode('utf-8')
return emb
# 这个函数用来获取元素的第一个值
def takeOne(self, elem):
return float(elem[0])
def save_m3u8(self, video, auto=1):
title, size, resolution, drm_type, key, stream_type, _, m3u8_url, _ = video
title = f"{title}_{resolution}_{size}_{stream_type}"
savepath = os.path.join(os.getcwd(), "/download/yk")
rm3u8_url = m3u8_url.replace("%", "%%")
if rm3u8_url.startswith("http"):
common_args = f"N_m3u8DL-RE.exe \"{rm3u8_url}\" --tmp-dir ./cache --save-name \"{title}\" --save-dir \"{savepath}\" --thread-count 16 --download-retry-count 30 --check-segments-count"
if auto:
common_args += " --auto-select"
if drm_type == "default":
cmd = common_args
elif drm_type == "cbcs":
cmd = f"{common_args} --key {key} -M format=mp4"
else:
key = key if ":" not in key else base64.b64encode(bytes.fromhex(key.split(":")[1])).decode()
txt = f'''
#OUT,{savepath}
#DECMETHOD,ECB
#KEY,{key}
{title},{m3u8_url}
'''
with open("{}.txt".format(title), "a", encoding="gbk") as f:
f.write(txt)
print("下载链接已生成")
return
else:
m3u8_path = "{}.m3u8".format(title)
with open(m3u8_path, "w", encoding="utf-8") as f:
f.write(m3u8_url)
common_args = f"N_m3u8DL-RE.exe \"{m3u8_path}\" --tmp-dir ./cache --save-name \"{title}\" --save-dir \"{savepath}\" --thread-count 16 --download-retry-count 30 --check-segments-count"
if ":" not in key:
uri = re.findall(r'(http.*)\n', m3u8_url)[0]
m3u8_text = requests.get(uri).text
keyid = re.findall(r'KEYID=0x(.*),IV', m3u8_text)[0].lower()
key = "--key {}:{}".format(keyid, base64.b64decode(key).hex())
cmd = f"{common_args} {key} -M format=mp4"
with open("{}.bat".format(title), "a", encoding="gbk") as f:
f.write(cmd)
f.write("\n")
f.close()
print("下载链接已生成")
def m3u8_url(self, t, params_data, sign, vid):
url = "https://acs.youku.com/h5/mtop.youku.play.ups.appinfo.get/1.1/"
params = {
"jsv": "2.5.8",
"appKey": "24679788",
"t": t,
"sign": sign,
"api": "mtop.youku.play.ups.appinfo.get",
"v": "1.1",
"timeout": "20000",
"YKPid": "20160317PLF000211",
"YKLoginRequest": "true",
"AntiFlood": "true",
"AntiCreep": "true",
"type": "jsonp",
"dataType": "jsonp",
"callback": "mtopjsonp1",
"data": params_data,
}
resp = requests.get(url=url, params=params)
result = resp.text
# print(result)
data = json.loads(result[12:-1])
# print(data)
ret = data["ret"]
video_lists = []
if ret == ["SUCCESS::调用成功"]:
stream = data["data"]["data"].get("stream", [])
title = data["data"]["data"].get("video", {}).get("title", "")
print("解析成功:")
keys = {}
tv_stream = self.get_TV_stream(vid)
stream.extend(tv_stream)
for video in stream:
m3u8_url = video["m3u8_url"]
width = video["width"]
height = video["height"]
size = video.get("size", 0)
size = '{:.1f}'.format(float(size) / 1048576)
drm_type = video["drm_type"]
audio_lang = video["audio_lang"]
audio = video['stream_ext'].get("audioGroupId", "") or "default"
if audio_lang == "default":
audio_lang = "guoyu"
language = []
language = re.findall(r'LANGUAGE="([\w\s]+)"', m3u8_url)
# print("language是------------------>>>>>", language)
if 'en' in language:
audio_lang = "en"
if video['drm_type'] == "default":
key = ""
elif audio_lang not in keys.keys():
if drm_type == "cbcs":
license_url = video["stream_ext"]["uri"]
key = self.get_cbcs_key(license_url, m3u8_url)
if key[0]:
key = key[1][0]
else:
encryptR_server = video['encryptR_server']
copyright_key = video['stream_ext']['copyright_key']
key = self.copyrightDRM(self.r, encryptR_server, copyright_key)
keys[audio_lang] = key
else:
key = keys[audio_lang]
video_lists.append(
[title, size + "M", f"{width}x{height}", drm_type, key, video["stream_type"],
audio + "_" + audio_lang, m3u8_url,
video.get("size", 0)])
video_lists = sorted(video_lists, key=lambda x: x[-1], reverse=True)
tb = tabulate([[*video_lists[i][:7]] for i in range(len(video_lists))],
headers=["标题", "视频大小", "分辨率", "drm_type", "base64key", "stream_type", "audio"],
tablefmt="pretty",
showindex=range(1, len(video_lists) + 1))
ch = input(f"{tb}\n请输入要下载的视频序号,输入0尝试自动选择最高清晰度视频：")
if ch == "0":
self.save_m3u82(video_lists)
return 0
ch = ch.split(",")
for i in ch:
video = video_lists[int(i) - 1]
self.save_m3u8(video)
elif ret == ["FAIL_SYS_ILLEGAL_ACCESS::非法请求"]:
print("请求参数错误")
elif ret == ["FAIL_SYS_TOKEN_EXOIRED::令牌过期"]:
print("Cookie过期")
return 10086
else:
print(ret[0])
return 0
def copyrightDRM(self, r, encryptR_server, copyright_key):
try:
crypto_1 = AES.new(r.encode(), AES.MODE_ECB)
key_2 = crypto_1.decrypt(base64.b64decode(encryptR_server))
crypto_2 = AES.new(key_2, AES.MODE_ECB)
return base64.b64encode(base64.b64decode(crypto_2.decrypt(base64.b64decode(copyright_key)))).decode()
except:
return ""
def get_cbcs_key(self, license_url, m3u8_url):
headers = {
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.82"
}
m3u8data = requests.get(m3u8_url, headers=headers).text
key_url = re.findall(r"URI=\"(.*?)\"", m3u8data)[0]
response = requests.get(key_url, headers=headers).text
pssh = response.split("data:text/plain;base64,").pop().split('",')[0]
wvdecrypt = WvDecrypt(init_data_b64=pssh, cert_data_b64="", device=deviceconfig.device_android_generic)
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.82",
}
dic = dict(parse_qsl(urlsplit(license_url).query))
url = license_url.split("?")[0]
dic["licenseRequest"] = base64.b64encode(wvdecrypt.get_challenge()).decode()
dic["drmType"] = "widevine"
response = requests.post(url, data=dic, headers=headers)
license_b64 = response.json()["data"]
wvdecrypt.update_license(license_b64)
Correct, keyswvdecrypt = wvdecrypt.start_process()
if Correct:
return Correct, keyswvdecrypt
def get_TV_stream(self, vid):
headers = {
"user-agent": "OTTSDK;1.0.8.6;Android;9;2203121C"
}
def getdata():
response = requests.get(url, headers=headers, params=params)
try:
data = response.json()["data"]
title = data['show']["title"]
streams = data["stream"]
streamss.extend(streams)
return title, streams
except Exception as e:
return None, []
url = "https://ups.youku.com/ups/get.json" # light_get.json
params = {
"ckey": "7B19C0AB12633B22E7FE81271162026020570708D6CC189E4924503C49D243A0DE6CD84A766832C2C99898FC5ED31F3709BB3CDD82C96492E721BDD381735026",
"client_ip": "192.168.3.1",
"client_ts": "1697343919",
"utid": self.utida,
"pid": "b777e6ae3c99e26",
# HAIER_PID = "36214723575196"; JIMI_PID = "3b777e6ae3c99e26";SONY_PID = "36281532078091";
"player_type": "dnahard", # system:hls,dnahard: cmfv
"app_ver": "11.4.6.4", # 2121104604,2121100600,11.0.6.0,11.4.6.4
"ccode": "0103010261", # sony :0103010261, jimi:010301025C,haier:0103010275 280
"player_source": "21", # 20 sdr 21hfr 22dolby 23bit10
"encryptR_client": "fTWuKHLOVUoOide+VH/h8w==",
"key_index": "key01",
"vid": vid,
"h265": "1",
"media_type": "standard,sei",
"client_id": "",
"ptoken": self.ptoken,
"drm_type": "7",
"extag": "EXT-X-PRIVINF",
"extag_fields": "STREAMTYPE",
"device_name": "XR-98X90L",
"play_ability": "405929984",
"preferClarity": "23",
"master_m3u8": "0",
"play_ability_v2": "2222222",
"site": "1",
"fu": "1",
"vs": "1.0",
"os": "android",
"osv": "12.1.1",
"bt": "tv",
"aw": "a",
"p": "27",
"mdl": "XR-98X90L",
"device_model": "XR-98X90L",
"": ""
}
streamss = []
player_source = [20, 23, 22, 21]
player_type = ["system", "dnahard"]
for i in player_source:
params["player_source"] = str(i)
for j in player_type:
params["player_type"] = j
getdata()
params["ccode"] = "0103010275"
params["player_source"] = "21"
params["player_type"] = "system"
getdata()
params.update({"app_ver": "11.4.7.0",
"play_ability": "274877906943",
"play_ability_v2": "1111111",
"pid": "52f8ca2b4982124b", })
play_ability_v2 = ["1111111111", "0111111111", 1111000000, 1101110000, 1101101000, 11101100100, 1101100010]
# "play_ability_v2": "1111111111", # 1:dolby_vision 2:hdr10 3:dtsc -5:dolby_atmos -4:dolby_around -3:dts -2:aac_hd3_51
#仅供测试，建议自行修改，减少请求次数，否则容易封，提示客户端无权播放
for v2 in play_ability_v2:
params["play_ability_v2"] = v2
getdata()
url = "https://ups.youku.com/ups/light_get.json"
params["ccode"] = "0103010280"
params["drm_type"] = 0
getdata()
streamss = sorted(streamss, key=lambda x: x["size"], reverse=True)
return streamss
def get(self, url):
t = str(int(time.time() * 1000))
user_info = self.utid()
userid = user_info["utid"]
page_info = self.page_parser(url)
emb = self.get_emb(page_info["videoId"])
steal_params = {
"ccode": "0502",
"utid": userid,
"version": "9.4.39",
"ckey": "DIl58SLFxFNndSV1GFNnMQVYkx1PP5tKe1siZu/86PR1u/Wh1Ptd+WOZsHHWxysSfAOhNJpdVWsdVJNsfJ8Sxd8WKVvNfAS8aS8fAOzYARzPyPc3JvtnPHjTdKfESTdnuTW6ZPvk2pNDh4uFzotgdMEFkzQ5wZVXl2Pf1/Y6hLK0OnCNxBj3+nb0v72gZ6b0td+WOZsHHWxysSo/0y9D2K42SaB8Y/+aD2K42SaB8Y/+ahU+WOZsHcrxysooUeND",
"client_ip": "192.168.1.1",
"client_ts": 1698373135
}
biz_params = {
"vid": page_info["vid"],
"h265": 1,
"preferClarity": 4,
"media_type": "standard,subtitle",
"app_ver": "9.4.39",
"extag": "EXT-X-PRIVINF",
"play_ability": 16782592,
"master_m3u8": 1,
"drm_type": 19,
"key_index": "web01",
"encryptR_client": self.R,
"skh": 1,
"last_clarity": 5,
"clarity_chg_ts": 1689341442,
"needad": 0,
}
ad_params = {
"vip": 1,
"needad":0,
}
params_data = {
"steal_params": json.dumps(steal_params),
"biz_params": json.dumps(biz_params),
"ad_params": json.dumps(ad_params),
}
params_data = json.dumps(params_data)
sign = self.youku_sign(t, params_data, user_info["token"])
return self.m3u8_url(t, params_data, sign, page_info["vid"])
def start(self, url=None):
url = input("请输入视频链接：") if url is None else url
url = self.redirect(url) if url.startswith("https://") else f"https://v.youku.com/v_show/id_{url}.html"
for i in range(3):
ret = self.get(url)
if ret:
continue
break
def save_m3u82(self, video_lists):
video_lists_a = [x for x in video_lists if "cmfv5hd" in x[5]]
if video_lists_a:
videoTypes = set()
audioTypes = []
langs = []
keys=""
m3u8data = "#EXTM3U\n"
for video in video_lists_a:
audioType = video[6]
lang = audioType.split("_")[-1]
m3u8_url = video[-2]
key=video[4]
if lang not in langs:
if ":" not in key:
uri = re.findall(r'(http.*)\n', m3u8_url)[0]
m3u8_text = requests.get(uri).text
keyid = re.findall(r'KEYID=0x(.*),IV', m3u8_text)[0].lower()
keys += " --key {}:{}".format(keyid, base64.b64decode(key).hex())
else:
keys += f" --key {key}"
langs.append(lang)
videoType = video[5].split("_")[1]
'''
if videoType not in videoTypes:
m3u8data += "\n".join(m3u8_url.split("\n")[2:4])
videoTypes.append(videoType)
elif audioType not in audioTypes:
m3u8data += m3u8_url.split("\n")[1]
audioTypes.append(audioType)
部分视频存在多个音频流,会出错
所以直接把所有的m3u8链接都写入m3u8文件，手动在下载时选择
'''
if videoType not in videoTypes or audioType not in audioTypes:
m3u8data += "\n".join(m3u8_url.split("\n")[1:-2])
videoTypes.add(videoType)
audioTypes.append(audioType)
m3u8data += "\n#EXT-X-ENDLIST"
video_lists_a[0][7] = m3u8data
video_lists_a[0][4] = keys
self.save_m3u8(video_lists_a[0])
else:
self.save_m3u8(video_lists[0])
if __name__ == '__main__':
cookie = ''
youku = YouKu(cookie)
youku.start("XNjE0OTU0NDU0NA==")

Write Preview
Loading…
Cancel
Save