how/script/data_processing/analyze_creation_origin.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
创作起点分析

整合数据准备 + AI分析两步流程：
1. 根据帖子图谱 + 人设图谱，准备待分析数据
2. 调用AI分析起点

输入：帖子图谱 + 人设图谱
输出：起点分析结果
"""

import asyncio
import json
from pathlib import Path
from typing import Dict, List, Optional
import sys

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

from agents import Agent, Runner, ModelSettings, trace
from agents.tracing.create import custom_span
from lib.client import get_model
from lib.my_trace import set_trace_smith as set_trace
from script.data_processing.path_config import PathConfig


# ===== 配置 =====
MODEL_NAME = "google/gemini-3-pro-preview"
# MODEL_NAME = "anthropic/claude-sonnet-4"

MATCH_SCORE_THRESHOLD = 0.8  # 匹配分数阈值
GLOBAL_RATIO_THRESHOLD = 0.8  # 全局占比阈值

agent = Agent(
    name="Creation Origin Analyzer",
    model=get_model(MODEL_NAME),
    model_settings=ModelSettings(
        temperature=0.0,
        max_tokens=8192,
    ),
    tools=[],
)


# ===== 数据加载 =====

def load_json(file_path: Path) -> Dict:
    """加载JSON文件"""
    with open(file_path, "r", encoding="utf-8") as f:
        return json.load(f)


def get_post_graph_files(config: PathConfig) -> List[Path]:
    """获取所有帖子图谱文件"""
    post_graph_dir = config.intermediate_dir / "post_graph"
    return sorted(post_graph_dir.glob("*_帖子图谱.json"))


def get_result_file(config: PathConfig, post_id: str) -> Path:
    """获取分析结果文件路径"""
    return config.intermediate_dir / "origin_analysis_result" / f"{post_id}_起点分析.json"


def is_already_processed(config: PathConfig, post_id: str) -> bool:
    """检查帖子是否已处理过"""
    result_file = get_result_file(config, post_id)
    return result_file.exists()


# ===== 第一步：数据准备 =====

def extract_post_detail(post_graph: Dict) -> Dict:
    """提取帖子详情（保留原始字段名）"""
    meta = post_graph.get("meta", {})
    post_detail = meta.get("postDetail", {})

    return {
        "postId": meta.get("postId", ""),
        "postTitle": meta.get("postTitle", ""),
        "body_text": post_detail.get("body_text", ""),
        "images": post_detail.get("images", []),
        "video": post_detail.get("video"),
        "publish_time": post_detail.get("publish_time", ""),
        "like_count": post_detail.get("like_count", 0),
        "collect_count": post_detail.get("collect_count", 0),
    }


def extract_analysis_nodes(post_graph: Dict, persona_graph: Dict) -> tuple:
    """
    提取待分析节点列表

    待分析节点 = 灵感点 + 目的点 + 关键点
    """
    nodes = post_graph.get("nodes", {})
    edges = post_graph.get("edges", {})
    persona_nodes = persona_graph.get("nodes", {})
    persona_index = persona_graph.get("index", {})

    # 1. 收集关键点信息（用于支撑信息）
    keypoints = {}
    for node_id, node in nodes.items():
        if node.get("type") == "标签" and node.get("dimension") == "关键点":
            keypoints[node_id] = {
                "名称": node.get("name", ""),
                "描述": node.get("detail", {}).get("description", ""),
            }

    # 2. 分析支撑关系：关键点 → 灵感点/目的点
    support_map = {}  # {target_node_id: [支撑的关键点信息]}
    for edge_id, edge in edges.items():
        if edge.get("type") == "支撑":
            source_id = edge.get("source", "")
            target_id = edge.get("target", "")
            if source_id in keypoints:
                if target_id not in support_map:
                    support_map[target_id] = []
                support_map[target_id].append(keypoints[source_id])

    # 3. 分析关联关系
    relation_map = {}  # {node_id: [关联的节点名称]}
    for edge_id, edge in edges.items():
        if edge.get("type") == "关联":
            source_id = edge.get("source", "")
            target_id = edge.get("target", "")
            source_name = nodes.get(source_id, {}).get("name", "")
            target_name = nodes.get(target_id, {}).get("name", "")

            # 双向记录
            if source_id not in relation_map:
                relation_map[source_id] = []
            relation_map[source_id].append(target_name)

            if target_id not in relation_map:
                relation_map[target_id] = []
            relation_map[target_id].append(source_name)

    # 4. 分析人设匹配
    match_map = {}  # {node_id: 匹配信息}
    persona_out_edges = persona_index.get("outEdges", {})

    def get_node_info(node_id: str) -> Optional[Dict]:
        """获取人设节点的标准信息"""
        node = persona_nodes.get(node_id, {})
        if not node:
            return None
        detail = node.get("detail", {})
        parent_path = detail.get("parentPath", [])
        return {
            "节点ID": node_id,
            "节点名称": node.get("name", ""),
            "节点分类": "/".join(parent_path) if parent_path else "",
            "节点维度": node.get("dimension", ""),
            "节点类型": node.get("type", ""),
            "人设全局占比": detail.get("probGlobal", 0),
            "父类下占比": detail.get("probToParent", 0),
        }

    def get_parent_category_id(node_id: str) -> Optional[str]:
        """通过属于边获取父分类节点ID"""
        belong_edges = persona_out_edges.get(node_id, {}).get("属于", [])
        for edge in belong_edges:
            target_id = edge.get("target", "")
            target_node = persona_nodes.get(target_id, {})
            if target_node.get("type") == "分类":
                return target_id
        return None

    for edge_id, edge in edges.items():
        if edge.get("type") == "匹配":
            source_id = edge.get("source", "")
            target_id = edge.get("target", "")

            # 只处理 帖子节点 → 人设节点 的匹配
            if source_id.startswith("帖子:") and target_id.startswith("人设:"):
                match_score = edge.get("score", 0)
                persona_node = persona_nodes.get(target_id, {})

                if persona_node:
                    node_type = persona_node.get("type", "")

                    # 获取匹配节点信息
                    match_node_info = get_node_info(target_id)
                    if not match_node_info:
                        continue

                    # 确定所属分类节点
                    if node_type == "标签":
                        # 标签：找父分类
                        category_id = get_parent_category_id(target_id)
                    else:
                        # 分类：就是自己
                        category_id = target_id

                    # 获取所属分类信息和常见搭配
                    category_info = None
                    if category_id:
                        category_node = persona_nodes.get(category_id, {})
                        if category_node:
                            category_detail = category_node.get("detail", {})
                            category_path = category_detail.get("parentPath", [])
                            category_info = {
                                "节点ID": category_id,
                                "节点名称": category_node.get("name", ""),
                                "节点分类": "/".join(category_path) if category_path else "",
                                "节点维度": category_node.get("dimension", ""),
                                "节点类型": "分类",
                                "人设全局占比": category_detail.get("probGlobal", 0),
                                "父类下占比": category_detail.get("probToParent", 0),
                                "历史共现分类": [],
                            }

                            # 获取分类共现节点（按共现度降序排列）
                            co_occur_edges = persona_out_edges.get(category_id, {}).get("分类共现", [])
                            co_occur_edges_sorted = sorted(co_occur_edges, key=lambda x: x.get("score", 0), reverse=True)
                            for co_edge in co_occur_edges_sorted[:5]:  # 取前5个
                                co_target_id = co_edge.get("target", "")
                                co_score = co_edge.get("score", 0)
                                co_node = persona_nodes.get(co_target_id, {})
                                if co_node:
                                    co_detail = co_node.get("detail", {})
                                    co_path = co_detail.get("parentPath", [])
                                    category_info["历史共现分类"].append({
                                        "节点ID": co_target_id,
                                        "节点名称": co_node.get("name", ""),
                                        "节点分类": "/".join(co_path) if co_path else "",
                                        "节点维度": co_node.get("dimension", ""),
                                        "节点类型": "分类",
                                        "人设全局占比": co_detail.get("probGlobal", 0),
                                        "父类下占比": co_detail.get("probToParent", 0),
                                        "共现度": round(co_score, 4),
                                    })

                    match_map[source_id] = {
                        "匹配节点": match_node_info,
                        "匹配分数": round(match_score, 4),
                        "所属分类": category_info,
                    }

    # 5. 构建待分析节点列表（灵感点、目的点、关键点）
    analysis_nodes = []
    for node_id, node in nodes.items():
        if node.get("type") == "标签" and node.get("domain") == "帖子":
            dimension = node.get("dimension", "")
            if dimension in ["灵感点", "目的点", "关键点"]:
                # 人设匹配信息
                match_info = match_map.get(node_id)

                analysis_nodes.append({
                    "节点ID": node_id,
                    "节点名称": node.get("name", ""),
                    "节点分类": node.get("category", ""),  # 根分类：意图/实质/形式
                    "节点维度": dimension,
                    "节点类型": node.get("type", ""),
                    "节点描述": node.get("detail", {}).get("description", ""),
                    "人设匹配": match_info,
                })

    # 6. 构建可能的关系列表
    relation_list = []

    # 支撑关系：关键点 → 灵感点/目的点
    for edge_id, edge in edges.items():
        if edge.get("type") == "支撑":
            source_id = edge.get("source", "")
            target_id = edge.get("target", "")
            if source_id in keypoints:
                relation_list.append({
                    "来源节点": source_id,
                    "目标节点": target_id,
                    "关系类型": "支撑",
                })

    # 关联关系：节点之间的关联（去重，只记录一次）
    seen_relations = set()
    for edge_id, edge in edges.items():
        if edge.get("type") == "关联":
            source_id = edge.get("source", "")
            target_id = edge.get("target", "")
            # 用排序后的元组作为key去重
            key = tuple(sorted([source_id, target_id]))
            if key not in seen_relations:
                seen_relations.add(key)
                relation_list.append({
                    "来源节点": source_id,
                    "目标节点": target_id,
                    "关系类型": "关联",
                })

    return analysis_nodes, relation_list


def prepare_analysis_data(post_graph: Dict, persona_graph: Dict) -> Dict:
    """
    准备完整的分析数据

    Returns:
        {
            "帖子详情": {...},
            "待分析节点列表": [...],
            "可能的关系列表": [...]
        }
    """
    analysis_nodes, relation_list = extract_analysis_nodes(post_graph, persona_graph)
    return {
        "帖子详情": extract_post_detail(post_graph),
        "待分析节点列表": analysis_nodes,
        "可能的关系列表": relation_list,
    }


# ===== 第二步：AI分析 =====

def build_context(data: Dict) -> Dict:
    """
    构造AI分析的上下文

    Returns:
        {
            "all_points": [...],  # 全部创意点（含详细信息）
            "candidates": [...],  # 起点候选集（名称列表）
            "constants": [...],   # 人设常量（名称列表）
        }
    """
    nodes = data.get("待分析节点列表", [])

    # 全部创意点（含详细信息）
    all_points = []
    for node in nodes:
        match_info = node.get("人设匹配")
        match_score = 0
        category_global_ratio = 0
        if match_info:
            match_score = match_info.get("匹配分数", 0)
            category_info = match_info.get("所属分类", {})
            if category_info:
                category_global_ratio = category_info.get("人设全局占比", 0)

        all_points.append({
            "名称": node["节点名称"],
            "分类": node.get("节点分类", ""),
            "维度": node.get("节点维度", ""),
            "描述": node.get("节点描述", ""),
            "人设匹配度": round(match_score, 2),
            "所属分类全局占比": round(category_global_ratio, 2),
        })

    # 起点候选集（灵感点 + 目的点）
    candidates = [
        node["节点名称"]
        for node in nodes
        if node["节点维度"] in ["灵感点", "目的点"]
    ]

    # 人设常量（匹配分数 > 0.8 且 全局占比 > 0.8）
    constants = []
    for node in nodes:
        match_info = node.get("人设匹配")
        if match_info:
            match_score = match_info.get("匹配分数", 0)
            match_node = match_info.get("匹配节点", {})
            global_ratio = match_node.get("人设全局占比", 0)

            if match_score > MATCH_SCORE_THRESHOLD and global_ratio > GLOBAL_RATIO_THRESHOLD:
                constants.append(node["节点名称"])

    return {
        "all_points": all_points,
        "candidates": candidates,
        "constants": constants,
    }


def format_prompt(context: Dict) -> str:
    """
    格式化为AI prompt
    """
    all_points = context["all_points"]
    candidates = context["candidates"]
    constants = context["constants"]

    # 格式化全部创意点为易读文本
    points_text = ""
    for p in all_points:
        points_text += f"- {p['名称']}\n"
        points_text += f"  维度: {p['维度']} | 分类: {p['分类']}\n"
        points_text += f"  描述: {p['描述']}\n"
        points_text += f"  人设匹配度: {p['人设匹配度']} | 所属分类全局占比: {p['所属分类全局占比']}\n"
        points_text += "\n"

    # 格式化起点候选集
    candidates_text = "、".join(candidates)

    # 格式化人设常量
    constants_text = "、".join(constants) if constants else "无"

    prompt = f"""# Role
