feat(kb-bilibili-content): Add article content extraction using Playwright
- Replace API-based article extraction with Playwright scraping - Support Bilibili article (专栏) content extraction - Extract full article text using .read-article-holder selector - Use async Playwright API for proper async/await support
This commit is contained in:
@@ -4,9 +4,9 @@
|
||||
Bilibili Content Downloader - B站内容获取工具
|
||||
|
||||
功能:
|
||||
- download_video(bvid, quality=480) - 下载低分辨率视频
|
||||
- download_video(bvid, quality=480) - 下载视频
|
||||
- extract_subtitle(bvid) - 提取字幕
|
||||
- extract_article(cvid) - 获取专栏全文
|
||||
- extract_article(cvid) - 获取专栏全文(使用Playwright)
|
||||
- 命令行参数支持
|
||||
- JSON 输出
|
||||
"""
|
||||
@@ -149,32 +149,51 @@ class BilibiliContent:
|
||||
return result
|
||||
|
||||
async def extract_article(self, cvid: str) -> Dict[str, Any]:
|
||||
"""获取专栏全文"""
|
||||
"""获取专栏全文(使用Playwright异步API)"""
|
||||
result = {"success": False, "cvid": cvid, "title": "", "content": "", "message": ""}
|
||||
|
||||
try:
|
||||
async with aiohttp.ClientSession(headers=self._get_headers()) as session:
|
||||
print(f"获取专栏: {cvid}")
|
||||
async with session.get("https://api.bilibili.com/x/article/viewinfo", params={'id': cvid}) as resp:
|
||||
data = await resp.json()
|
||||
from playwright.async_api import async_playwright
|
||||
|
||||
cvid_clean = cvid.replace('cv', '')
|
||||
url = f"https://www.bilibili.com/read/cv{cvid_clean}"
|
||||
print(f"获取专栏: {url}")
|
||||
|
||||
async with async_playwright() as p:
|
||||
browser = await p.chromium.launch(headless=True)
|
||||
page = await browser.new_page()
|
||||
|
||||
print(f"状态: {data.get('code')} - {data.get('message')}")
|
||||
await page.goto(url, wait_until="networkidle")
|
||||
await page.wait_for_timeout(3000)
|
||||
|
||||
if data.get('code') != 0:
|
||||
result["message"] = f"获取失败: {data.get('message')}"
|
||||
return result
|
||||
# 提取标题
|
||||
title_elem = await page.query_selector('h1')
|
||||
if title_elem:
|
||||
result["title"] = (await title_elem.inner_text()).strip()
|
||||
print(f"标题: {result['title'][:50]}...")
|
||||
|
||||
article = data.get('data', {})
|
||||
result["title"] = article.get('title', '')
|
||||
result["content"] = article.get('content', '')
|
||||
result["success"] = bool(result["title"] and result["content"])
|
||||
result["message"] = f"获取: {result['title']}" if result["success"] else "获取失败"
|
||||
# 提取正文内容
|
||||
content_elem = await page.query_selector('.read-article-holder')
|
||||
if content_elem:
|
||||
html_content = await content_elem.inner_html()
|
||||
text_content = await content_elem.inner_text()
|
||||
result["content"] = text_content.strip()
|
||||
result["html_content"] = html_content.strip()
|
||||
result["success"] = bool(result["title"] and result["content"])
|
||||
print(f"内容长度: {len(result['content'])} 字符")
|
||||
else:
|
||||
result["message"] = "未找到专栏内容"
|
||||
|
||||
if result["success"]:
|
||||
print(f"✅ {result['title']}")
|
||||
await browser.close()
|
||||
|
||||
if result["success"]:
|
||||
result["message"] = f"✅ 获取成功: {result['title'][:30]}..."
|
||||
else:
|
||||
result["message"] = f"获取失败: {result.get('message', '未知错误')}"
|
||||
|
||||
except Exception as e:
|
||||
result["message"] = f"错误: {str(e)}"
|
||||
print(f"错误: {e}")
|
||||
|
||||
return result
|
||||
|
||||
|
||||
Reference in New Issue
Block a user