From 4889ae8c9a341dbcf73669ff9e860294b2231bc8 Mon Sep 17 00:00:00 2001 From: XC-Pro <19840+xc-pro@noreply.example.org> Date: Fri, 14 Aug 2026 13:35:55 +0200 Subject: [PATCH] =?UTF-8?q?=E5=88=A0=E9=99=A4=20py/51=E5=90=B8=E7=93=9C?= =?UTF-8?q?=E5=8A=A8=E6=80=81=E7=89=88.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- py/51吸瓜动态版.py | 404 --------------------------------------------- 1 file changed, 404 deletions(-) delete mode 100644 py/51吸瓜动态版.py diff --git a/py/51吸瓜动态版.py b/py/51吸瓜动态版.py deleted file mode 100644 index 9d1e1c0..0000000 --- a/py/51吸瓜动态版.py +++ /dev/null @@ -1,404 +0,0 @@ -# -*- coding: utf-8 -*- -# 🌈 Love -import json -import random -import re -import sys -import threading -import time -from base64 import b64decode, b64encode -from urllib.parse import urlparse - -import requests -from Crypto.Cipher import AES -from Crypto.Util.Padding import unpad -from pyquery import PyQuery as pq -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.headers = { - 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36', - 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7', - 'Accept-Language': 'zh-CN,zh;q=0.9', - 'Connection': 'keep-alive', - 'Cache-Control': 'no-cache', - } - # Use working dynamic URLs directly - self.host = self.get_working_host() - self.headers.update({'Origin': self.host, 'Referer': f"{self.host}/"}) - self.log(f"使用站点: {self.host}") - print(f"使用站点: {self.host}") - pass - - def getName(self): - return "通用吸瓜" - - def isVideoFormat(self, url): - # Treat direct media formats as playable without parsing - return any(ext in (url or '') for ext in ['.m3u8', '.mp4', '.ts']) - - def manualVideoCheck(self): - return False - - def destroy(self): - pass - - def homeContent(self, filter): - try: - response = requests.get(self.host, headers=self.headers, proxies=self.proxies, timeout=15) - if response.status_code != 200: - return {'class': [], 'list': []} - - data = self.getpq(response.text) - result = {} - classes = [] - - # Try to get categories from different possible locations - category_selectors = [ - '.category-list ul li', - '.nav-menu li', - '.menu li', - 'nav ul li' - ] - - for selector in category_selectors: - for k in data(selector).items(): - link = k('a') - href = (link.attr('href') or '').strip() - name = (link.text() or '').strip() - # Skip placeholder or invalid entries - if not href or href == '#' or not name: - continue - classes.append({ - 'type_name': name, - 'type_id': href - }) - if classes: - break - - # If no categories found, create some default ones - if not classes: - classes = [ - {'type_name': '首页', 'type_id': '/'}, - {'type_name': '最新', 'type_id': '/latest/'}, - {'type_name': '热门', 'type_id': '/hot/'} - ] - - result['class'] = classes - result['list'] = self.getlist(data('#index article a')) - return result - - except Exception as e: - print(f"homeContent error: {e}") - return {'class': [], 'list': []} - - def homeVideoContent(self): - try: - response = requests.get(self.host, headers=self.headers, proxies=self.proxies, timeout=15) - if response.status_code != 200: - return {'list': []} - data = self.getpq(response.text) - return {'list': self.getlist(data('#index article a, #archive article a'))} - except Exception as e: - print(f"homeVideoContent error: {e}") - return {'list': []} - - def categoryContent(self, tid, pg, filter, extend): - try: - if '@folder' in tid: - id = tid.replace('@folder', '') - videos = self.getfod(id) - else: - # Build URL properly - if tid.startswith('/'): - if pg and pg != '1': - url = f"{self.host}{tid}page/{pg}/" - else: - url = f"{self.host}{tid}" - else: - url = f"{self.host}/{tid}" - - response = requests.get(url, headers=self.headers, proxies=self.proxies, timeout=15) - if response.status_code != 200: - return {'list': [], 'page': pg, 'pagecount': 1, 'limit': 90, 'total': 0} - - data = self.getpq(response.text) - videos = self.getlist(data('#archive article a, #index article a'), tid) - - result = {} - result['list'] = videos - result['page'] = pg - result['pagecount'] = 1 if '@folder' in tid else 99999 - result['limit'] = 90 - result['total'] = 999999 - return result - - except Exception as e: - print(f"categoryContent error: {e}") - return {'list': [], 'page': pg, 'pagecount': 1, 'limit': 90, 'total': 0} - - def detailContent(self, ids): - try: - url = f"{self.host}{ids[0]}" if not ids[0].startswith('http') else ids[0] - response = requests.get(url, headers=self.headers, proxies=self.proxies, timeout=15) - - if response.status_code != 200: - return {'list': [{'vod_play_from': '通用吸瓜', 'vod_play_url': f'页面加载失败${url}'}]} - - data = self.getpq(response.text) - vod = {'vod_play_from': '通用吸瓜'} - - # Get content/description - try: - clist = [] - if data('.tags .keywords a'): - for k in data('.tags .keywords a').items(): - title = k.text() - href = k.attr('href') - if title and href: - clist.append('[a=cr:' + json.dumps({'id': href, 'name': title}) + '/]' + title + '[/a]') - vod['vod_content'] = ' '.join(clist) if clist else data('.post-title').text() - except: - vod['vod_content'] = data('.post-title').text() or '通用吸瓜视频' - - # Get video URLs (build episode list when multiple players exist) - try: - plist = [] - used_names = set() - if data('.dplayer'): - for c, k in enumerate(data('.dplayer').items(), start=1): - config_attr = k.attr('data-config') - if config_attr: - try: - config = json.loads(config_attr) - video_url = config.get('video', {}).get('url', '') - # Determine a readable episode name from nearby headings if present - ep_name = '' - try: - parent = k.parents().eq(0) - # search up to a few ancestors for a heading text - for _ in range(3): - if not parent: break - heading = parent.find('h2, h3, h4').eq(0).text() or '' - heading = heading.strip() - if heading: - ep_name = heading - break - parent = parent.parents().eq(0) - except Exception: - ep_name = '' - base_name = ep_name if ep_name else f"视频{c}" - name = base_name - count = 2 - # Ensure the name is unique - while name in used_names: - name = f"{base_name} {count}" - count += 1 - used_names.add(name) - if video_url: - self.log(f"解析到视频: {name} -> {video_url}") - print(f"解析到视频: {name} -> {video_url}") - plist.append(f"{name}${video_url}") - except: - continue - - if plist: - self.log(f"拼装播放列表，共{len(plist)}个") - print(f"拼装播放列表，共{len(plist)}个") - vod['vod_play_url'] = '#'.join(plist) - else: - vod['vod_play_url'] = f"未找到视频源${url}" - - except Exception as e: - vod['vod_play_url'] = f"视频解析失败${url}" - - return {'list': [vod]} - - except Exception as e: - print(f"detailContent error: {e}") - return {'list': [{'vod_play_from': '通用吸瓜', 'vod_play_url': f'详情页加载失败${ids[0] if ids else ""}'}]} - - def searchContent(self, key, quick, pg="1"): - try: - url = f"{self.host}/search/{key}/{pg}" if pg != "1" else f"{self.host}/search/{key}/" - response = requests.get(url, headers=self.headers, proxies=self.proxies, timeout=15) - - if response.status_code != 200: - return {'list': [], 'page': pg} - - data = self.getpq(response.text) - videos = self.getlist(data('#archive article a, #index article a')) - return {'list': videos, 'page': pg} - - except Exception as e: - print(f"searchContent error: {e}") - return {'list': [], 'page': pg} - - def playerContent(self, flag, id, vipFlags): - url = id - p = 1 - if self.isVideoFormat(url): - # m3u8/mp4 direct play; when using proxy setting, wrap to proxy for m3u8 - if '.m3u8' in url: - url = self.proxy(url) - p = 0 - self.log(f"播放请求: parse={p}, url={url}") - print(f"播放请求: parse={p}, url={url}") - return {'parse': p, 'url': url, 'header': self.headers} - - def localProxy(self, param): - if param.get('type') == 'img': - res=requests.get(param['url'], headers=self.headers, proxies=self.proxies, timeout=10) - return [200,res.headers.get('Content-Type'),self.aesimg(res.content)] - elif param.get('type') == 'm3u8':return self.m3Proxy(param['url']) - else:return self.tsProxy(param['url']) - - def proxy(self, data, type='m3u8'): - if data and len(self.proxies):return f"{self.getProxyUrl()}&url={self.e64(data)}&type={type}" - else:return data - - def m3Proxy(self, url): - url=self.d64(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 - iskey=True - for index, string in enumerate(lines): - if iskey and 'URI' in string: - pattern = r'URI="([^"]*)"' - match = re.search(pattern, string) - if match: - lines[index] = re.sub(pattern, f'URI="{self.proxy(match.group(1), "mkey")}"', string) - iskey=False - continue - 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): - url = self.d64(url) - data = requests.get(url, headers=self.headers, proxies=self.proxies, stream=True) - return [200, data.headers['Content-Type'], data.content] - - 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 get_working_host(self): - """Get working host from known dynamic URLs""" - # Known working URLs from the dynamic gateway - dynamic_urls = [ - 'https://advise.nlwkmsv.cc/', - 'advise.nlwkmsv.cc', - # 'https://am.vgwtswi.xyz' - ] - - # Test each URL to find a working one - for url in dynamic_urls: - try: - response = requests.get(url, headers=self.headers, proxies=self.proxies, timeout=10) - if response.status_code == 200: - # Verify it has the expected content structure - data = self.getpq(response.text) - articles = data('#index article a') - if len(articles) > 0: - self.log(f"选用可用站点: {url}") - print(f"选用可用站点: {url}") - return url - except Exception as e: - continue - - # Fallback to first URL if none work (better than crashing) - self.log(f"未检测到可用站点，回退: {dynamic_urls[0]}") - print(f"未检测到可用站点，回退: {dynamic_urls[0]}") - return dynamic_urls[0] - - - def getlist(self, data, tid=''): - videos = [] - l = '/mrdg' in tid - for k in data.items(): - a = k.attr('href') - b = k('h2').text() - # Some pages might not include datePublished; use a fallback - c = k('span[itemprop="datePublished"]').text() or k('.post-meta, .entry-meta, time').text() - if a and b: - videos.append({ - 'vod_id': f"{a}{'@folder' if l else ''}", - 'vod_name': b.replace('\n', ' '), - 'vod_pic': self.getimg(k('script').text()), - 'vod_remarks': c or '', - 'vod_tag': 'folder' if l else '', - 'style': {"type": "rect", "ratio": 1.33} - }) - return videos - - def getfod(self, id): - url = f"{self.host}{id}" - data = self.getpq(requests.get(url, headers=self.headers, proxies=self.proxies).text) - vdata=data('.post-content[itemprop="articleBody"]') - r=['.txt-apps','.line','blockquote','.tags','.content-tabs'] - for i in r:vdata.remove(i) - p=vdata('p') - videos=[] - for i,x in enumerate(vdata('h2').items()): - c=i*2 - videos.append({ - 'vod_id': p.eq(c)('a').attr('href'), - 'vod_name': p.eq(c).text(), - 'vod_pic': f"{self.getProxyUrl()}&url={p.eq(c+1)('img').attr('data-xkrkllgl')}&type=img", - 'vod_remarks':x.text() - }) - return videos - - def getimg(self, text): - match = re.search(r"loadBannerDirect\('([^']+)'", text) - if match: - url = match.group(1) - return f"{self.getProxyUrl()}&url={url}&type=img" - else: - return '' - - def aesimg(self, word): - key = b'f5d965df75336270' - iv = b'97b60394abc2fbe1' - cipher = AES.new(key, AES.MODE_CBC, iv) - decrypted = unpad(cipher.decrypt(word), AES.block_size) - return decrypted - - def getpq(self, data): - try: - return pq(data) - except Exception as e: - print(f"{str(e)}") - return pq(data.encode('utf-8'))