hipy-server/timer/asyncSnifferPro.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# File  : asyncSnifferPro.py
# Author: DaShenHan&道长-----先苦后甜，任凭晚风拂柳颜------
# Date  : 2024/3/28
# desc 利用playwright实现的简易播放地址嗅探器,异步高性能

from playwright.async_api import async_playwright, Playwright
import re
import os
from urllib.parse import urlparse
from time import time
import asyncio

_description = r"""
pip install playwright
手动安装谷歌浏览器即可，不需要playwright install,因为它自带的三个浏览器都太垃圾了不好用
参考官方接口文档
https://playwright.dev/python/docs/intro
https://playwright.dev/python/docs/api/class-playwright
"""

# 储存驱动器列表,给接口缓存用
browser_drivers = []


# 全部毫秒为单位不需要转换
class Sniffer:
    # 正则嗅探匹配表达式
    urlRegex: str = 'http((?!http).){12,}?\\.(m3u8|mp4|flv|avi|mkv|rm|wmv|mpg|m4a|mp3)\\?.*|http((?!http).){12,}\\.(m3u8|mp4|flv|avi|mkv|rm|wmv|mpg|m4a|mp3)|http((?!http).)*?video/tos*'
    urlNoHead: str = 'http((?!http).){12,}?(ac=dm&url=)'
    # 每次嗅探间隔毫秒
    playwright = None
    browser = None
    main_page = None
    context = None
    requests = None
    user_agent: str = 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.50 Safari/537.36'
    pages = []
    web_timeout = 15000  # 获取页面源码不超过10秒
    sniffer_timeout = 20000  # 嗅探网页不超20秒
    wait_timeout = 3000  # 等待时间不超过3秒

    def __init__(self,
                 timeout=10000, head_timeout=200, user_agent=None,
                 custom_regex=None, headless=True, debug=False, use_chrome=True, is_pc=False):
        """
        初始化
        @param timeout: 全局嗅探超时
        @param head_timeout: head访问超时
        @param user_agent: 默认请求头
        @param custom_regex: 自定义嗅探正则
        @param headless: 无头模式
        @param debug: 启用debug打印日志
        """

        if user_agent is not None:
            self.user_agent = user_agent

        self.timeout = timeout
        self.head_timeout = head_timeout
        self.debug = debug
        self.custom_regex = custom_regex
        self.headless = headless
        self.channel = "chrome" if use_chrome else None
        self.is_pc = is_pc

    def log(self, *args):
        """
        打印日志print函数|根据类的实例化是否传入debug=True进行开启打印
        @param args:
        @return:
        """
        if self.debug:
            print(*args)

    async def __aenter__(self):
        # 在进入上下文管理器时调用异步函数
        print('在进入上下文管理器时调用异步函数')
        self.browser = await self.init_browser()
        return self

    async def __aexit__(self, exc_type, exc_value, traceback):
        # 在上下文管理器退出时执行清理工作
        print('在上下文管理器退出时执行清理工作')
        pass

    async def init_browser(self):
        """
        初始化驱动程序
        @return:
        """
        self.playwright = await async_playwright().start()

        # 用手动安装的chrome浏览器。不用它自带的三个垃圾浏览器
        browser = await self.playwright.chromium.launch(channel=self.channel, headless=self.headless)
        # print(self.playwright.devices)
        if not self.is_pc:
            # 模拟使用苹果手机
            iphone = self.playwright.devices["iPhone 14 Pro"]
            context = await browser.new_context(**iphone)
        else:
            context = await browser.new_context()
        # 开启一个主窗口方便后续的page新开和关闭不会退出程序
        self.main_page = await context.new_page()
        # 上下文自带的request库，跟requests库有点类似，但是用法也有差别
        self.requests = context.request
        # return browser
        return context

    async def setCookie(self, page, cookie=''):
        """
        设置cookie。可以在嗅探前或者获取源码前设置
        @param _dict:
        @return:
        """
        await page.set_extra_http_headers(headers={'Cookie': cookie})

    @staticmethod
    async def _route_interceptor(route):
        """
        全局路由拦截器,禁止加载某些资源
        @param route:
        @return:
        """
        excluded_resource_types = ["stylesheet", "image", "font"]
        resource_type = route.request.resource_type
        # print(resource_type)
        if resource_type in excluded_resource_types:
            # print('禁止加载资源:', excluded_resource_types, route.request.url, route.request.resource_type)
            await route.abort()
        else:
            await route.continue_()

    @staticmethod
    async def _on_dialog(dialog):
        """
        全局弹窗拦截器
        @param dialog:
        @return:
        """
        # print('on_dialog:', dialog)
        await dialog.accept()

    @staticmethod
    async def _on_pageerror(error):
        """
        全局页面请求错误拦截器
        @param error:
        @return:
        """
        # print('on_pageerror:', error)
        pass

    async def _get_page(self, headers=None):
        """
        新建一个页面。注入好相关依赖
        @param headers:
        @return:
        """
        page = await self.browser.new_page()
        # 设置全局导航超时
        page.set_default_navigation_timeout(self.timeout)
        # 设置全局等待超时
        page.set_default_timeout(self.timeout)

        # 添加初始化脚本 提高速度并且过无法播放的验证
        await page.add_init_script(path=os.path.join(os.path.dirname(__file__), './stealth.min.js'))
        await page.add_init_script(path=os.path.join(os.path.dirname(__file__), './devtools.js'))
        # 屏蔽控制台监听器 https://cdn.staticfile.net/devtools-detector/2.0.14/devtools-detector.min.js
        await page.route(re.compile(r"devtools-detector.*\.js$"), lambda route: route.abort())
        # 设置请求头
        if headers is not None:
            await page.set_extra_http_headers(headers=headers)
        else:
            await page.set_extra_http_headers(headers={'user-agent': self.user_agent})

        # 打开静态资源拦截器
        await page.route(re.compile(r"\.(png|jpg|jpeg|css|ttf)$"), self._route_interceptor)
        # await page.route(re.compile(r"\.(png|jpg|jpeg|ttf)$"), self._route_interceptor)
        # 打开弹窗拦截器
        page.on("dialog", self._on_dialog)
        # 打开页面错误监听
        page.on("pageerror", self._on_pageerror)

        # page.set_viewport_size({"width": 360, "height": 540})
        # 加入页面列表
        self.pages.append(page)
        return page

    @staticmethod
    def remove_element(array, element):
        """
        移除列表指定元素
        @param array:
        @param element:
        @return:
        """
        new_array = [x for x in array if x != element]
        return new_array

    async def close_page(self, page):
        """
        移除页面储存列表的指定page
        @param page:
        @return:
        """
        self.pages = self.remove_element(self.pages, page)
        await page.close()
        self.log('成功关闭page')

    async def close(self):
        """
        用完记得关闭驱动器
        @return:
        """
        await self.main_page.close()
        await self.browser.close()
        await self.playwright.stop()

    async def fetCodeByWebView(self, url, headers=None, timeout=None, is_pc=False, css=None, script=None):
        """
        利用webview请求得到渲染完成后的源码
        @param url: 待获取源码的网页链接
        @param headers: 访问网页的浏览器自定义请求头
        @param timeout: 访问网页的超时毫秒数
        @param is_pc: 是否用PC.此配置不生效。默认手机访问
        @param css: 等待出现定位器|如果不传css就等待加载页面状态为load
        @param script: 页面状态加载完毕后执行的网页脚本，可以点击网页元素之类的
        @return:

        """
        t1 = time()
        if timeout is None:
            timeout = self.timeout
        else:
            timeout = min([timeout, self.web_timeout])
        if not is_pc:
            is_pc = self.is_pc
        do_css = str(css).strip() if css and str(css).strip() else False
        page = await self._get_page(headers)
        # 设置全局导航超时
        page.set_default_navigation_timeout(timeout)
        # 设置全局等待超时
        page.set_default_timeout(timeout)
        response = {'content': '', 'headers': {'location': url}}
        try:
            await page.goto(url)
        except Exception as e:
            self.log(f'发生了错误:{e}')
        else:
            if do_css:
                await page.wait_for_selector(do_css)
            else:
                await page.wait_for_load_state('load')

            if script:
                try:
                    await page.evaluate("""(script) => {
                    eval(script);
                    }
                    """, script)
                    self.log(f'网页加载完成后成功执行脚本:{script}')
                except Exception as e:
                    self.log(f'网页加载完成后执行脚本:{script}发生错误:{e}')

            response['content'] = await page.content()
            response['headers']['location'] = page.url
        t2 = time()
        cost = round((t2 - t1) * 1000, 2)
        response['cost'] = cost
        await self.close_page(page)
        return response

    async def snifferMediaUrl(self, playUrl, mode=0, custom_regex=None, timeout=None, css=None, is_pc=False,
                              headers=None,
                              script=None):
        """
        输入播放地址，返回嗅探到的真实视频链接
        @param playUrl: 待嗅探的视频播放也地址
        @param mode: 模式:0 嗅探到一个就返回 1:在10秒内嗅探所有的返回列表
        @param custom_regex: 自定义嗅探正则
        @param timeout: 超时
        @param css: 等待出现定位器|如果不传css并且传了script就等待加载页面状态为load
        @param is_pc: 是否用PC.此配置不生效。默认手机访问
        @param headers: 访问网页的浏览器自定义请求头
        @param script: 页面状态加载完毕后执行的网页脚本，可以点击网页元素之类的
        @return:
        """
        t1 = time()
        if custom_regex is None:
            custom_regex = self.custom_regex

        if not is_pc:
            is_pc = self.is_pc

        do_css = str(css).strip() if css and str(css).strip() else False

        realUrls = []  # 真实链接列表，用于mode=1场景
        headUrls = []  # 已经head请求过的链接
        page = await self._get_page(headers)
        if timeout is None:
            timeout = self.timeout
        else:
            if mode == 1:  # 嗅探所有超时设定在10s内
                timeout = min([timeout, self.timeout])
            else:
                # 其他单个自定义时间不得超过20s
                timeout = min([timeout, self.sniffer_timeout])

        async def _on_request(request):
            nonlocal realUrls, headUrls
            url = request.url
            method = request.method
            headers = request.headers
            resource_type = request.resource_type
            self.log('on_request:', url, ' method:', method, ' resource_type:', resource_type)
            if custom_regex and re.search(custom_regex, url, re.M | re.I):
                _headers = {}
                if headers.get('referer'):
                    _headers['referer'] = headers['referer']
                if headers.get('user-agent'):
                    _headers['user-agent'] = headers['user-agent']
                realUrls.append({
                    'url': url,
                    'headers': _headers
                })
                await page.evaluate("""([url, _headers,realUrls]) => {
                window.realUrl = url
                window.realHeaders = _headers
                window.realUrls = realUrls
                }
                """, [url, _headers, realUrls])
                self.log('on_request通过custom_regex嗅探到真实地址:', url)
                if mode == 0:
                    page.remove_listener("request", _on_request)
                return True

            if re.search(self.urlRegex, url, re.M | re.I):
                if url.find('url=http') < 0 and url.find('v=http') < 0 and url.find('.css') < 0 and url.find(
                        '.html') < 0:
                    _headers = {}
                    if headers.get('referer'):
                        _headers['referer'] = headers['referer']
                    if headers.get('user-agent'):
                        _headers['user-agent'] = headers['user-agent']
                    realUrls.append({
                        'url': url,
                        'headers': _headers
                    })
                    await page.evaluate("""([url, _headers,realUrls]) => {
                    window.realUrl = url
                    window.realHeaders = _headers
                    window.realUrls = realUrls
                    }
                    """, [url, _headers, realUrls])
                    self.log('on_request通过默认正则已嗅探到真实地址:', url)
                    if mode == 0:
                        page.remove_listener("request", _on_request)
                    return True
            elif str(method).lower() == 'get' and str(url).startswith('http') and url != playUrl:
                parsed_url = urlparse(url)
                path = parsed_url.path
                filename = str(path.split('/')[-1])
                # 链接不含.并且正则匹配不在不head列表  或者 链接有.但是.后面没内容，也算空后缀
                if (filename and '.' not in filename and not re.search(self.urlNoHead, url, re.M | re.I)) or (
                        '.' in filename and len(filename) > 1 and not filename.split('.')[1]) and resource_type not in [
                    'script']:
                    # 如果链接没有进行过head请求。防止多次嗅探的时候重复去head请求
                    if url not in headUrls:
                        try:
                            r = await self.requests.head(url=url, timeout=self.head_timeout)
                            rheaders = r.headers
                            if rheaders.get('content-type') and rheaders[
                                'content-type'] == 'application/octet-stream' and '.m3u8' in rheaders[
                                'content-disposition']:
                                _headers = {}
                                if headers.get('referer'):
                                    _headers['referer'] = headers['referer']
                                if headers.get('user-agent'):
                                    _headers['user-agent'] = headers['user-agent']
                                realUrls.append({
                                    'url': url,
                                    'headers': _headers
                                })
                                await page.evaluate("""([url, _headers,realUrls]) => {
                                window.realUrl = url
                                window.realHeaders = _headers
                                window.realUrls = realUrls
                                }
                                """, [url, _headers, realUrls])
                                self.log('on_request通过head请求嗅探到真实地址:', url)
                                if mode == 0:
                                    page.remove_listener("request", _on_request)

                                return True
                        except Exception as e:
                            print(f'head请求访问: {url} 发生了错误:{e}')

                        headUrls.append(url)

        page.on('request', _on_request)
        # 设置全局导航超时
        page.set_default_navigation_timeout(timeout)
        # 设置全局等待超时
        page.set_default_timeout(timeout)
        await page.expose_function("log", lambda *args: print(*args))
        await page.add_init_script(path=os.path.join(os.path.dirname(__file__), './preload.js'))
        await page.evaluate("""
        window.realUrl = ''
        window.realHeaders = {}
        window.realUrls = []
        """)
        try:
            await page.goto(playUrl)
        except Exception as e:
            self.log('嗅探发生错误:', e)
            t2 = time()
            cost = round((t2 - t1) * 1000, 2)
            return {'url': '', 'headers': {}, 'from': playUrl, 'cost': cost, 'code': 404,
                    'msg': f'嗅探失败:{e}'}

        # 这里不需要另外分支去判断状态为load，因为嗅探无需等待页面加载完毕。异步就行。一般也不传css
        if do_css:
            await page.wait_for_selector(do_css)

        if script:
            if not do_css:
                await page.wait_for_load_state('load')
            try:
                await page.evaluate("""(script) => {
                eval(script);
                }
                """, script)
                self.log(f'网页加载完成后成功执行脚本:{script}')
            except Exception as e:
                self.log(f'网页加载完成后执行脚本:{script}发生错误:{e}')

        is_timeout = False
        if mode == 0:
            try:
                await page.wait_for_function("() => window.realUrl")
            except:
                is_timeout = True
        elif mode == 1:
            try:
                await page.wait_for_timeout(timeout)
            except:
                is_timeout = True

        realUrl = await page.evaluate('window.realUrl')
        realHeaders = await page.evaluate('window.realHeaders')
        realUrls = await page.evaluate('window.realUrls')

        t2 = time()
        cost = round((t2 - t1) * 1000, 2)
        cost_str = f'{cost} ms'
        self.log(f'共计耗时{cost}毫秒|{"已超时" if is_timeout else "未超时"}')
        self.log('realUrl:', realUrl)
        self.log('realHeaders:', realHeaders)
        await self.close_page(page)
        if mode == 0 and realUrl:
            return {'url': realUrl, 'headers': realHeaders, 'from': playUrl, 'cost': cost_str, 'code': 200,
                    'msg': '超级嗅探解析成功'}
        elif mode == 1 and realUrls:
            return {'urls': realUrls, 'code': 200, 'from': playUrl, 'cost': cost_str, 'msg': '超级嗅探解析成功'}
        else:
            return {'url': realUrl, 'headers': realHeaders, 'from': playUrl, 'cost': cost_str, 'code': 404,
                    'msg': '超级嗅探解析失败'}


