starmen1265/openclaw
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Auto backup before evolver run - 2026年 03月 10日 星期二 19:00:50 CST

Browse Source
This commit is contained in:
starmen1265
2026-03-10 19:00:50 +08:00
parent c69fcd1ace
commit 803bf3a6b6
101 changed files with 15557 additions and 868 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/kb-zhihu-content/SKILL.md
+86
View File
@@ -0,0 +1,86 @@
# KB Zhihu Content Skill
**kb-zhihu-content** - Download Zhihu videos and extract content.
## Features
- Video download using **yt-dlp**
- Subtitle extraction (returns no subtitle, triggers Whisper)
- **Read article content from MediaCrawler data** (NEW!)
- Supports `ZHIHU_COOKIE` environment variable
## Environment Variables
| Variable | Required | Description |
|----------|----------|-------------|
| `ZHIHU_COOKIE` | No | 知乎登录 Cookie (for member-only videos) |
## Commands
```bash
# Download video
python zhihu_content.py --id "12345678" --type video --output "/path/to/downloads/"
# Extract subtitle (returns no subtitle, triggers Whisper)
python zhihu_content.py --id "12345678" --type subtitle --output "/path/to/subtitles/"
# 从 MediaCrawler 读取文章内容 (⚡ NEW!)
python zhihu_content.py --type article-from-media --input "~/work/MediaCrawler/data/zhihu/json/detail_contents_2026-02-27.json" --output "~/work/knowledge-base/data/zhihu/"
```
## MediaCrawler 数据读取
**新功能**：从 MediaCrawler 爬取的数据中提取知乎文章内容
### 使用方法
```bash
# 从 MediaCrawler JSON 文件读取文章内容
python zhihu_content.py --type article-from-media \
--input "~/work/MediaCrawler/data/zhihu/json/detail_contents_2026-02-27.json" \
--output "~/work/knowledge-base/data/zhihu/"
# 查找最新的爬取数据
python zhihu_content.py --type article-from-media \
--output "~/work/knowledge-base/data/zhihu/"
```
### 参数说明
| 参数 | 说明 |
|------|------|
| `--type article-from-media` | 从 MediaCrawler 数据读取 |
| `--input` | MediaCrawler 输出文件路径（可选，不指定则自动查找最新） |
| `--output` | 输出目录 |
### 输出格式
```json
{
"success": true,
"type": "article",
"filename": "zhihu_article_如何高效的爬取微信公众号.md",
"filepath": "/home/xxx/work/knowledge-base/data/zhihu/zhihu_article_如何高效的爬取微信公众号.md",
"message": "文章内容已导出"
}
```
## 原有功能
### Video Download
Video download uses **yt-dlp**:
1. Parses `ZHIHU_COOKIE` environment variable (optional)
2. Creates temporary Netscape-format cookie file
3. Runs `yt-dlp -f best -o <output> <url>`
4. Cleans up temporary file
**Note**: 知乎视频通常没有官方字幕，将自动触发 Whisper 语音识别
## Installation
```bash
# Already included in kb-skills repo
# Ensure yt-dlp is installed
pip install yt-dlp aiohttp
```
skills/kb-zhihu-content/zhihu_content.py
+365
View File
@@ -0,0 +1,365 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
知乎内容获取工具
支持：
- zvideo: 知乎视频下载
- article-from-media: 从 MediaCrawler 数据读取知乎文章
- 从页面提取视频播放地址
"""
import argparse
import asyncio
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Optional, Dict, Any, List
# MediaCrawler 目录
MEDIACRAWLER_DIR = Path.home() / "work" / "MediaCrawler"
class ZhihuContent:
"""知乎内容获取器"""
def __init__(self, output_dir: str = "./output", cookie: str = None):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.cookie = cookie
def _create_cookie_file(self) -> str:
"""创建 Netscape 格式的 Cookie 文件"""
if not self.cookie:
return None
cookies = {}
for item in self.cookie.split(';'):
item = item.strip()
if '=' in item:
key, value = item.split('=', 1)
cookies[key.strip()] = value.strip()
cookie_content = """# Netscape HTTP Cookie File
# Generated by zhihu_content.py
"""
for name, value in cookies.items():
cookie_content += f".zhihu.com\tTRUE\t/\tFALSE\t0\t{name}\t{value}\n"
fd, path = tempfile.mkstemp(suffix='.txt')
with os.fdopen(fd, 'w') as f:
f.write(cookie_content)
return path
async def download_video(self, video_id: str, quality: int = 720) -> Dict[str, Any]:
"""下载知乎视频（使用 yt-dlp）"""
result = {"success": False, "video_id": video_id, "message": "", "filepath": None, "metadata": {}}
try:
print(f"获取知乎视频信息: {video_id}")
cookie_file = self._create_cookie_file()
url = f"https://www.zhihu.com/zvideo/{video_id}"
# 先获取视频标题
try:
if cookie_file:
cmd = ["yt-dlp", "--cookies", cookie_file, "--get-title", url]
else:
cmd = ["yt-dlp", "--get-title", url]
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await proc.communicate()
title = stdout.decode('utf-8').strip()
if not title or 'error' in title.lower():
title = f"zhihu_video_{video_id}"
except Exception as e:
title = f"zhihu_video_{video_id}"
# 构建 yt-dlp 命令
safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_')).strip()[:50]
print(f"下载视频: {title}")
output_template = str(self.output_dir / f"{safe_title}.%(ext)s")
# 构建下载命令
cmd = [
"yt-dlp",
"-f", "bestvideo+bestaudio/best",
"-o", output_template,
"--no-download-archive",
url
]
if cookie_file:
cmd.extend(["--cookies", cookie_file])
print(f"执行命令: {' '.join(cmd[:4])}...")
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await proc.communicate()
# 查找下载的文件
downloaded_files = list(self.output_dir.glob(f"{safe_title}.*"))
if downloaded_files:
video_file = downloaded_files[0]
result["success"] = True
result["filepath"] = str(video_file)
result["message"] = f"下载完成: {video_file.name}"
result["metadata"] = {"title": title, "video_id": video_id}
print(f"✅ {result['message']}")
else:
result["message"] = "未找到下载的文件"
print(f"❌ {result['message']}")
# 清理临时 Cookie 文件
if cookie_file and os.path.exists(cookie_file):
os.remove(cookie_file)
except Exception as e:
result["message"] = f"错误: {str(e)}"
print(f"❌ {result['message']}")
return result
async def extract_subtitle(self, video_id: str) -> Dict[str, Any]:
"""
提取字幕
知乎视频通常没有官方字幕，需要使用 Whisper 进行语音识别。
此函数返回 has_subtitle=False，让工作流继续执行 Whisper 步骤。
"""
result = {"success": True, "video_id": video_id, "has_subtitle": False, "message": "知乎视频无官方字幕，将使用 Whisper 进行语音识别"}
return result
def read_article_from_mediacrawler(self, input_file: str = None, output_dir: str = None) -> Dict[str, Any]:
"""
从 MediaCrawler 爬取的数据中读取知乎文章内容
Args:
input_file: MediaCrawler 输出文件路径，不指定则自动查找最新
output_dir: 输出目录
Returns:
处理结果字典
"""
# 确定输出目录
if output_dir:
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
# 查找数据文件
data_dir = MEDIACRAWLER_DIR / "data" / "zhihu" / "json"
if input_file:
# 使用指定的文件
json_file = Path(input_file)
if not json_file.exists():
return {"success": False, "message": f"文件不存在: {input_file}"}
else:
# 查找最新的数据文件
json_files = sorted(data_dir.glob("detail_contents_*.json"), reverse=True)
if not json_files:
return {"success": False, "message": "未找到 MediaCrawler 爬取的数据，请先运行爬虫"}
json_file = json_files[0]
try:
with open(json_file, encoding="utf-8") as f:
data = json.load(f)
if not data:
return {"success": False, "message": "数据文件为空"}
# 提取第一篇文章
item = data[0]
# 准备数据
title = item.get("title", "无标题")
content = item.get("content_text", "")
url = item.get("content_url", "")
author = item.get("user_nickname", "未知")
created_ts = item.get("created_time", 0)
created = datetime.fromtimestamp(created_ts).strftime("%Y-%m-%d %H:%M:%S") if created_ts else ""
voteup = item.get("voteup_count", 0)
comments = item.get("comment_count", 0)
# 生成文件名
safe_title = "".join(c for c in title if c.isalnum() or c in " -_")[:50]
filename = f"zhihu_article_{safe_title}.md"
output_path = self.output_dir / filename
# 写入 Markdown 文件
markdown_content = f"""---
title: "{title}"
author: "{author}"
url: "{url}"
created: "{created}"
voteup: {voteup}
comments: {comments}
platform: zhihu
type: article
---
# {title}
**作者**: {author}
**来源**: [知乎]({url})
**发布时间**: {created}
**点赞**: {voteup}
**评论**: {comments}
---
{content}
"""
with open(output_path, "w", encoding="utf-8") as f:
f.write(markdown_content)
return {
"success": True,
"type": "article",
"title": title,
"filename": filename,
"filepath": str(output_path),
"content_length": len(content),
"message": f"文章内容已导出: {title}"
}
except Exception as e:
return {"success": False, "message": f"读取失败: {str(e)}"}
async def main():
parser = argparse.ArgumentParser(description='知乎内容获取工具')
parser.add_argument('--id', '-i', help='知乎视频ID (用于 video/subtitle 类型)')
parser.add_argument('--type', '-t', required=True,
choices=['video', 'subtitle', 'article-from-media', 'zhuanlan', 'article'],
help='内容类型: video=视频下载, subtitle=字幕提取, article-from-media=从MediaCrawler读取文章, zhuanlan/article=爬取知乎专栏文章')
parser.add_argument('--output', '-o', default='./output', help='输出目录')
parser.add_argument('--quality', '-q', type=int, default=720, help='视频质量')
parser.add_argument('--input', '-in', help='MediaCrawler 数据文件路径 (用于 article-from-media 类型)')
args = parser.parse_args()
cookie = os.environ.get('ZHIHU_COOKIE')
downloader = ZhihuContent(output_dir=args.output, cookie=cookie)
if args.type == 'video':
result = await downloader.download_video(args.id, quality=args.quality)
elif args.type == 'subtitle':
result = await downloader.extract_subtitle(args.id)
elif args.type == 'article-from-media':
# 同步调用，不需要 asyncio
result = downloader.read_article_from_mediacrawler(
input_file=args.input,
output_dir=args.output
)
elif args.type in ('zhuanlan', 'article'):
# 知乎专栏文章：先爬取再读取
# 需要从 --id 或构造完整 URL
if args.id:
# 如果传入的是完整 URL
if args.id.startswith('http'):
article_url = args.id
else:
# 如果只传入 ID，构造 URL
article_url = f"https://zhuanlan.zhihu.com/p/{args.id}"
else:
result = {"success": False, "message": "zhuanlan/article 类型需要 --id 参数（文章ID或完整URL）"}
print(json.dumps(result, ensure_ascii=False, indent=2))
return result
# 调用 zhihu_article_crawler.py 爬取
crawler_script = Path(__file__).parent.parent / "kb-zhihu-metadata" / "zhihu_article_crawler.py"
if not crawler_script.exists():
# 尝试另一种路径
crawler_script = Path.home() / ".openclaw" / "agents" / "knowledge-base" / "workspace" / "skills" / "kb-zhihu-metadata" / "zhihu_article_crawler.py"
if crawler_script.exists():
import subprocess
cmd = ["python3", str(crawler_script), "--url", article_url, "--output", args.output]
print(f"调用爬虫: {' '.join(cmd)}")
proc = subprocess.run(cmd, capture_output=True, text=True)
if proc.returncode != 0:
result = {"success": False, "message": f"爬取失败: {proc.stderr[:200]}"}
else:
# 解析爬虫输出获取文件路径
try:
crawler_output = json.loads(proc.stdout.split("--- JSON 输出 ---")[-1].strip())
if crawler_output.get("success"):
# 直接使用爬虫输出的文件
result = {
"success": True,
"type": "article",
"title": crawler_output.get("title"),
"filename": Path(crawler_output["filepath"]).name,
"filepath": crawler_output["filepath"],
"content_length": crawler_output.get("content_length"),
"message": f"知乎专栏文章已爬取: {crawler_output.get('title')}"
}
else:
result = {"success": False, "message": crawler_output.get("error", "爬取失败")}
except Exception as e:
result = {"success": False, "message": f"解析爬虫输出失败: {str(e)}"}
else:
result = {"success": False, "message": f"未找到 zhihu_article_crawler.py: {crawler_script}"}
else:
result = {"success": False, "message": f"不支持的类型: {args.type}"}
# 统一输出格式
if result.get("success"):
output = {
"success": True,
"type": args.type,
"has_subtitle": result.get("has_subtitle", False),
"filename": result.get('filename') or (Path(result.get('filepath', '')).name if result.get('filepath') else None),
"filepath": str(Path(result.get('filepath', '')).resolve()) if result.get('filepath') else None,
"message": result.get('message', '处理成功')
}
if result.get('title'):
output['title'] = result['title']
if result.get('content_length'):
output['content_length'] = result['content_length']
else:
output = {
"success": False,
"error": result.get('message', '处理失败'),
"detail": f"ID: {args.id}, 类型: {args.type}"
}
print(json.dumps(output, ensure_ascii=False, indent=2))
return result
if __name__ == '__main__':
try:
result = asyncio.run(main())
sys.exit(0 if result.get("success") else 1)
except KeyboardInterrupt:
print(json.dumps({"success": False, "error": "用户取消"}, ensure_ascii=False, indent=2))
sys.exit(1)
except Exception as e:
print(json.dumps({"success": False, "error": str(e)}, ensure_ascii=False, indent=2))
sys.exit(1)