how/script/data_processing/extract_features_from_posts.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
从过去帖子解构结果目录中提取特征名称及其来源信息
仅支持新版数据结构（inspiration_final_result, purpose_final_result, keypoint_final）
"""

import json
from pathlib import Path
from typing import Dict, List, Optional, Set
import re
import sys

# 添加项目根目录到路径
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))

from script.detail import get_xiaohongshu_detail
from script.data_processing.path_config import PathConfig


def extract_post_id_from_filename(filename: str) -> str:
    """从文件名中提取帖子ID
    支持格式: 68a6b96f000000001d006058.json
    """
    return filename.replace('.json', '')


def get_post_detail(post_id: str) -> Optional[Dict]:
    """获取帖子详情"""
    try:
        detail = get_xiaohongshu_detail(post_id)
        return detail
    except Exception as e:
        print(f"  警告: 获取帖子 {post_id} 详情失败: {e}")
        return None


def process_single_file(file_path: Path) -> Dict[str, Dict[str, List[Dict]]]:
    """
    处理单个JSON文件，提取所有特征信息

    Args:
        file_path: JSON文件路径

    Returns:
        包含灵感点、目的点、关键点的特征字典
    """
    result = {
        "灵感点": {},
        "目的点": {},
        "关键点": {}
    }

    post_id = extract_post_id_from_filename(file_path.name)

    try:
        with open(file_path, "r", encoding="utf-8") as f:
            data = json.load(f)

        # 处理灵感点
        if "inspiration_final_result" in data:
            inspiration_data = data["inspiration_final_result"]
            for item in inspiration_data.get("最终灵感点列表", []):
                feature_name = item.get("灵感点", "")
                if not feature_name:
                    continue
                if feature_name not in result["灵感点"]:
                    result["灵感点"][feature_name] = []
                result["灵感点"][feature_name].append({
                    "点的名称": feature_name,
                    "点的描述": item.get("描述", ""),
                    "帖子id": post_id,
                    "点ID": item.get("id", ""),
                    "类型": item.get("类型", "")
                })

        # 处理目的点（意图+实质）
        if "purpose_final_result" in data:
            purpose_data = data["purpose_final_result"]

            # 处理意图列表
            for item in purpose_data.get("最终意图列表", []):
                feature_name = item.get("目的点", "")
                if not feature_name:
                    continue
                if feature_name not in result["目的点"]:
                    result["目的点"][feature_name] = []
                result["目的点"][feature_name].append({
                    "点的名称": feature_name,
                    "点的描述": item.get("描述", ""),
                    "帖子id": post_id,
                    "点ID": item.get("意图ID", ""),
                    "类型": "意图"
                })

            # 处理实质列表
            for item in purpose_data.get("最终实质列表", []):
                feature_name = item.get("目的点", "")
                if not feature_name:
                    continue
                if feature_name not in result["目的点"]:
                    result["目的点"][feature_name] = []
                result["目的点"][feature_name].append({
                    "点的名称": feature_name,
                    "点的描述": item.get("描述", ""),
                    "帖子id": post_id,
                    "点ID": item.get("实质ID", ""),
                    "类型": "实质",
                    "关联意图ID": item.get("关联意图ID", "")
                })

        # 处理关键点
        if "keypoint_final" in data:
            keypoint_data = data["keypoint_final"]
            for item in keypoint_data.get("最终关键点列表", []):
                feature_name = item.get("关键点", "")
                if not feature_name:
                    continue
                if feature_name not in result["关键点"]:
                    result["关键点"][feature_name] = []
                result["关键点"][feature_name].append({
                    "点的名称": feature_name,
                    "点的描述": item.get("描述", ""),
                    "帖子id": post_id,
                    "点ID": item.get("关键点ID", ""),
                    "类型": item.get("类型", ""),
                    "支撑的ID": item.get("支撑的ID", [])
                })

    except Exception as e:
        print(f"处理文件 {file_path.name} 时出错: {e}")

    return result


def merge_results(all_results: List[Dict]) -> Dict:
    """合并所有文件的提取结果"""
    merged = {
        "灵感点": {},
        "目的点": {},
        "关键点": {}
    }

    for result in all_results:
        for category in ["灵感点", "目的点", "关键点"]:
            for feature_name, sources in result[category].items():
                if feature_name not in merged[category]:
                    merged[category][feature_name] = {"来源": []}
                merged[category][feature_name]["来源"].extend(sources)

    return merged


def convert_to_array_format(
    merged_dict: Dict,
    fetch_details: bool = True,
    exclude_post_ids: Optional[Set[str]] = None
) -> Dict:
    """将字典格式转换为数组格式，并添加帖子详情"""
    result = {
        "灵感点": [],
        "目的点": [],
        "关键点": []
    }

    # 收集所有需要获取详情的帖子ID
    post_ids = set()
    if fetch_details:
        for category in ["灵感点", "目的点", "关键点"]:
            for feature_name, data in merged_dict[category].items():
                for source in data["来源"]:
                    post_ids.add(source["帖子id"])

        # 批量获取帖子详情
        print(f"\n正在获取 {len(post_ids)} 个帖子的详情...")
        post_details = {}
        for i, post_id in enumerate(post_ids, 1):
            print(f"[{i}/{len(post_ids)}] 获取帖子 {post_id} 的详情...")
            detail = get_post_detail(post_id)
            if detail:
                post_details[post_id] = detail

        print(f"成功获取 {len(post_details)} 个帖子详情")

        # 应用帖子ID过滤
        if exclude_post_ids:
            print(f"\n正在应用帖子ID过滤，排除 {len(exclude_post_ids)} 个当前帖子...")
            before_count = len(post_details)
            post_details = {pid: detail for pid, detail in post_details.items() if pid not in exclude_post_ids}
            filtered_count = before_count - len(post_details)
            if filtered_count > 0:
                print(f"  ⚠️  过滤掉 {filtered_count} 个当前帖子")
            print(f"保留 {len(post_details)} 个帖子")

    # 转换为数组格式并添加帖子详情
    for category in ["灵感点", "目的点", "关键点"]:
        for feature_name, data in merged_dict[category].items():
            enhanced_sources = []
            for source in data["来源"]:
                if fetch_details and exclude_post_ids and source["帖子id"] not in post_details:
                    continue
                enhanced_source = source.copy()
                if fetch_details and source["帖子id"] in post_details:
                    enhanced_source["帖子详情"] = post_details[source["帖子id"]]
                enhanced_sources.append(enhanced_source)

            if enhanced_sources:
                result[category].append({
                    "特征名称": feature_name,
                    "特征来源": enhanced_sources
                })

    return result


def get_current_post_ids(current_posts_dir: Path) -> Set[str]:
    """获取当前帖子目录中的所有帖子ID"""
    if not current_posts_dir.exists():
        print(f"警告: 当前帖子目录不存在: {current_posts_dir}")
        return set()

    json_files = list(current_posts_dir.glob("*.json"))
    if not json_files:
        print(f"警告: 当前帖子目录为空: {current_posts_dir}")
        return set()

    print(f"\n正在获取当前帖子ID...")
    print(f"找到 {len(json_files)} 个当前帖子")

    post_ids = set()
    for file_path in json_files:
        post_id = extract_post_id_from_filename(file_path.name)
        if post_id:
            post_ids.add(post_id)

    print(f"提取到 {len(post_ids)} 个帖子ID")
    return post_ids


def main():
    config = PathConfig()
    config.ensure_dirs()

    input_dir = config.historical_posts_dir
    current_posts_dir = config.current_posts_dir
    output_file = config.feature_source_mapping_file

    print(f"账号: {config.account_name}")
    print(f"过滤模式: {config.filter_mode}")
    print(f"过去帖子目录: {input_dir}")
    print(f"当前帖子目录: {current_posts_dir}")
    print(f"输出文件: {output_file}")
    print()

    print(f"\n正在扫描目录: {input_dir}")

    json_files = list(input_dir.glob("*.json"))
    print(f"找到 {len(json_files)} 个JSON文件")

    all_results = []
    for i, file_path in enumerate(json_files, 1):
        print(f"处理文件 [{i}/{len(json_files)}]: {file_path.name}")
        result = process_single_file(file_path)
        all_results.append(result)

    print("\n正在合并结果...")
    merged_result = merge_results(all_results)

    # 过滤当前帖子
    exclude_post_ids = None
    if config.filter_mode == "exclude_current_posts":
        print("\n应用过滤规则: 排除当前帖子ID")
        exclude_post_ids = get_current_post_ids(current_posts_dir)
    elif config.filter_mode == "none":
        print("\n过滤模式: none，不应用任何过滤")

    print("正在转换为数组格式...")
    final_result = convert_to_array_format(
        merged_result,
        fetch_details=True,
        exclude_post_ids=exclude_post_ids
    )

    print(f"\n提取统计:")
    for category in ["灵感点", "目的点", "关键点"]:
        feature_count = len(final_result[category])
        source_count = sum(len(item["特征来源"]) for item in final_result[category])
        print(f"  {category}: {feature_count} 个特征, {source_count} 个来源")

    print(f"\n正在保存结果到: {output_file}")
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(final_result, f, ensure_ascii=False, indent=4)

    print("完成!")


if __name__ == "__main__":
    main()