Xvd.py

# -*- coding: utf-8 -*-
# by @嗷呜
import json
import re
import sys
from urllib.parse import urlparse

import requests
from pyquery import PyQuery as pq
from base64 import b64decode, b64encode
from requests import Session
sys.path.append('..')
from base.spider import Spider


class Spider(Spider):

    def init(self, extend=""):
        try:self.proxies = json.loads(extend)
        except:self.proxies = {}
        self.session = Session()
        self.session.proxies.update(self.proxies)
        self.session.headers.update(self.headers)
        pass

    def getName(self):
        pass

    def isVideoFormat(self, url):
        pass

    def manualVideoCheck(self):
        pass

    def destroy(self):
        pass

    host = "https://www.xvideos.com"

    headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.5410.0 Safari/537.36",
        "pragma": "no-cache",
        "cache-control": "no-cache",
        "sec-ch-ua-platform": "\"Windows\"",
        "sec-ch-ua": "\"Not(A:Brand\";v=\"99\", \"Google Chrome\";v=\"133\", \"Chromium\";v=\"133\"",
        "dnt": "1",
        "origin":host,
        'referer':f'{host}/',
        "sec-ch-ua-mobile": "?0",
        "sec-fetch-site": "cross-site",
        "sec-fetch-mode": "cors",
        "sec-fetch-dest": "empty",
        "accept-language": "zh-CN,zh;q=0.9,en;q=0.8",
        "priority": "u=1, i"
    }

    def homeContent(self, filter):
        result = {}
        cateManual = {
            "最新": "/new",
            "最佳": "/best",
            "频道": "/channels-index",
            "标签": "/tags",
            "明星": "/pornstars-index"
        }
        classes = []
        for k in cateManual:
            classes.append({
                'type_name': k,
                'type_id': cateManual[k]
            })
        result['class'] = classes
        return result

    def homeVideoContent(self):
        data = self.getpq()
        return {'list':self.getlist(data(".mozaique .frame-block"))}

    def categoryContent(self, tid, pg, filter, extend):
        vdata = []
        result = {}
        page = f"/{int(pg) - 1}" if pg != '1' else ''
        result['page'] = pg
        result['pagecount'] = 9999
        result['limit'] = 90
        result['total'] = 999999
        if tid=='/new' or 'tags_click' in tid:
            if 'tags_click' in tid:tid=tid.split('click_')[-1]
            data=self.getpq(f'{tid}/{pg}')
            vdata=self.getlist(data(".mozaique .frame-block"))
        elif tid=='/best':
            if pg=='1':
                self.path=self.session.get(f'{self.host}{tid}',allow_redirects=False).headers['Location']
            data=self.getpq(f'{self.path}{page}')
            vdata=self.getlist(data(".mozaique .frame-block"))
        elif tid=='/channels-index' or tid=='/pornstars-index':
            data = self.getpq(f'{tid}{page}')
            vhtml=data(".mozaique .thumb-block")
            for i in vhtml.items():
                a = i('.thumb-inside .thumb a')
                match = re.search(r'src="([^"]+)"', a('script').text())
                img=''
                if match:
                    img = match.group(1).strip()
                vdata.append({
                    'vod_id': f"channels_click_{'/channels'if tid=='/channels-index' else ''}"+a.attr('href'),
                    'vod_name': a('.profile-name').text() or i('.profile-name').text().replace('\xa0','/'),
                    'vod_pic': self.proxy(img),
                    'vod_tag': 'folder',
                    'vod_remarks': i('.thumb-under .profile-counts').text(),
                    'style': {'ratio': 1.33, 'type': 'rect'}
                })
        elif tid=='/tags':
            result['pagecount'] = pg
            vhtml = self.getpq(tid)
            vhtml = vhtml('.tags-list')
            for d in vhtml.items():
                for i in d('li a').items():
                    vdata.append({
                        'vod_id': "tags_click_"+i.attr('href'),
                        'vod_name': i.attr('title') or i('b').text(),
                        'vod_pic': '',
                        'vod_tag': 'folder',
                        'vod_remarks': i('.navbadge').text(),
                        'style': {'ratio': 1.33, 'type': 'rect'}
                    })
        elif 'channels_click' in tid:
            tid=tid.split('click_')[-1]
            vhtml=self.session.post(f'{self.host}{tid}/videos/best/{int(pg)-1}').json()
            for i in vhtml['videos']:
                vdata.append({
                    'vod_id': i.get('u'),
                    'vod_name': i.get('tf'),
                    'vod_pic': self.proxy(i.get('il')),
                    'vod_year': i.get('n'),
                    'vod_remarks': i.get('d'),
                    'style': {'ratio': 1.33, 'type': 'rect'}
                })
        result['list'] = vdata
        return result

    def detailContent(self, ids):
        url = f"{self.host}{ids[0]}"
        data = self.getpq(ids[0])
        vn=data('meta[property="og:title"]').attr('content')
        dtext=data('.main-uploader a')
        href=dtext.attr('href')
        pdtitle=''
        if href and href.count('/') < 2:
            href=f'/channels{href}'
            pdtitle = '[a=cr:' + json.dumps({'id': 'channels_click_'+href, 'name': dtext('.name').text()}) + '/]' + dtext('.name').text() + '[/a]'
        vod = {
            'vod_name': vn,
            'vod_director':pdtitle,
            'vod_remarks': data('.page-title').text().replace(vn,''),
            'vod_play_from': 'Xvideos',
            'vod_play_url': ''
        }
        js_content = data("#video-player-bg script")
        jstr=''
        for script in js_content.items():
            content = script.text()
            if 'setVideoUrlLow' in content and 'html5player' in content:
                jstr = content
                break
        plist = [f"{vn}${self.e64(f'{1}@@@@{url}')}"]
        def extract_video_urls(js_content):
            try:
                low = re.search(r'setVideoUrlLow\([\'"]([^\'"]+)[\'"]\)', js_content)
                high = re.search(r'setVideoUrlHigh\([\'"]([^\'"]+)[\'"]\)', js_content)
                hls = re.search(r'setVideoHLS\([\'"]([^\'"]+)[\'"]\)', js_content)

                return {
                    'hls': hls.group(1) if hls else None,
                    'high': high.group(1) if high else None,
                    'low': low.group(1) if low else None
                }
            except Exception as e:
                print(f"提取视频URL失败: {str(e)}")
                return {}
        if jstr:
            try:
                urls = extract_video_urls(jstr)
                plist = [
                    f"{quality}${self.e64(f'{0}@@@@{url}')}"
                    for quality, url in urls.items()
                    if url
                ]
            except Exception as e:
                print(f"提取url失败: {str(e)}")
        vod['vod_play_url'] = '#'.join(plist)
        return {'list':[vod]}

    def searchContent(self, key, quick, pg="1"):
        data=self.getpq(f'/?k={key}&p={int(pg)-1}')
        return {'list':self.getlist(data(".mozaique .frame-block")),'page':pg}

    def playerContent(self, flag, id, vipFlags):
        ids=self.d64(id).split('@@@@')
        if '.m3u8' in ids[1]: ids[1] = self.proxy(ids[1], 'm3u8')
        return {'parse': int(ids[0]), 'url': ids[1], 'header': self.headers}

    def localProxy(self, param):
        url=self.d64(param['url'])
        if param.get('type') == 'm3u8':
            return self.m3Proxy(url)
        else:
            return self.tsProxy(url)

    def e64(self, text):
        try:
            text_bytes = text.encode('utf-8')
            encoded_bytes = b64encode(text_bytes)
            return encoded_bytes.decode('utf-8')
        except Exception as e:
            print(f"Base64编码错误: {str(e)}")
            return ""

    def d64(self,encoded_text):
        try:
            encoded_bytes = encoded_text.encode('utf-8')
            decoded_bytes = b64decode(encoded_bytes)
            return decoded_bytes.decode('utf-8')
        except Exception as e:
            print(f"Base64解码错误: {str(e)}")
            return ""

    def getlist(self, data):
        vlist=[]
        for i in data.items():
            a=i('.thumb-inside .thumb a')
            b=i('.thumb-under .title a')
            vlist.append({
                'vod_id': a.attr('href'),
                'vod_name': b('a').attr('title'),
                'vod_pic': self.proxy(a('img').attr('data-src')),
                'vod_year': a('.video-hd-mark').text(),
                'vod_remarks': b('.duration').text(),
                'style': {'ratio': 1.33, 'type': 'rect'}
            })
        return vlist

    def getpq(self, path=''):
        response = self.session.get(f'{self.host}{path}').text
        try:
            return pq(response)
        except Exception as e:
            print(f"{str(e)}")
            return pq(response.encode('utf-8'))

    def m3Proxy(self, url):
        ydata = requests.get(url, headers=self.headers, proxies=self.proxies, allow_redirects=False)
        data = ydata.content.decode('utf-8')
        if ydata.headers.get('Location'):
            url = ydata.headers['Location']
            data = requests.get(url, headers=self.headers, proxies=self.proxies).content.decode('utf-8')
        lines = data.strip().split('\n')
        last_r = url[:url.rfind('/')]
        parsed_url = urlparse(url)
        durl = parsed_url.scheme + "://" + parsed_url.netloc
        for index, string in enumerate(lines):
            if '#EXT' not in string:
                if 'http' not in string:
                    domain=last_r if string.count('/') < 2 else durl
                    string = domain + ('' if string.startswith('/') else '/') + string
                lines[index] = self.proxy(string, string.split('.')[-1].split('?')[0])
        data = '\n'.join(lines)
        return [200, "application/vnd.apple.mpegur", data]

    def tsProxy(self, url):
        data = requests.get(url, headers=self.headers, proxies=self.proxies, stream=True)
        return [200, data.headers['Content-Type'], data.content]

    def proxy(self, data, type='img'):
        if data and len(self.proxies):return f"{self.getProxyUrl()}&url={self.e64(data)}&type={type}"
        else:return data