Add files via upload

main
Tontonnow 11 months ago committed by GitHub
commit ecb94d6cc3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 296
      iqy.py
  2. 54
      main.py
  3. 17
      requirements.txt
  4. 586
      tx.py
  5. 251
      yk.py

296
iqy.py

@ -0,0 +1,296 @@
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()

@ -0,0 +1,54 @@
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 pywidevineb.L3.cdm import deviceconfig
from pywidevineb.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("暂不支持该链接")

@ -0,0 +1,17 @@
certifi==2023.7.22
charset-normalizer==3.3.1
idna==3.4
Nuitka==1.8.4
ordered-set==4.1.0
protobuf==3.20.0
pycryptodome==3.19.0
pycryptodomex==3.19.0
pyperclip==1.8.2
PyYAML==6.0.1
requests==2.31.0
tabulate==0.9.0
urllib3==2.0.7
wasmer==1.1.0
wasmer_compiler_cranelift==1.1.0
xmltodict==0.13.0
zstandard==0.21.0

586
tx.py

File diff suppressed because one or more lines are too long

251
yk.py

@ -0,0 +1,251 @@
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["size"]
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()
Loading…
Cancel
Save