starmen1265
3909bb0321
PR Test / lint (pull_request) Successful in 1s
- zhihu_content.py 添加 zhuanlan/article 类型支持 - 内部自动调用 zhihu_article_crawler.py 爬取新数据 - 更新 SKILL.md 文档，明确知乎专栏处理流程 - 解决读取旧数据导致分析错误内容的问题
366 lines
14 KiB
Python
366 lines
14 KiB
Python
#!/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)
|