async def main_test():
    t1 = time()
    urls = [
        # 'https://www.cs1369.com/play/2-1-94.html',
        # 'https://m.ting13.cc/play/19176_1_91258.html',
        'https://v.qq.com/x/page/i3038urj2mt.html',
        'http://www.mgtv.com/v/1/290346/f/3664551.html',
        'https://jx.jsonplayer.com/player/?url=https://m.iqiyi.com/v_1pj3ayb1n70.html',
        'https://jx.yangtu.top/?url=https://m.iqiyi.com/v_1pj3ayb1n70.html',
    ]
    _count = 0
    async with Sniffer(debug=True, headless=True) as browser:
        # 在这里，async_func已被调用并已完成
        pass
    for url in urls:
        _count += 1
        ret = await browser.snifferMediaUrl(url, timeout=15000)
        print(ret)

    _count += 1
    ret = await browser.snifferMediaUrl('https://jx.yangtu.top/?url=https://m.iqiyi.com/v_1pj3ayb1n70.html',
                                        custom_regex='http((?!http).){12,}?(download4|pcDownloadFile)')
    print(ret)
    # _count += 1
    # ret = await browser.fetCodeByWebView('https://www.freeok.pro/xplay/63170-8-12.html')
    # print(ret)

    await browser.close()
    t2 = time()
    print(f'嗅探{_count}个页面共计耗时:{round(t2 - t1, 2)}s')


if __name__ == '__main__':
    asyncio.run(main_test())