添加B站视频内容提取工具（反爬虫方案）
功能特性: - 使用Playwright + undetected-playwright绕过B站极验验证码 - 支持提取视频基本信息、统计数据、字幕 - 生成结构化Markdown笔记 - 多策略数据提取（JS变量→HTML解析→API备用） - 完善的错误处理和超时机制 文件说明: - bilibili_extractor.py: 主程序脚本（21KB） - BILIBILI_EXTRACTOR_README.md: 完整技术文档（6.0KB） - USAGE.md: 快速使用指南（1.5KB） - PROJECT_SUMMARY.md: 项目总结（2.9KB） - notes/: 生成的视频笔记目录 - .gitignore: 忽略Python虚拟环境等 测试验证: - 成功提取BV11t64BJEDZ视频内容 - 绕过B站极验验证码 - 生成完整的Markdown笔记
This commit is contained in:
OpenClaw Agent
@@ -0,0 +1,590 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
B站视频内容提取器 - 反爬虫解决方案
|
||||
|
||||
工具选择：Playwright + undetected-playwright (stealth_async)
|
||||
选择理由：
|
||||
1. Playwright 比 Selenium 更现代，支持更好的异步操作
|
||||
2. undetected-playwright 的 stealth_async 专门用于绕过反爬虫检测
|
||||
3. 可以自动处理极验验证码等反爬虫措施
|
||||
4. 与系统Chromium浏览器兼容性好
|
||||
5. 社区活跃，文档完善
|
||||
|
||||
安装依赖：
|
||||
python3 -m venv bilibili_venv
|
||||
source bilibili_venv/bin/activate
|
||||
pip install playwright undetected-playwright beautifulsoup4 lxml
|
||||
playwright install chromium
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from playwright.async_api import async_playwright
|
||||
from bs4 import BeautifulSoup
|
||||
from undetected_playwright import stealth_async
|
||||
|
||||
|
||||
class BilibiliExtractor:
|
||||
"""B站视频信息提取器"""
|
||||
|
||||
def __init__(self, headless=False):
|
||||
self.headless = headless
|
||||
self.playwright = None
|
||||
self.browser = None
|
||||
self.context = None
|
||||
self.page = None
|
||||
|
||||
async def init_browser(self):
|
||||
"""初始化浏览器"""
|
||||
print("正在初始化浏览器...")
|
||||
self.playwright = await async_playwright().start()
|
||||
|
||||
self.browser = await self.playwright.chromium.launch(
|
||||
headless=self.headless,
|
||||
args=[
|
||||
'--no-sandbox',
|
||||
'--disable-setuid-sandbox',
|
||||
'--disable-blink-features=AutomationControlled',
|
||||
'--disable-dev-shm-usage',
|
||||
'--start-maximized',
|
||||
]
|
||||
)
|
||||
|
||||
# 创建上下文
|
||||
self.context = await self.browser.new_context(
|
||||
viewport={'width': 1920, 'height': 1080},
|
||||
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',
|
||||
locale='zh-CN',
|
||||
timezone_id='Asia/Shanghai',
|
||||
extra_http_headers={
|
||||
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
|
||||
'Referer': 'https://www.bilibili.com/',
|
||||
}
|
||||
)
|
||||
|
||||
# 应用stealth模式（反检测）
|
||||
await stealth_async(self.context)
|
||||
|
||||
self.page = await self.context.new_page()
|
||||
print("浏览器初始化完成！")
|
||||
|
||||
async def extract_video_info(self, bvid):
|
||||
"""提取视频信息"""
|
||||
url = f"https://www.bilibili.com/video/{bvid}"
|
||||
print(f"\n正在访问: {url}")
|
||||
|
||||
try:
|
||||
# 访问页面
|
||||
response = await self.page.goto(url, wait_until='domcontentloaded', timeout=60000)
|
||||
|
||||
print(f"HTTP状态码: {response.status}")
|
||||
|
||||
# 等待页面加载
|
||||
print("等待页面加载...")
|
||||
await asyncio.sleep(5)
|
||||
|
||||
# 检查是否出现验证码
|
||||
if await self._check_captcha():
|
||||
print("检测到验证码，等待处理...")
|
||||
await self._handle_captcha()
|
||||
|
||||
# 再次等待
|
||||
await asyncio.sleep(3)
|
||||
|
||||
# 检查页面标题
|
||||
title = await self.page.title()
|
||||
print(f"页面标题: {title}")
|
||||
|
||||
# 提取页面源码
|
||||
content = await self.page.content()
|
||||
print(f"页面内容长度: {len(content)} 字符")
|
||||
|
||||
# 检查是否有错误页面
|
||||
if '错误' in title or '404' in title or '视频不存在' in content:
|
||||
print("⚠️ 检测到错误页面，视频可能不存在或已被删除")
|
||||
return None
|
||||
|
||||
soup = BeautifulSoup(content, 'lxml')
|
||||
|
||||
# 尝试多种方法提取视频信息
|
||||
video_info = await self._extract_from_methods(soup, self.page)
|
||||
|
||||
# 如果还是没有提取到，尝试直接从API获取
|
||||
if not video_info or not video_info.get('title'):
|
||||
print("HTML提取失败，尝试API方法...")
|
||||
video_info = await self._extract_from_api(bvid)
|
||||
|
||||
# 提取字幕
|
||||
subtitles = await self._extract_subtitles()
|
||||
|
||||
return {
|
||||
'video_info': video_info,
|
||||
'subtitles': subtitles
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
print(f"提取视频信息时出错: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return None
|
||||
|
||||
async def _extract_from_methods(self, soup, page):
|
||||
"""使用多种方法提取视频信息"""
|
||||
info = {}
|
||||
|
||||
# 方法1: 从JavaScript变量获取
|
||||
try:
|
||||
print("尝试方法1: 从 window.__INITIAL_STATE__ 获取...")
|
||||
initial_state = await page.evaluate('() => typeof window.__INITIAL_STATE__ !== "undefined" ? window.__INITIAL_STATE__ : null')
|
||||
if initial_state:
|
||||
print("✓ 找到 __INITIAL_STATE__")
|
||||
return self._parse_initial_state(initial_state)
|
||||
else:
|
||||
print("✗ 未找到 __INITIAL_STATE__")
|
||||
except Exception as e:
|
||||
print(f"✗ 方法1失败: {e}")
|
||||
|
||||
# 方法2: 从script标签中提取
|
||||
print("\n尝试方法2: 从script标签提取...")
|
||||
script_tags = soup.find_all('script')
|
||||
for i, script in enumerate(script_tags):
|
||||
if script.string and '__INITIAL_STATE__' in script.string:
|
||||
try:
|
||||
match = re.search(r'window\.__INITIAL_STATE__\s*=\s*({.+?});', script.string)
|
||||
if match:
|
||||
print("✓ 从script标签中找到 __INITIAL_STATE__")
|
||||
data = json.loads(match.group(1))
|
||||
return self._parse_initial_state(data)
|
||||
except Exception as e:
|
||||
print(f"✗ 解析script标签失败: {e}")
|
||||
|
||||
# 方法3: 从HTML元素提取
|
||||
print("\n尝试方法3: 从HTML元素提取...")
|
||||
|
||||
# 标题
|
||||
title_elem = soup.find('h1', class_='video-title')
|
||||
if title_elem:
|
||||
info['title'] = title_elem.get_text(strip=True)
|
||||
print(f"✓ 找到标题: {info['title']}")
|
||||
|
||||
# UP主
|
||||
up_elem = soup.find('a', class_='up-name')
|
||||
if up_elem:
|
||||
info['author'] = up_elem.get_text(strip=True)
|
||||
print(f"✓ 找到UP主: {info['author']}")
|
||||
|
||||
# BV号
|
||||
og_url = soup.find('meta', property='og:url')
|
||||
if og_url:
|
||||
url = og_url.get('content', '')
|
||||
info['bvid'] = url.split('/')[-1] if url else ''
|
||||
print(f"✓ 找到BV号: {info['bvid']}")
|
||||
|
||||
# 从meta标签提取
|
||||
if not info.get('title'):
|
||||
info['title'] = self._safe_extract(soup, 'meta[property="og:title"]', 'content')
|
||||
if not info.get('author'):
|
||||
info['author'] = self._safe_extract(soup, 'meta[property="og:video:author"]', 'content')
|
||||
|
||||
return info
|
||||
|
||||
async def _extract_from_api(self, bvid):
|
||||
"""通过API提取视频信息"""
|
||||
info = {}
|
||||
|
||||
try:
|
||||
print("\n尝试API方法获取视频信息...")
|
||||
|
||||
# B站视频信息API
|
||||
api_url = "https://api.bilibili.com/x/web-interface/view"
|
||||
response = await self.page.request.get(api_url, params={'bvid': bvid})
|
||||
|
||||
if response.ok:
|
||||
data = await response.json()
|
||||
if data.get('code') == 0:
|
||||
video_data = data.get('data', {})
|
||||
print("✓ API调用成功")
|
||||
|
||||
# 基本信息
|
||||
info['title'] = video_data.get('title', '')
|
||||
info['bvid'] = video_data.get('bvid', '')
|
||||
info['aid'] = video_data.get('aid', '')
|
||||
|
||||
# UP主信息
|
||||
owner = video_data.get('owner', {})
|
||||
info['author'] = owner.get('name', '')
|
||||
info['mid'] = owner.get('mid', '')
|
||||
|
||||
# 视频统计
|
||||
stat = video_data.get('stat', {})
|
||||
info['view'] = stat.get('view', 0)
|
||||
info['danmaku'] = stat.get('danmaku', 0)
|
||||
info['reply'] = stat.get('reply', 0)
|
||||
info['favorite'] = stat.get('favorite', 0)
|
||||
info['coin'] = stat.get('coin', 0)
|
||||
info['share'] = stat.get('share', 0)
|
||||
info['like'] = stat.get('like', 0)
|
||||
|
||||
# 时间信息
|
||||
info['pubdate'] = video_data.get('pubdate', 0)
|
||||
info['pubtime'] = datetime.fromtimestamp(info['pubdate']).strftime('%Y-%m-%d %H:%M:%S') if info['pubdate'] else '未知'
|
||||
|
||||
# 时长
|
||||
duration = video_data.get('duration', 0)
|
||||
info['duration'] = self._format_duration(duration)
|
||||
|
||||
# 简介
|
||||
info['desc'] = video_data.get('desc', '')
|
||||
|
||||
# 标签
|
||||
tags = []
|
||||
for tag in video_data.get('tag', []):
|
||||
tags.append(tag.get('tag_name', ''))
|
||||
info['tags'] = ', '.join(tags)
|
||||
|
||||
return info
|
||||
else:
|
||||
print(f"✗ API返回错误: {data.get('message', '未知错误')}")
|
||||
else:
|
||||
print(f"✗ API请求失败: HTTP {response.status}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ API方法失败: {e}")
|
||||
|
||||
return info
|
||||
|
||||
async def _check_captcha(self):
|
||||
"""检查是否出现验证码"""
|
||||
captcha_selectors = [
|
||||
'.captcha-box',
|
||||
'#captcha-box',
|
||||
'[class*="captcha"]',
|
||||
'[class*="geetest"]',
|
||||
]
|
||||
|
||||
for selector in captcha_selectors:
|
||||
try:
|
||||
element = await self.page.query_selector(selector)
|
||||
if element:
|
||||
return True
|
||||
except:
|
||||
continue
|
||||
|
||||
return False
|
||||
|
||||
async def _handle_captcha(self):
|
||||
"""处理验证码"""
|
||||
print("验证码检测到，等待 5 秒尝试自动绕过...")
|
||||
|
||||
for i in range(5):
|
||||
await asyncio.sleep(1)
|
||||
if not await self._check_captcha():
|
||||
print("验证码已自动处理！")
|
||||
return True
|
||||
|
||||
if not self.headless:
|
||||
print("验证码需要手动处理，请在浏览器中完成...")
|
||||
while await self._check_captcha():
|
||||
await asyncio.sleep(2)
|
||||
print("验证码已完成！")
|
||||
return True
|
||||
else:
|
||||
print("⚠️ Headless模式下无法手动处理验证码")
|
||||
print("建议：重新运行脚本，设置 headless=False")
|
||||
return False
|
||||
|
||||
def _parse_initial_state(self, data):
|
||||
"""解析INITIAL_STATE数据"""
|
||||
info = {}
|
||||
|
||||
try:
|
||||
video_data = data.get('videoData', {})
|
||||
|
||||
# 基本信息
|
||||
info['title'] = video_data.get('title', '')
|
||||
info['bvid'] = video_data.get('bvid', '')
|
||||
info['aid'] = video_data.get('aid', '')
|
||||
|
||||
# UP主信息
|
||||
owner = video_data.get('owner', {})
|
||||
info['author'] = owner.get('name', '')
|
||||
info['mid'] = owner.get('mid', '')
|
||||
|
||||
# 视频统计
|
||||
stat = video_data.get('stat', {})
|
||||
info['view'] = stat.get('view', 0)
|
||||
info['danmaku'] = stat.get('danmaku', 0)
|
||||
info['reply'] = stat.get('reply', 0)
|
||||
info['favorite'] = stat.get('favorite', 0)
|
||||
info['coin'] = stat.get('coin', 0)
|
||||
info['share'] = stat.get('share', 0)
|
||||
info['like'] = stat.get('like', 0)
|
||||
|
||||
# 时间信息
|
||||
info['pubdate'] = video_data.get('pubdate', 0)
|
||||
info['pubtime'] = datetime.fromtimestamp(info['pubdate']).strftime('%Y-%m-%d %H:%M:%S') if info['pubdate'] else '未知'
|
||||
|
||||
# 时长
|
||||
duration = video_data.get('duration', 0)
|
||||
info['duration'] = self._format_duration(duration)
|
||||
|
||||
# 简介
|
||||
info['desc'] = video_data.get('desc', '')
|
||||
|
||||
# 标签
|
||||
tags = []
|
||||
for tag in video_data.get('tag', []):
|
||||
tags.append(tag.get('tag_name', ''))
|
||||
info['tags'] = ', '.join(tags)
|
||||
|
||||
except Exception as e:
|
||||
print(f"解析INITIAL_STATE时出错: {e}")
|
||||
|
||||
return info
|
||||
|
||||
def _format_duration(self, seconds):
|
||||
"""格式化时长"""
|
||||
if not seconds:
|
||||
return "未知"
|
||||
|
||||
hours = seconds // 3600
|
||||
minutes = (seconds % 3600) // 60
|
||||
secs = seconds % 60
|
||||
|
||||
if hours > 0:
|
||||
return f"{hours:02d}:{minutes:02d}:{secs:02d}"
|
||||
else:
|
||||
return f"{minutes:02d}:{secs:02d}"
|
||||
|
||||
async def _extract_subtitles(self):
|
||||
"""提取视频字幕"""
|
||||
subtitles = []
|
||||
|
||||
try:
|
||||
# 尝试通过API获取
|
||||
bvid = await self.page.evaluate('() => window.__INITIAL_STATE__?.videoData?.bvid')
|
||||
aid = await self.page.evaluate('() => window.__INITIAL_STATE__?.videoData?.aid')
|
||||
|
||||
if bvid and aid:
|
||||
print(f"正在获取字幕 (aid={aid})...")
|
||||
subtitles = await self._fetch_subtitle_api(aid, bvid)
|
||||
except Exception as e:
|
||||
print(f"提取字幕时出错: {e}")
|
||||
|
||||
return subtitles
|
||||
|
||||
async def _fetch_subtitle_api(self, aid, bvid):
|
||||
"""通过API获取字幕"""
|
||||
try:
|
||||
api_url = "https://api.bilibili.com/x/player/v2"
|
||||
response = await self.page.request.get(
|
||||
api_url,
|
||||
params={'aid': aid, 'bvid': bvid}
|
||||
)
|
||||
|
||||
if response.ok:
|
||||
data = await response.json()
|
||||
if data.get('code') == 0:
|
||||
subtitle_info = data.get('data', {}).get('subtitle', {})
|
||||
subtitles_list = subtitle_info.get('subtitles', [])
|
||||
|
||||
if subtitles_list:
|
||||
print(f"找到 {len(subtitles_list)} 个字幕")
|
||||
# 获取第一个字幕
|
||||
subtitle_url = subtitles_list[0].get('subtitle_url', '')
|
||||
if subtitle_url:
|
||||
full_url = f"https:{subtitle_url}"
|
||||
print(f"下载字幕: {full_url}")
|
||||
return await self._fetch_subtitle_content(full_url)
|
||||
except Exception as e:
|
||||
print(f"通过API获取字幕失败: {e}")
|
||||
|
||||
return []
|
||||
|
||||
async def _fetch_subtitle_content(self, url):
|
||||
"""获取字幕内容"""
|
||||
try:
|
||||
response = await self.page.request.get(url)
|
||||
if response.ok:
|
||||
data = await response.json()
|
||||
return self._parse_subtitle_json(data)
|
||||
except Exception as e:
|
||||
print(f"获取字幕内容失败: {e}")
|
||||
return []
|
||||
|
||||
def _parse_subtitle_json(self, data):
|
||||
"""解析字幕JSON"""
|
||||
subtitles = []
|
||||
|
||||
try:
|
||||
body = data.get('body', [])
|
||||
for item in body:
|
||||
from_time = item.get('from', 0)
|
||||
to_time = item.get('to', 0)
|
||||
content = item.get('content', '')
|
||||
|
||||
subtitles.append({
|
||||
'from': self._format_timestamp(from_time),
|
||||
'to': self._format_timestamp(to_time),
|
||||
'content': content
|
||||
})
|
||||
except Exception as e:
|
||||
print(f"解析字幕JSON失败: {e}")
|
||||
|
||||
return subtitles
|
||||
|
||||
def _format_timestamp(self, seconds):
|
||||
"""格式化时间戳"""
|
||||
minutes = int(seconds // 60)
|
||||
secs = int(seconds % 60)
|
||||
return f"{minutes:02d}:{secs:02d}"
|
||||
|
||||
def _safe_extract(self, soup, selector, attr):
|
||||
"""安全提取元素属性"""
|
||||
try:
|
||||
elem = soup.select_one(selector)
|
||||
if elem:
|
||||
return elem.get(attr, '').strip()
|
||||
except:
|
||||
pass
|
||||
return ''
|
||||
|
||||
def generate_markdown(self, data, output_path):
|
||||
"""生成Markdown笔记"""
|
||||
if not data or not data.get('video_info'):
|
||||
print("没有视频数据可生成笔记")
|
||||
return False
|
||||
|
||||
info = data['video_info']
|
||||
subtitles = data.get('subtitles', [])
|
||||
|
||||
md_content = []
|
||||
md_content.append(f"# {info.get('title', '未知标题')}\n")
|
||||
md_content.append(f"**视频链接**: https://www.bilibili.com/video/{info.get('bvid', '')}\n")
|
||||
md_content.append("---\n\n")
|
||||
|
||||
# 基本信息
|
||||
md_content.append("## 基本信息\n\n")
|
||||
md_content.append(f"- **标题**: {info.get('title', '未知')}\n")
|
||||
md_content.append(f"- **UP主**: {info.get('author', '未知')}\n")
|
||||
md_content.append(f"- **BV号**: {info.get('bvid', '未知')}\n")
|
||||
md_content.append(f"- **时长**: {info.get('duration', '未知')}\n")
|
||||
md_content.append(f"- **发布时间**: {info.get('pubtime', '未知')}\n")
|
||||
|
||||
# 统计数据
|
||||
md_content.append("\n## 数据统计\n\n")
|
||||
md_content.append(f"- **播放量**: {info.get('view', 0):,}\n")
|
||||
md_content.append(f"- **点赞**: {info.get('like', 0):,}\n")
|
||||
md_content.append(f"- **投币**: {info.get('coin', 0):,}\n")
|
||||
md_content.append(f"- **收藏**: {info.get('favorite', 0):,}\n")
|
||||
md_content.append(f"- **分享**: {info.get('share', 0):,}\n")
|
||||
md_content.append(f"- **弹幕**: {info.get('danmaku', 0):,}\n")
|
||||
md_content.append(f"- **评论**: {info.get('reply', 0):,}\n")
|
||||
|
||||
# 标签
|
||||
if info.get('tags'):
|
||||
md_content.append(f"\n## 标签\n\n{info['tags']}\n")
|
||||
|
||||
# 简介
|
||||
if info.get('desc'):
|
||||
md_content.append(f"\n## 简介\n\n{info['desc']}\n")
|
||||
|
||||
# 字幕
|
||||
if subtitles:
|
||||
md_content.append("\n## 视频字幕\n\n")
|
||||
for sub in subtitles[:100]: # 限制字幕数量
|
||||
md_content.append(f"**{sub['from']} - {sub['to']}**: {sub['content']}\n\n")
|
||||
if len(subtitles) > 100:
|
||||
md_content.append(f"\n*(仅显示前100条字幕，共{len(subtitles)}条)*\n")
|
||||
|
||||
md_content.append("\n---\n")
|
||||
md_content.append(f"*笔记生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*\n")
|
||||
|
||||
# 写入文件
|
||||
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(output_path, 'w', encoding='utf-8') as f:
|
||||
f.writelines(md_content)
|
||||
|
||||
print(f"\n✅ Markdown笔记已保存到: {output_path}")
|
||||
return True
|
||||
|
||||
async def close(self):
|
||||
"""关闭浏览器"""
|
||||
if self.playwright:
|
||||
await self.playwright.stop()
|
||||
print("浏览器已关闭")
|
||||
|
||||
|
||||
async def main():
|
||||
"""主函数"""
|
||||
bvid = "BV11t64BJEDZ"
|
||||
output_file = "/home/admin/.openclaw/workspace/notes/时代真变了glm4.7+playwright-cli-AI自主进行自动化测试.md"
|
||||
|
||||
# 创建提取器（True=无头模式，False=显示浏览器窗口）
|
||||
# 无头模式更快，但如果遇到验证码可能无法处理
|
||||
extractor = BilibiliExtractor(headless=True)
|
||||
|
||||
try:
|
||||
await extractor.init_browser()
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"开始提取视频: {bvid}")
|
||||
print(f"{'='*60}")
|
||||
|
||||
data = await extractor.extract_video_info(bvid)
|
||||
|
||||
if data and data.get('video_info'):
|
||||
info = data['video_info']
|
||||
print(f"\n{'='*60}")
|
||||
print("视频信息提取成功！")
|
||||
print(f"{'='*60}")
|
||||
print(f"标题: {info.get('title', '未知')}")
|
||||
print(f"UP主: {info.get('author', '未知')}")
|
||||
print(f"时长: {info.get('duration', '未知')}")
|
||||
print(f"播放量: {info.get('view', 0):,}")
|
||||
print(f"点赞: {info.get('like', 0):,}")
|
||||
|
||||
# 生成Markdown笔记
|
||||
extractor.generate_markdown(data, output_file)
|
||||
|
||||
# 打印预览
|
||||
print(f"\n{'='*60}")
|
||||
print("Markdown笔记预览 (前50行):")
|
||||
print(f"{'='*60}")
|
||||
try:
|
||||
with open(output_file, 'r', encoding='utf-8') as f:
|
||||
lines = f.readlines()
|
||||
for line in lines[:50]:
|
||||
print(line.rstrip())
|
||||
if len(lines) > 50:
|
||||
print(f"\n... (共 {len(lines)} 行)")
|
||||
except Exception as e:
|
||||
print(f"无法读取文件: {e}")
|
||||
else:
|
||||
print("\n❌ 提取视频信息失败")
|
||||
print("可能原因：")
|
||||
print("1. 视频不存在或已被删除")
|
||||
print("2. 触发了验证码且未完成")
|
||||
print("3. 网络问题")
|
||||
print("\n建议：")
|
||||
print("- 检查视频链接是否正确")
|
||||
print("- 如果出现验证码，请手动完成")
|
||||
print("- 尝试重新运行脚本")
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n\n⏹️ 用户中断")
|
||||
except Exception as e:
|
||||
print(f"\n❌ 发生错误: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
finally:
|
||||
await extractor.close()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user