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>
407 lines
18 KiB
Python
407 lines
18 KiB
Python
"""xchina.co 图集(photos series)采集 —— 抓系列页里的套图详情页,提取标题/封面/全部图片 URL。
|
|
|
|
图片 URL 模板(实测):https://img.xchina.io/photos2/{detailId}/{NNNN}_600x0.webp
|
|
- NNNN: 四位补零的 1-based 序号
|
|
- _600x0: 600px 宽缩放变体 —— 就用它,体积小(几十 KB),不追原图
|
|
- 旧套图前缀可能是 .../photos/{id}/...(不带 2) —— 从页面渲染出的实际 bg-image 取,不硬编码
|
|
详情页本身分页:/photo/id-{detailId}/{N}.html,每页约 14 张。
|
|
|
|
复用 XChinaScraper 的导航/反爬 helper(_visit / _assert_not_blocked / _page_url_candidates / _check_cancel)。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import re
|
|
from urllib.parse import urlparse
|
|
|
|
from playwright.async_api import TimeoutError as PlaywrightTimeoutError
|
|
|
|
from .media import extract_detail_id, head_ok
|
|
from .models import ScrapedItem, ScrapeResult
|
|
from .scraper import ScrapeCancelled, XChinaScraper
|
|
|
|
# 详情页一页大约多少张图(用于从分页数估总数)
|
|
_PER_DETAIL_PAGE = 14
|
|
# 解析图片 URL:.../{NNNN}{_WxH}?.{ext}
|
|
_IMG_URL_RE = re.compile(
|
|
r"^(?P<prefix>https?://[^\s'\"]+?)/(?P<seg>\d{2,6})(?P<size>_\d+x\d+)?\.(?P<ext>webp|jpe?g|png)(?:\?[^\s'\"]*)?$",
|
|
re.I,
|
|
)
|
|
_BAD_IMG_PARTS = ("/images/sites/", "/model/", "/media/", "/css/avatars", "/upload/", "logo", "banner", "icon", "avatar")
|
|
|
|
|
|
def is_photo_detail_url(url: str) -> bool:
|
|
path = urlparse(url).path.lower()
|
|
return path.endswith(".html") and re.search(r"/photo/id-[a-z0-9_-]+(?:/\d+)?\.html$", path) is not None
|
|
|
|
|
|
def is_photo_series_url(url: str) -> bool:
|
|
return "/photos/series-" in urlparse(url).path.lower()
|
|
|
|
|
|
def _canon_detail_url(url: str) -> str:
|
|
"""/photo/id-XXX/3.html -> /photo/id-XXX.html"""
|
|
return re.sub(r"(/photo/id-[A-Za-z0-9_-]+)/\d+\.html$", r"\1.html", url.split("#", 1)[0])
|
|
|
|
|
|
class XChinaPhotoScraper(XChinaScraper):
|
|
def __init__(self, context, is_cancelled=lambda: False, log=lambda _m: None,
|
|
items_per_page: int = 14, max_images: int = 100):
|
|
super().__init__(context, is_cancelled=is_cancelled, log=log, items_per_page=items_per_page)
|
|
self.max_images = max(1, int(max_images or 100))
|
|
|
|
# ---- top level -------------------------------------------------------
|
|
|
|
async def run(self, series_url: str, start_page: int = 1, end_page: int = 1) -> ScrapeResult:
|
|
result = ScrapeResult()
|
|
if is_photo_detail_url(series_url):
|
|
self.log(f"输入是单个图集详情页,直接采集: {series_url}")
|
|
item = await self._scrape_photo_safe(_canon_detail_url(series_url), 1, result)
|
|
if item:
|
|
result.resources.append(item)
|
|
result.pages_scanned = 1
|
|
return result
|
|
|
|
list_page = await self.context.new_page()
|
|
try:
|
|
for page_no in range(start_page, end_page + 1):
|
|
self._check_cancel()
|
|
ok = await self._go_series_page_photos(list_page, series_url, page_no)
|
|
if not ok:
|
|
self.log(f"第 {page_no} 页不存在或无图集,停止翻页。")
|
|
break
|
|
result.pages_scanned += 1
|
|
links = await self._collect_photo_links(list_page)
|
|
if not links:
|
|
self.log(f"第 {page_no} 页没识别到图集链接。")
|
|
continue
|
|
self.log(f"第 {page_no} 页识别到 {len(links)} 个图集链接。")
|
|
for detail_url, hint_title in links:
|
|
self._check_cancel()
|
|
item = await self._scrape_photo_safe(detail_url, page_no, result, hint_title=hint_title)
|
|
if item:
|
|
result.resources.append(item)
|
|
return result
|
|
finally:
|
|
await list_page.close()
|
|
|
|
async def _scrape_photo_safe(self, detail_url, page_no, result, hint_title=None):
|
|
try:
|
|
return await self.scrape_photo_detail(detail_url, page_no, hint_title=hint_title)
|
|
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
|
|
|
|
# ---- list page -------------------------------------------------------
|
|
|
|
async def _go_series_page_photos(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_photo_links(page):
|
|
return True
|
|
except RuntimeError:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001
|
|
self.log(f"访问候选分页失败: {candidate} -> {exc}")
|
|
return False
|
|
|
|
async def _collect_photo_links(self, page) -> list[tuple[str, str]]:
|
|
await self._assert_not_blocked(page)
|
|
current = page.url
|
|
rows = await page.evaluate(
|
|
"""() => {
|
|
const out = [];
|
|
for (const item of document.querySelectorAll('.item.photo, .list .item')) {
|
|
const a = item.querySelector('a[href*="/photo/id-"]');
|
|
if (!a) continue;
|
|
const href = new URL(a.getAttribute('href'), location.href).href;
|
|
let t = '';
|
|
const te = item.querySelector('.title a') || item.querySelector('.title');
|
|
if (te) t = te.innerText || '';
|
|
if (!t) t = a.getAttribute('title') || '';
|
|
out.push({href: href, title: t.replace(/\\s+/g, ' ').trim()});
|
|
}
|
|
return out;
|
|
}"""
|
|
)
|
|
seen, links = set(), []
|
|
for row in rows:
|
|
href = _canon_detail_url(row.get("href") or "")
|
|
if not href or href in seen or href == current or not is_photo_detail_url(href):
|
|
continue
|
|
seen.add(href)
|
|
links.append((href, row.get("title") or ""))
|
|
if len(links) >= self.items_per_page:
|
|
break
|
|
return links
|
|
|
|
# ---- detail page -----------------------------------------------------
|
|
|
|
async def scrape_photo_detail(self, detail_url: str, page_no: int, hint_title: str | None = None) -> ScrapedItem:
|
|
detail_url = _canon_detail_url(detail_url)
|
|
page = await self.context.new_page()
|
|
try:
|
|
await self._visit(page, detail_url)
|
|
await self._assert_not_blocked(page)
|
|
await self._wait_photo_detail(page, detail_url)
|
|
|
|
title = await self._extract_photo_title(page, hint_title)
|
|
meta = await self._extract_photo_meta(page) # {model, series, date, tags, count}
|
|
bg_first = await self._photo_bg_images(page) # 第 1 页渲染出的图片 URL(按 DOM 顺序)
|
|
detail_id = extract_detail_id(detail_url)
|
|
|
|
template = self._parse_img_template(bg_first, detail_id)
|
|
total = int(meta.get("count") or 0)
|
|
if not total:
|
|
total = await self._estimate_total_from_pager(page) or len(bg_first)
|
|
total = max(total, len(bg_first))
|
|
|
|
wanted = min(total, self.max_images) if total else self.max_images
|
|
images: list[str] = []
|
|
if template and wanted > 0:
|
|
images = [template.format(n=i) for i in range(1, wanted + 1)]
|
|
if not head_ok(images[-1], referer=detail_url):
|
|
self.log(f"图集 {detail_id}: 构造 URL 末张不可达,回退逐页抓取")
|
|
images = []
|
|
if not images:
|
|
images = await self._scrape_all_pages_images(page, detail_url, limit=wanted or self.max_images)
|
|
if not images:
|
|
images = bg_first[: self.max_images]
|
|
|
|
capped = bool(total) and total > len(images)
|
|
cover = images[0] if images else (bg_first[0] if bg_first else None)
|
|
eff_total = total or len(images)
|
|
content = self._compose_content(title, meta, eff_total)
|
|
summary = self._compose_summary(meta, eff_total) or None
|
|
|
|
self.log(
|
|
f"图集采集: {title} | 图 {len(images)}/{total or '?'}"
|
|
f"{' (已截断)' if capped else ''} | cover={'有' if cover else '无'}"
|
|
)
|
|
return ScrapedItem(
|
|
source_url=detail_url,
|
|
title=title,
|
|
media_kind="photo",
|
|
cover_image=cover,
|
|
images=images,
|
|
image_total=eff_total,
|
|
image_capped=capped,
|
|
summary=summary,
|
|
content=content,
|
|
page_no=page_no,
|
|
video_id=detail_id,
|
|
extra={
|
|
"base_prefix": (template.split("/{", 1)[0] if template else ""),
|
|
"detail_id": detail_id,
|
|
"img_template": template or "",
|
|
},
|
|
)
|
|
finally:
|
|
await page.close()
|
|
|
|
async def _wait_photo_detail(self, page, expected_url: str) -> None:
|
|
"""图集详情页没有 .video-detail/.tab-nav;等图片容器出现,顺手处理中转广告页。"""
|
|
sel = ".list.photo-items, .item.photo-image, div.img[role='img']"
|
|
want_id = (re.search(r"/photo/id-([A-Za-z0-9_-]+)", expected_url) or [None, ""])[1]
|
|
for _ in range(3):
|
|
self._check_cancel()
|
|
await page.wait_for_timeout(900)
|
|
parsed = urlparse(page.url)
|
|
cur_id = (re.search(r"/photo/id-([A-Za-z0-9_-]+)", page.url) or [None, ""])[1]
|
|
on_detail = parsed.netloc.endswith("xchina.co") and cur_id == want_id and bool(want_id)
|
|
if on_detail:
|
|
try:
|
|
if await page.locator(sel).count():
|
|
return
|
|
except Exception:
|
|
pass
|
|
else:
|
|
# 可能被弹到广告/中转页
|
|
try:
|
|
await page.go_back(wait_until="domcontentloaded", timeout=20000)
|
|
await page.wait_for_timeout(1000)
|
|
except Exception:
|
|
await self._visit(page, expected_url)
|
|
cur_id = (re.search(r"/photo/id-([A-Za-z0-9_-]+)", page.url) or [None, ""])[1]
|
|
if cur_id != want_id:
|
|
await self._visit(page, expected_url)
|
|
try:
|
|
await page.wait_for_selector(sel, timeout=10000)
|
|
except PlaywrightTimeoutError:
|
|
pass
|
|
|
|
async def _photo_bg_images(self, page) -> list[str]:
|
|
"""页面里所有套图缩略 div(div.img[role=img] 等)的 background-image URL,按 DOM 顺序去重。"""
|
|
urls = await page.evaluate(
|
|
"""() => {
|
|
const out = [];
|
|
const els = document.querySelectorAll("div.img[role='img'], .item.photo-image .img, .photo-items .img");
|
|
for (const el of els) {
|
|
let s = (getComputedStyle(el).backgroundImage || el.style.backgroundImage || '');
|
|
const m = s.match(/url\\((["']?)(.*?)\\1\\)/);
|
|
if (m && m[2]) { try { out.push(new URL(m[2], location.href).href); } catch (e) {} }
|
|
}
|
|
return out;
|
|
}"""
|
|
)
|
|
seen, res = set(), []
|
|
for u in urls:
|
|
if not u or u in seen:
|
|
continue
|
|
low = u.lower()
|
|
if not any(d in low for d in ("xchina.io", "xchina.co", "xchina.click")):
|
|
continue
|
|
if any(w in low for w in _BAD_IMG_PARTS):
|
|
continue
|
|
seen.add(u)
|
|
res.append(u)
|
|
return res
|
|
|
|
def _parse_img_template(self, urls: list[str], detail_id: str) -> str | None:
|
|
"""从一批图片 URL 推 .../{NNNN}{size}.{ext} 模板,返回带 {n} 占位的 .format 串。"""
|
|
best = None
|
|
for u in urls:
|
|
m = _IMG_URL_RE.match(u)
|
|
if not m:
|
|
continue
|
|
prefix, seg, size, ext = m.group("prefix"), m.group("seg"), m.group("size") or "", m.group("ext")
|
|
width = len(seg)
|
|
tmpl = f"{prefix}/{{n:0{width}d}}{size}.{ext}"
|
|
if detail_id and detail_id in prefix:
|
|
return tmpl
|
|
best = best or tmpl
|
|
return best
|
|
|
|
async def _estimate_total_from_pager(self, page) -> int:
|
|
try:
|
|
nums = await page.eval_on_selector_all(
|
|
".pager a.pager-num, .pager .pager-num, [class*=pager] a",
|
|
"els => els.map(e => parseInt((e.innerText || '').trim(), 10)).filter(n => !isNaN(n))",
|
|
)
|
|
if nums:
|
|
return max(nums) * _PER_DETAIL_PAGE
|
|
except Exception:
|
|
pass
|
|
return 0
|
|
|
|
async def _scrape_all_pages_images(self, page, detail_url: str, limit: int) -> list[str]:
|
|
"""逐页加载 /photo/id-{id}/{N}.html 收集 bg-image,直到够 limit 张或没有新图。"""
|
|
base = re.sub(r"(/photo/id-[A-Za-z0-9_-]+)(?:/\d+)?\.html$", r"\1", detail_url)
|
|
collected: list[str] = list(await self._photo_bg_images(page)) # 第 1 页就是当前 page
|
|
max_pages = max(1, math.ceil(max(limit, 1) / _PER_DETAIL_PAGE)) + 2
|
|
n = 2
|
|
while len(collected) < limit and n <= max_pages:
|
|
self._check_cancel()
|
|
url = f"{base}/{n}.html"
|
|
try:
|
|
await self._visit(page, url)
|
|
await self._assert_not_blocked(page)
|
|
await page.wait_for_timeout(400)
|
|
except Exception as exc: # noqa: BLE001
|
|
self.log(f"图集翻页失败 {url}: {exc}")
|
|
break
|
|
new = [u for u in await self._photo_bg_images(page) if u not in collected]
|
|
if not new:
|
|
break
|
|
collected.extend(new)
|
|
n += 1
|
|
seen, res = set(), []
|
|
for u in collected:
|
|
if u not in seen:
|
|
seen.add(u)
|
|
res.append(u)
|
|
return res[:limit]
|
|
|
|
# ---- title / meta / content -----------------------------------------
|
|
|
|
async def _extract_photo_title(self, page, hint_title: str | None) -> str:
|
|
t = await page.evaluate(
|
|
"""() => {
|
|
function clean(s){ return (s || '').replace(/\\s+/g, ' ').trim(); }
|
|
const h = document.querySelector('h1, .photo-title, .video-detail .text');
|
|
const ht = clean(h && h.innerText);
|
|
if (ht && ht.length >= 2 && ht.length <= 200) return ht;
|
|
const og = document.querySelector('meta[property="og:title"], meta[name="twitter:title"]');
|
|
const ot = clean(og && og.getAttribute('content'));
|
|
if (ot) return ot;
|
|
return clean(document.title);
|
|
}"""
|
|
)
|
|
t = self._strip_title_suffix(t)
|
|
bad = ("cloudflare", "attention required", "you have been blocked")
|
|
if not (t and len(t.strip()) >= 2) or any(b in t.lower() for b in bad):
|
|
t = self._strip_title_suffix(hint_title or "")
|
|
if not (t and len(t.strip()) >= 2):
|
|
m = re.search(r"/photo/id-([A-Za-z0-9_-]+)", page.url)
|
|
t = f"photos-{m.group(1)}" if m else "未命名图集"
|
|
return t[:500]
|
|
|
|
@staticmethod
|
|
def _strip_title_suffix(t: str) -> str:
|
|
# "标题 - 系列 - 中国工作室 - 小黄书 xChina - 第 2 页" -> "标题"
|
|
t = (t or "").strip()
|
|
t = re.sub(r"\s*-\s*第\s*\d+\s*页\s*$", "", t)
|
|
t = re.split(r"\s*-\s*(小黄书|xChina|XChina)\b.*$", t)[0]
|
|
parts = [p.strip() for p in t.split(" - ") if p.strip()]
|
|
return (parts[0] if parts else t).strip()
|
|
|
|
async def _extract_photo_meta(self, page) -> dict:
|
|
return await page.evaluate(
|
|
"""() => {
|
|
function clean(s){ return (s || '').replace(/\\s+/g, ' ').trim(); }
|
|
const out = {model: '', series: '', date: '', tags: [], count: 0};
|
|
const ma = document.querySelector('.model-container a, a.model-item, a[href*="/model/id-"]');
|
|
out.model = clean(ma && ma.innerText);
|
|
// 系列名优先取面包屑 / .subs 里的(侧边栏分类也是 /photos/series- 链接,会误命中)
|
|
const sa = document.querySelector(
|
|
'.breadcrumb a[href*="/photos/series-"], .subs a[href*="/photos/series-"], '
|
|
+ '.path a[href*="/photos/series-"], .nav-path a[href*="/photos/series-"]')
|
|
|| document.querySelector('a[href*="/photos/series-"]');
|
|
out.series = clean(sa && sa.innerText);
|
|
const tags = [];
|
|
for (const a of document.querySelectorAll('a[href*="/tags/"], .tags a, .tag 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') || h.pathname.indexOf('prepare.html') >= 0) continue;
|
|
const t = clean(a.innerText);
|
|
if (t && t.length <= 20 && tags.indexOf(t) < 0) tags.push(t);
|
|
}
|
|
out.tags = tags.slice(0, 30);
|
|
const bt = clean(document.body.innerText);
|
|
let m = bt.match(/(\\d{4})[.\\-/](\\d{1,2})[.\\-/](\\d{1,2})/);
|
|
if (m) out.date = m[1] + '.' + ('0' + m[2]).slice(-2) + '.' + ('0' + m[3]).slice(-2);
|
|
m = bt.match(/(\\d{1,5})\\s*P\\b/);
|
|
if (m) out.count = parseInt(m[1], 10);
|
|
return out;
|
|
}"""
|
|
) or {}
|
|
|
|
@staticmethod
|
|
def _compose_content(title: str, meta: dict, total: int) -> str:
|
|
parts = []
|
|
if meta.get("model"):
|
|
parts.append(f"模特: {meta['model']}")
|
|
if meta.get("series"):
|
|
parts.append(f"系列: {meta['series']}")
|
|
if meta.get("date"):
|
|
parts.append(f"日期: {meta['date']}")
|
|
if total:
|
|
parts.append(f"共 {total} 张图")
|
|
if meta.get("tags"):
|
|
parts.append("标签: " + " ".join(meta["tags"]))
|
|
body = " · ".join(parts)
|
|
return (f"{title}\n{body}".strip() if body else (title or "").strip()) or (title or "")
|
|
|
|
@staticmethod
|
|
def _compose_summary(meta: dict, total: int) -> str:
|
|
bits = [b for b in (meta.get("model"), meta.get("series")) if b]
|
|
if total:
|
|
bits.append(f"{total}张")
|
|
return " · ".join(bits)
|