From 193d2b375d8c7b49bd10992085e0f972652b4f3a Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 27 Feb 2026 17:42:21 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=20MediaCrawler=20?= =?UTF-8?q?=E6=95=B4=E5=90=88=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - kb-zhihu-metadata: 新增 zhihu_article_crawler.py 支持爬取知乎专栏文章 - kb-zhihu-content: 添加 article-from-media 类型支持从 MediaCrawler 数据读取 - 更新 SKILL.md 文档说明新功能 --- kb-zhihu-content/SKILL.md | 64 ++++-- kb-zhihu-content/zhihu_content.py | 126 +++++++++++- kb-zhihu-metadata/SKILL.md | 41 +++- kb-zhihu-metadata/zhihu_article_crawler.py | 215 +++++++++++++++++++++ 4 files changed, 423 insertions(+), 23 deletions(-) create mode 100644 kb-zhihu-metadata/zhihu_article_crawler.py diff --git a/kb-zhihu-content/SKILL.md b/kb-zhihu-content/SKILL.md index 8d5151f..9009c3e 100644 --- a/kb-zhihu-content/SKILL.md +++ b/kb-zhihu-content/SKILL.md @@ -6,6 +6,7 @@ - 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 @@ -22,20 +23,60 @@ python zhihu_content.py --id "12345678" --type video --output "/path/to/download # 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/" ``` -## Output +## 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": "video", - "filename": "video_title.mp4", - "filepath": "/path/to/downloads/video_title.mp4", - "message": "下载完成: video_title.mp4" + "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 ` +4. Cleans up temporary file + +**Note**: 知乎视频通常没有官方字幕，将自动触发 Whisper 语音识别 + ## Installation ```bash @@ -43,16 +84,3 @@ python zhihu_content.py --id "12345678" --type subtitle --output "/path/to/subti # Ensure yt-dlp is installed pip install yt-dlp aiohttp ``` - -## How It Works - -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 ` -4. Cleans up temporary file - -## Notes - -- 知乎视频通常没有官方字幕，将自动触发 Whisper 语音识别 -- 视频下载需要能访问知乎（可能需要代理） diff --git a/kb-zhihu-content/zhihu_content.py b/kb-zhihu-content/zhihu_content.py index e909198..6e85292 100644 --- a/kb-zhihu-content/zhihu_content.py +++ b/kb-zhihu-content/zhihu_content.py @@ -5,6 +5,7 @@ 支持： - zvideo: 知乎视频下载 +- article-from-media: 从 MediaCrawler 数据读取知乎文章 - 从页面提取视频播放地址 """ @@ -13,6 +14,7 @@ import asyncio import json import os import re +import shutil import subprocess import sys import tempfile @@ -21,6 +23,10 @@ from pathlib import Path from typing import Optional, Dict, Any, List +# MediaCrawler 目录 +MEDIACRAWLER_DIR = Path.home() / "work" / "MediaCrawler" + + class ZhihuContent: """知乎内容获取器""" @@ -146,13 +152,113 @@ class ZhihuContent: 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', required=True, help='知乎视频ID') - parser.add_argument('--type', '-t', required=True, choices=['video', 'subtitle'], help='内容类型') + parser.add_argument('--id', '-i', help='知乎视频ID (用于 video/subtitle 类型)') + parser.add_argument('--type', '-t', required=True, + choices=['video', 'subtitle', 'article-from-media'], + help='内容类型: video=视频下载, subtitle=字幕提取, article-from-media=从MediaCrawler读取文章') 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() @@ -161,8 +267,16 @@ async def main(): if args.type == 'video': result = await downloader.download_video(args.id, quality=args.quality) - else: + 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 + ) + else: + result = {"success": False, "message": f"不支持的类型: {args.type}"} # 统一输出格式 if result.get("success"): @@ -170,10 +284,14 @@ async def main(): "success": True, "type": args.type, "has_subtitle": result.get("has_subtitle", False), - "filename": Path(result.get('filepath', '')).name if result.get('filepath') else None, + "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, diff --git a/kb-zhihu-metadata/SKILL.md b/kb-zhihu-metadata/SKILL.md index 9b1a9cb..789e29c 100644 --- a/kb-zhihu-metadata/SKILL.md +++ b/kb-zhihu-metadata/SKILL.md @@ -1,12 +1,13 @@ # KB Zhihu Metadata Skill -**kb-zhihu-metadata** - Extract Zhihu question/answer/user metadata. +**kb-zhihu-metadata** - Extract Zhihu question/answer/user metadata and crawl articles. ## Features - Question metadata extraction - Answer metadata extraction - User information extraction +- **Article crawling using MediaCrawler** (NEW!) - JSON output format - Markdown conversion support - **Cookie 支持**：可配置 ZHIHU_COOKIE 访问需要登录的内容 @@ -36,8 +37,46 @@ python zhihu_metadata.py --id "266309957" --type question # Answer metadata python zhihu_metadata.py --id "2003427891925509507" --type answer + +# 专栏文章爬取 (使用 MediaCrawler) ⚡ NEW! +python zhihu_article_crawler.py --url "https://zhuanlan.zhihu.com/p/1894054073625461813" --output "~/work/knowledge-base/data/zhihu/" ``` +## MediaCrawler 爬取文章 + +**新功能**：使用 MediaCrawler 爬取知乎专栏文章 + +### 环境要求 + +MediaCrawler 已安装在 `~/work/MediaCrawler` + +### 使用方法 + +```bash +# 爬取单篇文章 +python zhihu_article_crawler.py --url "https://zhuanlan.zhihu.com/p/XXXXX" + +# 指定输出目录 +python zhihu_article_crawler.py --url "https://zhuanlan.zhihu.com/p/XXXXX" --output "/path/to/output/" + +# 输出示例 +{ + "success": true, + "type": "article", + "title": "如何高效的爬取微信公众号", + "filepath": "/home/xxx/work/knowledge-base/data/zhihu/zhihu_article_如何高效的爬取微信公众号.md", + "content_length": 4563, + "message": "爬取成功: 如何高效的爬取微信公众号" +} +``` + +### 工作原理 + +1. 调用 MediaCrawler 爬取文章 +2. 从 JSON 输出中提取内容 +3. 转换为 Markdown 格式（带 frontmatter） +4. 保存到指定目录 + ## Output ```json diff --git a/kb-zhihu-metadata/zhihu_article_crawler.py b/kb-zhihu-metadata/zhihu_article_crawler.py new file mode 100644 index 0000000..8498cf0 --- /dev/null +++ b/kb-zhihu-metadata/zhihu_article_crawler.py @@ -0,0 +1,215 @@ +#!/usr/bin/env python3 +""" +知乎文章爬取脚本 - 基于 MediaCrawler +用于爬取知乎专栏文章并导出到知识库格式 +""" +import argparse +import json +import os +import sys +import subprocess +import shutil +from pathlib import Path +from datetime import datetime + +# 配置 +MEDIACRAWLER_DIR = Path.home() / "work" / "MediaCrawler" +DEFAULT_OUTPUT_DIR = Path.home() / "work" / "knowledge-base" / "data" / "zhihu" + + +def load_env(): + """加载环境变量""" + env_file = Path.home() / ".kb_env" + if env_file.exists(): + with open(env_file) as f: + for line in f: + line = line.strip() + if line and not line.startswith("#") and "=" in line: + key, val = line.split("=", 1) + os.environ[key.strip()] = val.strip() + + +def get_zhihu_cookie(): + """获取知乎 Cookie""" + return os.environ.get("ZHIHU_COOKIE", "") + + +def crawl_zhihu_article(article_url: str, output_dir: Path) -> dict: + """ + 使用 MediaCrawler 爬取知乎文章 + + Args: + article_url: 知乎文章链接 + output_dir: 输出目录 + + Returns: + dict: 爬取结果 + """ + # 清理旧数据 + data_dir = MEDIACRAWLER_DIR / "data" / "zhihu" / "json" + data_dir.mkdir(parents=True, exist_ok=True) + + # 备份现有数据 + backup_dir = data_dir / "backup" + backup_dir.mkdir(exist_ok=True) + for f in data_dir.glob("detail_contents_*.json"): + shutil.move(str(f), str(backup_dir / f.name)) + + # 配置 MediaCrawler + config_file = MEDIACRAWLER_DIR / "config" / "zhihu_config.py" + + # 读取配置文件并修改 + with open(config_file) as f: + config_content = f.read() + + # 检查是否已配置指定 ID + if "ZHIHU_SPECIFIED_ID_LIST" not in config_content or "[]" in config_content: + # 需要配置文章 URL + article_id = article_url.split("/")[-1] + new_config = config_content.replace( + 'ZHIHU_SPECIFIED_ID_LIST = []', + f'ZHIHU_SPECIFIED_ID_LIST = ["{article_url}"]' + ) + with open(config_file, "w") as f: + f.write(new_config) + + # 运行 MediaCrawler + env = os.environ.copy() + cookie = get_zhihu_cookie() + if cookie: + env["ZHIHU_COOKIE"] = cookie + + cmd = [ + "python3", "main.py", + "--platform", "zhihu", + "--lt", "cookie", + "--type", "detail" + ] + + print(f"运行命令: {' '.join(cmd)}") + print(f"工作目录: {MEDIACRAWLER_DIR}") + + result = subprocess.run( + cmd, + cwd=MEDIACRAWLER_DIR, + env=env, + capture_output=True, + text=True, + timeout=120 + ) + + print(f"返回码: {result.returncode}") + if result.returncode != 0: + print(f"stderr: {result.stderr[:500]}") + + # 查找最新的输出文件 + output_files = sorted(data_dir.glob("detail_contents_*.json"), reverse=True) + + if output_files: + latest_file = output_files[0] + with open(latest_file) as f: + data = json.load(f) + + if data: + item = data[0] + return { + "success": True, + "title": item.get("title", ""), + "content": item.get("content_text", ""), + "url": item.get("content_url", ""), + "author": item.get("user_nickname", ""), + "created_time": item.get("created_time", 0), + "voteup": item.get("voteup_count", 0), + "comments": item.get("comment_count", 0), + "content_type": item.get("content_type", "article"), + } + + return {"success": False, "error": "未获取到数据"} + + +def save_to_markdown(data: dict, output_path: Path) -> str: + """保存为 Markdown 格式""" + created = datetime.fromtimestamp(data["created_time"]).strftime("%Y-%m-%d %H:%M:%S") if data.get("created_time") else "" + + content = f"""--- +title: "{data['title']}" +author: "{data['author']}" +url: "{data['url']}" +created: "{created}" +voteup: {data.get('voteup', 0)} +comments: {data.get('comments', 0)} +platform: zhihu +type: article +--- + +# {data['title']} + +**作者**: {data['author']} +**来源**: [知乎专栏]({data['url']}) +**发布时间**: {created} +**点赞**: {data.get('voteup', 0)} +**评论**: {data.get('comments', 0)} + +--- + +{data['content']} +""" + + with open(output_path, "w", encoding="utf-8") as f: + f.write(content) + + return str(output_path) + + +def main(): + parser = argparse.ArgumentParser(description="知乎文章爬取工具 (基于 MediaCrawler)") + parser.add_argument("--url", "-u", required=True, help="知乎文章链接") + parser.add_argument("--output", "-o", default=str(DEFAULT_OUTPUT_DIR), help="输出目录") + + args = parser.parse_args() + + # 加载环境变量 + load_env() + + # 确保输出目录存在 + output_dir = Path(args.output) + output_dir.mkdir(parents=True, exist_ok=True) + + print(f"目标文章: {args.url}") + print(f"输出目录: {output_dir}") + + # 爬取文章 + print("\n开始爬取...") + result = crawl_zhihu_article(args.url, output_dir) + + if not result.get("success"): + print(f"爬取失败: {result.get('error', '未知错误')}") + sys.exit(1) + + print(f"\n爬取成功!") + print(f"标题: {result['title']}") + print(f"内容长度: {len(result['content'])} 字符") + + # 生成文件名 + safe_title = "".join(c for c in result['title'] if c.isalnum() or c in " -_")[:50] + filename = f"zhihu_article_{safe_title}.md" + output_path = output_dir / filename + + # 保存为 Markdown + save_to_markdown(result, output_path) + print(f"\n已保存到: {output_path}") + + # 输出 JSON 格式 + print("\n--- JSON 输出 ---") + print(json.dumps({ + "success": True, + "type": "article", + "title": result["title"], + "filepath": str(output_path), + "content_length": len(result["content"]), + "message": f"爬取成功: {result['title']}" + }, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main()