yongge/ge
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

更新 wmf.txt

Browse Source
This commit is contained in:
yongge
2026-08-24 07:04:41 +02:00
parent 4f749c5444
commit bcd4accabe
1 changed files with 0 additions and 404 deletions
Download Patch File Download Diff File Expand all files Collapse all files
wmf.txt
-404
View File
@@ -17,410 +17,6 @@
"quickSearch": 1,
"filterable": 0,
"ext": "https://www.vip1949.com/"
},
{
"key": "SA视频",
"name": "七味网",
"type": 3,
"py":"# -*- coding: utf-8 -*-
# - SA 影视 (https://www.lsjys11.com/)
# 优化版：图片直链 + 视频代理 + 单线路
import re
import json
import sys
import requests
from urllib.parse import quote, unquote, urljoin, urlparse
from html import unescape
sys.path.append('..')
from base.spider import Spider
class Spider(Spider):
def getName(self):
return "SA影视"
def init(self, extend=""):
self.host = "https://www.lsjys11.com"
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',
'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',
'Referer': self.host,
}
self.session = requests.Session()
adapter = requests.adapters.HTTPAdapter(pool_connections=10, pool_maxsize=30, max_retries=0)
self.session.mount('http://', adapter)
self.session.mount('https://', adapter)
self.timeout = 8
self.UA = self.headers['User-Agent']
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def destroy(self):
if hasattr(self, 'session'):
self.session.close()
CATEGORIES = {
"movie": {"name": "电影", "cat_id": 13},
"tv": {"name": "连续剧", "cat_id": 12},
"variety": {"name": "综艺", "cat_id": 11},
"anime": {"name": "动漫", "cat_id": 14},
"short": {"name": "短剧", "cat_id": 16},
"documentary": {"name": "纪录片", "cat_id": 15},
}
def _proxy_url(self, url, typ="m3u8"):
url = str(url or "").strip()
if not url:
return ""
try:
return self.getProxyUrl() + "&type=" + typ + "&url=" + quote(url, safe="")
except Exception:
return url
def _play_headers(self, url=""):
host = ""
try:
u = urlparse(url or self.host)
host = u.scheme + "://" + u.netloc + "/" if u.scheme and u.netloc else self.host
except Exception:
host = self.host
return {
"User-Agent": self.UA,
"Accept": "*/*",
"Connection": "keep-alive",
"Referer": self.host,
"Origin": self.host.rstrip("/"),
}
def _parse_nuxt(self, html):
m = re.search(r'<script[^>]*id="__NUXT_DATA__"[^>]*>(.*?)</script>', html, re.DOTALL)
if not m:
return None
try:
return json.loads(m.group(1))
except:
return None
def _extract_videos(self, data):
if not data:
return []
videos = []
for item in data:
if not isinstance(item, dict):
continue
if 'id' not in item or 'name' not in item or 'score' not in item:
continue
vid = item['id']
name = item['name']
if isinstance(vid, int) and vid < len(data):
vid = data[vid]
if isinstance(name, int) and name < len(data):
name = data[name]
if not isinstance(vid, str) or not isinstance(name, str):
continue
if len(vid) < 5:
continue
cover = ""
img_idx = item.get('img')
if isinstance(img_idx, int) and img_idx < len(data):
cover = data[img_idx]
if isinstance(cover, str) and cover.startswith("//"):
cover = "https:" + cover
score = ""
score_idx = item.get('score')
if isinstance(score_idx, int) and score_idx < len(data):
score = str(data[score_idx])
episodes = ""
if item.get('number'):
n = item['number']
if isinstance(n, int) and n < len(data):
episodes = str(data[n])
videos.append({
"vod_id": f"/movie/detail/{vid}",
"vod_name": name,
"vod_pic": cover if isinstance(cover, str) else "",
"vod_remarks": f"{score}分 {episodes}集".strip() if score else "",
})
return videos
def homeContent(self, filter):
classes = []
filters = {}
for cid, info in self.CATEGORIES.items():
classes.append({"type_id": cid, "type_name": info["name"]})
filters[cid] = []
result = {"class": classes, "filters": filters}
try:
rsp = self.fetch(self.host, headers=self.headers)
html = rsp.text
data = self._parse_nuxt(html)
videos = self._extract_videos(data)
result["list"] = videos[:50]
except Exception as e:
self.log(f"首页获取出错: {str(e)}")
result["list"] = []
return result
def homeVideoContent(self):
try:
rsp = self.fetch(self.host, headers=self.headers)
html = rsp.text
data = self._parse_nuxt(html)
videos = self._extract_videos(data)
return {"list": videos[:50]}
except Exception as e:
self.log(f"首页视频获取出错: {str(e)}")
return {"list": []}
def categoryContent(self, tid, pg, filter, extend):
result = {"list": [], "page": int(pg), "pagecount": 999, "limit": 24, "total": 9999}
info = self.CATEGORIES.get(tid)
if not info:
return result
try:
page_num = int(pg) if pg and int(pg) > 0 else 1
if page_num > 1:
url = f"{self.host}/movie/list/{page_num}?cat_id={info['cat_id']}&position=movie&page={page_num}"
else:
url = f"{self.host}/movie/list?cat_id={info['cat_id']}&position=movie"
rsp = self.fetch(url, headers=self.headers)
html = rsp.text
data = self._parse_nuxt(html)
result["list"] = self._extract_videos(data)
except Exception as e:
self.log(f"分类获取出错: {str(e)}")
return result
def _get_str(self, data, val):
if isinstance(val, int) and val < len(data):
return str(data[val])
return str(val)
def _extract_episodes(self, data, item):
episodes = []
links_idx = item.get('links')
if not isinstance(links_idx, int) or links_idx >= len(data):
return episodes
links = data[links_idx]
if not isinstance(links, list):
return episodes
for link_idx in links:
if not isinstance(link_idx, int) or link_idx >= len(data):
continue
link_data = data[link_idx]
if not isinstance(link_data, dict):
continue
items_idx = link_data.get('items')
if not isinstance(items_idx, int) or items_idx >= len(data):
continue
items = data[items_idx]
if not isinstance(items, list):
continue
for item_idx in items:
if not isinstance(item_idx, int) or item_idx >= len(data):
continue
ep = data[item_idx]
if not isinstance(ep, dict):
continue
ep_id = self._get_str(data, ep.get('id', ''))
ep_name = self._get_str(data, ep.get('name', ''))
if ep_id and ep_name:
episodes.append({'name': ep_name, 'id': ep_id})
return episodes
def detailContent(self, ids):
if not ids or not ids[0]:
return {"list": []}
vid = ids[0]
url = f"{self.host}{vid}"
try:
rsp = self.fetch(url, headers=self.headers)
html = rsp.text
title = ""
desc = ""
cover = ""
category = ""
director = ""
actors = ""
genre = ""
ld_m = re.search(r'<script[^>]*type="application/ld\+json"[^>]*>(.*?)</script>', html, re.DOTALL)
if ld_m:
try:
ld = json.loads(ld_m.group(1))
if ld.get("@type") in ["Movie", "TVSeries", "Episode"]:
title = ld.get("name", "")
desc = ld.get("description", "")
cover = ld.get("image", "")
d = ld.get("director", [])
if d and isinstance(d, list):
director = ", ".join([x.get("name", "") for x in d if isinstance(x, dict)])
a = ld.get("actor", [])
if a and isinstance(a, list):
actors = ", ".join([x.get("name", "") for x in a if isinstance(x, dict)])
g = ld.get("genre", [])
if g and isinstance(g, list):
genre = ", ".join(g)
category = genre
except:
pass
if not title:
hm = re.search(r'<meta[^>]+property="og:title"[^>]+content="([^"]*)"', html)
if hm:
title = unescape(hm.group(1))
title = re.sub(r'\s+在线观看.*', '', title)
if not desc:
dm = re.search(r'<meta[^>]+name="description"[^>]+content="([^"]*)"', html)
if dm:
desc = unescape(dm.group(1))
if not cover:
cm = re.search(r'<meta[^>]+property="og:image"[^>]+content="([^"]*)"', html)
if cm:
cover = unescape(cm.group(1))
h1m = re.search(r'<h1[^>]*>\s*([^<]+)\s*</h1>', html)
if h1m:
title = h1m.group(1).strip()
if not category:
category = "电影" if "/movie/" in vid else "未知"
# === 只保留一条线路 ===
play_from = ["SA影视"]
play_urls_parts = []
data = self._parse_nuxt(html)
if data:
for item in data:
if not isinstance(item, dict):
continue
if 'play_links' not in item or 'links' not in item:
continue
episodes = self._extract_episodes(data, item)
if episodes:
ep_list = [f"{ep['name']}${url}" for ep in episodes]
play_urls_parts.append('#'.join(ep_list))
else:
play_urls_parts.append(f"SA影视${url}")
break
if not play_urls_parts:
play_urls_parts = [f"SA影视${url}"]
vod = {
"vod_id": vid,
"vod_name": title,
"type_name": category,
"vod_pic": cover,
"vod_content": desc,
"vod_play_from": "$$$".join(play_from),
"vod_play_url": "$$$".join(play_urls_parts),
"vod_director": director,
"vod_actor": actors,
"vod_class": genre,
}
return {"list": [vod]}
except Exception as e:
self.log(f"详情获取出错: {str(e)}")
return {"list": []}
def searchContent(self, key, quick, pg="1"):
try:
url = f"{self.host}/search?keyword={key}"
rsp = self.fetch(url, headers=self.headers)
html = rsp.text
data = self._parse_nuxt(html)
videos = self._extract_videos(data)
return {"list": videos, "page": pg}
except Exception as e:
self.log(f"搜索出错: {str(e)}")
return {"list": [], "page": pg}
def playerContent(self, flag, id, vipFlags):
url = str(id or "").strip()
is_direct = any(url.lower().endswith(ext) for ext in ['.m3u8', '.mp4', '.ts', '.flv', '.mkv']) if url else False
if is_direct:
proxy_url = self._proxy_url(url, "m3u8" if ".m3u8" in url.lower() else "media")
return {
"parse": 0,
"playUrl": "",
"url": proxy_url,
"header": self._play_headers(url),
}
else:
return {
"parse": 1,
"url": url,
"header": {"User-Agent": self.UA}
}
def localProxy(self, params):
try:
typ = params.get("type") if isinstance(params, dict) else ""
if typ == "m3u8":
return self.proxyM3u8(params)
if typ in ("media", "ts", "key"):
return self.proxyMedia(params)
except Exception as e:
return [500, "text/plain", str(e).encode("utf-8"), {}]
return None
def proxyM3u8(self, params):
url = unquote(params.get("url", ""))
r = self.session.get(url, headers=self._play_headers(url), timeout=self.timeout, allow_redirects=True)
r.encoding = "utf-8"
base = r.url or url
text = r.text or ""
def repl_uri(m):
raw = m.group(1)
abs_url = urljoin(base, raw)
ptype = "m3u8" if ".m3u8" in abs_url.lower() else "media"
return 'URI="' + self._proxy_url(abs_url, ptype) + '"'
out = []
for line in text.splitlines():
s = line.strip()
if not s:
out.append(line)
continue
if s.startswith("#"):
if "URI=" in s:
s = re.sub(r'URI="([^"]+)"', repl_uri, s)
out.append(s)
continue
abs_url = urljoin(base, s)
ptype = "m3u8" if ".m3u8" in abs_url.lower() else "media"
out.append(self._proxy_url(abs_url, ptype))
body = ("\n".join(out) + "\n").encode("utf-8")
return [200, "application/vnd.apple.mpegurl", body, {"Access-Control-Allow-Origin": "*"}]
def proxyMedia(self, params):
url = unquote(params.get("url", ""))
r = self.session.get(url, headers=self._play_headers(url), timeout=self.timeout, stream=False, allow_redirects=True)
ctype = r.headers.get("Content-Type") or "application/octet-stream"
return [200, ctype, r.content, {"Access-Control-Allow-Origin": "*"}]
def liveContent(self, url):
pass
",
"searchable": 1,
"quickSearch": 0,
},
{