huyueandClaude Opus 4.7
21c11fb04a
xchina 视频详情页没有真正的简介文本,.video-detail 面板就是 标题/模特/分类/系列名 + 一整块 「推广：…」广告位(a.cpc-text → xchina.click/prepare.html 的联盟跳转,常含 VPN/楼凤/约炮乃至 「幼女」「学生妹」等违规词)。旧 _extract_description 把整段面板 innerText + 那些"标签"链接一锅端 进 content,导致主站内容审核把整批视频资源 block 掉(实测一个 11 视频的系列页 0 条入库)。 改为:视频 content 只抓模特名 + 分类/系列名(限同源 xchina.co,排除 a.cpc-text / xchina.click / prepare.html),拼成「标题\n模特: X · 分类: Y / Z」,抓不到就回退用标题;不再用面板 innerText、 也不用站点级 boilerplate 的 meta/jsonld 描述。图集 scraper 的标签查询也加上同样的广告/外站过滤(防御)。 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
628 lines
28 KiB
Python
628 lines
28 KiB
Python
"""xchina.co 详情页采集 —— 无头、只抓元数据(标题 / 封面图 URL / m3u8 URL),不下载文件。
|
|
|
|
逻辑大体从原 GUI `xchina_gui_downloader.py` 的 Downloader 里抽出并精简:
|
|
- 去掉 yt-dlp / ffmpeg / 文件落盘 / GUI 回调 / 跳过记录
|
|
- stop/skip 的 threading.Event 换成一个 asyncio cancel 检查
|
|
- 无头环境下遇到 Cloudflare/验证码直接抛错(没有人工过验证的退路)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Callable, Optional
|
|
from urllib.parse import urljoin, urlparse
|
|
|
|
from playwright.async_api import BrowserContext
|
|
from playwright.async_api import Error as PlaywrightError
|
|
from playwright.async_api import TimeoutError as PlaywrightTimeoutError
|
|
|
|
from .models import ScrapedItem, ScrapeResult
|
|
|
|
# 真正的 Cloudflare 质询页 body 很短(几十~几百字),且含这些特征串之一。
|
|
# 不用 "verify" / "验证" 这种泛词 —— 正常详情页(年龄确认、表单等)也会命中,会误判。
|
|
_BLOCK_MARKERS = [
|
|
"just a moment", "checking your browser", "cloudflare",
|
|
"attention required", "you have been blocked", "verify you are human",
|
|
"请稍候", "正在验证您的浏览器", "你已被屏蔽",
|
|
]
|
|
_BLOCK_BODY_MAXLEN = 800
|
|
|
|
|
|
class ScrapeCancelled(Exception):
|
|
pass
|
|
|
|
|
|
class XChinaScraper:
|
|
def __init__(
|
|
self,
|
|
context: BrowserContext,
|
|
is_cancelled: Callable[[], bool] = lambda: False,
|
|
log: Callable[[str], None] = lambda _msg: None,
|
|
items_per_page: int = 12,
|
|
):
|
|
self.context = context
|
|
self.is_cancelled = is_cancelled
|
|
self.log = log
|
|
self.items_per_page = items_per_page
|
|
|
|
# ---- cancel ----------------------------------------------------------
|
|
|
|
def _check_cancel(self) -> None:
|
|
if self.is_cancelled():
|
|
raise ScrapeCancelled()
|
|
|
|
# ---- top level -------------------------------------------------------
|
|
|
|
async def run(self, series_url: str, start_page: int = 1, end_page: int = 1) -> ScrapeResult:
|
|
result = ScrapeResult()
|
|
series_page = await self.context.new_page()
|
|
try:
|
|
if self.is_detail_url(series_url):
|
|
self.log(f"输入是单个详情页,直接采集: {series_url}")
|
|
item = await self._scrape_detail_safe(series_url, 1, result)
|
|
if item:
|
|
result.resources.append(item)
|
|
result.pages_scanned = 1
|
|
return result
|
|
|
|
for page_no in range(start_page, end_page + 1):
|
|
self._check_cancel()
|
|
ok = await self._go_series_page(series_page, series_url, page_no)
|
|
if not ok:
|
|
self.log(f"第 {page_no} 页不存在或无视频,停止翻页。")
|
|
break
|
|
result.pages_scanned += 1
|
|
links = await self._collect_video_links(series_page)
|
|
if not links:
|
|
self.log(f"第 {page_no} 页没识别到详情链接。")
|
|
continue
|
|
self.log(f"第 {page_no} 页识别到 {len(links)} 个详情链接。")
|
|
for detail_url in links:
|
|
self._check_cancel()
|
|
item = await self._scrape_detail_safe(detail_url, page_no, result)
|
|
if item:
|
|
result.resources.append(item)
|
|
return result
|
|
finally:
|
|
await series_page.close()
|
|
|
|
async def _scrape_detail_safe(self, detail_url: str, page_no: int, result: ScrapeResult) -> Optional[ScrapedItem]:
|
|
try:
|
|
return await self.scrape_detail(detail_url, page_no)
|
|
except ScrapeCancelled:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001
|
|
self.log(f"详情页采集失败,跳过: {detail_url} -> {exc}")
|
|
result.item_errors.append({"source_url": detail_url, "reason": f"{type(exc).__name__}: {exc}"})
|
|
return None
|
|
|
|
# ---- url helpers -----------------------------------------------------
|
|
|
|
@staticmethod
|
|
def is_detail_url(url: str) -> bool:
|
|
path = urlparse(url).path.lower()
|
|
if not path.endswith(".html"):
|
|
return False
|
|
if any(part in path for part in ["/series-", "/category", "/tags", "/search"]):
|
|
return False
|
|
return "/video/" in path or "/videos/" in path
|
|
|
|
def _page_url_candidates(self, current_url: str, page_no: int) -> list[str]:
|
|
parsed = urlparse(current_url)
|
|
base = current_url.split("#", 1)[0].split("?", 1)[0]
|
|
candidates: list[str] = []
|
|
if base.endswith(".html"):
|
|
m = re.match(r"(.*/series-[^/]+)(?:/\d+)?\.html$", base)
|
|
if m:
|
|
candidates.append(f"{m.group(1)}.html" if page_no == 1 else f"{m.group(1)}/{page_no}.html")
|
|
else:
|
|
stem = base[:-5]
|
|
if page_no == 1:
|
|
candidates.append(base)
|
|
else:
|
|
candidates.extend([
|
|
f"{stem}/{page_no}.html",
|
|
f"{stem}-{page_no}.html",
|
|
f"{stem}_{page_no}.html",
|
|
f"{stem}/page/{page_no}.html",
|
|
])
|
|
candidates.extend([f"{base}?page={page_no}", f"{base}?p={page_no}"])
|
|
root = f"{parsed.scheme}://{parsed.netloc}"
|
|
try:
|
|
series_part = Path(parsed.path).stem.split("series-")[-1]
|
|
candidates.append(urljoin(root, f"/videos/series-{series_part}-{page_no}.html"))
|
|
except Exception:
|
|
pass
|
|
unique: list[str] = []
|
|
for url in candidates:
|
|
if url not in unique:
|
|
unique.append(url)
|
|
return unique
|
|
|
|
def _extract_detail_urls_from_text(self, text: str, base_url: str) -> list[str]:
|
|
if not text:
|
|
return []
|
|
text = text.replace("\\/", "/")
|
|
parsed_base = urlparse(base_url)
|
|
site_root = f"{parsed_base.scheme}://{parsed_base.netloc}"
|
|
patterns = [
|
|
r"https?://[^'\"<>\s]+/video/[A-Za-z0-9_-]+\.html",
|
|
r"https?://[^'\"<>\s]+/videos/[A-Za-z0-9_-]+\.html",
|
|
r"/video/[A-Za-z0-9_-]+\.html",
|
|
r"/videos/[A-Za-z0-9_-]+\.html",
|
|
r"video/[A-Za-z0-9_-]+\.html",
|
|
r"videos/[A-Za-z0-9_-]+\.html",
|
|
]
|
|
results: list[str] = []
|
|
for pattern in patterns:
|
|
for match in re.findall(pattern, text, flags=re.I):
|
|
if match.startswith(("video/", "videos/")):
|
|
match = "/" + match
|
|
url = urljoin(site_root + "/", match)
|
|
if self.is_detail_url(url) and url not in results:
|
|
results.append(url)
|
|
return results
|
|
|
|
def _extract_media_urls_from_text(self, text: str) -> list[str]:
|
|
if not text:
|
|
return []
|
|
text = text.replace("\\/", "/").replace("&", "&")
|
|
patterns = [
|
|
r"https?://[^'\"<>\s]+\.m3u8(?:\?[^'\"<>\s]+)?",
|
|
r"https?://[^'\"<>\s]+\.mp4(?:\?[^'\"<>\s]+)?",
|
|
r"https?://[^'\"<>\s]+\.webm(?:\?[^'\"<>\s]+)?",
|
|
r"https?://[^'\"<>\s]+\.mov(?:\?[^'\"<>\s]+)?",
|
|
]
|
|
urls: list[str] = []
|
|
for pattern in patterns:
|
|
for match in re.findall(pattern, text, flags=re.I):
|
|
if "upload.xchina.io/media/" in match:
|
|
continue
|
|
if match not in urls:
|
|
urls.append(match)
|
|
return urls
|
|
|
|
# ---- navigation ------------------------------------------------------
|
|
|
|
async def _visit(self, page, url: str) -> None:
|
|
last_error = None
|
|
attempt = 0
|
|
for attempt in range(1, 4):
|
|
self._check_cancel()
|
|
try:
|
|
await page.goto(url, wait_until="domcontentloaded", timeout=60000)
|
|
break
|
|
except PlaywrightError as exc:
|
|
last_error = exc
|
|
txt = str(exc)
|
|
retryable = any(m in txt for m in [
|
|
"ERR_CONNECTION_CLOSED", "ERR_CONNECTION_RESET", "ERR_TIMED_OUT", "Timeout",
|
|
])
|
|
if attempt >= 3 or not retryable:
|
|
raise
|
|
await page.wait_for_timeout(2500 * attempt)
|
|
if last_error and attempt >= 3:
|
|
raise last_error
|
|
try:
|
|
await page.wait_for_load_state("networkidle", timeout=15000)
|
|
except PlaywrightTimeoutError:
|
|
pass
|
|
|
|
async def _looks_blocked(self, page) -> Optional[str]:
|
|
"""返回命中的特征串(说明当前停在 Cloudflare 质询页),否则 None。"""
|
|
try:
|
|
body = await page.locator("body").inner_text(timeout=8000)
|
|
except Exception:
|
|
body = ""
|
|
try:
|
|
title = await page.title()
|
|
except Exception:
|
|
title = ""
|
|
low = body.lower()
|
|
low_t = title.lower()
|
|
hit = next((w for w in _BLOCK_MARKERS if w in low or w in low_t), None)
|
|
# 真正的质询页内容很少;内容丰富的正常页即使含某串也放过
|
|
if hit and len(body.strip()) <= _BLOCK_BODY_MAXLEN:
|
|
return hit
|
|
return None
|
|
|
|
async def _assert_not_blocked(self, page) -> None:
|
|
"""若停在 Cloudflare 质询页,等它的 JS 质询自动放行(通常几秒);仍卡住才报错。"""
|
|
hit = await self._looks_blocked(page)
|
|
if not hit:
|
|
return
|
|
self.log(f"检测到 Cloudflare 质询页(命中 {hit!r}),等待其自动放行…")
|
|
for _ in range(8): # 最多 ~24s
|
|
self._check_cancel()
|
|
await page.wait_for_timeout(3000)
|
|
try:
|
|
await page.wait_for_load_state("networkidle", timeout=5000)
|
|
except PlaywrightTimeoutError:
|
|
pass
|
|
if not await self._looks_blocked(page):
|
|
self.log("Cloudflare 质询已通过。")
|
|
return
|
|
raise RuntimeError(
|
|
f"被 Cloudflare 质询拦住未能自动放行(命中 {hit!r};无头环境过不了交互式验证,"
|
|
f"请用桌面浏览器登录/过一次验证后把 browser_profile 目录预置进容器卷)"
|
|
)
|
|
|
|
async def _go_series_page(self, page, series_url: str, page_no: int) -> bool:
|
|
for candidate in self._page_url_candidates(series_url, page_no):
|
|
try:
|
|
await self._visit(page, candidate)
|
|
await page.wait_for_timeout(300)
|
|
await self._assert_not_blocked(page)
|
|
if await self._collect_video_links(page):
|
|
return True
|
|
except RuntimeError:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001
|
|
self.log(f"访问候选分页失败: {candidate} -> {exc}")
|
|
return False
|
|
|
|
# ---- list page -------------------------------------------------------
|
|
|
|
async def _collect_video_links(self, page) -> list[str]:
|
|
await self._assert_not_blocked(page)
|
|
current = page.url
|
|
structured = await page.evaluate(
|
|
"""() => {
|
|
const rows = [];
|
|
for (const item of document.querySelectorAll('.item.video')) {
|
|
const a = item.querySelector('a[href*="/video/"]');
|
|
if (!a) continue;
|
|
const href = new URL(a.getAttribute('href'), location.href).href;
|
|
rows.push({href});
|
|
}
|
|
return rows;
|
|
}"""
|
|
)
|
|
seen: set[str] = set()
|
|
links: list[str] = []
|
|
for row in structured:
|
|
href = row.get("href", "")
|
|
if href in seen or href == current or not self.is_detail_url(href):
|
|
continue
|
|
seen.add(href)
|
|
links.append(href)
|
|
if len(links) >= self.items_per_page:
|
|
break
|
|
if links:
|
|
return links
|
|
|
|
# 兜底:从整页 HTML 里正则抠 /video/*.html
|
|
try:
|
|
html = await page.content()
|
|
except Exception:
|
|
html = ""
|
|
for href in self._extract_detail_urls_from_text(html, current):
|
|
if href in seen or href == current:
|
|
continue
|
|
seen.add(href)
|
|
links.append(href)
|
|
if len(links) >= self.items_per_page:
|
|
break
|
|
return links
|
|
|
|
# ---- detail page -----------------------------------------------------
|
|
|
|
async def scrape_detail(self, detail_url: str, page_no: int) -> ScrapedItem:
|
|
page = await self.context.new_page()
|
|
media_urls: list[str] = []
|
|
|
|
def remember_response(response):
|
|
url = response.url
|
|
lower = url.lower().split("?", 1)[0]
|
|
if any(lower.endswith(ext) for ext in (".mp4", ".m3u8", ".mov", ".webm")):
|
|
media_urls.append(url)
|
|
|
|
page.on("response", remember_response)
|
|
try:
|
|
await self._visit(page, detail_url)
|
|
await self._assert_not_blocked(page)
|
|
await self._handle_interstitial_ad(page, detail_url)
|
|
title = await self._extract_title(page)
|
|
summary, content = await self._extract_description(page, title)
|
|
cover_url = await self._find_cover_url(page)
|
|
video_url = await self._find_video_url(page, media_urls)
|
|
self.log(f"采集: {title} | cover={'有' if cover_url else '无'} | video={video_url or '无'} | content={len(content or '')}字")
|
|
from .media import extract_video_id
|
|
return ScrapedItem(
|
|
source_url=detail_url,
|
|
title=title,
|
|
summary=summary,
|
|
content=content,
|
|
cover_image=cover_url,
|
|
video_url=video_url,
|
|
page_no=page_no,
|
|
video_id=extract_video_id(detail_url),
|
|
)
|
|
finally:
|
|
await page.close()
|
|
|
|
# ---- ad / overlay handling (无头时只是尽力,不卡死) ----------------------
|
|
|
|
async def _has_ad_overlay(self, page) -> bool:
|
|
try:
|
|
return bool(await page.evaluate(
|
|
"""() => {
|
|
const bodyText = document.body.innerText || '';
|
|
if (bodyText.includes('请点击广告后继续浏览')) return true;
|
|
const nodes = Array.from(document.querySelectorAll('div, a, img'));
|
|
return nodes.some(el => {
|
|
const rect = el.getBoundingClientRect();
|
|
const style = getComputedStyle(el);
|
|
const z = Number.parseInt(style.zIndex || '0', 10) || 0;
|
|
const text = el.innerText || el.alt || '';
|
|
const visible = rect.width > 100 && rect.height > 40 &&
|
|
style.display !== 'none' && style.visibility !== 'hidden' &&
|
|
Number(style.opacity || '1') > 0.2;
|
|
return visible && z >= 100 && /请点击广告|继续浏览|点击后当天不再出现/.test(text);
|
|
});
|
|
}"""
|
|
))
|
|
except Exception:
|
|
return False
|
|
|
|
async def _handle_interstitial_ad(self, page, expected_url: str) -> None:
|
|
expected_path = urlparse(expected_url).path
|
|
for _ in range(3):
|
|
self._check_cancel()
|
|
await page.wait_for_timeout(1000)
|
|
parsed = urlparse(page.url)
|
|
on_detail = parsed.netloc.endswith("xchina.co") and parsed.path == expected_path
|
|
if on_detail and not await self._has_ad_overlay(page):
|
|
try:
|
|
if await page.locator(".tab-nav, .video-detail").count():
|
|
return
|
|
except Exception:
|
|
return
|
|
# 跳到了广告/中转页 -> 后退或重开
|
|
parsed = urlparse(page.url)
|
|
if not (parsed.netloc.endswith("xchina.co") and parsed.path == expected_path):
|
|
try:
|
|
await page.go_back(wait_until="domcontentloaded", timeout=20000)
|
|
await page.wait_for_timeout(1200)
|
|
except Exception:
|
|
await self._visit(page, expected_url)
|
|
parsed = urlparse(page.url)
|
|
if not (parsed.netloc.endswith("xchina.co") and parsed.path == expected_path):
|
|
await self._visit(page, expected_url)
|
|
try:
|
|
await page.wait_for_selector(".video-detail, .tab-nav", timeout=10000)
|
|
except PlaywrightTimeoutError:
|
|
pass
|
|
|
|
async def _click_tab_by_text(self, page, text: str) -> None:
|
|
for selector in (
|
|
f"text={text}", f"a:has-text('{text}')", f"button:has-text('{text}')",
|
|
f"li:has-text('{text}')", f"div:has-text('{text}')",
|
|
):
|
|
try:
|
|
loc = page.locator(selector).first
|
|
if await loc.count() and await loc.is_visible():
|
|
await loc.click(timeout=5000)
|
|
return
|
|
except Exception:
|
|
continue
|
|
|
|
# ---- title -----------------------------------------------------------
|
|
|
|
async def _extract_title(self, page) -> str:
|
|
title = ""
|
|
for _ in range(4):
|
|
await self._click_tab_by_text(page, "影片信息")
|
|
await page.wait_for_timeout(600)
|
|
title = await page.evaluate(
|
|
"""() => {
|
|
function clean(t){ return (t || '').replace(/\\s+/g, ' ').trim(); }
|
|
const infoTitle = document.querySelector('.video-detail .item .text');
|
|
const infoText = clean(infoTitle && infoTitle.innerText);
|
|
if (infoText && infoText.length >= 2 && infoText.length <= 180) return infoText;
|
|
for (const script of document.querySelectorAll('script[type="application/ld+json"]')) {
|
|
try {
|
|
const data = JSON.parse(script.textContent || 'null');
|
|
const rows = Array.isArray(data) ? data : [data];
|
|
for (const row of rows) {
|
|
if (row && row['@type'] === 'VideoObject' && row.name) return clean(row.name);
|
|
}
|
|
} catch (_) {}
|
|
}
|
|
const og = document.querySelector('meta[property="og:title"], meta[name="twitter:title"]');
|
|
const ogText = clean(og && og.getAttribute('content'));
|
|
if (ogText) return ogText;
|
|
const h = document.querySelector('h1, h2, .video-title, .movie-title');
|
|
const hText = clean(h && h.innerText);
|
|
if (hText) return hText;
|
|
return clean(document.title);
|
|
}"""
|
|
)
|
|
title = self._clean_title(title)
|
|
if self._is_good_title(title):
|
|
break
|
|
await page.wait_for_timeout(1200)
|
|
if not self._is_good_title(title):
|
|
m = re.search(r"/video/id-([A-Za-z0-9_-]+)\.html", page.url)
|
|
title = f"video-{m.group(1)}" if m else "未命名"
|
|
return title[:500]
|
|
|
|
@staticmethod
|
|
def _clean_title(title: str) -> str:
|
|
title = re.split(r"\s*[-_|]\s*(XChina|xchina|AV视频|成人视频|小黄书).*$", title or "", flags=re.I)[0]
|
|
return re.sub(r"^【[^】]+】", "", title).strip()
|
|
|
|
@staticmethod
|
|
def _is_good_title(title: str) -> bool:
|
|
if not title or len(title.strip()) < 2:
|
|
return False
|
|
bad = ["广告", "推广", "澳门", "新葡京", "blocked", "cloudflare", "attention required", "未命名"]
|
|
low = title.lower()
|
|
return not any(w.lower() in low for w in bad)
|
|
|
|
# ---- description / content -------------------------------------------
|
|
|
|
async def _extract_description(self, page, title: str) -> tuple[str | None, str]:
|
|
"""返回 (summary, content)。content 保证非空(实在没有就用 title)——主站资源库 ingest 对空 content 直接拒收。
|
|
|
|
xchina 视频详情页**没有真正的简介文本**:`.video-detail` 面板就是 标题 / 模特 / 分类 / 系列名,
|
|
再加一整块「推广：...」广告位(里面是 `a.cpc-text` → xchina.click/prepare.html 的联盟跳转,常含
|
|
VPN/楼凤/约炮乃至「幼女」「学生妹」这类违规词,会被主站内容审核 block)。`meta description` 也只是
|
|
站点级 boilerplate。所以这里只抓干净的结构化字段:模特名 + 分类/系列名(限同源 xchina.co,排除 cpc-text /
|
|
xchina.click / prepare.html),拼成 content;不要面板 innerText、不要 boilerplate meta/jsonld。
|
|
"""
|
|
data = await page.evaluate(
|
|
"""() => {
|
|
function clean(t){ return (t || '').replace(/\\s+/g,' ').trim(); }
|
|
const out = {model: '', cats: []};
|
|
const ma = document.querySelector('.video-detail a[href*="/model/id-"], a.model-item, .model-container a[href*="/model/id-"]');
|
|
out.model = clean(ma && ma.innerText);
|
|
const seen = new Set();
|
|
for (const a of document.querySelectorAll('.video-detail a, .video-info a, .breadcrumb a')) {
|
|
if (a.classList && a.classList.contains('cpc-text')) continue; // 广告位链接
|
|
let h; try { h = new URL(a.href, location.href); } catch(e){ continue; }
|
|
if (!h.host.endsWith('xchina.co')) continue; // 排除 xchina.click 等广告域
|
|
if (h.pathname.indexOf('prepare.html') >= 0) continue;
|
|
const ok = h.pathname.indexOf('/videos/series-') >= 0 || h.pathname.indexOf('/photos/series-') >= 0
|
|
|| h.pathname.indexOf('/tags/') === 0 || h.pathname.indexOf('/tag/') === 0;
|
|
if (!ok) continue;
|
|
const t = clean(a.innerText);
|
|
if (t && t.length <= 24 && !seen.has(t)) { seen.add(t); out.cats.push(t); }
|
|
}
|
|
out.cats = out.cats.slice(0, 8);
|
|
return out;
|
|
}"""
|
|
)
|
|
model = (data.get("model") or "").strip()
|
|
cats = data.get("cats") or []
|
|
parts: list[str] = []
|
|
if model:
|
|
parts.append(f"模特: {model}")
|
|
if cats:
|
|
parts.append("分类: " + " / ".join(cats))
|
|
content = ("\n".join([title, " · ".join(parts)]).strip() if parts else (title or "").strip()) or (title or "")
|
|
bits = [b for b in [model, *cats[:3]] if b]
|
|
summary = " · ".join(bits) if bits else None
|
|
return summary, content
|
|
|
|
# ---- cover -----------------------------------------------------------
|
|
|
|
# 明显不是封面的图: 站点 logo / 占位图 / 图标 / 广告
|
|
_COVER_BLOCKLIST = ("logo", "avatar", "icon", "banner", "ads", "adserver", "/images/sites/", "/static/", "placeholder")
|
|
|
|
def _bad_cover(self, src: str) -> bool:
|
|
if not src or src.startswith("data:"):
|
|
return True
|
|
low = src.lower()
|
|
return any(w in low for w in self._COVER_BLOCKLIST)
|
|
|
|
async def _find_cover_url(self, page) -> str | None:
|
|
# 1) 截屏 tab 里的截图
|
|
await self._click_tab_by_text(page, "截屏")
|
|
await page.wait_for_timeout(800)
|
|
screenshot_cover = await page.evaluate(
|
|
"""() => {
|
|
const img = document.querySelector('.tab-content.active .screenshot-container img, .screenshot-container img');
|
|
if (!img) return null;
|
|
const src = img.currentSrc || img.src || img.getAttribute('data-src') || img.getAttribute('data-original') || '';
|
|
return src ? new URL(src, location.href).href : null;
|
|
}"""
|
|
)
|
|
if screenshot_cover and not self._bad_cover(screenshot_cover):
|
|
return screenshot_cover
|
|
|
|
# 2) og:image / twitter:image / JSON-LD thumbnailUrl / <video poster> —— 通常就是真封面
|
|
meta_cover = await page.evaluate(
|
|
"""() => {
|
|
function abs(u){ try { return u ? new URL(u, location.href).href : null; } catch(_) { return null; } }
|
|
const og = document.querySelector('meta[property="og:image"], meta[name="og:image"], meta[name="twitter:image"], meta[property="twitter:image"]');
|
|
if (og && og.getAttribute('content')) return abs(og.getAttribute('content'));
|
|
for (const s of document.querySelectorAll('script[type="application/ld+json"]')) {
|
|
try {
|
|
const data = JSON.parse(s.textContent || 'null');
|
|
const rows = Array.isArray(data) ? data : [data];
|
|
for (const r of rows) {
|
|
if (!r) continue;
|
|
let t = r.thumbnailUrl || r.thumbnail || (r.image && (r.image.url || r.image));
|
|
if (Array.isArray(t)) t = t[0];
|
|
if (typeof t === 'string') return abs(t);
|
|
}
|
|
} catch(_){}
|
|
}
|
|
const v = document.querySelector('video[poster]');
|
|
if (v && v.getAttribute('poster')) return abs(v.getAttribute('poster'));
|
|
return null;
|
|
}"""
|
|
)
|
|
if meta_cover and not self._bad_cover(meta_cover):
|
|
return meta_cover
|
|
|
|
# 3) 兜底: 页面里最靠上/最大的一张正常图
|
|
images = await page.evaluate(
|
|
"""() => {
|
|
const rows = [];
|
|
for (const img of document.querySelectorAll('img')) {
|
|
const rect = img.getBoundingClientRect();
|
|
const style = getComputedStyle(img);
|
|
const src = img.currentSrc || img.src || img.getAttribute('data-src') || img.getAttribute('data-original') || '';
|
|
if (!src || style.display === 'none' || style.visibility === 'hidden') continue;
|
|
rows.push({
|
|
src: new URL(src, location.href).href,
|
|
width: img.naturalWidth || rect.width,
|
|
height: img.naturalHeight || rect.height,
|
|
x: rect.x, y: rect.y,
|
|
area: Math.max(rect.width,1) * Math.max(rect.height,1)
|
|
});
|
|
}
|
|
return rows;
|
|
}"""
|
|
)
|
|
candidates = [
|
|
img for img in images
|
|
if not self._bad_cover(img.get("src", "")) and img.get("width", 0) >= 180 and img.get("height", 0) >= 120
|
|
]
|
|
if not candidates:
|
|
return meta_cover or screenshot_cover or None
|
|
candidates.sort(key=lambda x: (x.get("y", 99999), x.get("x", 99999), -x.get("area", 0)))
|
|
return candidates[0]["src"]
|
|
|
|
# ---- video url -------------------------------------------------------
|
|
|
|
async def _find_video_url(self, page, media_urls: list[str]) -> str | None:
|
|
await self._click_tab_by_text(page, "下载")
|
|
await page.wait_for_timeout(800)
|
|
try:
|
|
page_source = await page.content()
|
|
except Exception:
|
|
page_source = ""
|
|
for url in self._extract_media_urls_from_text(page_source):
|
|
return url
|
|
direct = await page.evaluate(
|
|
"""() => {
|
|
const attrs = ['href', 'src', 'data-url', 'data-src', 'data-href'];
|
|
const nodes = Array.from(document.querySelectorAll('a, button, video, source, [data-url], [data-src], [data-href]'));
|
|
const good = [];
|
|
for (const node of nodes) {
|
|
for (const attr of attrs) {
|
|
const raw = node.getAttribute(attr);
|
|
if (!raw) continue;
|
|
let url; try { url = new URL(raw, location.href).href; } catch (_) { continue; }
|
|
const lower = url.toLowerCase();
|
|
if (/(\\.mp4|\\.m3u8|\\.webm|\\.mov)(\\?|$)/.test(lower) || /download|downurl/.test(lower)) good.push(url);
|
|
}
|
|
}
|
|
return good;
|
|
}"""
|
|
)
|
|
for url in direct:
|
|
if url:
|
|
return url
|
|
# 网络请求里嗅到的媒体
|
|
if media_urls:
|
|
return media_urls[-1]
|
|
await page.wait_for_timeout(2000)
|
|
if media_urls:
|
|
return media_urls[-1]
|
|
return None
|