Sync all projects

This commit is contained in:
github-actions[bot]
2026-07-27 11:54:56 +00:00
parent 7023648359
commit 4e4e437247
83 changed files with 32938 additions and 13627 deletions
+163
View File
@@ -0,0 +1,163 @@
# -*- coding: utf-8 -*-
# 8x8x官网: https://www.7xb38c.com/
import sys,re,json,base64
from urllib.parse import quote
sys.path.append('..')
try:
from base.spider import Spider as _Base
except ImportError:
class _Base:
def fetch(self,url,headers=None,**kw):
import requests as rq
kw.pop('timeout',None);r=rq.get(url,headers=headers,timeout=15,**kw)
r.encoding='utf-8';return r
try:
import curl_cffi.requests as cr
_HAS_CFFI=True
except ImportError:
_HAS_CFFI=False
import requests as cr
_H=[104,116,116,112,115,58,47,47,119,119,119,46,51,97,98,102,117,103,57,50,100,46,99,111,109]
H=bytes(_H).decode()
U="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
CATS={1:"大陆",2:"日韩",3:"欧美",4:"动漫",5:"三级"}
TC=None
# ── 动态发现 body path ──
def _discover_body(session):
"""从 SPA 壳中提取 body 路径前缀, 失败返回默认值"""
try:
kw={"timeout":10}
if _HAS_CFFI:kw["impersonate"]="chrome120"
r=session.get(H+"/",**kw)
r.raise_for_status()
js_name=re.search(r'src=/assets/(app\.[a-f0-9]+\.js)',r.text)
if not js_name:return"/cou345w"
r2=session.get(H+"/assets/"+js_name.group(1),**kw)
r2.raise_for_status()
seg=re.search(r'atob\("([^"]+)"\)',r2.text)
if not seg:return"/cou345w"
return"/"+base64.b64decode(seg.group(1)).decode()
except Exception as e:
print(f"[8x8x] body discover failed: {e}, using default")
return"/cou345w"
class Spider(_Base):
def init(self,extend=""):
self._s=cr.Session()
self._s.headers.update({"User-Agent":U,"Accept-Language":"zh-CN,zh;q=0.9"})
self._s.verify=False
# ★ 动态发现 body path
self._bd=_discover_body(self._s)
print(f"[8x8x] body path: {self._bd}")
def getName(self):return"8x8x"
def isVideoFormat(self,u):return".m3u8"in u or".mp4"in u
def manualVideoCheck(self):return False
def _get(self,url,timeout=15):
if not url.startswith("http"):url=H+self._bd+url
try:
kw={"timeout":timeout}
if _HAS_CFFI:kw["impersonate"]="chrome120"
r=self._s.get(url,**kw);r.raise_for_status()
if hasattr(r,'encoding'):r.encoding='utf-8'
return r.text
except Exception as e:
print(f"[8x8x]GET {url[:60]} -> {e}")
return""
def _tags(self):
global TC
if TC is not None:return TC
h=self._get("/");g={}
if not h:TC=g;return g
for gm in re.finditer(r'<div class=tag-group data-group-id=\d+><span class=tag-group-label>([^<]+)</span>(.*?)</div>',h,re.DOTALL):
gn=gm.group(1);inner=gm.group(2)
ts=re.findall(r'<a href=(/tags/[^/]+/)\s[^>]*>([^<]+)</a>',inner)
if ts:g[gn]=ts
TC=g;return g
def _cards(self,h):
v=[]
for m in re.finditer(r'<a href=(/vd/(\d+)/)\s[^>]*>(.*?)</a>',h,re.DOTALL):
inner=m.group(3)
tm=re.search(r'<div class=card-title>(.*?)</div>',inner)
im=re.search(r'data-src=([^\s>]+)',inner)
v.append({"vod_id":m.group(1),"vod_name":tm.group(1)if tm else"N/A","vod_pic":im.group(1)if im else"","vod_remarks":""})
return v
def homeContent(self,filter=False):
t=self._tags();cs=[]
for cid,cn in sorted(CATS.items()):cs.append({"type_id":str(cid),"type_name":cn})
for gn in sorted(t.keys()):
for url,name in t[gn]:cs.append({"type_id":url,"type_name":f"[{gn}] {name}"})
return{"class":cs}
def homeVideoContent(self):
h=self._get("/")
return{"list":self._cards(h)[:30]if h else[]}
def categoryContent(self,tid,pg=1,filter=False,extend=None):
try:
pn=max(int(str(pg)),1)
base=tid.rstrip("/")if tid.startswith("/tags/")else f"/category/{int(tid)}"
h=self._get(f"{base}/page/{pn}/")
if not h:return{"list":[],"page":pg,"pagecount":1}
mp=re.search(r'data-max=(\d+)',h);pc=int(mp.group(1))if mp else pn
v=self._cards(h)
return{"list":v,"page":pn,"pagecount":pc,"limit":len(v),"total":pc*len(v)if v else 0}
except Exception as e:print(f"[8x8x]cat:{e}");return{"list":[],"page":pg,"pagecount":1}
def detailContent(self,ids):
vid=ids[0]
if not vid.startswith("/vd/"):vid=f"/vd/{vid.strip('/')}/"
h=self._get(vid)
if not h:return{"list":[]}
tm=re.search(r'<title>(.*?)</title>',h);title=tm.group(1).replace(" - 8x8x","")if tm else""
pm=re.search(r'data-poster=([^\s>]+)',h);pic=pm.group(1).strip('"').strip("'")if pm else""
mm=re.search(r'data-m3u8=([^\s>]+)',h)
if not mm:return{"list":[{"vod_id":ids[0],"vod_name":title,"vod_pic":pic}]}
m3u8_path=mm.group(1).strip('"').strip("'")
pf=[];pu=[]
for i,rn in enumerate(["data-route1","data-route2","data-route3"],1):
rm=re.search(rf'{rn}=([^\s>]+)',h)
if rm:
rt=rm.group(1).strip('"').strip("'")
full_m3u8=rt.rstrip("/")+"/"+m3u8_path.lstrip("/")
pf.append(f"线路{i}")
pu.append(f"线路{i}${full_m3u8}")
return{"list":[{"vod_id":ids[0],"vod_name":title,"vod_pic":pic,"type_name":"","vod_year":"","vod_area":"","vod_remarks":"","vod_actor":"","vod_director":"","vod_content":"","vod_play_from":"$$$".join(pf),"vod_play_url":"$$$".join(pu)}]}
def playerContent(self,flag,id,vipFlags=None):
if id and".m3u8"in id:return{"url":id,"header":json.dumps({"User-Agent":U,"Referer":H+"/"})}
d=self.detailContent([id])
if d and d.get("list"):
urls=d["list"][0].get("vod_play_url","").split("$$$")
if urls:
first=urls[0]
if"$"in first:first=first.split("$",1)[1]
return{"url":first,"header":json.dumps({"User-Agent":U,"Referer":H+"/"})}
return{"url":""}
def searchContent(self,key,quick=False,pg=1):
try:
pn=max(int(str(pg)),1)
url=f"{H}/api/search/video?keyword={quote(key)}&page={pn}"
kw={"timeout":15}
if _HAS_CFFI:kw["impersonate"]="chrome120"
r=self._s.get(url,**kw);r.raise_for_status()
data=r.json()
if data.get("code")!=0:return{"list":[]}
dl=data.get("data",{})
videos=[]
for item in dl.get("list",[]):
videos.append({"vod_id":f"/vd/{item['id']}/","vod_name":item.get("title",""),"vod_pic":item.get("litpic",""),"vod_remarks":item.get("typename","")})
return{"list":videos,"page":pn,"pagecount":dl.get("total_pages",pn),"limit":len(videos),"total":dl.get("total",0)}
except Exception as e:print(f"[8x8x]search:{e}");return{"list":[],"page":pg,"pagecount":1}
def localProxy(self,param):pass
+192
View File
@@ -0,0 +1,192 @@
#coding=utf-8
#!/usr/bin/python
import sys
sys.path.append('..')
import json
import re
from base.spider import Spider
class Spider(Spider):
def getName(self):
return "91Porn"
def init(self, extend):
self.baseUrl = "https://91porn.com"
self.header = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept-Language": "zh-CN,zh;q=0.9",
"Referer": self.baseUrl + "/index.php"
}
self.cookies = {}
# ✨ 核心改造:动态从 config.json 的 ext 参数中读取配置
self.username = ""
self.password = ""
self.email = ""
try:
if extend:
extendDict = json.loads(extend) if isinstance(extend, str) else extend
self.username = str(extendDict.get('username', '')).strip()
self.password = str(extendDict.get('password', '')).strip()
self.email = str(extendDict.get('email', '')).strip()
except Exception:
pass
self.is_ready = False
def safe_bypass_and_verify(self):
if self.is_ready:
return True
try:
rsp = self.fetch(self.baseUrl + "/email_verify.php", headers=self.header, timeout=5)
self.cookies.update(rsp.cookies.get_dict())
self.cookies['language'] = 'cn_CN'
self.cookies['CNAM'] = '1'
# 如果外部没有配置邮箱,则直接作为普通游客会话放行
if not self.email:
self.is_ready = True
return True
post_data = {"email": self.email, "recover": "Submit", "submit": "true"}
v_hd = self.header.copy()
v_hd["Content-Type"] = "application/x-www-form-urlencoded"
v_hd["Referer"] = self.baseUrl + "/email_verify.php"
rsp_v = self.post(self.baseUrl + "/email_verify.php", data=post_data, headers=v_hd, cookies=self.cookies, timeout=5)
self.cookies.update(rsp_v.cookies.get_dict())
self.is_ready = True
return True
except Exception:
self.is_ready = True
return False
def homeContent(self, filter):
result = {}
classList = [
{"type_name": "今日排行", "type_id": "hot"},
{"type_name": "最近更新", "type_id": "rp"},
{"type_name": "本月最热", "type_id": "md"}
]
result['class'] = classList
return result
def homeVideoContent(self):
return {'list': []}
def categoryContent(self, cid, page, filter, ext):
self.safe_bypass_and_verify()
result = {'page': int(page), 'pagecount': 1, 'limit': 0, 'total': 0, 'list': []}
page = int(page)
url = self.baseUrl + "/v.php?category=" + cid + "&page=" + str(page)
if cid == "hot":
url = self.baseUrl + "/index.php"
elif cid == "rp":
url = self.baseUrl + "/v.php?next=watch&page=" + str(page)
try:
rsp = self.fetch(url, headers=self.header, cookies=self.cookies, timeout=5)
html = self.html(rsp.text)
items = html.xpath("//div[contains(@class, 'well')]")
videos = []
for item in items:
try:
a_tag = item.xpath(".//a[contains(@href, 'view_video.php')]")
if not a_tag:
continue
href = a_tag[0].get('href', '')
v_match = re.search(r'viewkey=([a-zA-Z0-9]+)', href)
if not v_match:
continue
vod_id = v_match.group(1)
name_nodes = item.xpath(".//span[contains(@class, 'video-title')]/text()")
vod_name = name_nodes[0].strip() if name_nodes else "精彩视频"
img_nodes = item.xpath(".//img[contains(@class, 'img-responsive')]/@src")
vod_pic = img_nodes[0].strip() if img_nodes else ""
if vod_pic.startswith('//'):
vod_pic = "https:" + vod_pic
remark_nodes = item.xpath(".//span[@class='duration']/text()")
vod_remarks = remark_nodes[0].strip() if remark_nodes else "完整版"
videos.append({"vod_id": vod_id, "vod_name": self.cleanText(self.removeHtmlTags(vod_name)), "vod_pic": vod_pic, "vod_remarks": vod_remarks})
except:
continue
result['list'] = videos
result['limit'] = len(videos)
result['pagecount'] = page + 1 if len(videos) >= 10 else page
except Exception:
pass
return result
def detailContent(self, did):
self.safe_bypass_and_verify()
tid = did[0]
url = self.baseUrl + "/view_video.php?viewkey=" + tid
try:
rsp = self.fetch(url, headers=self.header, cookies=self.cookies, timeout=5)
html_text = rsp.text.replace('&amp;', '&')
root = self.html(html_text)
title_nodes = root.xpath("//h4[contains(@class, 'login_register_header')]/text() | //title/text()")
title = title_nodes[0].strip().replace(" - 91porn", "").strip() if title_nodes else "精彩视频"
cover_nodes = root.xpath("//video/@poster")
cover_pic = cover_nodes[0] if cover_nodes else ""
real_video_url = ""
strencode_match = re.search(r'strencode2\([\"\']([^\"\'\)]+)[\"\']\)', html_text)
if strencode_match:
ciphertext = strencode_match.group(1)
try:
from urllib.parse import unquote
decrypted_html = unquote(ciphertext)
except:
import urllib
decrypted_html = urllib.unquote(ciphertext)
src_match = re.search(r"src=['\"]([^'\"]+)['\"]", decrypted_html)
if src_match:
real_video_url = src_match.group(1)
if not real_video_url:
src_nodes = root.xpath("//video/source/@src | //video/@src")
if src_nodes:
real_video_url = src_nodes[0]
if not real_video_url:
real_video_url = url
vod = {"vod_id": tid, "vod_name": self.cleanText(self.removeHtmlTags(title)), "vod_pic": cover_pic, "type_name": "在线视频", "vod_content": "资源解析就绪", "vod_play_from": "91Porn秒解流", "vod_play_url": "直连资源源$" + real_video_url}
return {'list': [vod]}
except Exception:
pass
return {'list': []}
def playerContent(self, flag, pid, vipFlags):
need_parse = 1 if "view_video.php" in pid else 0
return {"url": pid, "header": self.header, "parse": need_parse}
def searchContent(self, key, quick):
return {'list': []}
def searchContentPage(self, key, quick, page):
return {'list': []}
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def localProxy(self, params):
return None
def destroy(self):
pass
+255
View File
@@ -0,0 +1,255 @@
#author Kyle
import sys
import time
import json
import re
import urllib.parse
from base64 import b64decode, b64encode
sys.path.append('..')
from base.spider import Spider
class Spider(Spider):
def init(self, extend=""):
self.host = "https://www.eporner.com"
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': self.host + '/',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8'
}
self.timeout = 10
self.retries = 2
self.cookies = {"EPRNS": "1"}
def getName(self):
return "EP涩"
def isVideoFormat(self, url):
return True
def manualVideoCheck(self):
return False
def destroy(self):
pass
def homeContent(self, filter):
result = {}
result['class'] = [
{"type_id": "/cat/all/", "type_name": "最新"},
{"type_id": "/best-videos/", "type_name": "最佳视频"},
{"type_id": "/top-rated/", "type_name": "最高评分"},
{"type_id": "/cat/4k-porn/", "type_name": "4K"},
]
return result
def homeVideoContent(self):
result = {}
videos = self._api_search(query="all", page=1, order="latest", gay="0", per_page=20)
result['list'] = videos
return result
def categoryContent(self, tid, pg, filter, extend):
result = {}
params = self._map_tid_to_api(tid)
videos = self._api_search(page=int(pg), **params)
result['list'] = videos
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = len(videos) or 20
result['total'] = 999999
return result
def searchContent(self, key, quick, pg="1"):
result = {}
videos = self._api_search(query=key or "all", page=int(pg), order="latest", gay="0")
result['list'] = videos
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = len(videos) or 20
result['total'] = 999999
return result
def detailContent(self, ids):
result = {}
url = ids[0]
if not url.startswith('http'):
url = self.host + url
r = self.fetch(url, headers=self.headers)
root = self.html(r.text)
og_title = root.xpath('//meta[@property="og:title"]/@content')
if og_title:
title = og_title[0].replace(' - EPORNER', '').strip()
else:
h1 = "".join(root.xpath('//h1//text()'))
title = re.sub(r"\s*(\d+\s*min.*)$", "", h1).strip() or "爱看AV"
img_elem = root.xpath('//meta[@property="og:image"]/@content') or root.xpath('//img[@id="mainvideoimg"]/@src')
thumbnail = img_elem[0] if img_elem else ""
meta_desc = (root.xpath('//meta[@name="description"]/@content') or root.xpath('//meta[@property="og:description"]/@content'))
desc_text = meta_desc[0] if meta_desc else ""
dur_match = re.search(r"Duration:\s*([0-9:]+)", desc_text)
duration = dur_match.group(1) if dur_match else ""
encoded_url = self.e64(url)
play_url = f"播放${encoded_url}"
vod = {
"vod_id": url,
"vod_name": title,
"vod_pic": thumbnail,
"vod_remarks": duration,
"vod_content": desc_text.strip(),
"vod_play_from": "🍑Play",
"vod_play_url": play_url
}
result['list'] = [vod]
return result
def playerContent(self, flag, id, vipFlags):
result = {}
url = self.d64(id)
if not url.startswith('http'):
url = self.host + url
r = self.fetch(url, headers=self.headers)
html_content = r.text
pattern = r"vid\s*=\s*'([^']+)';\s*[\w*\.]+hash\s*=\s*['\"]([\da-f]{32})"
match = re.search(pattern, html_content)
if match:
vid, hash_val = match.groups()
hash_code = ''.join((self.encode_base_n(int(hash_val[i:i + 8], 16), 36) for i in range(0, 32, 8)))
xhr_url = f"{self.host}/xhr/video/{vid}?hash={hash_code}&device=generic&domain=www.eporner.com&fallback=false&embed=false&supportedFormats=mp4"
xhr_headers = {
**self.headers,
'X-Requested-With': 'XMLHttpRequest',
'Accept': 'application/json, text/javascript, */*; q=0.01'
}
resp = self.fetch(xhr_url, headers=xhr_headers)
data = json.loads(resp.text)
if data.get("available", True):
sources_block = data.get("sources", {})
sources = sources_block.get("mp4", {})
final_url = None
if sources:
quality_sorted = sorted(sources.keys(), key=lambda q: int(re.sub(r"\D", "", q) or 0), reverse=True)
best_quality = quality_sorted[0]
final_url = sources[best_quality].get("src")
else:
hls = sources_block.get("hls")
if isinstance(hls, dict):
final_url = hls.get("src")
elif isinstance(hls, list) and hls:
final_url = hls[0].get("src")
best_quality = "hls"
if final_url:
result["parse"] = 0
result["url"] = final_url
result["header"] = {
'User-Agent': self.headers['User-Agent'],
'Referer': url,
'Origin': self.host,
'Accept': '*/*'
}
return result
def encode_base_n(self, num, n, table=None):
FULL_TABLE = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
if not table:
table = FULL_TABLE[:n]
if n > len(table):
raise ValueError('base %d exceeds table length %d' % (n, len(table)))
if num == 0:
return table[0]
ret = ''
while num:
ret = table[num % n] + ret
num = num // n
return ret
def e64(self, text):
return b64encode(text.encode()).decode()
def d64(self, encoded_text):
return b64decode(encoded_text.encode()).decode()
def html(self, content):
from lxml import etree
return etree.HTML(content)
def fetch(self, url, headers=None, timeout=None):
import requests, ssl
ssl._create_default_https_context = ssl._create_unverified_context
if headers is None:
headers = self.headers
if timeout is None:
timeout = self.timeout
for _ in range(self.retries + 1):
resp = requests.get(url, headers=headers, timeout=timeout, verify=False, cookies=getattr(self, 'cookies', None))
resp.encoding = 'utf-8'
return resp
time.sleep(1)
raise Exception("Fetch failed")
def _parse_video_list(self, root):
videos = []
items = root.xpath('//div[contains(@class, "mb") and @data-id]')
for item in items:
link = item.xpath('.//a[contains(@href, "/video-")]/@href')
if not link:
continue
vod_id = self.host + link[0]
title = "".join(item.xpath('.//p[contains(@class, "mbtit")]//text()')).strip()
img = item.xpath('.//img/@src')
thumbnail = img[0] if img else ""
duration = "".join(item.xpath('.//span[contains(@class,"mbtim")]/text()')).strip()
videos.append({
"vod_id": vod_id,
"vod_name": title or "爱看AV",
"vod_pic": thumbnail,
"vod_remarks": duration
})
return videos
def _map_tid_to_api(self, tid: str):
params = {"query": "all", "order": "latest", "gay": "0", "per_page": 30}
t = (tid or '').strip('/').lower()
if t.startswith('best-videos'):
params["order"] = "most-popular"
elif t.startswith('top-rated'):
params["order"] = "top-rated"
elif t.startswith('cat/4k-porn'):
params["query"] = "4k"
elif t.startswith('cat/gay'):
params["gay"] = "2"
params["order"] = "latest"
else:
params["gay"] = "0"
return params
def _api_search(self, query="all", page=1, order="latest", gay="0", per_page=30, thumbsize="medium"):
base = f"{self.host}/api/v2/video/search/"
q = {
"query": query,
"per_page": per_page,
"page": page,
"thumbsize": thumbsize,
"order": order,
"format": "json"
}
if gay is not None:
q["gay"] = gay
url = base + "?" + urllib.parse.urlencode(q)
r = self.fetch(url, headers={**self.headers, 'X-Requested-With': 'XMLHttpRequest'})
data = json.loads(r.text)
return self._parse_api_list(data)
def _parse_api_list(self, data: dict):
videos = []
for v in (data or {}).get('videos', []):
vurl = v.get('url') or ''
title = v.get('title') or ''
thumb = (v.get('default_thumb') or {}).get('src') or ''
remarks = v.get('length_min') or ''
videos.append({
"vod_id": vurl if vurl.startswith('http') else (self.host + vurl),
"vod_name": title or "爱看AV",
"vod_pic": thumb,
"vod_remarks": remarks
})
return videos
+335
View File
@@ -0,0 +1,335 @@
import requests
from bs4 import BeautifulSoup
import re
from base.spider import Spider
import sys
import json
import base64
import urllib.parse
from Crypto.Cipher import ARC4
from Crypto.Util.Padding import unpad
import binascii
sys.path.append('..')
xurl = "https://www.fullhd.xxx/zh/"
headerx = {
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.87 Safari/537.36'
}
pm = ''
class Spider(Spider):
global xurl
global headerx
def getName(self):
return "首页"
def init(self, extend):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def extract_middle_text(self, text, start_str, end_str, pl, start_index1: str = '', end_index2: str = ''):
if pl == 3:
plx = []
while True:
start_index = text.find(start_str)
if start_index == -1:
break
end_index = text.find(end_str, start_index + len(start_str))
if end_index == -1:
break
middle_text = text[start_index + len(start_str):end_index]
plx.append(middle_text)
text = text.replace(start_str + middle_text + end_str, '')
if len(plx) > 0:
purl = ''
for i in range(len(plx)):
matches = re.findall(start_index1, plx[i])
output = ""
for match in matches:
match3 = re.search(r'(?:^|[^0-9])(\d+)(?:[^0-9]|$)', match[1])
if match3:
number = match3.group(1)
else:
number = 0
if 'http' not in match[0]:
output += f"#{'📽️' + match[1]}${number}{xurl}{match[0]}"
else:
output += f"#{'📽️' + match[1]}${number}{match[0]}"
output = output[1:]
purl = purl + output + "$$$"
purl = purl[:-3]
return purl
else:
return ""
else:
start_index = text.find(start_str)
if start_index == -1:
return ""
end_index = text.find(end_str, start_index + len(start_str))
if end_index == -1:
return ""
if pl == 0:
middle_text = text[start_index + len(start_str):end_index]
return middle_text.replace("\\", "")
if pl == 1:
middle_text = text[start_index + len(start_str):end_index]
matches = re.findall(start_index1, middle_text)
if matches:
jg = ' '.join(matches)
return jg
if pl == 2:
middle_text = text[start_index + len(start_str):end_index]
matches = re.findall(start_index1, middle_text)
if matches:
new_list = [f'{item}' for item in matches]
jg = '$$$'.join(new_list)
return jg
def homeContent(self, filter):
result = {}
result = {"class": [{"type_id": "latest-updates", "type_name": "最新视频🌠"},
{"type_id": "top-rated", "type_name": "最佳视频🌠"},
{"type_id": "most-popular", "type_name": "热门影片🌠"}],
}
return result
def homeVideoContent(self):
videos = []
try:
detail = requests.get(url=xurl, headers=headerx)
detail.encoding = "utf-8"
res = detail.text
doc = BeautifulSoup(res, "lxml")
# Get videos from different sections
sections = {
"latest-updates": "最新视频",
"top-rated": "最佳视频",
"most-popular": "热门影片"
}
for section_id, section_name in sections.items():
section = doc.find('div', id=f"list_videos_videos_watched_right_now_items")
if not section:
continue
vods = section.find_all('div', class_="item")
for vod in vods:
names = vod.find_all('a')
name = names[0]['title'] if names and 'title' in names[0].attrs else section_name
ids = vod.find_all('a')
id = ids[0]['href'] if ids else ""
pics = vod.find('img', class_="lazyload")
pic = pics['data-src'] if pics and 'data-src' in pics.attrs else ""
if pic and 'http' not in pic:
pic = xurl + pic
remarks = vod.find('span', class_="duration")
remark = remarks.text.strip() if remarks else ""
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
result = {'list': videos}
return result
except Exception as e:
print(f"Error in homeVideoContent: {str(e)}")
return {'list': []}
def categoryContent(self, cid, pg, filter, ext):
result = {}
videos = []
try:
if pg and int(pg) > 1:
url = f'{xurl}/{cid}/{pg}/'
else:
url = f'{xurl}/{cid}/'
detail = requests.get(url=url, headers=headerx)
detail.encoding = "utf-8"
res = detail.text
doc = BeautifulSoup(res, "lxml")
section = doc.find('div', class_="list-videos")
if section:
vods = section.find_all('div', class_="item")
for vod in vods:
names = vod.find_all('a')
name = names[0]['title'] if names and 'title' in names[0].attrs else ""
ids = vod.find_all('a')
id = ids[0]['href'] if ids else ""
pics = vod.find('img', class_="lazyload")
pic = pics['data-src'] if pics and 'data-src' in pics.attrs else ""
if pic and 'http' not in pic:
pic = xurl + pic
remarks = vod.find('span', class_="duration")
remark = remarks.text.strip() if remarks else ""
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
except Exception as e:
print(f"Error in categoryContent: {str(e)}")
result = {
'list': videos,
'page': pg,
'pagecount': 9999,
'limit': 90,
'total': 999999
}
return result
def detailContent(self, ids):
global pm
did = ids[0]
result = {}
videos = []
playurl = ''
if 'http' not in did:
did = xurl + did
res1 = requests.get(url=did, headers=headerx)
res1.encoding = "utf-8"
res = res1.text
content = '👉' + self.extract_middle_text(res,'<h1>','</h1>', 0)
yanuan = self.extract_middle_text(res, '<span>Pornstars:</span>','</div>',1, 'href=".*?">(.*?)</a>')
bofang = did
videos.append({
"vod_id": did,
"vod_actor": yanuan,
"vod_director": '',
"vod_content": content,
"vod_play_from": '💗4K💗',
"vod_play_url": bofang
})
result['list'] = videos
return result
def playerContent(self, flag, id, vipFlags):
parts = id.split("http")
xiutan = 0
if xiutan == 0:
if len(parts) > 1:
before_https, after_https = parts[0], 'http' + parts[1]
res = requests.get(url=after_https, headers=headerx)
res = res.text
url2 = self.extract_middle_text(res, '<video', '</video>', 0).replace('\\', '')
soup = BeautifulSoup(url2, 'html.parser')
first_source = soup.find('source')
src_value = first_source.get('src')
response = requests.head(src_value, allow_redirects=False)
if response.status_code == 302:
redirect_url = response.headers['Location']
response = requests.head(redirect_url, allow_redirects=False)
if response.status_code == 302:
redirect_url = response.headers['Location']
result = {}
result["parse"] = xiutan
result["playUrl"] = ''
result["url"] = redirect_url
result["header"] = headerx
return result
def searchContentPage(self, key, quick, page):
result = {}
videos = []
if not page:
page = '1'
if page == '1':
url = f'{xurl}/search/{key}/'
else:
url = f'{xurl}/search/{key}/{str(page)}/'
try:
detail = requests.get(url=url, headers=headerx)
detail.encoding = "utf-8"
res = detail.text
doc = BeautifulSoup(res, "lxml")
section = doc.find('div', class_="list-videos")
if section:
vods = section.find_all('div', class_="item")
for vod in vods:
names = vod.find_all('a')
name = names[0]['title'] if names and 'title' in names[0].attrs else ""
ids = vod.find_all('a')
id = ids[0]['href'] if ids else ""
pics = vod.find('img', class_="lazyload")
pic = pics['data-src'] if pics and 'data-src' in pics.attrs else ""
if pic and 'http' not in pic:
pic = xurl + pic
remarks = vod.find('span', class_="duration")
remark = remarks.text.strip() if remarks else ""
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
except Exception as e:
print(f"Error in searchContentPage: {str(e)}")
result = {
'list': videos,
'page': page,
'pagecount': 9999,
'limit': 90,
'total': 999999
}
return result
def searchContent(self, key, quick):
return self.searchContentPage(key, quick, '1')
def localProxy(self, params):
if params['type'] == "m3u8":
return self.proxyM3u8(params)
elif params['type'] == "media":
return self.proxyMedia(params)
elif params['type'] == "ts":
return self.proxyTs(params)
return None
+71
View File
@@ -0,0 +1,71 @@
import re
import requests
class Spider():
def getName(self):
return 'JAV36'
def init(self, extend=''):
pass
def getDependence(self):
return []
def homeContent(self, filter):
return {
'class': [
{'type_name': '最新更新', 'type_id': 'latest-updates/'},
{'type_name': '4K高清', 'type_id': 'tags/4k/'}
],
'list': []
}
def homeVideoContent(self):
return self.categoryContent('latest-updates/', 1, False, {})
def categoryContent(self, tid, pg, filter, extend):
url = f'https://jav36.com/{tid}'
if int(pg) > 1:
url = f'https://jav36.com/{tid}{pg}/'
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'}
try:
res = requests.get(url, headers=headers, timeout=15)
return {'list': self.parse_list(res.text)}
except: return {'list': []}
def searchContent(self, keyword, quick, pg=1):
url = f'https://jav36.com/search/{keyword}/'
if int(pg) > 1:
url = f'https://jav36.com/search/{keyword}/{pg}/'
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'}
try:
res = requests.get(url, headers=headers, timeout=15)
return {'list': self.parse_list(res.text)}
except: return {'list': []}
def parse_list(self, html):
vod_list = []
pattern = r'href=\"https://jav36\.com/videos/(?P<id_path>\d+/(?P<id>[^/]+)/)\" title=\"(?P<name>[^\"]+)\".*?data-original=\"(?P<pic>[^\"]+)\"'
for m in re.finditer(pattern, html, re.S):
vod_list.append({
'vod_id': m.group('id_path'),
'vod_name': m.group('name').strip(),
'vod_pic': m.group('pic'),
'vod_remarks': 'Full HD'
})
return vod_list
def detailContent(self, ids):
return {'list': [{'vod_id': ids[0], 'vod_name': 'Video', 'vod_play_from': 'Direct', 'vod_play_url': 'Play$'+ids[0]}]}
def playerContent(self, flag, id, vipFlags):
url = f'https://jav36.com/videos/{id}'
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'}
try:
res = requests.get(url, headers=headers, timeout=10)
match = re.search(r'\"contentUrl\"\s*:\s*\"([^\"]+)\"', res.text)
if match:
return {'parse': 0, 'url': match.group(1), 'header': {'User-Agent': headers['User-Agent'], 'Referer': 'https://jav36.com/'}}
except: pass
return {'parse': 0, 'url': url}
+161
View File
@@ -0,0 +1,161 @@
# -*- coding: utf-8 -*-
import re
import urllib.parse
import requests
try:
from base.spider import Spider as BaseSpider
except ImportError:
class BaseSpider:
pass
class Spider(BaseSpider):
BASE_URL = "https://jable.sbs"
FALLBACK_URLS = ["https://jable.sbs", "https://jable.tv"]
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Referer": BASE_URL + "/",
}
def __init__(self):
super().__init__()
self.name = "JableTV"
self.session = requests.Session()
self.session.headers.update(self.HEADERS)
self._class_cache = None
def init(self, extend="{}"):
return None
def getName(self):
return self.name
def homeContent(self, filter):
html = self._get(self.BASE_URL + "/latest-updates/")
return {"class": self._classes(), "filters": {}, "list": self._parse_list(html), "parse": 0, "jx": 0}
def homeVideoContent(self):
return {"list": self._parse_list(self._get(self.BASE_URL + "/latest-updates/"))}
def categoryContent(self, tid, pg, filter, extend):
page = self._to_int(pg, 1)
path = str(tid or "latest-updates").strip("/")
url = self.BASE_URL + "/" + path + "/" if page <= 1 else self.BASE_URL + "/" + path + "/" + str(page) + "/"
data = self._parse_list(self._get(url))
return {"page": page, "pagecount": page if len(data) < 10 else page + 1, "limit": 24, "total": 99999, "list": data, "parse": 0, "jx": 0}
def detailContent(self, ids):
result = {"list": [], "parse": 0, "jx": 0}
if not ids:
return result
url = self._fix_url(ids[0] if str(ids[0]).startswith("http") else self.BASE_URL + "/videos/" + str(ids[0]).strip("/") + "/")
html = self._get(url)
name = self._clean(self._match(html, r'<h4[^>]*>(.*?)</h4>') or self._match(html, r'<meta[^>]+property=["\']og:title["\'][^>]+content=["\']([^"\']+)') or self._match(html, r'<title>(.*?)</title>').split("-")[0])
pic = self._match(html, r'<meta[^>]+property=["\']og:image["\'][^>]+content=["\']([^"\']+)') or self._match(html, r'<video[^>]+poster=["\']([^"\']+)') or self._match(html, r'<img[^>]+(?:data-src|src)=["\']([^"\']+)')
tags = ",".join([self._clean(x) for x in re.findall(r'<a[^>]+href=["\'][^"\']*/tags/[^"\']+["\'][^>]*>(.*?)</a>', html, re.S)])
remarks = self._clean(" ".join(re.findall(r'<h6[^>]*>(.*?)</h6>', html, re.S)[:3]))
content = self._clean(self._match(html, r'<div[^>]+class=["\'][^"\']*(?:description|info|text)[^"\']*["\'][^>]*>(.*?)</div>') or remarks or name)
m3u8 = self._m3u8(html)
result["list"].append({"vod_id": url, "vod_name": name, "vod_pic": urllib.parse.urljoin(self.BASE_URL, pic), "type_name": tags, "vod_year": "", "vod_area": "", "vod_remarks": remarks, "vod_actor": tags, "vod_director": "", "vod_content": content, "vod_play_from": "Jable", "vod_play_url": "正片$" + (m3u8 or url)})
return result
def searchContent(self, key, quick, pg="1"):
page = self._to_int(pg, 1)
q = urllib.parse.quote(str(key))
url = self.BASE_URL + "/search/" + q + "/" if page <= 1 else self.BASE_URL + "/search/" + q + "/" + str(page) + "/"
data = self._parse_list(self._get(url))
return {"page": page, "pagecount": page if len(data) < 10 else page + 1, "limit": 24, "total": 99999, "list": data, "parse": 0, "jx": 0}
def playerContent(self, flag, id, vipFlags):
result = {"parse": 0, "playUrl": "", "url": id or "", "jx": 0, "header": {"User-Agent": self.HEADERS["User-Agent"], "Referer": self.BASE_URL + "/"}}
if not id:
return result
if ".m3u8" in id or ".mp4" in id:
return result
play_page = self._fix_url(id if str(id).startswith("http") else self.BASE_URL + "/videos/" + str(id).strip("/") + "/")
html = self._get(play_page)
m3u8 = self._m3u8(html)
if m3u8:
result["url"] = m3u8
result["header"] = {"User-Agent": self.HEADERS["User-Agent"], "Referer": play_page, "Origin": self.BASE_URL}
else:
result["url"] = play_page
result["parse"] = 1
return result
def _classes(self, html=None):
if self._class_cache:
return self._class_cache
self._class_cache = [
{"type_id": "latest-updates", "type_name": "最近更新"},
{"type_id": "hot", "type_name": "热门影片"},
{"type_id": "new-release", "type_name": "全新上市"},
{"type_id": "tags/chinese-subtitle", "type_name": "中文字幕"},
{"type_id": "tags/drama", "type_name": "剧情"},
{"type_id": "tags/cosplay", "type_name": "角色扮演"},
]
return self._class_cache
def _parse_list(self, html):
data, seen = [], set()
cards = re.findall(r'(<div[^>]+class=["\'][^"\']*video-img-box[^"\']*["\'][\s\S]*?</h6>[\s\S]*?</div>\s*</div>)', html or "", re.S | re.I)
if not cards:
cards = re.findall(r'(<a[^>]+href=["\'][^"\']*/videos/[^"\']+["\'][\s\S]*?</a>)', html or "", re.S | re.I)
for item in cards:
href = self._match(item, r'href=["\']([^"\']*/videos/[^"\']+)["\']')
if not href:
continue
name = self._clean(self._match(item, r'<h6[^>]*class=["\'][^"\']*title[^"\']*["\'][^>]*>\s*<a[^>]*>(.*?)</a>') or self._match(item, r'title=["\']([^"\']+)') or self._match(item, r'alt=["\']([^"\']+)'))
pic = self._match(item, r'(?:data-src|data-original|data-lazy-src|data-lazyload)=["\']([^"\']+)') or self._match(item, r'<img[^>]+src=["\']([^"\']+)')
remarks = self._clean(self._match(item, r'<span[^>]+class=["\'][^"\']*(?:duration|label|badge)[^"\']*["\'][^>]*>(.*?)</span>') or self._match(item, r'(\d{1,2}:\d{2}(?::\d{2})?)'))
full = self._fix_url(urllib.parse.urljoin(self.BASE_URL, href))
if full not in seen and name and not re.fullmatch(r'\d{1,2}:\d{2}(?::\d{2})?', name):
seen.add(full)
data.append({"vod_id": full, "vod_name": name, "vod_pic": urllib.parse.urljoin(self.BASE_URL, pic), "vod_remarks": remarks})
return data
def _get(self, url, headers=None):
for real in self._candidate_urls(self._fix_url(url)):
h = dict(self.HEADERS)
h["Referer"] = self.BASE_URL + "/"
if headers:
h.update(headers)
try:
r = self.session.get(real, headers=h, timeout=15, verify=False)
r.encoding = "utf-8"
if r.status_code < 400 and "Just a moment" not in r.text and "cf-browser-verification" not in r.text:
return r.text
except Exception:
continue
return ""
def _candidate_urls(self, url):
urls = [url]
for host in self.FALLBACK_URLS:
p = urllib.parse.urlparse(url)
if p.netloc and host not in url:
urls.append(host + p.path + ("?" + p.query if p.query else ""))
return list(dict.fromkeys(urls))
def _fix_url(self, url):
return str(url or "").replace("https://jable.tv", self.BASE_URL).replace("http://jable.tv", self.BASE_URL).replace("https://www.jable.tv", self.BASE_URL)
def _m3u8(self, html):
return self._match(html, r'var\s+hlsUrl\s*=\s*["\']([^"\']+\.m3u8[^"\']*)') or self._match(html, r'["\'](https?://[^"\']+\.m3u8[^"\']*)["\']')
def _match(self, text, pattern):
m = re.search(pattern, text or "", re.S | re.I)
return m.group(1).strip() if m else ""
def _clean(self, text):
text = re.sub(r'<.*?>', '', text or '')
text = text.replace('&nbsp;', ' ').replace('&amp;', '&').replace('&#038;', '&').replace('&quot;', '"')
return re.sub(r'\s+', ' ', text).strip()
def _to_int(self, value, default=0):
try:
return int(value)
except Exception:
return default
+301
View File
@@ -0,0 +1,301 @@
# -*- coding: utf-8 -*-
import json
import re
import sys
from base64 import b64decode, b64encode
from urllib.parse import urlparse
import requests
from pyquery import PyQuery as pq
from requests import Session
sys.path.append('..')
from base.spider import Spider
class Spider(Spider):
def init(self, extend=""):
'''
内置代理配置:真心jar为例
{
"key": "Phb",
"name": "Phb",
"type": 3,
"searchable": 1,
"quickSearch": 1,
"filterable": 1,
"api": "./py/Phb.py",
"ext": {
"http": "http://127.0.0.1:1072",
"https": "http://127.0.0.1:1072"
}
},
注:http(s)代理都是http
'''
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/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',
'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',
}
self.host = self.gethost()
self.headers.update({'referer': f'{self.host}/', 'origin': self.host})
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
def homeContent(self, filter):
result = {}
cateManual = {
"视频": "/video",
"片单": "/playlists",
"频道": "/channels",
"分类": "/categories",
"明星": "/pornstars"
}
classes = []
filters = {}
for k in cateManual:
classes.append({
'type_name': k,
'type_id': cateManual[k]
})
result['class'] = classes
result['filters'] = filters
return result
def homeVideoContent(self):
data = self.getpq('/recommended')
vhtml = data("#recommendedListings .pcVideoListItem .phimage")
return {'list': self.getlist(vhtml)}
def categoryContent(self, tid, pg, filter, extend):
vdata = []
result = {}
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
if tid == '/video' or '_this_video' in tid:
pagestr = f'&' if '?' in tid else f'?'
tid = tid.split('_this_video')[0]
data = self.getpq(f'{tid}{pagestr}page={pg}')
vdata = self.getlist(data('#videoCategory .pcVideoListItem'))
elif tid == '/playlists':
data = self.getpq(f'{tid}?page={pg}')
vhtml = data('#playListSection li')
vdata = []
for i in vhtml.items():
vdata.append({
'vod_id': 'playlists_click_' + i('.thumbnail-info-wrapper .display-block a').attr('href'),
'vod_name': i('.thumbnail-info-wrapper .display-block a').attr('title'),
'vod_pic': self.proxy(i('.largeThumb').attr('src')),
'vod_tag': 'folder',
'vod_remarks': i('.playlist-videos .number').text(),
'style': {"type": "rect", "ratio": 1.33}
})
elif tid == '/channels':
data = self.getpq(f'{tid}?o=rk&page={pg}')
vhtml = data('#filterChannelsSection li .description')
vdata = []
for i in vhtml.items():
vdata.append({
'vod_id': 'director_click_' + i('.avatar a').attr('href'),
'vod_name': i('.avatar img').attr('alt'),
'vod_pic': self.proxy(i('.avatar img').attr('src')),
'vod_tag': 'folder',
'vod_remarks': i('.descriptionContainer ul li').eq(-1).text(),
'style': {"type": "rect", "ratio": 1.33}
})
elif tid == '/categories' and pg == '1':
result['pagecount'] = 1
data = self.getpq(f'{tid}')
vhtml = data('.categoriesListSection li .relativeWrapper')
vdata = []
for i in vhtml.items():
vdata.append({
'vod_id': i('a').attr('href') + '_this_video',
'vod_name': i('a').attr('alt'),
'vod_pic': self.proxy(i('a img').attr('src')),
'vod_tag': 'folder',
'style': {"type": "rect", "ratio": 1.33}
})
elif tid == '/pornstars':
data = self.getpq(f'{tid}?o=t&page={pg}')
vhtml = data('#popularPornstars .performerCard .wrap')
vdata = []
for i in vhtml.items():
vdata.append({
'vod_id': 'pornstars_click_' + i('a').attr('href'),
'vod_name': i('.performerCardName').text(),
'vod_pic': self.proxy(i('a img').attr('src')),
'vod_tag': 'folder',
'vod_year': i('.performerVideosViewsCount span').eq(0).text(),
'vod_remarks': i('.performerVideosViewsCount span').eq(-1).text(),
'style': {"type": "rect", "ratio": 1.33}
})
elif 'playlists_click' in tid:
tid = tid.split('click_')[-1]
if pg == '1':
hdata = self.getpq(tid)
self.token = hdata('#searchInput').attr('data-token')
vdata = self.getlist(hdata('#videoPlaylist .pcVideoListItem .phimage'))
else:
tid = tid.split('playlist/')[-1]
data = self.getpq(f'/playlist/viewChunked?id={tid}&token={self.token}&page={pg}')
vdata = self.getlist(data('.pcVideoListItem .phimage'))
elif 'director_click' in tid:
tid = tid.split('click_')[-1]
data = self.getpq(f'{tid}/videos?page={pg}')
vdata = self.getlist(data('#showAllChanelVideos .pcVideoListItem .phimage'))
elif 'pornstars_click' in tid:
tid = tid.split('click_')[-1]
data = self.getpq(f'{tid}/videos?page={pg}')
vdata = self.getlist(data('#mostRecentVideosSection .pcVideoListItem .phimage'))
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('.userInfo .usernameWrap a')
pdtitle = '[a=cr:' + json.dumps(
{'id': 'director_click_' + dtext.attr('href'), 'name': dtext.text()}) + '/]' + dtext.text() + '[/a]'
vod = {
'vod_name': vn,
'vod_director': pdtitle,
'vod_remarks': (data('.userInfo').text() + ' / ' + data('.ratingInfo').text()).replace('\n', ' / '),
'vod_play_from': '老僧酿酒',
'vod_play_url': ''
}
js_content = data("#player script").eq(0).text()
plist = [f"{vn}${self.e64(f'{1}@@@@{url}')}"]
try:
pattern = r'"mediaDefinitions":\s*(\[.*?\]),\s*"isVertical"'
match = re.search(pattern, js_content, re.DOTALL)
if match:
json_str = match.group(1)
udata = json.loads(json_str)
plist = [
f"{media['height']}${self.e64(f'{0}@@@@{url}')}"
for media in udata[:-1]
if (url := media.get('videoUrl'))
]
except Exception as e:
print(f"提取mediaDefinitions失败: {str(e)}")
vod['vod_play_url'] = '#'.join(plist)
return {'list': [vod]}
def searchContent(self, key, quick, pg="1"):
data = self.getpq(f'/video/search?search={key}&page={pg}')
return {'list': self.getlist(data('#videoSearchResult .pcVideoListItem .phimage'))}
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.get('url'))
if param.get('type') == 'm3u8':
return self.m3Proxy(url)
else:
return self.tsProxy(url)
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 gethost(self):
try:
response = requests.get('https://www.pornhub.com', headers=self.headers, proxies=self.proxies,
allow_redirects=False)
return response.headers['Location'][:-1]
except Exception as e:
print(f"获取主页失败: {str(e)}")
return "https://www.pornhub.com"
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():
vlist.append({
'vod_id': i('a').attr('href'),
'vod_name': i('a').attr('title'),
'vod_pic': self.proxy(i('img').attr('src')),
'vod_remarks': i('.bgShadeEffect').text() or i('.duration').text(),
'style': {'ratio': 1.33, 'type': 'rect'}
})
return vlist
def getpq(self, path):
try:
response = self.session.get(f'{self.host}{path}').text
return pq(response.encode('utf-8'))
except Exception as e:
print(f"请求失败: , {str(e)}")
return None
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
+124
View File
@@ -0,0 +1,124 @@
import sys
import re
import json
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
from base.spider import Spider
class Spider(Spider):
def getName(self):
return "TOPTV"
def init(self, extend=""):
super().init(extend)
self.site_url = "https://toptv15.cyou"
self.headers = {
"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",
"Referer": self.site_url
}
self.sess = requests.Session()
self.sess.mount("https://", HTTPAdapter(max_retries=Retry(total=3, backoff_factor=1)))
def fetch(self, url, timeout=10):
try:
res = self.sess.get(url, headers=self.headers, timeout=timeout, verify=False)
res.encoding = "utf-8"
return res
except:
return None
def homeContent(self, filter):
cate_list = [
{"type_name": "国产自拍", "type_id": "1"},
{"type_name": "国产传媒", "type_id": "2"},
{"type_name": "探花系列", "type_id": "3"},
{"type_name": "人妻熟女", "type_id": "4"},
{"type_name": "日本无码", "type_id": "5"},
{"type_name": "美乳巨乳", "type_id": "6"},
{"type_name": "强制侵犯", "type_id": "7"},
{"type_name": "制服诱惑", "type_id": "8"},
{"type_name": "绝色佳人", "type_id": "9"},
{"type_name": "家庭乱伦", "type_id": "10"},
{"type_name": "绝顶潮吹", "type_id": "11"},
{"type_name": "网红主播", "type_id": "12"}
]
return {"class": cate_list}
def categoryContent(self, tid, pg, filter, extend):
if not hasattr(self, 'site_url'): self.init()
pg = int(pg) if str(pg).isdigit() else 1
list_url = f"{self.site_url}/index.php/vod/type/id/{tid}/page/{pg}.html"
res = self.fetch(list_url)
video_list = []
if res:
pattern = r'href="(/index.php/vod/detail/id/(\d+).html)".*?data-original="(.*?)".*?vod-name.*?>(.*?)<'
matches = re.findall(pattern, res.text, re.S)
for href, v_id, pic, name in matches:
video_list.append({
"vod_id": v_id,
"vod_name": name.strip(),
"vod_pic": pic if pic.startswith("http") else self.site_url + pic,
"vod_remarks": ""
})
return {'list': video_list, 'page': pg, 'pagecount': 999, 'limit': 20, 'total': 9999}
def detailContent(self, ids):
if not hasattr(self, 'site_url'): self.init()
vod_id = ids[0]
res = self.fetch(f"{self.site_url}/index.php/vod/detail/id/{vod_id}.html")
if not res: return {}
html = res.text
name_match = re.search(r'vod-name.*?>(.*?)<', html) or re.search(r'title-box.*?>(.*?)<', html)
pic_match = re.search(r'detail-pic.*?src="(.*?)"', html) or re.search(r'data-original="(.*?)"', html)
play_matches = re.findall(r'href="(/index.php/vod/play/id/(\d+)/sid/(\d+)/nid/(\d+).html)">(.*?)<', html)
play_urls = []
for m in play_matches:
play_urls.append(f"{m[4]}${m[1]}-{m[2]}-{m[3]}")
if not play_urls:
play_urls.append(f"立即播放${vod_id}-1-1")
vod = {
"vod_id": vod_id,
"vod_name": name_match.group(1).strip() if name_match else "视频详情",
"vod_pic": pic_match.group(1) if pic_match else "",
"vod_play_from": "TOP-TV",
"vod_play_url": "#".join(play_urls)
}
return {"list": [vod]}
def playerContent(self, flag, id, vipFlags):
if not hasattr(self, 'site_url'): self.init()
parts = id.split('-')
if len(parts) == 3:
v_id, s_id, n_id = parts
play_url = f"{self.site_url}/index.php/vod/play/id/{v_id}/sid/{s_id}/nid/{n_id}.html"
else:
play_url = f"{self.site_url}/index.php/vod/play/id/{id}.html"
res = self.fetch(play_url)
if res:
data_json = re.search(r'var player_aaaa=(.*?)</script>', res.text)
if data_json:
try:
url = json.loads(data_json.group(1)).get("url", "")
return {"parse": 0, "url": url, "header": self.headers}
except:
pass
return {"parse": 1, "url": play_url}
def searchContent(self, key, quick, pg=1):
if not hasattr(self, 'site_url'): self.init()
res = self.fetch(f"{self.site_url}/index.php/vod/search/page/{pg}/wd/{key}.html")
video_list = []
if res:
pattern = r'href="(/index.php/vod/detail/id/(\d+).html)".*?data-original="(.*?)".*?vod-name.*?>(.*?)<'
matches = re.findall(pattern, res.text, re.S)
for href, v_id, pic, name in matches:
video_list.append({
"vod_id": v_id,
"vod_name": name.strip(),
"vod_pic": pic if pic.startswith("http") else self.site_url + pic
})
return {"list": video_list}
+303
View File
@@ -0,0 +1,303 @@
# coding: utf-8
import json
import sys
import re
import urllib.request
import urllib.parse
import ssl
sys.path.append('..')
from base.spider import Spider
VERSION = '2.0.0'
SITE_URL = 'https://newxvideos.pages.dev'
API_URL = 'https://newxvideos.pages.dev/api'
CATEGORIES = [
{"type_id": "Arab-159", "type_name": "阿拉伯"},
{"type_id": "Mature-38", "type_name": "成熟"},
{"type_id": "Cuckold-237", "type_name": "出轨背叛"},
{"type_id": "Femdom-235", "type_name": "调教"},
{"type_id": "Anal-12", "type_name": "肛交"},
{"type_id": "Brunette-25", "type_name": "褐发"},
{"type_id": "Black_Woman-30", "type_name": "黑人"},
{"type_id": "Redhead-31", "type_name": "红发"},
{"type_id": "Fucked_Up_Family-81", "type_name": "家庭乱搞"},
{"type_id": "Blonde-20", "type_name": "金发"},
{"type_id": "Big_Cock-34", "type_name": "巨屌"},
{"type_id": "Big_Tits-23", "type_name": "巨乳"},
{"type_id": "Big_Ass-24", "type_name": "巨臀"},
{"type_id": "Blowjob-15", "type_name": "口交"},
{"type_id": "Latina-16", "type_name": "拉丁裔"},
{"type_id": "Milf-19", "type_name": "辣妈"},
{"type_id": "Gapes-167", "type_name": "裂开"},
{"type_id": "Ass-14", "type_name": "美臀"},
{"type_id": "Lesbian-26", "type_name": "女同"},
{"type_id": "bbw-51", "type_name": "胖女"},
{"type_id": "Squirting-56", "type_name": "喷出"},
{"type_id": "Fisting-165", "type_name": "拳交"},
{"type_id": "Gangbang-69", "type_name": "群交"},
{"type_id": "Teen-13", "type_name": "少女"},
{"type_id": "Cumshot-18", "type_name": "射颜"},
{"type_id": "Cam_Porn-58", "type_name": "摄像头"},
{"type_id": "Bi_Sexual-62", "type_name": "双性恋"},
{"type_id": "Stockings-28", "type_name": "丝袜"},
{"type_id": "Oiled-22", "type_name": "涂油"},
{"type_id": "Lingerie-83", "type_name": "性感内衣"},
{"type_id": "Asian_Woman-32", "type_name": "亚洲"},
{"type_id": "Amateur-65", "type_name": "业余"},
{"type_id": "Interracial-27", "type_name": "异族"},
{"type_id": "Indian-89", "type_name": "印度"},
{"type_id": "Creampie-40", "type_name": "中出"},
{"type_id": "Solo_and_Masturbation-33", "type_name": "自慰"},
{"type_id": "AI-239", "type_name": "AI"},
{"type_id": "ASMR-229", "type_name": "ASMR"},
]
class Spider(Spider):
def getName(self):
return "V-HUB[成人]"
def init(self, extend):
if extend:
self.host = extend.get('host', SITE_URL)
else:
self.host = SITE_URL
self.api_url = self.host.rstrip('/') + '/api'
self.headers = {
'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',
'Referer': self.host + '/',
'Origin': self.host
}
self._ssl_context = ssl.create_default_context()
self._ssl_context.check_hostname = False
self._ssl_context.verify_mode = ssl.CERT_NONE
def _xhttp(self, params):
"""使用标准库urllib发起HTTP GET请求"""
try:
qs = urllib.parse.urlencode(params)
full_url = self.api_url + '?' + qs
req = urllib.request.Request(full_url, headers=self.headers, method='GET')
resp = urllib.request.urlopen(req, context=self._ssl_context, timeout=15)
data = json.loads(resp.read().decode('utf-8'))
if isinstance(data, list):
return data
elif isinstance(data, dict) and 'data' in data:
return data['data']
return []
except Exception as e:
print('_xhttp error: %s' % str(e), file=sys.stderr)
return []
def _format_time_cn(self, time_str):
"""将英文时间格式转为中文,如 '11 min' -> '11分钟'"""
if not time_str:
return ''
m = re.match(r'^(\d+)\s*min\s*$', time_str.strip(), re.IGNORECASE)
if m:
return m.group(1) + '分钟'
m = re.match(r'^(\d+)\s*h(?:our)?s?\s*(\d+)?\s*min\s*$', time_str.strip(), re.IGNORECASE)
if m:
h = m.group(1)
mi = m.group(2)
if mi:
return h + '小时' + mi + '分钟'
return h + '小时'
return time_str
def _extract_xvid(self, url):
"""从视频URL的查询参数中提取xvid值"""
if not url:
return ''
parsed = urllib.parse.urlparse(url)
qs = urllib.parse.parse_qs(parsed.query)
if 'xvid' in qs:
return qs['xvid'][0]
return ''
def _build_vod_list(self, raw_data):
"""将API返回的原始数据构造为vod列表"""
videos = []
for item in raw_data:
title = item.get('title', '')
clean_title = re.sub(r'^AVOTC资源网[—-]+\s*', '', title).strip()
if not clean_title:
clean_title = title
url = item.get('url', '')
vod_id = self._extract_xvid(url)
if not vod_id:
vod_id = str(item.get('videoid', ''))
videos.append({
'vod_id': vod_id,
'vod_name': clean_title,
'vod_pic': item.get('img', ''),
'vod_remarks': self._format_time_cn(item.get('time', '')),
'vod_url': url
})
return videos
def homeContent(self, filter):
"""首页:返回分类列表 + 首页视频"""
classes = []
for cat in CATEGORIES:
classes.append({'type_id': cat['type_id'], 'type_name': cat['type_name']})
raw_data = self._xhttp({'play': 'list', 'page': 1})
videos = self._build_vod_list(raw_data)
return {'class': classes, 'list': videos}
def homeVideoContent(self):
return {'list': []}
def categoryContent(self, tid, pg, filter, extend):
"""分类内容"""
raw_data = self._xhttp({'play': 'class', 'c': tid, 'page': pg})
videos = self._build_vod_list(raw_data)
type_name = tid
for cat in CATEGORIES:
if cat['type_id'] == tid:
type_name = cat['type_name']
break
return {
'page': int(pg),
'pagecount': 9999,
'limit': 90,
'total': 9999,
'type_name': type_name,
'list': videos
}
def detailContent(self, array):
"""详情:通过xvid获取视频播放地址"""
result = {}
if not array or not array[0]:
return result
xvid = array[0]
vod = {
'vod_id': xvid,
'vod_name': '视频详情',
'vod_pic': '',
'vod_remarks': '',
'vod_play_from': 'newxvideos',
'vod_play_url': ''
}
try:
qs = urllib.parse.urlencode({'xvid': xvid})
full_url = self.api_url + '?' + qs
req = urllib.request.Request(full_url, headers=self.headers, method='GET')
resp = urllib.request.urlopen(req, context=self._ssl_context, timeout=15)
data = json.loads(resp.read().decode('utf-8'))
except Exception as e:
print('detailContent error: %s' % str(e), file=sys.stderr)
result['list'] = [vod]
return result
play_urls = []
if isinstance(data, dict):
item = data
if 'data' in data and isinstance(data['data'], dict):
item = data['data']
hls_url = item.get('hls') or item.get('m3u8') or ''
hight_url = item.get('hight') or item.get('high') or item.get('hd') or ''
low_url = item.get('low') or item.get('sd') or ''
if hls_url:
play_urls.append('高清HLS$' + hls_url)
if hight_url:
play_urls.append('高清MP4$' + hight_url)
if low_url:
play_urls.append('低清MP4$' + low_url)
title = item.get('title', '')
if title:
clean_title = re.sub(r'^AVOTC资源网[—-]+\s*', '', title).strip()
if clean_title:
vod['vod_name'] = clean_title
img = item.get('img', '')
if img:
vod['vod_pic'] = img
time_str = item.get('time', '')
if time_str:
vod['vod_remarks'] = self._format_time_cn(time_str)
elif isinstance(data, list):
for item in data:
hls_url = item.get('hls') or item.get('m3u8') or ''
hight_url = item.get('hight') or item.get('high') or item.get('hd') or ''
low_url = item.get('low') or item.get('sd') or ''
if hls_url:
play_urls.append('高清HLS$' + hls_url)
if hight_url:
play_urls.append('高清MP4$' + hight_url)
if low_url:
play_urls.append('低清MP4$' + low_url)
if vod['vod_name'] == '视频详情':
title = item.get('title', '')
if title:
clean_title = re.sub(r'^AVOTC资源网[—-]+\s*', '', title).strip()
if clean_title:
vod['vod_name'] = clean_title
img = item.get('img', '')
if img:
vod['vod_pic'] = img
time_str = item.get('time', '')
if time_str:
vod['vod_remarks'] = self._format_time_cn(time_str)
if play_urls:
vod['vod_play_url'] = '#'.join(play_urls)
result['list'] = [vod]
return result
def searchContent(self, key, quick, pg='1'):
"""搜索"""
raw_data = self._xhttp({'play': 'k', 'k': key, 'page': pg})
videos = self._build_vod_list(raw_data)
return {
'page': int(pg),
'pagecount': 9999,
'limit': 90,
'total': 9999,
'list': videos
}
def playerContent(self, flag, id, vipFlags):
"""播放地址解析 - 直接返回用户选择的清晰度地址"""
if id and (id.startswith('http://') or id.startswith('https://')):
return {
'parse': 0,
'playUrl': '',
'url': id,
'header': {
'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',
'Referer': self.host + '/'
}
}
return {'parse': 0, 'playUrl': '', 'url': '', 'header': {}}
def isVideoFormat(self, url):
return False
def manualVideoCheck(self):
return False
def localProxy(self, param):
return {}
+1 -1
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
# by @嗷呜
# by @6666
import json
import re
import sys
+3 -3
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
# by @嗷呜
import json
import sys
from base64 import b64decode, b64encode
@@ -172,11 +172,11 @@ class Spider(Spider):
def gethost(self):
try:
response = self.fetch('https://xhamster.com', headers=self.headers, allow_redirects=False)
response = self.fetch('https://zh.xhamster1.desi/', headers=self.headers, allow_redirects=False)
return response.headers['Location']
except Exception as e:
print(f"获取主页失败: {str(e)}")
return "https://zn.xhamster.com"
return "https://zh.xhamster1.desi/"
def e64(self, text):
try:
+360
View File
@@ -0,0 +1,360 @@
# coding=utf-8
import sys
import os
import re
import json
import urllib.parse
from base.spider import Spider
from bs4 import BeautifulSoup
class Spider(Spider):
def getName(self):
return "\u4e45\u4e45\u7f51"
def init(self, extend=""):
self.host = "https://ww.jiujiu.one"
print(f"Initialized with host: {self.host}")
def header(self):
return {
'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',
'Referer': self.host,
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
}
def homeContent(self, filter):
"""\u8fd4\u56de\u5206\u7c7b\u5217\u8868"""
result = {}
classes = [
{"type_name": "亞洲無碼", "type_id": "68"},
{"type_name": "日本女優", "type_id": "67"},
{"type_name": "日本無碼", "type_id": "23"},
{"type_name": "中文字幕", "type_id": "9"},
{"type_name": "日本有碼", "type_id": "24"},
{"type_name": "日韓無碼", "type_id": "82"},
{"type_name": "無碼專區", "type_id": "113"},
{"type_name": "AV明星", "type_id": "78"},
{"type_name": "倫理影片", "type_id": "269"},
{"type_name": "日本片商", "type_id": "90"},
{"type_name": "國產自拍", "type_id": "80"},
{"type_name": "傳媒原創", "type_id": "231"},
{"type_name": "國產精品", "type_id": "63"},
{"type_name": "國產情色", "type_id": "77"},
{"type_name": "美女主播", "type_id": "105"},
{"type_name": "強姦亂倫", "type_id": "33"},
{"type_name": "國產主播", "type_id": "36"},
{"type_name": "亞洲有碼", "type_id": "66"},
{"type_name": "偷拍自拍", "type_id": "3"},
{"type_name": "抖陰視頻", "type_id": "91"},
{"type_name": "制服誘惑", "type_id": "31"},
{"type_name": "黑料不打烊", "type_id": "10"},
{"type_name": "歐美精品", "type_id": "25"},
]
result["class"] = classes
result["list"] = []
return result
def homeVideoContent(self):
"""\u9996\u9875\u63a8\u8350\u89c6\u9891"""
try:
print("Fetching home page...")
rsp = self.fetch(self.host, headers=self.header())
print(f"Response status: {rsp.status}")
# \u4f7f\u7528\u6b63\u5219\u8865\u6551\u65b9\u6848\uff0c\u76f4\u63a5\u4eceHTML\u4e2d\u63d0\u53d6\u89c6\u9891\u4fe1\u606f
html = rsp.text
# \u5339\u914d\u89c6\u9891\u9879\u6a21\u5f0f
# \u67e5\u627e\u6240\u6709 div.item
videos = []
# \u4f7f\u7528BeautifulSoup
soup = BeautifulSoup(html, 'html.parser')
# \u76f4\u63a5\u67e5\u627e\u6240\u6709\u5e26\u6709\u89c6\u9891\u7684\u5361\u7247
items = soup.find_all('div', class_=lambda c: c and 'item' in c.split())
print(f"Found {len(items)} items with class containing 'item'")
for item in items:
try:
# \u627e\u5230\u94fe\u63a5
links = item.find_all('a', href=True)
if len(links) < 2:
continue
# \u6807\u9898\u94fe\u63a5\u901a\u5e38\u662f\u7b2c\u4e8c\u4e2a
title_link = links[1] if len(links) > 1 else links[0]
href = title_link.get('href', '')
if not href:
continue
# \u6784\u5efa\u5b8c\u6574\u7684URL
if href.startswith('/'):
vod_id = self.host + href
else:
vod_id = href
# \u6807\u9898
vod_name = title_link.get_text(strip=True)
# \u5c01\u9762\u56fe
vod_pic = ''
img = item.find('img')
if img:
vod_pic = img.get('src') or img.get('data-src') or ''
if vod_pic and not vod_pic.startswith('http'):
if vod_pic.startswith('/'):
vod_pic = self.host + vod_pic
else:
vod_pic = 'https:' + vod_pic if vod_pic.startswith('//') else vod_pic
# \u5907\u6ce8
vod_remarks = ''
badge = item.find('span', class_='badge')
if badge:
vod_remarks = badge.get_text(strip=True)
if vod_name and vod_id:
videos.append({
'vod_id': vod_id,
'vod_name': vod_name,
'vod_pic': vod_pic,
'vod_remarks': vod_remarks
})
print(f"Added video: {vod_name}")
except Exception as e:
print(f"Error processing item: {e}")
continue
print(f"Total videos extracted: {len(videos)}")
return {'list': videos}
except Exception as e:
print(f"Error in homeVideoContent: {e}")
return {'list': []}
def categoryContent(self, tid, pg, filter, extend):
"""\u5206\u7c7b\u9875\u5185\u5bb9"""
try:
# \u6784\u5efa\u5206\u7c7b\u9875URL\uff0c\u652f\u6301\u5206\u9875
if pg == "1":
url = f"{self.host}/c/{tid}"
else:
url = f"{self.host}/c/{tid}?page={pg}"
print(f"Fetching category URL: {url}")
rsp = self.fetch(url, headers=self.header())
soup = BeautifulSoup(rsp.text, 'html.parser')
videos = []
items = soup.find_all('div', class_=lambda c: c and 'item' in c.split())
for item in items:
try:
links = item.find_all('a', href=True)
if len(links) < 2:
continue
title_link = links[1] if len(links) > 1 else links[0]
href = title_link.get('href', '')
if not href:
continue
if href.startswith('/'):
vod_id = self.host + href
else:
vod_id = href
vod_name = title_link.get_text(strip=True)
vod_pic = ''
img = item.find('img')
if img:
vod_pic = img.get('src') or img.get('data-src') or ''
if vod_pic and not vod_pic.startswith('http'):
if vod_pic.startswith('/'):
vod_pic = self.host + vod_pic
else:
vod_pic = 'https:' + vod_pic if vod_pic.startswith('//') else vod_pic
vod_remarks = ''
badge = item.find('span', class_='badge')
if badge:
vod_remarks = badge.get_text(strip=True)
if vod_name and vod_id:
videos.append({
'vod_id': vod_id,
'vod_name': vod_name,
'vod_pic': vod_pic,
'vod_remarks': vod_remarks
})
except Exception as e:
print(f"Error processing item: {e}")
continue
# \u83b7\u53d6\u603b\u9875\u6570
total_pages = 1
pagination = soup.find('ul', class_='pagination')
if pagination:
page_links = pagination.find_all('a')
for link in page_links:
text = link.get_text(strip=True)
if text.isdigit():
page_num = int(text)
if page_num > total_pages:
total_pages = page_num
return {
'list': videos,
'page': int(pg),
'pagecount': total_pages,
'limit': len(videos),
'total': total_pages * len(videos) if total_pages > 0 else len(videos)
}
except Exception as e:
print(f"Error in categoryContent: {e}")
return {'list': [], 'page': int(pg), 'pagecount': 1, 'limit': 0, 'total': 0}
def detailContent(self, ids):
"""\u8be6\u60c5\u9875\u5185\u5bb9"""
try:
vod_id = ids[0]
if not vod_id.startswith('http'):
vod_id = self.host + vod_id
print(f"Fetching detail page: {vod_id}")
rsp = self.fetch(vod_id, headers=self.header())
soup = BeautifulSoup(rsp.text, 'html.parser')
vod = {
'vod_id': vod_id,
'vod_name': '',
'vod_pic': '',
'vod_actor': '',
'vod_director': '',
'vod_content': '',
'vod_play_from': '\u4e45\u4e45\u7f51',
'vod_play_url': ''
}
# \u6807\u9898
title = soup.find('h1')
if title:
vod['vod_name'] = title.get_text(strip=True)
# \u5c01\u9762
img = soup.find('meta', property='og:image')
if img and img.get('content'):
vod['vod_pic'] = img.get('content')
else:
img = soup.find('img', class_='card-img-top')
if img:
vod['vod_pic'] = img.get('src') or ''
# \u64ad\u653e\u5730\u5740 - \u67e5\u627evideo\u6807\u7b64\u6216iframe
play_url = ''
video = soup.find('video')
if video:
source = video.find('source')
if source and source.get('src'):
play_url = source.get('src')
if not play_url:
iframe = soup.find('iframe')
if iframe and iframe.get('src'):
play_url = iframe.get('src')
if play_url:
vod['vod_play_url'] = f'\u6b63\u7247${play_url}'
return {'list': [vod]}
except Exception as e:
print(f"Error in detailContent: {e}")
return {'list': []}
def searchContent(self, key, quick, pg=1):
"""\u641c\u7d22\u529f\u80fd"""
try:
search_url = f"{self.host}/node/search?q={urllib.parse.quote(keyword)}"
print(f"Search URL: {search_url}")
rsp = self.fetch(search_url, headers=self.header())
soup = BeautifulSoup(rsp.text, 'html.parser')
videos = []
items = soup.find_all('div', class_=lambda c: c and 'item' in c.split())
for item in items:
try:
links = item.find_all('a', href=True)
if len(links) < 2:
continue
title_link = links[1] if len(links) > 1 else links[0]
href = title_link.get('href', '')
if not href:
continue
if href.startswith('/'):
vod_id = self.host + href
else:
vod_id = href
vod_name = title_link.get_text(strip=True)
vod_pic = ''
img = item.find('img')
if img:
vod_pic = img.get('src') or img.get('data-src') or ''
if vod_pic and not vod_pic.startswith('http'):
if vod_pic.startswith('/'):
vod_pic = self.host + vod_pic
else:
vod_pic = 'https:' + vod_pic if vod_pic.startswith('//') else vod_pic
vod_remarks = ''
badge = item.find('span', class_='badge')
if badge:
vod_remarks = badge.get_text(strip=True)
if vod_name and vod_id:
videos.append({
'vod_id': vod_id,
'vod_name': vod_name,
'vod_pic': vod_pic,
'vod_remarks': vod_remarks
})
except Exception as e:
print(f"Error processing search item: {e}")
continue
return {'list': videos}
except Exception as e:
print(f"Error in searchContent: {e}")
return {'list': []}
def playerContent(self, flag, id, vipFlags):
"""\u8fd4\u56de\u64ad\u653e\u5730\u5740"""
return {
'parse': 0,
'playUrl': '',
'url': id
}
def isVideoFormat(self, url):
"""\u5224\u65ad\u662f\u5426\u4e3a\u89c6\u9891\u683c\u5f0f"""
video_extensions = ['.mp4', '.m3u8', '.flv', '.avi', '.mkv', '.wmv', '.mov']
lower_url = url.lower()
for ext in video_extensions:
if ext in lower_url:
return True
return False
def localProxy(self, param):
return None
def destroy(self):
pass
+276
View File
@@ -0,0 +1,276 @@
# -*- coding: utf-8 -*-
# 爬虫源: 怦然心动 (prshinezenx.blog)
# 站点类型: SPA + 服务端渲染,数据通过 Base64 编码嵌入 HTML
# 开发者: AI Assistant
# 日期: 2026-07-22
import re
import json
import base64
from urllib.parse import urljoin, quote
from base.spider import Spider as BaseSpider
class Spider(BaseSpider):
def __init__(self):
self.host = "https://prshinezenx.blog"
self.headers = {
"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",
"Referer": self.host + "/",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9"
}
# 分类列表(从首页提取,本地硬编码保证首页秒出)
self.classes = [
{"type_id": "43", "type_name": "国产精选"},
{"type_id": "31", "type_name": "束缚玩法"},
{"type_id": "35", "type_name": "中字专区"},
{"type_id": "33", "type_name": "女优精选"},
{"type_id": "53", "type_name": "传媒拍摄"},
{"type_id": "29", "type_name": "变性纪实"},
{"type_id": "21", "type_name": "同志日常"},
{"type_id": "23", "type_name": "百合情境"},
{"type_id": "39", "type_name": "欧美精选"},
{"type_id": "45", "type_name": "虚拟换脸"},
{"type_id": "47", "type_name": "少女幻想"},
{"type_id": "49", "type_name": "主播日记"},
{"type_id": "51", "type_name": "约会实录"},
{"type_id": "55", "type_name": "伦理剧场"},
{"type_id": "57", "type_name": "黑料档案"},
{"type_id": "63", "type_name": "自拍实录"},
]
# 无筛选功能
self.filters = {}
def getName(self):
return "怦然心动"
def getDependence(self):
return []
def init(self, extend=""):
"""初始化,零网络"""
pass
def _fetch(self, url):
"""请求页面,返回 HTML 文本"""
try:
rsp = self.fetch(url, headers=self.headers, timeout=15000)
if rsp and hasattr(rsp, 'text'):
return rsp.text
return None
except Exception:
return None
def _extract_vod_data(self, html):
"""从 HTML 中提取 window.__vod_data__ 的 Base64 数据并解码"""
if not html:
return None
pattern = r"const binaryStr = atob\('([^']+)'\)"
match = re.search(pattern, html)
if not match:
return None
base64_str = match.group(1)
try:
json_str = base64.b64decode(base64_str).decode('utf-8')
return json.loads(json_str)
except Exception:
return None
def _parse_list(self, items):
"""解析视频列表项,打包数据到 vod_id 以便详情页快速展示"""
if not items:
return []
result = []
for item in items:
vod_id = str(item.get("vod_id", ""))
if not vod_id:
continue
vod_name = item.get("vod_name", "未知标题")
vod_pic = item.get("vod_pic", "")
if vod_pic and not vod_pic.startswith("http"):
vod_pic = urljoin(self.host, vod_pic)
vod_remark = item.get("vod_duration", "")
type_id = str(item.get("type_id", ""))
# 打包数据到 vod_id,方便详情页快速返回
packed_id = f"{vod_id}|$|{vod_name}|$|{vod_pic}|$|{vod_remark}|$|{type_id}"
result.append({
"vod_id": packed_id,
"vod_name": vod_name,
"vod_pic": vod_pic,
"vod_remarks": vod_remark,
})
return result
def homeContent(self, filter=False):
"""首页:返回分类列表,零网络"""
return {
"class": self.classes,
"filters": self.filters if filter else {}
}
def getHomeContent(self, filter=False):
return self.homeContent(filter)
def homeVideoContent(self):
"""首页推荐视频"""
html = self._fetch(self.host + "/")
if not html:
return {"list": []}
data = self._extract_vod_data(html)
if not data:
return {"list": []}
items = data.get("other_request_data", {}).get("random_list", [])
if not items:
items = data.get("request_data", {}).get("list", [])
return {"list": self._parse_list(items[:20])}
def categoryContent(self, tid, pg=1, filter=False, extend=None):
"""分类列表页"""
page = pg or 1
url = f"{self.host}/vodlist/type/{tid}/keyword/all/orderby/default/page/{page}.html"
html = self._fetch(url)
if not html:
return {"list": [], "page": page, "pagecount": 1, "limit": 20, "total": 0}
data = self._extract_vod_data(html)
if not data:
return {"list": [], "page": page, "pagecount": 1, "limit": 20, "total": 0}
items = data.get("request_data", {}).get("list", [])
total = data.get("request_data", {}).get("total", 0)
limit = data.get("limit", 20)
total_pages = (total + limit - 1) // limit if total > 0 else 1
return {
"list": self._parse_list(items),
"page": page,
"pagecount": total_pages,
"limit": limit,
"total": total
}
def detailContent(self, ids):
"""视频详情 - 从打包的 vod_id 中解析数据,快速返回"""
if not ids:
return {"list": []}
raw = str(ids[0])
# 解析打包的数据: vod_id|$|vod_name|$|vod_pic|$|vod_remark|$|type_id
parts = raw.split("|$|")
if len(parts) >= 5:
vod_id = parts[0]
vod_name = parts[1] if len(parts) > 1 else "未知标题"
vod_pic = parts[2] if len(parts) > 2 else ""
vod_remark = parts[3] if len(parts) > 3 else ""
type_id = parts[4] if len(parts) > 4 else ""
else:
# 兼容旧格式:直接传数字ID
vod_id = raw
vod_name = ""
vod_pic = ""
vod_remark = ""
type_id = ""
# 获取播放地址:请求详情页提取 vod_play_url
if type_id:
detail_url = f"{self.host}/voddetail/type/{type_id}/id/{vod_id}.html"
else:
detail_url = f"{self.host}/voddetail/type/all/id/{vod_id}.html"
html = self._fetch(detail_url)
play_page = ""
if html:
data = self._extract_vod_data(html)
if data:
vod_info = data.get("vod_info", {})
play_page = vod_info.get("vod_play_url", "")
if not vod_name:
vod_name = vod_info.get("vod_name", "未知标题")
if not vod_pic:
vod_pic = vod_info.get("vod_pic", "")
if not vod_remark:
vod_remark = vod_info.get("vod_duration", "")
if not play_page:
return {
"list": [{
"vod_id": vod_id,
"vod_name": vod_name or "未知标题",
"vod_pic": vod_pic,
"vod_remarks": vod_remark,
"vod_content": "",
"vod_play_from": "播放",
"vod_play_url": ""
}]
}
# 如果是 ao jie xi 包装,提取真实 m3u8
if "aojiexi.com" in play_page:
match = re.search(r'url=([^&]+)', play_page)
if match:
real_url = match.group(1)
real_url = re.sub(r'%([0-9A-Fa-f]{2})', lambda m: chr(int(m.group(1), 16)), real_url)
play_page = real_url
# 构造播放数据:单线路单集
# 格式参考 tmcrownxlift 成功案例
return {
"list": [{
"vod_id": vod_id,
"vod_name": vod_name or "未知标题",
"vod_pic": vod_pic,
"vod_remarks": vod_remark,
"vod_content": "",
"vod_play_from": "播放",
"vod_play_url": "播放$" + play_page
}]
}
def searchContent(self, key, quick=False, pg="1"):
"""搜索"""
if not key:
return {"list": []}
page = pg or 1
url = f"{self.host}/vodlist/type/all/keyword/{quote(key)}/orderby/default/page/{page}.html"
html = self._fetch(url)
if not html:
return {"list": []}
data = self._extract_vod_data(html)
if not data:
return {"list": []}
items = data.get("request_data", {}).get("list", [])
return {"list": self._parse_list(items)}
def playerContent(self, flag, vid, vipFlags=None):
"""播放地址解析"""
if not vid:
return {"parse": 0, "url": ""}
if vid.endswith((".m3u8", ".mp4")):
return {"parse": 0, "url": vid, "header": self.headers}
if "aojiexi.com" in vid:
match = re.search(r'url=([^&]+)', vid)
if match:
real_url = match.group(1)
real_url = re.sub(r'%([0-9A-Fa-f]{2})', lambda m: chr(int(m.group(1), 16)), real_url)
return {"parse": 0, "url": real_url, "header": self.headers}
if vid.startswith("http"):
html = self._fetch(vid)
if html:
match = re.search(r'https?://[^\s"\']+\.m3u8[^\s"\']*', html)
if match:
return {"parse": 0, "url": match.group(0), "header": self.headers}
return {"parse": 1, "url": vid}
def isVideoFormat(self, url):
if not url:
return False
return url.endswith((".m3u8", ".mp4", ".m3u8?"))
def destroy(self):
pass
+461
View File
@@ -0,0 +1,461 @@
# coding=utf-8
import sys
import json
import re
import requests
import base64
from bs4 import BeautifulSoup
from urllib.parse import unquote, urljoin
try:
from base.spider import Spider as BaseSpider
except ImportError:
class BaseSpider():
def fetch(self, url, headers=None, timeout=10):
try:
res = requests.get(url, headers=headers, timeout=timeout, allow_redirects=True)
res.encoding = 'utf-8'
return res
except Exception as e:
print(f"fetch error: {e}")
return None
class Spider(BaseSpider):
def getName(self):
return "撸一天"
def init(self, extend=""):
self.host = "https://luyitian.com"
self.session = requests.Session()
self.session.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',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Connection': 'keep-alive'
})
def homeVideoContent(self):
return {"list": []}
def localProxy(self, params):
return [200, "video/MP2T", ""]
def isVideoFormat(self, url):
return False
def manualVideoCheck(self):
return False
def fetch(self, url, headers=None, timeout=5):
try:
req_headers = headers or self.session.headers
res = self.session.get(url, headers=req_headers, timeout=timeout, allow_redirects=True)
res.encoding = 'utf-8'
return res
except Exception as e:
print(f"fetch error: {e}")
return None
def _get_topic_filters(self):
"""
从 /topic/ 页面提取所有专题入口,生成子分类列表。
适配真实 URL/topicdetail-7/
"""
url = f"{self.host}/topic/"
res = self.fetch(url, timeout=5)
if not res:
return []
soup = BeautifulSoup(res.text, 'html.parser')
# 专门匹配 /topicdetail- 链接
topic_links = soup.select('a[href*="/topicdetail-"]')
if not topic_links:
# 兜底:可能写成 /topicdetail/ 格式
topic_links = soup.select('a[href*="/topicdetail"]')
filters = []
seen = set()
for a in topic_links:
href = a.get('href', '')
# 提取数字 ID(例如 /topicdetail-7/
match = re.search(r'/topicdetail-(\d+)', href)
if not match:
match = re.search(r'/topicdetail/(\d+)', href)
if not match:
continue
tid = match.group(1)
name = a.get_text(strip=True) or a.get('title', '') or f"专题{tid}"
if len(name) < 2:
continue
if tid not in seen:
seen.add(tid)
filters.append({"n": name, "v": tid})
return filters
def homeContent(self, filter):
classes = [
{"type_name": "最近更新", "type_id": "new"},
{"type_name": "热门影片", "type_id": "hot"},
{"type_name": "影片专题", "type_id": "topic"},
{"type_name": "中文字幕", "type_id": "28"},
{"type_name": "国产", "type_id": "20"},
{"type_name": "日本有码", "type_id": "21"},
{"type_name": "日本无码", "type_id": "22"},
{"type_name": "欧美", "type_id": "23"},
{"type_name": "动漫", "type_id": "24"},
{"type_name": "伦理", "type_id": "25"},
{"type_name": "韩国", "type_id": "36"},
{"type_name": "另类", "type_id": "41"}
]
filters = {
"28": [{"key": "tid", "name": "子分类", "value": [
{"n": "全部", "v": "28"},
{"n": "日本中字", "v": "51"}
]}],
"20": [{"key": "tid", "name": "子分类", "value": [
{"n": "全部", "v": "20"},
{"n": "国产精品", "v": "26"},
{"n": "国产剧情", "v": "27"},
{"n": "国产自拍", "v": "29"},
{"n": "国产主播", "v": "35"},
{"n": "国模私拍", "v": "85"},
{"n": "网红明星", "v": "91"},
{"n": "国产SM", "v": "105"},
{"n": "台湾辣妹", "v": "107"},
{"n": "香港正妹", "v": "108"}
]}],
"21": [{"key": "tid", "name": "子分类", "value": [
{"n": "全部", "v": "21"},
{"n": "人妻", "v": "31"},
{"n": "素人", "v": "44"},
{"n": "口爆颜射", "v": "46"},
{"n": "萝莉少女", "v": "47"},
{"n": "美乳巨乳", "v": "48"},
{"n": "制服诱惑", "v": "52"},
{"n": "调教", "v": "57"},
{"n": "出轨", "v": "58"},
{"n": "有码精品", "v": "101"}
]}],
"22": [{"key": "tid", "name": "子分类", "value": [
{"n": "全部", "v": "22"},
{"n": "无码精品", "v": "102"}
]}],
"23": [{"key": "tid", "name": "子分类", "value": [
{"n": "全部", "v": "23"},
{"n": "欧美精品", "v": "104"}
]}],
"24": [{"key": "tid", "name": "子分类", "value": [
{"n": "全部", "v": "24"},
{"n": "动漫精品", "v": "103"}
]}],
"25": [{"key": "tid", "name": "子分类", "value": [
{"n": "全部", "v": "25"},
{"n": "综合三级", "v": "39"}
]}],
"36": [{"key": "tid", "name": "子分类", "value": [
{"n": "全部", "v": "36"},
{"n": "韩国主播", "v": "37"}
]}],
"41": [{"key": "tid", "name": "子分类", "value": [
{"n": "全部", "v": "41"},
{"n": "Cosplay", "v": "106"}
]}]
}
# 动态注入专题子分类(例如:2018必看、2019必看...
topic_values = self._get_topic_filters()
if topic_values:
filters["topic"] = [{
"key": "tid",
"name": "专题",
"value": topic_values
}]
return {'class': classes, 'filters': filters}
def categoryContent(self, tid, pg, filter, extend):
pg = int(pg)
result = {"list": [], "page": pg, "pagecount": 999, "limit": 20, "total": 9999}
real_tid = extend.get('tid', tid)
soup = None
# ---------- 影片专题下的具体专题(/topicdetail-7/ ----------
if real_tid.isdigit() and tid == "topic":
urls_to_try = [
f"{self.host}/topicdetail-{real_tid}/",
f"{self.host}/topicdetail-{real_tid}.html",
f"{self.host}/topicdetail/{real_tid}/",
f"{self.host}/topicdetail/{real_tid}.html",
f"{self.host}/topicdetail-{real_tid}-{pg}/",
f"{self.host}/topicdetail/{real_tid}-{pg}/"
]
for url in urls_to_try:
res = self.fetch(url, headers={'Referer': self.host})
if res and res.status_code == 200 and ('video-img-box' in res.text or 'vodlist' in res.text or 'vodplay' in res.text):
soup = BeautifulSoup(res.text, 'html.parser')
break
if not soup:
return result
# ---------- new / hot / topic 总入口 ----------
elif real_tid in ["new", "hot", "topic"]:
if real_tid == "topic":
return result
if pg > 1:
url = f"{self.host}/label/{real_tid}/page/{pg}/"
else:
url = f"{self.host}/label/{real_tid}/"
res = self.fetch(url, headers={'Referer': self.host})
if not res:
url = f"{self.host}/label/{real_tid}/"
res = self.fetch(url, headers={'Referer': self.host})
if not res:
return result
soup = BeautifulSoup(res.text, 'html.parser')
# ---------- 普通分类 ----------
else:
urls_to_try = [
f"{self.host}/vodtype/{real_tid}-{pg}.html",
f"{self.host}/vodtype/{real_tid}-{pg}/",
f"{self.host}/type/{real_tid}-{pg}.html",
f"{self.host}/type/{real_tid}-{pg}/",
f"{self.host}/vodtype/{real_tid}/",
f"{self.host}/vodtype/{real_tid}.html"
]
res = None
for url in urls_to_try:
res = self.fetch(url, headers={'Referer': self.host})
if res and res.status_code == 200:
if 'video-img-box' in res.text or 'vodlist' in res.text or 'item' in res.text:
break
res = None
if not res:
return result
soup = BeautifulSoup(res.text, 'html.parser')
# ---------- 统一解析视频列表 ----------
vod_list = []
items = soup.select('.video-img-box') or soup.select('.video-film-list .video-item') or soup.select('.vodlist_item') or soup.select('.item')
for item in items:
a = item.select_one('a')
if not a:
continue
href = a.get('href', '')
vid_match = re.search(r'/vodplay/(\d+)', href) or \
re.search(r'/voddetail/(\d+)', href) or \
re.search(r'/vod/(\d+)', href) or \
re.search(r'/play/(\d+)', href)
vid = vid_match.group(1) if vid_match else href
name = ""
img = item.select_one('img')
if img and img.get('alt'):
name = img['alt']
if not name and a.get('title'):
name = a['title']
if not name:
title_elem = item.select_one('.title a') or item.select_one('.detail .title a')
if title_elem:
name = title_elem.get_text(strip=True)
if not name:
name = a.get_text(strip=True)
if not name:
name = "未知标题"
pic = ""
if img:
pic = img.get('data-src') or img.get('src', '')
if pic and not pic.startswith('http'):
pic = urljoin(self.host, pic)
remark = ""
remark_elem = item.select_one('.sub-title') or item.select_one('.remarks') or item.select_one('.video-remarks')
if remark_elem:
remark = remark_elem.get_text(strip=True)
if len(remark) > 20:
remark = remark[:20]
else:
text = item.get_text(strip=True)
parts = [p.strip() for p in text.split('\n') if p.strip()]
if parts:
remark = parts[-1][:20]
vod_list.append({
"vod_id": vid,
"vod_name": name.strip(),
"vod_pic": pic,
"vod_remarks": remark
})
result['list'] = vod_list
# ---------- 分页信息 ----------
page_elem = soup.select_one('.pagination a:last-child') or soup.select_one('.page a:last-child')
if page_elem and page_elem.get('href'):
try:
nums = re.findall(r'(\d+)', page_elem['href'])
if nums:
result['pagecount'] = max(int(nums[-1]), 1)
except:
pass
return result
def detailContent(self, ids):
vid = ids[0]
url = f"{self.host}/vodplay/{vid}-1-1/"
res = self.fetch(url, headers={'Referer': self.host})
if not res:
return {"list": []}
soup = BeautifulSoup(res.text, 'html.parser')
raw_title = soup.title.text.split('|')[0].replace('在线播放在线观看','').replace('','').replace('','').strip()
vod = {
"vod_id": vid,
"vod_name": raw_title,
"vod_type": "视频",
"vod_content": "资源来自于网络",
"vod_play_from": "Luyitian",
"vod_play_url": f"播放${vid}-1-1"
}
return {"list": [vod]}
def searchContent(self, key, quick, pg=1):
url = f"{self.host}/vodsearch/{key}----------{pg}---/"
res = self.fetch(url, headers={'Referer': self.host})
if not res:
return {"list": []}
soup = BeautifulSoup(res.text, 'html.parser')
vod_list = []
items = soup.select('.video-img-box') or soup.select('.video-film-list .video-item')
for item in items:
a = item.select_one('a')
if not a:
continue
href = a.get('href', '')
vid_match = re.search(r'/vodplay/(\d+)', href) or re.search(r'/voddetail/(\d+)', href)
vid = vid_match.group(1) if vid_match else href
name = ""
img = item.select_one('img')
if img and img.get('alt'):
name = img['alt']
if not name and a.get('title'):
name = a['title']
if not name:
title_elem = item.select_one('.title a')
if title_elem:
name = title_elem.get_text(strip=True)
if not name:
name = a.get_text(strip=True)
if not name:
name = "搜索结果"
pic = ""
if img:
pic = img.get('data-src') or img.get('src', '')
vod_list.append({
"vod_id": vid,
"vod_name": name.strip(),
"vod_pic": pic,
"vod_remarks": ""
})
return {"list": vod_list}
def _js_decode(self, js_str):
b64_match = re.search(r'atob\s*\(\s*["\']([^"\']+)["\']\s*\)', js_str)
if b64_match:
try:
decoded = base64.b64decode(b64_match.group(1)).decode('utf-8')
return decoded
except:
pass
unescape_match = re.search(r'unescape\s*\(\s*["\']([^"\']+)["\']\s*\)', js_str)
if unescape_match:
try:
decoded = unquote(unescape_match.group(1))
return decoded
except:
pass
url_match = re.search(r'(https?://[^\s"\']+\.m3u8[^\s"\']*)', js_str, re.I)
if url_match:
return url_match.group(1)
return None
def _sniff_xhr(self, html, page_url):
patterns = [
r'fetch\s*\(\s*["\']([^"\']+\.m3u8[^"\']*)["\']',
r'XMLHttpRequest.*?\.open\s*\(\s*["\']GET["\']\s*,\s*["\']([^"\']+\.m3u8[^"\']*)["\']',
r'\.get\s*\(\s*["\']([^"\']+\.m3u8[^"\']*)["\']',
r'url\s*:\s*["\']([^"\']+\.m3u8[^"\']*)["\']',
r'src\s*=\s*["\']([^"\']+\.m3u8[^"\']*)["\']',
]
for pat in patterns:
match = re.search(pat, html, re.I)
if match:
url = match.group(1)
if not url.startswith('http'):
url = urljoin(page_url, url)
return url
scripts = re.findall(r'<script[^>]*>(.*?)</script>', html, re.I | re.S)
for script_content in scripts:
if script_content.strip():
found = self._js_decode(script_content)
if found and '.m3u8' in found:
return found
return None
def playerContent(self, flag, id, vipFlags=None):
play_url = f"{self.host}/vodplay/{id}/"
res = self.fetch(play_url, headers={'Referer': self.host}, timeout=5)
if not res:
return {"parse": 1, "url": play_url}
html = res.text
m3u8_url = None
match = re.search(r'var\s+player_aaaa\s*=\s*(\{.*?\});', html, re.DOTALL)
if match:
try:
json_str = match.group(1).strip()
if json_str.endswith(','):
json_str = json_str[:-1]
config = json.loads(json_str)
m3u8_url = config.get('url', '')
except:
pass
if not m3u8_url:
m3u8_url = self._js_decode(html)
if not m3u8_url:
m3u8_url = self._sniff_xhr(html, play_url)
if not m3u8_url:
return {"parse": 1, "url": play_url}
m3u8_url = unquote(m3u8_url)
if m3u8_url.startswith('//'):
m3u8_url = 'https:' + m3u8_url
elif not m3u8_url.startswith('http'):
m3u8_url = urljoin(self.host, m3u8_url)
return {
"parse": 0,
"playUrl": "",
"url": m3u8_url,
"header": {
"User-Agent": self.session.headers['User-Agent'],
"Referer": play_url,
"Origin": self.host
}
}
+10 -1
View File
@@ -19,7 +19,7 @@ YOUTUBE_CLASSES = [
{'type_id': '最新新聞', 'type_name': '新聞'},
{'type_id': '新聞直播', 'type_name': '新聞直播'},
{'type_id': '歌曲', 'type_name': '歌曲'},
{'type_id': '動畫', 'type_name': '動畫'},
{'type_id': '動畫', 'type_name': '動畫'},
{'type_id': '短劇', 'type_name': '短劇'},
{'type_id': '劇集', 'type_name': '劇集'},
{'type_id': '電影', 'type_name': '電影'},
@@ -28,6 +28,15 @@ YOUTUBE_CLASSES = [
{'type_id': '16K HDR', 'type_name': '16K HDR'},
{'type_id': '科技', 'type_name': '科技'},
{'type_id': '解說', 'type_name': '解說'},
{'type_id': '台劇', 'type_name': '台劇'},
{'type_id': '陸劇', 'type_name': '陸劇'},
{'type_id': '綜藝', 'type_name': '綜藝'},
{'type_id': '網紅', 'type_name': '網紅'},
{'type_id': '靈異', 'type_name': '靈異'},
{'type_id': '探險', 'type_name': '探險'},
{'type_id': '旅遊', 'type_name': '旅遊'},
{'type_id': '美食', 'type_name': '美食'},
]
CATEGORY_QUERY = {
+162
View File
@@ -0,0 +1,162 @@
# -*- coding: utf-8 -*-
import re
import urllib.parse
import requests
try:
from base.spider import Spider as BaseSpider
except ImportError:
class BaseSpider:
def __init__(self):
return None
class Spider(BaseSpider):
BASE_URL = "https://maomi66.cc"
HEADERS = {
"User-Agent": "Mozilla/5.0 (Linux; Android 12; Mobile) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Mobile Safari/537.36",
"Referer": "https://maomi66.cc/",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
}
def __init__(self):
self.session = requests.Session()
self.session.headers.update(self.HEADERS)
self._class_cache = []
def getName(self):
return "猫咪AV"
def init(self, extend=""):
return None
def isVideoFormat(self, url):
return bool(re.search(r'\.(m3u8|mp4|flv|avi|mkv|mov)(\?|$)', url or '', re.I))
def manualVideoCheck(self):
return True
def homeContent(self, filter):
html = self._get(self.BASE_URL)
classes = self._classes(html)
return {"class": classes, "list": self._parse_list(html), "filters": {}, "parse": 0, "jx": 0}
def homeVideoContent(self):
return {"list": self._parse_list(self._get(self.BASE_URL))}
def categoryContent(self, tid, pg, filter, extend):
page = self._to_int(pg, 1)
html = self._get(self.BASE_URL + "/list/%s-%s.html" % (tid, page))
data = self._parse_list(html)
return {"list": data, "page": page, "pagecount": page + 1 if data else page, "limit": len(data) or 20, "total": (page + 1) * (len(data) or 20)}
def detailContent(self, ids):
vid = ids[0] if isinstance(ids, list) and ids else str(ids)
url = vid if str(vid).startswith("http") else self.BASE_URL + "/video/%s.html" % vid
html = self._get(url)
title = self._clean(self._match(html, r'<h1[^>]*>(.*?)</h1>') or self._match(html, r'<h2[^>]*>(.*?)</h2>') or self._match(html, r'<title[^>]*>(.*?)</title>'))
if not title:
title = "视频%s" % re.sub(r'\D+', '', str(vid))
pic = self._fix(self._match(html, r'<meta[^>]+property=["\']og:image["\'][^>]+content=["\']([^"\']+)') or self._match(html, r'<video[^>]+poster=["\']([^"\']+)') or self._match(html, r'(?:data-original|data-src|src)=["\']([^"\']+\.(?:jpg|jpeg|png|webp|gif)[^"\']*)'))
play = self._extract_play(html)
tags = []
for x in re.findall(r'<a[^>]+href=["\']/list/\d+-1\.html["\'][^>]*>(.*?)</a>', html, re.S):
t = self._clean(x)
if t and t not in tags:
tags.append(t)
content = self._clean(self._match(html, r'<div[^>]+class=["\'][^"\']*(?:des|intro|content|info)[^"\']*["\'][^>]*>(.*?)</div>')) or title
vod = {
"vod_id": str(vid).split("/")[-1].replace(".html", ""),
"vod_name": title,
"vod_pic": pic,
"type_name": "/".join(tags[:3]),
"vod_year": "",
"vod_area": "",
"vod_remarks": "",
"vod_actor": "",
"vod_director": "",
"vod_content": content,
"vod_play_from": "默认",
"vod_play_url": "播放$%s" % (play or url)
}
return {"list": [vod]}
def searchContent(self, key, quick, pg="1"):
q = urllib.parse.quote(str(key or ""))
page = self._to_int(pg, 1)
html = self._get(self.BASE_URL + "/search.php?content=%s&type=1&page=%s" % (q, page))
data = self._parse_list(html)
if not data:
html = self._get(self.BASE_URL + "/search.php?content=%s&type=1" % q)
data = self._parse_list(html)
return {"list": data, "page": page, "pagecount": page + 1 if data else page, "limit": len(data) or 20, "total": (page + 1) * (len(data) or 20)}
def playerContent(self, flag, id, vipFlags):
url = urllib.parse.unquote(str(id or ""))
if "/video/" in url or re.fullmatch(r'\d+', url):
page = url if url.startswith("http") else self.BASE_URL + "/video/%s.html" % url
play = self._extract_play(self._get(page))
url = play or page
return {"parse": 0, "playUrl": "", "url": self._fix(url), "header": self.HEADERS}
def _classes(self, html):
arr = []
for tid, name in re.findall(r'href=["\']/list/(\d+)-1\.html["\'][^>]*>(.*?)</a>', html or "", re.S):
name = self._clean(name)
if tid and name and not any(x["type_id"] == tid for x in arr):
arr.append({"type_id": tid, "type_name": name})
if not arr:
arr = [
{"type_id": "69829818", "type_name": "国产精品"},
{"type_id": "71188148", "type_name": "国产自拍"},
{"type_id": "43659662", "type_name": "日本精品"},
{"type_id": "37440125", "type_name": "欧美极品"},
{"type_id": "19211697", "type_name": "中文字幕"},
{"type_id": "77777777", "type_name": "动漫精品"}
]
self._class_cache = arr
return arr
def _parse_list(self, html):
out = []
blocks = re.findall(r'<li[\s\S]*?</li>', html or "", re.I)
if not blocks:
blocks = re.findall(r'<a[^>]+href=["\']/video/\d+\.html["\'][\s\S]*?</a>', html or "", re.I)
for item in blocks:
vid = self._match(item, r'href=["\'][^"\']*/video/(\d+)\.html["\']')
if not vid:
continue
name = self._clean(self._match(item, r'<h5[^>]*>\s*<a[^>]*>(.*?)</a>') or self._match(item, r'title=["\']([^"\']+)') or self._match(item, r'alt=["\']([^"\']+)'))
pic = self._fix(self._match(item, r'data-original=["\']([^"\']+)') or self._match(item, r'data-src=["\']([^"\']+)') or self._match(item, r'<img[^>]+src=["\']([^"\']+)'))
remark = self._clean(self._match(item, r'<span[^>]*>(.*?)</span>') or self._match(item, r'<em[^>]*>(.*?)</em>'))
if not name:
name = "视频%s" % vid
vod = {"vod_id": vid, "vod_name": name, "vod_pic": pic, "vod_remarks": remark}
if not any(x["vod_id"] == vid for x in out):
out.append(vod)
return out
def _extract_play(self, html):
play = self._match(html, r'hls\.loadSource\(["\']([^"\']+)["\']\)') or self._match(html, r'video\.src\s*=\s*["\']([^"\']+)["\']') or self._match(html, r'<source[^>]+src=["\']([^"\']+)["\']') or self._match(html, r'["\'](https?://[^"\']+play\.php\?[^"\']+)["\']') or self._match(html, r'["\'](/play\.php\?[^"\']+)["\']')
return self._fix(play)
def _get(self, url):
if not url:
return ""
url = self._fix(url)
headers = dict(self.HEADERS)
headers["Referer"] = self.BASE_URL + "/"
try:
r = self.session.get(url, headers=headers, timeout=12, verify=False)
if not r.encoding or r.encoding.lower() == "iso-8859-1":
r.encoding = r.apparent_encoding or "utf-8"
return r.text
except requests.RequestException:
return ""
def _match(self, text, pattern, default=""):
m = re.search(pattern, text or "", re.S | re.I)
if not m:
return default
return m.group(1) if m.lastindex else m.group(0)
def _clean(self, text):
text = re.sub(r'<script[\s\S]*?</script>|<style[\s\S]*?</style>', ' ', text or '', flags=re.I)
text = re.sub(r'<[^>]+>', ' ', text)
text = text.replace('&nbsp;', ' ').replace('&amp;amp;', '&').replace('&amp;', '&').replace('&#038;', '&').replace('&quot;', '"').replace('&#39;', "'").replace('&lt;', '<').replace('&gt;', '>')
return re.sub(r'\s+', ' ', text).strip()
def _fix(self, url):
url = (url or "").strip().replace("\\/", "/")
if not url:
return ""
if url.startswith("//"):
return "https:" + url
if url.startswith("/"):
return self.BASE_URL + url
return url
def _to_int(self, value, default=1):
try:
return int(value)
except Exception:
return default
+347
View File
@@ -0,0 +1,347 @@
# -*- coding: utf-8 -*-
import re
import json
import urllib.parse
from urllib.parse import urljoin, quote
import sys
sys.path.append('..')
from base.spider import Spider
from bs4 import BeautifulSoup
class Spider(Spider):
def getName(self):
return "香肠派对"
def init(self, extend=""):
self.host = "https://xiang512.xiang.party/xcpd"
pass
def header(self):
return {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36',
'Referer': self.host
}
def homeContent(self, filter):
result = {"class": [], "filters": {}, "list": []}
# 分类列表
classes = [
{"type_id": "1", "type_name": "在线看片"},
{"type_id": "2", "type_name": "无需等待"},
{"type_id": "3", "type_name": "不用下载"},
{"type_id": "4", "type_name": "全部免费"}
]
result["class"] = classes
# 获取首页视频
url = f"{self.host}/"
rsp = self.fetch(url, headers=self.header())
if rsp.status_code != 200:
return result
root = BeautifulSoup(rsp.text, 'html.parser')
videos = []
# 查找视频列表
items = root.select('ul.thumbnail-group.clearfix li')
for item in items:
try:
a = item.select_one('a.thumbnail')
if not a:
continue
href = a.get('href', '')
vod_id = re.search(r'/vod(?:detail|play)/(\d+)', href)
if not vod_id:
continue
vod_id = vod_id.group(1)
img = a.select_one('img')
pic = img.get('src', '') if img else ''
info = item.select_one('.video-info')
if info:
h5 = info.select_one('h5 a')
name = h5.get('title', '') if h5 else ''
if not name:
name = h5.text.strip() if h5 else ''
p = info.select_one('p')
remarks = p.text.strip() if p else ''
else:
name = a.get('title', '')
remarks = ''
videos.append({
"vod_id": vod_id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remarks
})
if len(videos) >= 20:
break
except:
continue
result["list"] = videos
return result
def homeVideoContent(self):
return self.homeContent(False)
def categoryContent(self, tid, pg, filter, extend):
p = int(pg)
# 修复分页URL格式
url = f"{self.host}/vodtype/{tid}-{p}.html"
rsp = self.fetch(url, headers=self.header())
if rsp.status_code != 200:
return {"list": [], "page": p, "pagecount": 1}
root = BeautifulSoup(rsp.text, 'html.parser')
videos = []
# 提取总页数(从“共XX条数据,当前X/913页”)
pagecount = p
page_info = root.find(string=re.compile(r'\d+条数据,当前\d+/(\d+)页'))
if page_info:
try:
pagecount = int(re.search(r'/(\d+)页', page_info).group(1))
except:
pass
# 提取视频列表
items = root.select('ul.thumbnail-group.clearfix li')
for item in items:
try:
a = item.select_one('a.thumbnail')
if not a:
continue
href = a.get('href', '')
vod_id = re.search(r'/vod(?:detail|play)/(\d+)', href)
if not vod_id:
continue
vod_id = vod_id.group(1)
img = a.select_one('img')
pic = img.get('src', '') if img else ''
info = item.select_one('.video-info')
if info:
h5 = info.select_one('h5 a')
name = h5.get('title', '') if h5 else ''
if not name:
name = h5.text.strip() if h5 else ''
p_elem = info.select_one('p')
remarks = p_elem.text.strip() if p_elem else ''
else:
name = a.get('title', '')
remarks = ''
videos.append({
"vod_id": vod_id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remarks
})
except:
continue
return {
"list": videos,
"page": p,
"pagecount": pagecount
}
def detailContent(self, ids):
vid = ids[0] if isinstance(ids, list) else ids.split(",")[0]
url = f"{self.host}/voddetail/{vid}.html"
rsp = self.fetch(url, headers=self.header())
if rsp.status_code != 200:
return {"list": []}
root = BeautifulSoup(rsp.text, 'html.parser')
# 标题
title = ""
h1 = root.select_one('h1.appel-title')
if h1:
title = h1.text.strip()
if not title:
title_elem = root.select_one('title')
if title_elem:
title = title_elem.text.replace('视频介绍--香肠派对', '').strip()
# 图片
pic = ""
img = root.select_one('img.appel-img')
if img:
pic = img.get('src', '')
if not pic:
img = root.select_one('.detail-poster img')
if img:
pic = img.get('src', '')
# 描述
desc = ""
desc_elem = root.select_one('.detail-content')
if desc_elem:
desc = desc_elem.text.strip()
if not desc:
desc_elem = root.select_one('.appel-content')
if desc_elem:
desc = desc_elem.text.strip()
# 播放列表
play_from_list = []
play_url_list = []
# 查找线路
tabs = root.select('.detail-tab li a')
play_blocks = root.select('ul.detail-play-list')
for i, block in enumerate(play_blocks):
line_name = tabs[i].text.strip() if i < len(tabs) else f"线路{i+1}"
urls = []
for a in block.select('a'):
href = a.get('href', '')
if href:
full_url = urljoin(self.host, href)
name = a.text.strip() or f"{len(urls)+1}"
urls.append(f"{name}${full_url}")
if urls:
play_from_list.append(line_name)
play_url_list.append("#".join(urls))
# 如果没有找到,尝试其他选择器
if not play_from_list:
lines = root.select('.ff-playurl-tab li a')
for i, block in enumerate(root.select('.ff-playurl-tab-pane')):
line_name = lines[i].text.strip() if i < len(lines) else f"线路{i+1}"
urls = []
for a in block.select('a'):
href = a.get('href', '')
if href:
full_url = urljoin(self.host, href)
name = a.text.strip() or f"{len(urls)+1}"
urls.append(f"{name}${full_url}")
if urls:
play_from_list.append(line_name)
play_url_list.append("#".join(urls))
vod_play_from = "$$$".join(play_from_list) if play_from_list else ""
vod_play_url = "$$$".join(play_url_list) if play_url_list else ""
return {"list": [{
"vod_id": vid,
"vod_name": title,
"vod_pic": pic,
"vod_content": desc,
"vod_play_from": vod_play_from,
"vod_play_url": vod_play_url
}]}
def searchContent(self, key, quick, pg="1"):
p = int(pg)
url = f"{self.host}/vodsearch/-------------.html?wd={quote(key)}&page={p}"
rsp = self.fetch(url, headers=self.header())
if rsp.status_code != 200:
return {"list": []}
root = BeautifulSoup(rsp.text, 'html.parser')
videos = []
items = root.select('ul.thumbnail-group.clearfix li')
for item in items:
try:
a = item.select_one('a.thumbnail')
if not a:
continue
href = a.get('href', '')
vod_id = re.search(r'/vod(?:detail|play)/(\d+)', href)
if not vod_id:
continue
vod_id = vod_id.group(1)
img = a.select_one('img')
pic = img.get('src', '') if img else ''
info = item.select_one('.video-info')
if info:
h5 = info.select_one('h5 a')
name = h5.get('title', '') if h5 else ''
if not name:
name = h5.text.strip() if h5 else ''
p_elem = info.select_one('p')
remarks = p_elem.text.strip() if p_elem else ''
else:
name = a.get('title', '')
remarks = ''
videos.append({
"vod_id": vod_id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remarks
})
except:
continue
return {"list": videos, "page": p}
def playerContent(self, flag, id, vipFlags):
# 构建播放页URL
if id.startswith('http'):
play_url = id
else:
play_url = urljoin(self.host, id)
rsp = self.fetch(play_url, headers=self.header())
if rsp.status_code != 200:
return {"parse": 0, "playUrl": play_url}
html = rsp.text
# 方法1:从 player_aaaa 提取
match = re.search(r'var player_aaaa\s*=\s*({.*?});', html, re.DOTALL)
if match:
try:
js_str = match.group(1)
js_str = re.sub(r'(\w+):', r'"\1":', js_str)
player_data = json.loads(js_str)
if player_data.get('url'):
return {"parse": 0, "playUrl": player_data['url']}
except:
pass
# 方法2:从 iframe 提取(关键修复)
# 匹配 id="playleft" 的 td 中的 iframe
iframe_match = re.search(r'<td[^>]*id="playleft"[^>]*>.*?<iframe[^>]+src="([^"]+)"', html, re.DOTALL)
if iframe_match:
iframe_url = iframe_match.group(1)
# 提取 url 参数
m3u8_match = re.search(r'[?&]url=([^&]+)', iframe_url)
if m3u8_match:
m3u8_url = urllib.parse.unquote(m3u8_match.group(1))
return {"parse": 0, "playUrl": m3u8_url}
# 如果 iframe 本身就是 m3u8
if '.m3u8' in iframe_url:
return {"parse": 0, "playUrl": iframe_url}
# 方法3:直接查找 iframe
iframe_match2 = re.search(r'<iframe[^>]+src="([^"]+)"', html)
if iframe_match2:
iframe_url = iframe_match2.group(1)
m3u8_match = re.search(r'[?&]url=([^&]+)', iframe_url)
if m3u8_match:
m3u8_url = urllib.parse.unquote(m3u8_match.group(1))
return {"parse": 0, "playUrl": m3u8_url}
# 方法4:直接查找 m3u8
m3u8 = re.search(r'https?://[^"\']+\.m3u8[^"\']*', html)
if m3u8:
return {"parse": 0, "playUrl": m3u8.group(0)}
# 方法5:让系统解析
return {"parse": 1, "url": play_url}
def localProxy(self, params):
return [200, "video/MP2T", ""]
+137
View File
@@ -0,0 +1,137 @@
# -*- coding: utf-8 -*-
import re
import urllib.parse
import requests
try:
from base.spider import Spider as BaseSpider
except ImportError:
class BaseSpider:
pass
class Spider(BaseSpider):
BASE_URL = "https://madou.club"
DASH_URL = "https://dash.madou.club"
HEADERS = {
"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",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Referer": BASE_URL + "/",
}
def __init__(self):
super().__init__()
self.name = "麻豆社"
self.session = requests.Session()
self.session.headers.update(self.HEADERS)
self._class_cache = None
def init(self, extend="{}"):
return None
def getName(self):
return self.name
def homeContent(self, filter):
html = self._get(self.BASE_URL + "/")
return {"class": self._classes(html), "filters": {}, "list": self._parse_list(html), "parse": 0, "jx": 0}
def homeVideoContent(self):
return {"list": self._parse_list(self._get(self.BASE_URL + "/"))}
def categoryContent(self, tid, pg, filter, extend):
page = self._to_int(pg, 1)
base = tid if str(tid).startswith("http") else self.BASE_URL + "/category/" + str(tid).strip("/")
url = base.rstrip("/") if page <= 1 else base.rstrip("/") + "/page/" + str(page)
data = self._parse_list(self._get(url))
return {"page": page, "pagecount": page if len(data) < 10 else page + 1, "limit": 20, "total": 99999, "list": data, "parse": 0, "jx": 0}
def detailContent(self, ids):
result = {"list": [], "parse": 0, "jx": 0}
if not ids:
return result
url = ids[0]
html = self._get(url)
name = self._clean(self._match(html, r'<h1[^>]*class=["\']article-title["\'][^>]*>(.*?)</h1>') or self._match(html, r'<title>(.*?)</title>').split("-")[0])
pic = self._match(html, r'shareimage\s*:\s*["\']([^"\']+)') or self._match(html, r'<img[^>]+data-src=["\']([^"\']+)') or self._match(html, r'<img[^>]+src=["\']([^"\']+)')
cate = self._clean(self._match(html, r'分类:\s*<a[^>]*>(.*?)</a>'))
remarks = self._clean(self._match(html, r'观看\((.*?)\)'))
tag_block = self._match(html, r'<div[^>]+class=["\']article-tags["\'][^>]*>(.*?)</div>')
tags = ",".join([self._clean(x) for x in re.findall(r'<a[^>]*>(.*?)</a>', tag_block, re.S)])
iframe = self._match(html, r'<iframe[^>]+src=["\']?([^"\'\s>]+)')
play_id = urllib.parse.urljoin(self.BASE_URL, iframe or url)
result["list"].append({"vod_id": url, "vod_name": name, "vod_pic": urllib.parse.urljoin(self.BASE_URL, pic), "type_name": cate, "vod_year": "", "vod_area": "", "vod_remarks": remarks, "vod_actor": tags, "vod_director": "", "vod_content": name, "vod_play_from": "DPlayer", "vod_play_url": name + "$" + play_id})
return result
def searchContent(self, key, quick, pg="1"):
page = self._to_int(pg, 1)
q = urllib.parse.quote(str(key))
url = self.BASE_URL + "/?s=" + q if page <= 1 else self.BASE_URL + "/page/" + str(page) + "?s=" + q
data = self._parse_list(self._get(url))
return {"page": page, "pagecount": page if len(data) < 10 else page + 1, "limit": 20, "total": 99999, "list": data, "parse": 0, "jx": 0}
def playerContent(self, flag, id, vipFlags):
result = {"parse": 0, "playUrl": "", "url": id or "", "jx": 0, "header": {"User-Agent": self.HEADERS["User-Agent"], "Referer": self.BASE_URL + "/"}}
if not id:
return result
play_page = id
if "dash.madou.club/share/" not in play_page:
html = self._get(play_page)
play_page = urllib.parse.urljoin(self.BASE_URL, self._match(html, r'<iframe[^>]+src=["\']?([^"\'\s>]+)') or play_page)
html = self._get(play_page, {"Referer": self.BASE_URL + "/"})
token = self._match(html, r'var\s+token\s*=\s*["\']([^"\']*)')
m3u8 = self._match(html, r'var\s+m3u8\s*=\s*["\']([^"\']+\.m3u8)["\']')
if m3u8:
url = urllib.parse.urljoin(self.DASH_URL, m3u8)
result["url"] = url + (("&" if "?" in url else "?") + "token=" + token if token else "")
result["header"] = {"User-Agent": self.HEADERS["User-Agent"], "Referer": play_page, "Origin": self.BASE_URL}
return result
def _classes(self, html=None):
if self._class_cache:
return self._class_cache
html = html or self._get(self.BASE_URL + "/")
classes, seen = [], set()
for href, name in re.findall(r'<a[^>]+href=["\'](https://madou\.club/category/[^"\']+)["\'][^>]*>(.*?)</a>', html, re.S):
name = self._clean(name)
key = href.rstrip("/")
if key not in seen and name:
seen.add(key)
classes.append({"type_id": href, "type_name": name})
self._class_cache = classes
return classes
def _parse_list(self, html):
data = []
blocks = re.findall(r'<article\b.*?</article>', html, re.S) or re.findall(r'<li>.*?</li>', html, re.S)
for item in blocks:
href = self._match(item, r'<a[^>]+href=["\']([^"\']+\.html)["\']')
name = self._clean(self._match(item, r'<h2[^>]*>.*?<a[^>]*>(.*?)</a>') or self._match(item, r'<a[^>]*>(?:<span.*?</span>)?\s*(.*?)</a>'))
pic = self._match(item, r'<img[^>]+data-src=["\']([^"\']+)') or self._match(item, r'<img[^>]+src=["\']([^"\']+)')
remarks = self._clean(self._match(item, r'<time[^>]*>(.*?)</time>') or self._match(item, r'观看\((.*?)\)'))
if href and name:
data.append({"vod_id": urllib.parse.urljoin(self.BASE_URL, href), "vod_name": name, "vod_pic": urllib.parse.urljoin(self.BASE_URL, pic), "vod_remarks": remarks})
return data
def _get(self, url, headers=None):
h = dict(self.HEADERS)
if headers:
h.update(headers)
try:
return self.session.get(url, headers=h, timeout=15, verify=False).text
except Exception:
return ""
def _match(self, text, pattern):
m = re.search(pattern, text or "", re.S | re.I)
return m.group(1).strip() if m else ""
def _clean(self, text):
text = re.sub(r'<.*?>', '', text or '')
text = text.replace('&nbsp;', ' ').replace('&amp;', '&').replace('&#038;', '&').replace('"', '"')
return re.sub(r'\s+', ' ', text).strip()
def _to_int(self, value, default=0):
try:
return int(value)
except Exception:
return default
+11 -1
View File
@@ -1922,7 +1922,17 @@
"ua": "okhttp",
"url": "https://raw.githubusercontent.com/FGBLH/HKL/refs/heads/main/xxx视频资源.txt"
},
{
"name": "🔞午夜剧场",
"type": 0,
"playerType": 2,
"epg": "https://epg.112114.xyz/?ch={name}&date={date}",
"logo": "https://epg.112114.xyz/logo/{name}.png",
"url": "https://raw.githubusercontent.com/FGBLH/HKL/refs/heads/main/午夜剧场.txt",
"header": {
"X-Api-key": "8692bd33270440d4d2941b814b81a25a1eb84a7205532971714c4299c57449e0..0..1738210005..5cf2c4ad-043b-4234-83a2-e84ae917"
}
},
{
"name": "台灣景點直播(freeman)",
"playerType": 2,
+11
View File
@@ -132,6 +132,17 @@
"ua": "okhttp",
"url": "https://raw.githubusercontent.com/FGBLH/HKL/refs/heads/main/xxx视频资源.txt"
},
{
"name": "🔞午夜剧场",
"type": 0,
"playerType": 2,
"epg": "https://epg.112114.xyz/?ch={name}&date={date}",
"logo": "https://epg.112114.xyz/logo/{name}.png",
"url": "https://raw.githubusercontent.com/FGBLH/HKL/refs/heads/main/午夜剧场.txt",
"header": {
"X-Api-key": "8692bd33270440d4d2941b814b81a25a1eb84a7205532971714c4299c57449e0..0..1738210005..5cf2c4ad-043b-4234-83a2-e84ae917"
}
},
{
"name": "台灣景點直播(freeman)",
"playerType": 2,
+109 -1
View File
@@ -10,9 +10,117 @@
],
"logo" : ".美.gif",
"wallpaper": "https://picsum.photos/1080/",
"notice" : "版本:115.07.26.Ver.2",
"notice" : "版本:115.07.27.Ver.2",
"sites" : [
{
"key": "真实人妻",
"name": "🔞真实人妻.py",
"type": 3,
"api": "./py/真实人妻.py"
},
{
"key": "猫咪TV",
"name": "🔞咪TV.py",
"type": 3,
"api": "./py/猫咪TV.py"
},
{
"key": "麻豆社",
"name": "🔞麻豆社.py",
"type": 3,
"api": "./py/麻豆社.py"
},
{
"key": "prsd",
"name": "🔞怦然心动.py",
"type": 3,
"api": "./py/怦然心动.py"
},
{
"key": "lyt",
"name": "🔞撸一天.py",
"type": 3,
"api": "./py/撸一天.py"
},
{
"key": "scpd",
"name": "🔞香肠派对.py",
"type": 3,
"api": "./py/香肠派对.py"
},
{
"key": "8X8X",
"name": "🔞8X8X.py",
"type": 3,
"api": "./py/8X8X.py"
},
{
"key": "Jable",
"name": "🔞Jable.py",
"type": 3,
"api": "./py/Jable.py"
},
{
"key": "Xvideos",
"name": "🔞Xvideos.py",
"type": 3,
"api": "./py/Xvideos.py"
},
{
"key": "Pornhub",
"name": "🔞Pornhub.py",
"type": 3,
"api": "./py/Pornhub.py"
},
{
"key": "xhamster",
"name": "🔞xhamster.py",
"type": 3,
"api": "./py/xhamster.py"
},
{
"key": "eporner",
"name": "🔞EPORNER.py",
"type": 3,
"api": "./py/EPORNER.py"
},
{
"key": "91porn",
"name": "🔞91porn.py",
"type": 3,
"api": "./py/91porn.py"
},
{
"key": "fullhd",
"name": "🔞FullHD.py",
"type": 3,
"api": "./py/FullHD.py"
},
{
"key": "JAV36",
"name": "🔞JAV36.py",
"type": 3,
"api": "./py/JAV36.py"
},
{
"key": "ggsp",
"name": "🔞久久視頻.py",
"type": 3,
"api": "./py/久久視頻.py"
},
{
"key": "TOPTV",
"name": "🔞TOPTV.py",
"type": 3,
"api": "./py/TOPTV.py"
},
{
"key": "VHUB",
"name": "🔞VHUB.py",
"type": 3,
"api": "./py/VHUB[成人].py"
},
{
"key" : "麻豆",