你是小红书爆款内容的"逆向工程"专家。你的核心能力是透过内容的表象（视觉/形式），还原创作者最初的脑回路（动机/实质）。

# Task
我提供一组笔记的【创意标签】和一个【起点候选集】。
请推理出哪些选项是真正的**创意起点**。


# Input Data

## 全部创意点

{points_text}

## 起点候选集
{candidates_text}

## 来自人设的常量
{constants_text}


# 推理约束

1. 实质推形式，而不是形式推实质，除非形式是一切创意的起点
2. 因推果而不是果推因
3. 无法被其他项或人设推理出的点，即为起点

# Output Format

请输出一个标准的 JSON 格式。
- Key: 候选集中的词。
- Value: 一个对象，包含：
  - `score`: 0.0 到 1.0 的浮点数（代表是起点的可能性）。
  - `analysis`: 一句话推理"""

    return prompt


# ===== 显示函数 =====

def display_context(context: Dict, post_id: str):
    """显示构造的上下文"""
    print(f"\n帖子: {post_id}")
    print(f"\n全部创意点 ({len(context['all_points'])} 个):")
    for p in context['all_points']:
        print(f"  - {p['名称']} ({p['维度']}/{p['分类']}) 匹配度={p['人设匹配度']}, 分类占比={p['所属分类全局占比']}")
    print(f"\n起点候选集 ({len(context['candidates'])} 个):")
    print(f"  {context['candidates']}")
    print(f"\n人设常量 ({len(context['constants'])} 个):")
    print(f"  {context['constants']}")


def display_result(result: Dict):
    """显示分析结果"""
    output = result.get("输出")
    if output:
        print("\n起点分析结果:")
        # 按score降序排列
        sorted_items = sorted(output.items(), key=lambda x: x[1].get("score", 0), reverse=True)
        for name, info in sorted_items:
            score = info.get("score", 0)
            analysis = info.get("analysis", "")
            marker = "★" if score >= 0.7 else "○"
            print(f"  {marker} {name}: {score:.2f}")
            print(f"      {analysis}")
    else:
        print(f"  分析失败: {result.get('错误', 'N/A')}")


# ===== 处理函数 =====

async def process_single_post(
    post_file: Path,
    persona_graph: Dict,
    config: PathConfig,
    current_time: str = None,
    log_url: str = None,
    force: bool = False,
) -> Dict:
    """
    处理单个帖子（数据准备 + AI分析）
    """
    # 加载帖子图谱
    post_graph = load_json(post_file)
    post_id = post_graph.get("meta", {}).get("postId", "unknown")

    # 检查是否已处理
    if not force and is_already_processed(config, post_id):
        print(f"\n跳过帖子 {post_id}（已处理，使用 --force 强制重新分析）")
        # 返回已有结果
        result_file = get_result_file(config, post_id)
        return load_json(result_file)

    print(f"\n{'=' * 60}")
    print(f"处理帖子: {post_id}")
    print("-" * 60)

    # 第一步：准备数据
    data = prepare_analysis_data(post_graph, persona_graph)

    # 构造上下文
    context = build_context(data)
    display_context(context, post_id)

    # 格式化prompt
    prompt = format_prompt(context)

    # 第二步：调用AI
    print("\n调用AI分析中...")
    with custom_span(
        name=f"创作起点分析 - {post_id}",
        data={
            "帖子id": post_id,
            "候选数": len(context["candidates"]),
            "模型": MODEL_NAME
        }
    ):
        result = await Runner.run(agent, input=prompt)
        output_text = result.final_output

    # 解析JSON
    try:
        if "```json" in output_text:
            json_start = output_text.find("```json") + 7
            json_end = output_text.find("```", json_start)
            json_str = output_text[json_start:json_end].strip()
        elif "{" in output_text and "}" in output_text:
            json_start = output_text.find("{")
            json_end = output_text.rfind("}") + 1
            json_str = output_text[json_start:json_end]
        else:
            json_str = output_text

        analysis_result = json.loads(json_str)

        result_data = {
            "帖子id": post_id,
            "模型": MODEL_NAME,
            "输入": context,
            "输出": analysis_result
        }
    except Exception as e:
        result_data = {
            "帖子id": post_id,
            "模型": MODEL_NAME,
            "输入": context,
            "输出": None,
            "错误": str(e),
            "原始输出": output_text
        }

    # 显示结果
    display_result(result_data)

    # 保存结果
    output_dir = config.intermediate_dir / "origin_analysis_result"
    output_dir.mkdir(parents=True, exist_ok=True)

    output_with_meta = {
        "元数据": {
            "current_time": current_time,
            "log_url": log_url,
            "model": MODEL_NAME
        },
        **result_data
    }

    output_file = output_dir / f"{post_id}_起点分析.json"
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(output_with_meta, f, ensure_ascii=False, indent=2)

    print(f"\n已保存: {output_file.name}")

    return result_data


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

async def main(
    post_id: str = None,
    all_posts: bool = False,
    force: bool = False,
):
    """
    主函数

    Args:
        post_id: 帖子ID，可选
        all_posts: 是否处理所有帖子
        force: 强制重新分析已处理的帖子
    """
    # 设置 trace
    current_time, log_url = set_trace()

    config = PathConfig()

    print(f"账号: {config.account_name}")
    print(f"使用模型: {MODEL_NAME}")
    print(f"Trace URL: {log_url}")

    # 加载人设图谱
    persona_graph_file = config.intermediate_dir / "人设图谱.json"
    if not persona_graph_file.exists():
        print(f"错误: 人设图谱文件不存在: {persona_graph_file}")
        return

    persona_graph = load_json(persona_graph_file)
    print(f"人设图谱节点数: {len(persona_graph.get('nodes', {}))}")

    # 获取帖子图谱文件
    post_graph_files = get_post_graph_files(config)
    if not post_graph_files:
        print("错误: 没有找到帖子图谱文件")
        return

    # 确定要处理的帖子
    if post_id:
        target_file = next(
            (f for f in post_graph_files if post_id in f.name),
            None
        )
        if not target_file:
            print(f"错误: 未找到帖子 {post_id}")
            return
        files_to_process = [target_file]
    elif all_posts:
        files_to_process = post_graph_files
    else:
        files_to_process = [post_graph_files[0]]

    print(f"待处理帖子数: {len(files_to_process)}")

    # 处理
    with trace("创作起点分析"):
        results = []
        skipped = 0
        for i, post_file in enumerate(files_to_process, 1):
            print(f"\n{'#' * 60}")
            print(f"# 处理帖子 {i}/{len(files_to_process)}")
            print(f"{'#' * 60}")

            result = await process_single_post(
                post_file=post_file,
                persona_graph=persona_graph,
                config=config,
                current_time=current_time,
                log_url=log_url,
                force=force,
            )

            # 检查是否是跳过的
            if not force and "元数据" in result:
                skipped += 1

            results.append(result)

    # 汇总
    print(f"\n{'#' * 60}")
    print(f"# 完成! 共处理 {len(results)} 个帖子 (跳过 {skipped} 个已处理)")
    print(f"{'#' * 60}")
    print(f"Trace: {log_url}")

    print("\n汇总（score >= 0.7 的起点）:")
    for result in results:
        post_id = result.get("帖子id")
        output = result.get("输出")
        if output:
            origins = [f"{k}({v['score']:.2f})" for k, v in output.items() if v.get("score", 0) >= 0.7]
            print(f"  {post_id}: {', '.join(origins) if origins else '无高置信起点'}")
        else:
            print(f"  {post_id}: 分析失败")


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="创作起点分析")
    parser.add_argument("--post-id", type=str, help="帖子ID")
    parser.add_argument("--all-posts", action="store_true", help="处理所有帖子")
    parser.add_argument("--force", action="store_true", help="强制重新分析已处理的帖子")
    args = parser.parse_args()

    asyncio.run(main(
        post_id=args.post_id,
        all_posts=args.all_posts,
        force=args.force,
    ))