yk.py

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()