From bcd4accabe4fabebe1d9e51e06f66741b016c162 Mon Sep 17 00:00:00 2001 From: yongge <18926+yongge@noreply.example.org> Date: Mon, 24 Aug 2026 07:04:41 +0200 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0=20wmf.txt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- wmf.txt | 404 -------------------------------------------------------- 1 file changed, 404 deletions(-) diff --git a/wmf.txt b/wmf.txt index be07d75..94cd1f3 100644 --- a/wmf.txt +++ b/wmf.txt @@ -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']*id="__NUXT_DATA__"[^>]*>(.*?)', 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']*type="application/ld\+json"[^>]*>(.*?)', 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']+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']+name="description"[^>]+content="([^"]*)"', html) - if dm: - desc = unescape(dm.group(1)) - - if not cover: - cm = re.search(r']+property="og:image"[^>]+content="([^"]*)"', html) - if cm: - cover = unescape(cm.group(1)) - - h1m = re.search(r']*>\s*([^<]+)\s*', 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, - }, {