how/script/data_processing/build_post_graph.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
构建帖子图谱

================================================================================
输入文件:
================================================================================
filtered_results/*_filtered.json - 帖子解构结果（过滤后的how解构）

================================================================================
输出文件: post_graph/{post_id}_帖子图谱.json（每个帖子一个文件）
================================================================================
{
    "meta": {                    # 元信息
        "postId": "帖子ID",
        "postTitle": "帖子标题",
        "postDetail": {...},
        "createdAt": "时间戳",
        "stats": { ... }
    },
    "nodes": {                   # 节点字典 (nodeId -> nodeData)
        "{domain}:{dimension}:{type}:{name}": {
            "name": "显示名称",
            "type": "帖子|灵感点|目的点|关键点|点|标签",
            "domain": "帖子",
            "dimension": "帖子|灵感点|目的点|关键点",
            "detail": { ... }
        }
    },
    "edges": {                   # 边字典 (edgeId -> edgeData)
        "{source}|{type}|{target}": {
            "source": "源节点ID",
            "target": "目标节点ID",
            "type": "属于|包含",
            "score": 1.0,
            "detail": { ... }
        }
    },
    "index": {                   # 游走索引
        "outEdges": { nodeId: { edgeType: [{ target, score }] } },
        "inEdges": { nodeId: { edgeType: [{ source, score }] } }
    },
    "tree": { ... }              # 嵌套树结构
}

================================================================================
核心逻辑:
================================================================================
1. 从 filtered_results 读取帖子解构结果
2. 提取点节点和标签节点
3. 添加根节点（帖子）和维度节点（灵感点/目的点/关键点）
4. 构建属于/包含边
5. 构建索引和嵌套树

================================================================================
层级对应（人设 vs 帖子）:
================================================================================
| 人设   | 帖子   |
|--------|--------|
| 人设   | 帖子   |
| 维度   | 维度   |
| 分类   | 点     |
| 标签   | 标签   |

================================================================================
节点ID格式: {domain}:{dimension}:{type}:{name}
================================================================================
- 根节点:   帖子:帖子:帖子:{post_id}
- 维度节点: 帖子:灵感点:灵感点:灵感点
- 点节点:   帖子:灵感点:点:{point_name}
- 标签节点: 帖子:灵感点:标签:{tag_name}

================================================================================
边类型:
================================================================================
- 属于: 子节点 -> 父节点（层级关系）
- 包含: 父节点 -> 子节点（层级关系）
- 匹配: 帖子标签 <-> 人设标签（双向，score为相似度）

================================================================================
匹配边说明:
================================================================================
帖子图谱包含与人设图谱的匹配边，通过节点ID关联：
- 帖子标签ID: 帖子:灵感点:标签:{tag_name}
- 人设标签ID: 人设:灵感点:标签:{persona_tag_name}

使用方式：从帖子标签出发，沿"匹配"边游走到人设标签ID，
再从人设图谱.json中查找该ID的详细信息。

================================================================================
"""

import json
from pathlib import Path
from typing import Dict, List, Set
from datetime import datetime
import sys

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

from script.data_processing.path_config import PathConfig


# ==================== 节点和边构建工具 ====================

def build_node_id(domain: str, dimension: str, node_type: str, name: str) -> str:
    """构建节点ID"""
    return f"{domain}:{dimension}:{node_type}:{name}"


def build_edge_id(source: str, edge_type: str, target: str) -> str:
    """构建边ID"""
    return f"{source}|{edge_type}|{target}"


def create_node(
    domain: str,
    dimension: str,
    node_type: str,
    name: str,
    detail: Dict = None
) -> Dict:
    """创建节点"""
    return {
        "name": name,
        "type": node_type,
        "dimension": dimension,
        "domain": domain,
        "detail": detail or {}
    }


def create_edge(
    source: str,
    target: str,
    edge_type: str,
    score: float = None,
    detail: Dict = None
) -> Dict:
    """创建边"""
    return {
        "source": source,
        "target": target,
        "type": edge_type,
        "score": score,
        "detail": detail or {}
    }


# ==================== 从帖子解构结果提取节点和匹配边 ====================

def extract_points_tags_and_matches(filtered_data: Dict) -> tuple:
    """
    从帖子解构结果中提取点节点、标签节点和匹配边

    Returns:
        (点节点字典, 标签节点字典, 标签到点的映射, 匹配边字典)
    """
    point_nodes = {}  # nodeId -> nodeData
    tag_nodes = {}    # nodeId -> nodeData
    tag_to_point = {} # tagId -> [pointId, ...]
    match_edges = {}  # edgeId -> edgeData

    how_result = filtered_data.get("how解构结果", {})

    dimension_mapping = {
        "灵感点列表": "灵感点",
        "目的点列表": "目的点",
        "关键点列表": "关键点"
    }

    for list_key, dimension in dimension_mapping.items():
        points = how_result.get(list_key, [])

        for point in points:
            point_name = point.get("名称", "")
            point_desc = point.get("描述", "")

            if not point_name:
                continue

            # 创建点节点
            point_id = build_node_id("帖子", dimension, "点", point_name)
            point_nodes[point_id] = create_node(
                domain="帖子",
                dimension=dimension,
                node_type="点",
                name=point_name,
                detail={
                    "description": point_desc
                }
            )

            # 遍历how步骤列表，提取标签和匹配
            how_steps = point.get("how步骤列表", [])

            for step in how_steps:
                step_name = step.get("步骤名称", "")
                features = step.get("特征列表", [])

                for feature in features:
                    tag_name = feature.get("特征名称", "")
                    weight = feature.get("权重", 1.0)

                    if not tag_name:
                        continue

                    # 创建标签节点
                    tag_id = build_node_id("帖子", dimension, "标签", tag_name)

                    if tag_id not in tag_nodes:
                        tag_nodes[tag_id] = create_node(
                            domain="帖子",
                            dimension=dimension,
                            node_type="标签",
                            name=tag_name,
                            detail={
                                "weight": weight,
                                "stepName": step_name,
                                "pointNames": [point_name]
                            }
                        )
                    else:
                        # 同一标签可能属于多个点
                        if point_name not in tag_nodes[tag_id]["detail"]["pointNames"]:
                            tag_nodes[tag_id]["detail"]["pointNames"].append(point_name)

                    # 记录标签到点的映射
                    if tag_id not in tag_to_point:
                        tag_to_point[tag_id] = []
                    if point_id not in tag_to_point[tag_id]:
                        tag_to_point[tag_id].append(point_id)

                    # 提取匹配边
                    matches = feature.get("匹配结果", [])
                    for match in matches:
                        persona_name = match.get("人设特征名称", "")
                        persona_dimension = match.get("人设特征层级", "")
                        persona_type = match.get("特征类型", "标签")
                        match_detail = match.get("匹配结果", {})
                        similarity = match_detail.get("相似度", 0)

                        if not persona_name or not persona_dimension:
                            continue

                        # 构建人设节点ID
                        persona_id = build_node_id("人设", persona_dimension, persona_type, persona_name)

                        # 创建双向匹配边
                        # 帖子标签 -> 人设标签
                        edge_id_1 = build_edge_id(tag_id, "匹配", persona_id)
                        match_edges[edge_id_1] = create_edge(
                            source=tag_id,
                            target=persona_id,
                            edge_type="匹配",
                            score=similarity,
                            detail={}
                        )

                        # 人设标签 -> 帖子标签
                        edge_id_2 = build_edge_id(persona_id, "匹配", tag_id)
                        match_edges[edge_id_2] = create_edge(
                            source=persona_id,
                            target=tag_id,
                            edge_type="匹配",
                            score=similarity,
                            detail={}
                        )

    return point_nodes, tag_nodes, tag_to_point, match_edges


# ==================== 构建边 ====================

def build_belong_contain_edges(
    point_nodes: Dict[str, Dict],
    tag_nodes: Dict[str, Dict],
    tag_to_point: Dict[str, List[str]],
    dimension_node_ids: Dict[str, str]
) -> Dict[str, Dict]:
    """
    构建属于/包含边

    Returns:
        边字典 { edgeId: edgeData }
    """
    edges = {}

    # 1. 点 -> 维度（属于/包含）
    for point_id, point_data in point_nodes.items():
        dimension = point_data["dimension"]
        dim_node_id = dimension_node_ids[dimension]

        # 属于边：点 -> 维度
        edge_id = build_edge_id(point_id, "属于", dim_node_id)
        edges[edge_id] = create_edge(
            source=point_id,
            target=dim_node_id,
            edge_type="属于",
            score=1.0
        )

        # 包含边：维度 -> 点
        edge_id_contain = build_edge_id(dim_node_id, "包含", point_id)
        edges[edge_id_contain] = create_edge(
            source=dim_node_id,
            target=point_id,
            edge_type="包含",
            score=1.0
        )

    # 2. 标签 -> 点（属于/包含）
    for tag_id, point_ids in tag_to_point.items():
        for point_id in point_ids:
            # 属于边：标签 -> 点
            edge_id = build_edge_id(tag_id, "属于", point_id)
            edges[edge_id] = create_edge(
                source=tag_id,
                target=point_id,
                edge_type="属于",
                score=1.0
            )

            # 包含边：点 -> 标签
            edge_id_contain = build_edge_id(point_id, "包含", tag_id)
            edges[edge_id_contain] = create_edge(
                source=point_id,
                target=tag_id,
                edge_type="包含",
                score=1.0
            )

    return edges


# ==================== 构建索引 ====================

def build_index(edges: Dict[str, Dict]) -> Dict:
    """
    构建游走索引

    Returns:
        {
            "outEdges": { nodeId: { edgeType: [{ target, score }] } },
            "inEdges": { nodeId: { edgeType: [{ source, score }] } }
        }
    """
    out_edges = {}
    in_edges = {}

    for edge_data in edges.values():
        source = edge_data["source"]
        target = edge_data["target"]
        edge_type = edge_data["type"]
        score = edge_data["score"]

        # outEdges
        if source not in out_edges:
            out_edges[source] = {}
        if edge_type not in out_edges[source]:
            out_edges[source][edge_type] = []
        out_edges[source][edge_type].append({
            "target": target,
            "score": score
        })

        # inEdges
        if target not in in_edges:
            in_edges[target] = {}
        if edge_type not in in_edges[target]:
            in_edges[target][edge_type] = []
        in_edges[target][edge_type].append({
            "source": source,
            "score": score
        })

    return {
        "outEdges": out_edges,
        "inEdges": in_edges
    }


# ==================== 构建嵌套树 ====================

def build_nested_tree(nodes: Dict[str, Dict], edges: Dict[str, Dict], root_id: str) -> Dict:
    """
    从根节点开始，沿"包含"边递归构建嵌套树结构

    Returns:
        嵌套的树结构
    """
    # 从"包含"边构建 父节点 -> [子节点] 的映射
    parent_to_children = {}

    for edge_data in edges.values():
        if edge_data["type"] == "包含":
            parent_id = edge_data["source"]
            child_id = edge_data["target"]

            if parent_id not in parent_to_children:
                parent_to_children[parent_id] = []
            parent_to_children[parent_id].append(child_id)

    # 递归构建子树
    def build_subtree(node_id: str) -> Dict:
        node_data = nodes[node_id]

        subtree = {
            "id": node_id,
            "name": node_data["name"],
            "type": node_data["type"],
            "domain": node_data["domain"],
            "dimension": node_data["dimension"],
            "detail": node_data.get("detail", {}),
            "children": []
        }

        # 获取子节点
        child_ids = parent_to_children.get(node_id, [])

        for child_id in child_ids:
            if child_id in nodes:
                subtree["children"].append(build_subtree(child_id))

        return subtree

    return build_subtree(root_id)


# ==================== 图游走工具 ====================

def walk_graph(
    index: Dict,
    start_node: str,
    edge_types: List[str],
    direction: str = "out",
    min_score: float = None
) -> Set[str]:
    """
    从起始节点出发，按指定边类型序列游走N步

    Args:
        index: 游走索引 {"outEdges": {...}, "inEdges": {...}}
        start_node: 起始节点ID
        edge_types: 边类型序列，如 ["属于", "包含"]
        direction: 游走方向 "out"(沿出边) / "in"(沿入边)
        min_score: 最小分数过滤

    Returns:
        到达的节点ID集合
    """
    edge_index = index["outEdges"] if direction == "out" else index["inEdges"]
    target_key = "target" if direction == "out" else "source"

    current_nodes = {start_node}

    for edge_type in edge_types:
        next_nodes = set()
        for node in current_nodes:
            neighbors = edge_index.get(node, {}).get(edge_type, [])
            for neighbor in neighbors:
                if min_score is not None and neighbor.get("score", 0) < min_score:
                    continue
                next_nodes.add(neighbor[target_key])
        current_nodes = next_nodes

        if not current_nodes:
            break

    return current_nodes


def get_neighbors(
    index: Dict,
    node_id: str,
    edge_type: str = None,
    direction: str = "out",
    min_score: float = None
) -> List[Dict]:
    """
    获取节点的邻居

    Args:
        index: 游走索引
        node_id: 节点ID
        edge_type: 边类型（可选，不指定则返回所有类型）
        direction: 方向 "out" / "in"
        min_score: 最小分数过滤

    Returns:
        邻居列表 [{"target": "...", "score": 0.5}, ...]
    """
    edge_index = index["outEdges"] if direction == "out" else index["inEdges"]
    node_edges = edge_index.get(node_id, {})

    if edge_type:
        neighbors = node_edges.get(edge_type, [])
    else:
        neighbors = []
        for edges in node_edges.values():
            neighbors.extend(edges)

    if min_score is not None:
        neighbors = [n for n in neighbors if n.get("score", 0) >= min_score]

    return neighbors


# ==================== 处理单个帖子 ====================

def process_single_post(filtered_file: Path, output_dir: Path) -> Dict:
    """
    处理单个帖子，生成帖子图谱

    Returns:
        处理结果统计
    """
    # 读取数据
    with open(filtered_file, "r", encoding="utf-8") as f:
        filtered_data = json.load(f)

    post_id = filtered_data.get("帖子id", "")
    post_detail = filtered_data.get("帖子详情", {})
    post_title = post_detail.get("title", "")

    # 初始化节点和边
    all_nodes = {}
    all_edges = {}

    # 1. 提取点节点、标签节点和匹配边
    point_nodes, tag_nodes, tag_to_point, match_edges = extract_points_tags_and_matches(filtered_data)

    # 2. 添加根节点
    root_id = build_node_id("帖子", "帖子", "帖子", post_id)
    all_nodes[root_id] = create_node(
        domain="帖子",
        dimension="帖子",
        node_type="帖子",
        name=post_id,
        detail={
            "postTitle": post_title,
            "postDetail": post_detail
        }
    )

    # 3. 添加维度节点
    dimensions = ["灵感点", "目的点", "关键点"]
    dimension_node_ids = {}

    for dim in dimensions:
        dim_id = build_node_id("帖子", dim, dim, dim)
        dimension_node_ids[dim] = dim_id

        all_nodes[dim_id] = create_node(
            domain="帖子",
            dimension=dim,
            node_type=dim,
            name=dim,
            detail={}
        )

        # 维度 -> 根 的属于边
        edge_id = build_edge_id(dim_id, "属于", root_id)
        all_edges[edge_id] = create_edge(
            source=dim_id,
            target=root_id,
            edge_type="属于",
            score=1.0
        )

        # 根 -> 维度 的包含边
        edge_id_contain = build_edge_id(root_id, "包含", dim_id)
        all_edges[edge_id_contain] = create_edge(
            source=root_id,
            target=dim_id,
            edge_type="包含",
            score=1.0
        )

    # 4. 添加点节点和标签节点
    all_nodes.update(point_nodes)
    all_nodes.update(tag_nodes)

    # 5. 构建属于/包含边
    belong_contain_edges = build_belong_contain_edges(
        point_nodes, tag_nodes, tag_to_point, dimension_node_ids
    )
    all_edges.update(belong_contain_edges)

    # 6. 添加匹配边
    all_edges.update(match_edges)

    # 7. 构建索引
    index = build_index(all_edges)

    # 8. 构建嵌套树
    tree = build_nested_tree(all_nodes, all_edges, root_id)

    # 统计
    point_count = len(point_nodes)
    tag_count = len(tag_nodes)
    match_count = len(match_edges) // 2  # 双向边，除以2得到实际匹配数

    dimension_stats = {}
    for dim in dimensions:
        dim_points = sum(1 for n in point_nodes.values() if n["dimension"] == dim)
        dim_tags = sum(1 for n in tag_nodes.values() if n["dimension"] == dim)
        dimension_stats[dim] = {
            "pointCount": dim_points,
            "tagCount": dim_tags
        }

    # 构建输出
    output_data = {
        "meta": {
            "postId": post_id,
            "postTitle": post_title,
            "postDetail": post_detail,
            "createdAt": datetime.now().isoformat(),
            "stats": {
                "nodeCount": len(all_nodes),
                "edgeCount": len(all_edges),
                "pointCount": point_count,
                "tagCount": tag_count,
                "matchCount": match_count,
                "dimensions": dimension_stats
            }
        },
        "nodes": all_nodes,
        "edges": all_edges,
        "index": index,
        "tree": tree
    }

    # 保存
    output_file = output_dir / f"{post_id}_帖子图谱.json"
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(output_data, f, ensure_ascii=False, indent=2)

    return {
        "postId": post_id,
        "postTitle": post_title,
        "nodeCount": len(all_nodes),
        "edgeCount": len(all_edges),
        "pointCount": point_count,
        "tagCount": tag_count,
        "matchCount": match_count,
        "outputFile": str(output_file)
    }


# ==================== 主函数 ====================

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

    print(f"账号: {config.account_name}")
    print(f"输出版本: {config.output_version}")
    print()

    # 输入目录
    filtered_results_dir = config.intermediate_dir / "filtered_results"

    # 输出目录
    output_dir = config.intermediate_dir / "post_graph"
    output_dir.mkdir(parents=True, exist_ok=True)

    print(f"输入目录: {filtered_results_dir}")
    print(f"输出目录: {output_dir}")
    print()

    # 获取所有帖子文件
    filtered_files = list(filtered_results_dir.glob("*_filtered.json"))
    print(f"找到 {len(filtered_files)} 个帖子文件")
    print()

    # 处理每个帖子
    results = []
    for i, filtered_file in enumerate(filtered_files, 1):
        print(f"[{i}/{len(filtered_files)}] 处理: {filtered_file.name}")
        result = process_single_post(filtered_file, output_dir)
        results.append(result)
        print(f"  节点: {result['nodeCount']}, 边: {result['edgeCount']}")
        print(f"  点: {result['pointCount']}, 标签: {result['tagCount']}, 匹配: {result['matchCount']}")
        print(f"  → {Path(result['outputFile']).name}")
        print()

    # 汇总统计
    print("=" * 60)
    print("处理完成!")
    print(f"  帖子数: {len(results)}")
    print(f"  总节点数: {sum(r['nodeCount'] for r in results)}")
    print(f"  总边数: {sum(r['edgeCount'] for r in results)}")
    print(f"  总点数: {sum(r['pointCount'] for r in results)}")
    print(f"  总标签数: {sum(r['tagCount'] for r in results)}")
    print(f"  总匹配数: {sum(r['matchCount'] for r in results)}")
    print(f"\n输出目录: {output_dir}")


if __name__ == "__main__":
    main()