feat: Add KB knowledge base series skills

- kb-link-parser: Multi-platform URL parser (Bilibili, YouTube, Zhihu)
- kb-bilibili-metadata: Bilibili video/article/up主 metadata extractor
- kb-zhihu-metadata: Zhihu question/answer/user metadata extractor
- kb-bilibili-content: Bilibili video download and content extraction
- kb-ocr-processor: OCR text extraction from video frames
- kb-deep-analyzer: AI-powered deep content analysis
- kb-workflow: Workflow orchestration (inbox, archive, subscriptions, report)

Also add:
- ARCHITECTURE_FLOWCHART.md: System architecture diagrams
- DETAILED_DESIGN.md: Complete design documentation
This commit is contained in:
OpenClaw Agent
2026-02-10 12:18:52 +08:00
parent 2f5a269866
commit 5507b1d986
28 changed files with 7715 additions and 0 deletions
+44
View File
@@ -0,0 +1,44 @@
# KB Zhihu Metadata Skill
**kb-zhihu-metadata** - Extract Zhihu question/answer/user metadata.
## Features
- Question metadata extraction
- Answer metadata extraction
- User information extraction
- JSON output format
- Markdown conversion support
## Commands
```bash
# Question metadata
python zhihu_metadata.py --id "266309957" --type question
# Answer metadata
python zhihu_metadata.py --id "2003427891925509507" --type answer
```
## Output
```json
{
"success": true,
"data": {
"platform": "zhihu",
"type": "question",
"id": "266309957",
"title": "问题标题",
"author": "作者名字",
"answer_count": 50,
"followers": 1000
}
}
```
## Installation
```bash
cp -r kb-zhihu-metadata/ ~/skills/
```
@@ -0,0 +1,17 @@
from markdownify import MarkdownConverter as _MarkdownConverter
class MarkdownConverter:
"""HTML 转 Markdown 转换器,保留完整格式"""
def __init__(self):
self.converter = _MarkdownConverter(
heading_style="ATX",
bullets="*,-",
code_blocks=True,
tables=True,
strip=["script", "style", "nav", "footer", "aside"]
)
def convert(self, html: str) -> str:
"""转换 HTML 为 Markdown"""
return self.converter.convert(html)
+385
View File
@@ -0,0 +1,385 @@
"""
知乎 Playwright 内容提取器
使用 Playwright 浏览器自动化工具抓取知乎内容
"""
import re
import sys
import os
from typing import Dict, Optional, Any
from urllib.parse import urlparse
# 添加当前目录到 Python 路径以支持绝对导入
current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
sys.path.insert(0, current_dir)
try:
from .markdown_converter import MarkdownConverter
except ImportError:
from markdown_converter import MarkdownConverter
class PlaywrightBrowser:
"""Playwright 浏览器管理器"""
def __init__(self, headless: bool = True, timeout: int = 30):
self.headless = headless
self.timeout = timeout * 1000
self.browser = None
self.context = None
self.page = None
def start(self):
"""启动浏览器"""
from playwright.sync_api import sync_playwright
self.playwright = sync_playwright().start()
self.browser = self.playwright.chromium.launch(headless=self.headless)
self.context = self.browser.new_context(
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'
)
self.page = self.context.new_page()
self.page.set_default_timeout(self.timeout)
def close(self):
"""关闭浏览器"""
if self.page:
self.page.close()
if self.context:
self.context.close()
if self.browser:
self.browser.close()
if hasattr(self, 'playwright'):
self.playwright.stop()
def goto(self, url: str, wait_until: str = "networkidle"):
"""导航到指定URL"""
return self.page.goto(url, wait_until=wait_until)
def wait_for_selector(self, selector: str, timeout: int = None):
"""等待选择器出现"""
timeout_ms = timeout * 1000 if timeout else self.timeout
return self.page.wait_for_selector(selector, timeout=timeout_ms)
def query_selector(self, selector: str):
"""查询选择器"""
return self.page.query_selector(selector)
def query_selector_all(self, selector: str):
"""查询所有匹配的选择器"""
return self.page.query_selector_all(selector)
def evaluate(self, script: str):
"""执行 JavaScript 代码"""
return self.page.evaluate(script)
class ZhihuPlaywrightExtractor:
"""知乎内容提取器 - 使用 Playwright"""
URL_PATTERNS = {
'answer': r'zhihu\.com/question/\d+/answer/(\d+)',
'article': r'zhuanlan\.zhihu\.com/p/(\d+)',
'question': r'zhihu\.com/question/(\d+)(?:$|/)',
}
def __init__(self, headless: bool = True, timeout: int = 30):
"""
初始化提取器
Args:
headless: 是否使用无头浏览器
timeout: 超时时间(秒)
"""
self.browser = PlaywrightBrowser(headless=headless, timeout=timeout)
self.markdown_converter = MarkdownConverter()
def identify_url_type(self, url: str) -> Optional[str]:
"""
识别 URL 类型
Returns:
'answer', 'article', 'question' 或 None
"""
for url_type, pattern in self.URL_PATTERNS.items():
if re.search(pattern, url):
return url_type
return None
def extract_id(self, url: str) -> Optional[str]:
"""
从 URL 中提取内容 ID
Returns:
内容 ID 或 None
"""
url_type = self.identify_url_type(url)
if url_type:
pattern = self.URL_PATTERNS[url_type]
match = re.search(pattern, url)
if match:
return match.group(1)
return None
def _extract_text_content(self, element) -> str:
"""提取元素的文本内容"""
if element is None:
return ''
return element.inner_text() if hasattr(element, 'inner_text') else ''
def _extract_html_content(self, element) -> str:
"""提取元素的 HTML 内容"""
if element is None:
return ''
return element.inner_html() if hasattr(element, 'inner_html') else ''
def _extract_answer_content(self) -> Dict[str, Any]:
"""提取回答内容"""
result = {
'type': 'answer',
'title': '',
'author': '',
'content': '',
'content_markdown': '',
'publish_time': '',
'vote_up': 0,
}
# 打印页面标题用于调试
page_title = self.browser.page.title()
print(f"页面标题: {page_title}")
# 尝试保存页面快照用于调试
try:
import os
debug_dir = '/home/admin/.openclaw/workspace/debug'
os.makedirs(debug_dir, exist_ok=True)
self.browser.page.screenshot(path=os.path.join(debug_dir, 'zhihu_screenshot.png'))
with open(os.path.join(debug_dir, 'zhihu_page.html'), 'w', encoding='utf-8') as f:
f.write(self.browser.page.content())
print(f"页面快照已保存到 {debug_dir}")
except Exception as e:
print(f"保存页面快照时出错: {e}")
# 尝试多个可能的选择器
selectors = {
'title': ['h1.QuestionHeader-title', 'h1', '.QuestionHeader-main'],
'author': ['.AuthorInfo-name', '.AuthorInfo .Text', '[itemprop="name"]'],
'content': ['.RichContent-inner', '.RichContent', '.List-itemText', '[data-zop-feedlist] article'],
'time': ['.ContentItem-time', '[itemprop="dateCreated"]'],
'vote': ['.VoteButton--up .VoteButton-label', '.VoteButton--up']
}
for selector in selectors['title']:
title_elem = self.browser.query_selector(selector)
if title_elem:
result['title'] = self._extract_text_content(title_elem)
print(f"找到标题(使用选择器 {selector}): {result['title'][:50]}...")
break
for selector in selectors['author']:
author_elem = self.browser.query_selector(selector)
if author_elem:
result['author'] = self._extract_text_content(author_elem)
print(f"找到作者(使用选择器 {selector}): {result['author']}")
break
for selector in selectors['content']:
content_elem = self.browser.query_selector(selector)
if content_elem:
html_content = self._extract_html_content(content_elem)
result['content'] = html_content
result['content_markdown'] = self.markdown_converter.convert(html_content)
print(f"找到内容(使用选择器 {selector}),长度: {len(result['content_markdown'])}")
break
for selector in selectors['time']:
time_elem = self.browser.query_selector(selector)
if time_elem:
result['publish_time'] = time_elem.get_attribute('data-tooltip') or self._extract_text_content(time_elem)
break
for selector in selectors['vote']:
vote_elem = self.browser.query_selector(selector)
if vote_elem:
vote_text = self._extract_text_content(vote_elem)
try:
result['vote_up'] = int(vote_text.replace(',', ''))
except (ValueError, AttributeError):
pass
break
return result
def _extract_article_content(self) -> Dict[str, Any]:
"""提取文章内容"""
result = {
'type': 'article',
'title': '',
'author': '',
'content': '',
'content_markdown': '',
'publish_time': '',
'vote_up': 0,
}
# 提取文章标题
title_elem = self.browser.query_selector('.Post-Title')
if title_elem:
result['title'] = self._extract_text_content(title_elem)
# 提取作者信息
author_elem = self.browser.query_selector('.AuthorInfo-name')
if author_elem:
result['author'] = self._extract_text_content(author_elem)
# 提取文章内容
content_elem = self.browser.query_selector('.Post-RichText')
if content_elem:
html_content = self._extract_html_content(content_elem)
result['content'] = html_content
result['content_markdown'] = self.markdown_converter.convert(html_content)
# 提取发布时间
time_elem = self.browser.query_selector('.ContentItem-time')
if time_elem:
result['publish_time'] = time_elem.get_attribute('data-tooltip') or self._extract_text_content(time_elem)
# 提取点赞数
vote_elem = self.browser.query_selector('.VoteButton--up .VoteButton-label')
if vote_elem:
vote_text = self._extract_text_content(vote_elem)
try:
result['vote_up'] = int(vote_text.replace(',', ''))
except (ValueError, AttributeError):
pass
return result
def _extract_question_content(self) -> Dict[str, Any]:
"""提取问题内容(问题页本身)"""
result = {
'type': 'question',
'title': '',
'author': '',
'content': '',
'content_markdown': '',
'publish_time': '',
'vote_up': 0,
}
# 提取问题标题
title_elem = self.browser.query_selector('h1.QuestionHeader-title')
if title_elem:
result['title'] = self._extract_text_content(title_elem)
# 提取问题详情
detail_elem = self.browser.query_selector('.QuestionHeader-detail')
if detail_elem:
html_content = self._extract_html_content(detail_elem)
result['content'] = html_content
result['content_markdown'] = self.markdown_converter.convert(html_content)
# 问题的作者通常是提问者
author_elem = self.browser.query_selector('.AuthorInfo-name')
if author_elem:
result['author'] = self._extract_text_content(author_elem)
return result
def extract(self, url: str) -> Optional[Dict[str, Any]]:
"""
从 URL 提取知乎内容
Args:
url: 知乎 URL
Returns:
包含提取内容的字典,结构如下:
{
'type': 'answer' | 'article' | 'question',
'title': str,
'author': str,
'content': str (HTML),
'content_markdown': str (Markdown),
'publish_time': str,
'vote_up': int,
}
如果提取失败返回 None
"""
try:
# 识别 URL 类型
url_type = self.identify_url_type(url)
if not url_type:
print(f"无法识别 URL 类型: {url}")
return None
print(f"URL 类型: {url_type}")
print(f"启动浏览器...")
# 启动浏览器
self.browser.start()
print(f"访问页面: {url}")
# 访问页面
self.browser.goto(url, wait_until="domcontentloaded")
print("等待页面加载...")
import time
time.sleep(3) # 等待 JavaScript 执行
# 尝试多个可能的选择器
if url_type == 'answer':
selectors = ['.RichContent-inner', '.RichContent', '.List-itemText', '[data-zop-feedlist]']
print(f"尝试查找选择器: {selectors}")
result = self._extract_answer_content()
elif url_type == 'article':
selectors = ['.Post-RichText', '.Post-Title']
print(f"尝试查找选择器: {selectors}")
result = self._extract_article_content()
else: # question
selectors = ['.QuestionHeader-title', 'h1']
print(f"尝试查找选择器: {selectors}")
result = self._extract_question_content()
# 添加额外信息
result['url'] = url
result['content_id'] = self.extract_id(url)
print(f"提取成功! 标题: {result['title'][:50]}...")
return result
except Exception as e:
print(f"提取知乎内容时出错: {e}")
import traceback
traceback.print_exc()
return None
finally:
# 关闭浏览器
self.browser.close()
def __enter__(self):
"""上下文管理器入口"""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""上下文管理器出口"""
self.browser.close()
return False
# 便捷函数
def extract_zhihu_content(url: str, headless: bool = True, timeout: int = 30) -> Optional[Dict[str, Any]]:
"""
便捷函数:提取知乎内容
Args:
url: 知乎 URL
headless: 是否使用无头浏览器
timeout: 超时时间(秒)
Returns:
提取的内容字典
"""
with ZhihuPlaywrightExtractor(headless=headless, timeout=timeout) as extractor:
return extractor.extract(url)
+258
View File
@@ -0,0 +1,258 @@
#!/usr/bin/env python3
"""
知乎元数据获取工具
支持获取知乎问题、回答、用户的元数据信息。
"""
import argparse
import json
import sys
import os
from typing import Dict, Optional, Any
# 添加当前目录到 Python 路径
current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
sys.path.insert(0, current_dir)
from zhihu_extractor import ZhihuPlaywrightExtractor
def extract_question(question_id: str, headless: bool = True) -> Dict[str, Any]:
"""
获取知乎问题元数据
Args:
question_id: 问题 ID (数字字符串)
headless: 是否使用无头浏览器
Returns:
问题元数据字典
"""
url = f"https://www.zhihu.com/question/{question_id}"
return _extract_metadata(url, "question", headless)
def extract_answer(answer_id: str, headless: bool = True) -> Dict[str, Any]:
"""
获取知乎回答元数据
Args:
answer_id: 回答 ID (数字字符串)
headless: 是否使用无头浏览器
Returns:
回答元数据字典
"""
# 回答 URL 格式: https://www.zhihu.com/question/{question_id}/answer/{answer_id}
# 需要先获取问题ID,这里简化处理,尝试从回答ID构建URL
# 实际上知乎回答URL需要问题ID,这里用模拟URL
url = f"https://www.zhihu.com/question/0/answer/{answer_id}"
return _extract_metadata(url, "answer", headless)
def extract_user(url_token: str, headless: bool = True) -> Dict[str, Any]:
"""
获取知乎用户信息
Args:
url_token: 用户 URL token
headless: 是否使用无头浏览器
Returns:
用户信息字典
"""
url = f"https://www.zhihu.com/people/{url_token}"
return _extract_user_metadata(url, headless)
def _extract_metadata(url: str, content_type: str, headless: bool) -> Dict[str, Any]:
"""
内部函数:提取元数据
Args:
url: 知乎 URL
content_type: 内容类型 (question/answer)
headless: 是否使用无头浏览器
Returns:
元数据字典
"""
extractor = ZhihuPlaywrightExtractor(headless=headless)
result = extractor.extract(url)
if result:
# 标准化输出格式
metadata = {
"type": content_type,
"id": result.get("content_id", ""),
"url": result.get("url", url),
"title": result.get("title", ""),
"author": result.get("author", ""),
"publish_time": result.get("publish_time", ""),
"vote_up": result.get("vote_up", 0),
"content_markdown": result.get("content_markdown", ""),
"raw_html": result.get("content", ""),
}
return metadata
else:
return {
"type": content_type,
"error": "Failed to extract content",
"url": url
}
def _extract_user_metadata(url: str, headless: bool) -> Dict[str, Any]:
"""
内部函数:提取用户信息
Args:
url: 用户主页 URL
headless: 是否使用无头浏览器
Returns:
用户信息字典
"""
extractor = ZhihuPlaywrightExtractor(headless=headless)
try:
extractor.browser.start()
extractor.browser.goto(url, wait_until="networkidle")
import time
time.sleep(3)
# 提取用户名
username = ""
username_elem = extractor.browser.query_selector('.ProfileHeader-name')
if username_elem:
username = username_elem.inner_text()
# 提取个人简介
headline = ""
headline_elem = extractor.browser.query_selector('.ProfileHeader-headline')
if headline_elem:
headline = headline_elem.inner_text()
# 提取简介详情
description = ""
desc_elem = extractor.browser.query_selector('.ProfileHeader-description')
if desc_elem:
description = desc_elem.inner_text()
# 提取关注者数
followers = 0
follower_elem = extractor.browser.query_selector('.FollowshipCard-followersNum')
if follower_elem:
follower_text = follower_elem.inner_text()
try:
followers = int(follower_text.replace(',', ''))
except ValueError:
pass
# 提取关注者数
following = 0
following_elem = extractor.browser.query_selector('.FollowshipCard-followingsNum')
if following_elem:
following_text = following_elem.inner_text()
try:
following = int(following_text.replace(',', ''))
except ValueError:
pass
# 提取回答数
answers = 0
answers_elem = extractor.browser.query_selector('.Profile-mainColumn .Tabs-item span')
if answers_elem:
answers_text = answers_elem.inner_text()
try:
answers = int(answers_text.replace(',', ''))
except ValueError:
pass
metadata = {
"type": "user",
"url": url,
"username": username,
"headline": headline,
"description": description,
"followers": followers,
"following": following,
"answers_count": answers,
}
return metadata
except Exception as e:
return {
"type": "user",
"error": str(e),
"url": url
}
finally:
extractor.browser.close()
def main():
"""命令行入口"""
parser = argparse.ArgumentParser(
description="知乎元数据获取工具",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
python zhihu_metadata.py --id "266309957" --type question
python zhihu_metadata.py --id "2003427891925509507" --type answer
python zhihu_metadata.py --id "example-user" --type user
"""
)
parser.add_argument(
"--id", "-i",
required=True,
help="知乎内容 ID (问题ID、回答ID或用户URL token)"
)
parser.add_argument(
"--type", "-t",
required=True,
choices=["question", "answer", "user"],
help="内容类型"
)
parser.add_argument(
"--output", "-o",
default=None,
help="输出文件路径 (默认: stdout)"
)
parser.add_argument(
"--headless",
action="store_true",
default=True,
help="使用无头浏览器模式 (默认: True)"
)
args = parser.parse_args()
# 提取元数据
if args.type == "question":
result = extract_question(args.id, headless=args.headless)
elif args.type == "answer":
result = extract_answer(args.id, headless=args.headless)
else: # user
result = extract_user(args.id, headless=args.headless)
# 输出结果
json_output = json.dumps(result, ensure_ascii=False, indent=2)
if args.output:
with open(args.output, 'w', encoding='utf-8') as f:
f.write(json_output)
print(f"结果已保存到: {args.output}")
else:
print(json_output)
if __name__ == "__main__":
main()