openclaw-skills/skills/kb-zhihu-metadata/zhihu_metadata.py

#!/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()