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

更新 'yk.py'

Browse Source
main
Violet 6 months ago
parent 903d188823
commit 1f617717a6
1 changed files with 428 additions and 251 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 279
      yk.py

279
yk.py
Unescape View File

@ -5,8 +5,8 @@ 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 pywidevine.L3.cdm import deviceconfig
from pywidevine.L3.decrypt.wvdecryptcustom import WvDecrypt
from tools import get_pssh, dealck
requests = requests.Session()
@ -23,6 +23,8 @@ class YouKu:
}
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' # 固定值
@ -75,7 +77,49 @@ class YouKu:
def takeOne(self, elem):
return float(elem[0])
def m3u8_url(self, t, params_data, sign):
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 = {
@ -103,9 +147,12 @@ class YouKu:
ret = data["ret"]
video_lists = []
if ret == ["SUCCESS::调用成功"]:
stream = data["data"]["data"]["stream"]
title = data["data"]["data"]["video"]["title"]
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"]
@ -113,48 +160,47 @@ class YouKu:
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 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]
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:
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",
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请输入要下载的视频序号：")
ch = input(f"{tb}\n请输入要下载的视频序号,输入0尝试自动选择最高清晰度视频：")
if ch == "0":
self.save_m3u82(video_lists)
return 0
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("下载链接已生成")
video = video_lists[int(i) - 1]
self.save_m3u8(video)
elif ret == ["FAIL_SYS_ILLEGAL_ACCESS::非法请求"]:
print("请求参数错误")
elif ret == ["FAIL_SYS_TOKEN_EXOIRED::令牌过期"]:
@ -165,10 +211,13 @@ class YouKu:
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()
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 = {
@ -193,6 +242,90 @@ class YouKu:
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()
@ -221,10 +354,12 @@ class YouKu:
"encryptR_client": self.R,
"skh": 1,
"last_clarity": 5,
"clarity_chg_ts": 1689341442
"clarity_chg_ts": 1689341442,
"needad": 0,
}
ad_params = {
"vip": 1,
"needad":0,
}
params_data = {
"steal_params": json.dumps(steal_params),
@ -233,19 +368,61 @@ class YouKu:
}
params_data = json.dumps(params_data)
sign = self.youku_sign(t, params_data, user_info["token"])
return self.m3u8_url(t, params_data, sign)
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)
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 =""
cookie = ''
youku = YouKu(cookie)
youku.start()
youku.start("XNjE0OTU0NDU0NA==")

Write Preview
Loading…
Cancel
Save