Agent/examples_how/overall_derivation/tools/find_tree_node.py

"""
查找树节点 Tool - 人设树节点查询

功能：
1. 获取人设树的常量节点（全局常量、局部常量）
2. 获取符合条件概率阈值的节点（按条件概率排序返回 topN）

平台库人设树（第二节输出）流水线（由 build_platform_tree_section_items 聚合）：
  xiaohongshu/tree → 与账号相同的条件概率计算 → xiaohongshu/match_data 按匹配分过滤选题点
  → 剔除与账号段同名的节点。
"""

import json
import sys
from pathlib import Path
from typing import Any, Optional

# 保证直接运行或作为包加载时都能解析 utils / tools（IDE 可跳转）
_root = Path(__file__).resolve().parent.parent
if str(_root) not in sys.path:
    sys.path.insert(0, str(_root))
from utils.conditional_ratio_calc import (  # noqa: E402
    build_node_post_index,
    build_node_post_index_from_tree_dir,
    calc_node_conditional_ratio,
    load_persona_trees_from_dir,
)
from tools.point_match import (  # noqa: E402
    DEFAULT_MATCH_THRESHOLD,
    match_derivation_to_post_points,
)

try:
    from agent.tools import tool, ToolResult, ToolContext
except ImportError:
    def tool(*args, **kwargs):
        return lambda f: f
    ToolResult = None  # 仅用 main() 测核心逻辑时可无 agent
    ToolContext = None

# 相对本文件：tools -> overall_derivation，input / output 在 overall_derivation 下
_BASE_INPUT = Path(__file__).resolve().parent.parent / "input"
_BASE_OUTPUT = Path(__file__).resolve().parent.parent / "output"


def _dimension_analysis_log_dir(account_name: str, post_id: str, log_id: str) -> Path:
    """推导日志目录：output/{account_name}/推导日志/{post_id}/{log_id}/"""
    return _BASE_OUTPUT / account_name / "推导日志" / post_id / log_id


def _load_derived_dim_tree_node_names(
    account_name: str, post_id: str, log_id: str, round: int
) -> list[str]:
    """
    读取当前轮次对应的维度分析 JSON（优先 {round}_维度分析.json，不存在则 {round-1}_维度分析.json），
    返回 derived_dims 中每项的 tree_node_name（已推导出的维度节点，人设树中层次较高）。
    无可用文件时返回空列表。
    """
    if not log_id or not str(log_id).strip():
        return []
    log_dir = _dimension_analysis_log_dir(account_name, post_id, str(log_id).strip())
    for r in (round, round - 1):
        if r < 1:
            continue
        path = log_dir / f"{r}_维度分析.json"
        if not path.is_file():
            continue
        try:
            with open(path, "r", encoding="utf-8") as f:
                data = json.load(f)
        except Exception:
            continue
        dims = data.get("derived_dims") or []
        names: list[str] = []
        for d in dims:
            if isinstance(d, dict):
                tn = d.get("tree_node_name")
                if tn is not None and str(tn).strip():
                    names.append(str(tn).strip())
        return names
    return []


def _descendant_names_under_tree_nodes(
    account_name: str, anchor_node_names: list[str]
) -> tuple[set[str], dict[str, str]]:
    """
    在每个人设维度树根上 DFS，收集所有锚点（derived_dims.tree_node_name）之下的**全部后代**（不含锚点自身）。

    同时记录「所属维度」：对路径上每个后代节点，取从维度根到该节点路径上**最深的**那个锚点
    （与原先沿父链向上找最近 derived_dim 一致；多个锚点呈祖孙时取更深者）。
    """
    if not anchor_node_names:
        return set(), {}
    S = set(anchor_node_names)
    allowed: set[str] = set()
    dim_map: dict[str, str] = {}

    for dim_root_name, root in _load_trees(account_name):

        def dfs(node_name: str, node_dict: dict, parent_deepest_s: Optional[str]) -> None:
            d_self = node_name if node_name in S else parent_deepest_s
            for cname, cnode in (node_dict.get("children") or {}).items():
                if not isinstance(cnode, dict):
                    continue
                if cname not in S and d_self is not None:
                    allowed.add(cname)
                    dim_map[cname] = d_self
                dfs(cname, cnode, d_self)

        dfs(dim_root_name, root, None)

    return allowed, dim_map


def _tree_dir(account_name: str) -> Path:
    """人设树目录：../input/{account_name}/处理后数据/tree/"""
    return _BASE_INPUT / account_name / "处理后数据" / "tree"


def _load_trees(account_name: str) -> list[tuple[str, dict]]:
    """加载该账号下所有维度的人设树。返回 [(维度名, 根节点 dict), ...]。"""
    td = _tree_dir(account_name)
    if not td.is_dir():
        return []
    result = []
    for p in td.glob("*.json"):
        try:
            with open(p, "r", encoding="utf-8") as f:
                data = json.load(f)
            for dim_name, root in data.items():
                if isinstance(root, dict):
                    result.append((dim_name, root))
                break
        except Exception:
            continue
    return result


def _iter_all_nodes(account_name: str):
    """遍历该账号下所有人设树节点，产出 (节点名称, 父节点名称, 节点 dict)。"""
    for dim_name, root in _load_trees(account_name):
        def walk(parent_name: str, node_dict: dict):
            for name, child in (node_dict.get("children") or {}).items():
                if not isinstance(child, dict):
                    continue
                yield (name, parent_name, child)
                yield from walk(name, child)

        yield from walk(dim_name, root)


# ---------------------------------------------------------------------------
# 1. 获取人设树常量节点
# ---------------------------------------------------------------------------

def get_constant_nodes(account_name: str) -> list[dict[str, Any]]:
    """
    获取人设树的常量节点。
    - 全局常量：_is_constant=True
    - 局部常量：_is_local_constant=True 且 _is_constant=False
    返回列表项：节点名称、概率(_ratio)、常量类型。
    """
    result = []
    for node_name, _parent, node in _iter_all_nodes(account_name):
        is_const = node.get("_is_constant") is True
        is_local = node.get("_is_local_constant") is True
        if is_const:
            const_type = "全局常量"
        elif is_local and not is_const:
            const_type = "局部常量"
        else:
            continue
        ratio = node.get("_ratio")
        result.append({
            "节点名称": node_name,
            "概率": ratio,
            "常量类型": const_type,
        })
    result.sort(key=lambda x: (x["概率"] is None, -(x["概率"] or 0)))
    return result


# ---------------------------------------------------------------------------
# 2. 获取符合条件概率阈值的节点
# ---------------------------------------------------------------------------

def get_nodes_by_conditional_ratio(
    account_name: str,
    derived_list: list[tuple[str, str]],
    threshold: float,
    top_n: int,
    allowed_node_names: Optional[set[str]] = None,
    node_belonging_dim: Optional[dict[str, str]] = None,
) -> list[dict[str, Any]]:
    """
    获取人设树中条件概率 >= threshold 的节点，按条件概率降序，返回前 top_n 个。
    derived_list: 已推导列表，每项 (已推导的选题点, 推导来源人设树节点)；为空时使用节点自身的 _ratio 作为条件概率。
    allowed_node_names: 若给定，仅保留节点名称在该集合内的结果。
    node_belonging_dim: 与 allowed 同步生成（见 _descendant_names_under_tree_nodes），节点名 -> 所属已推导维度；不传则所属维度均为「—」。
    返回列表项：节点名称、条件概率、父节点名称、所属维度。
    """
    base_dir = _BASE_INPUT
    node_to_parent: dict[str, str] = {}
    if derived_list:
        for n, p, _ in _iter_all_nodes(account_name):
            node_to_parent[n] = p

    def dim_for(node_name: str) -> str:
        if not node_belonging_dim:
            return "—"
        return node_belonging_dim.get(node_name) or "—"

    scored: list[tuple[str, float, str, str]] = []

    if not derived_list:
        for node_name, parent_name, node in _iter_all_nodes(account_name):
            if allowed_node_names is not None and node_name not in allowed_node_names:
                continue
            ratio = node.get("_ratio")
            if ratio is None:
                ratio = 0.0
            else:
                ratio = float(ratio)
            if ratio >= threshold:
                scored.append((node_name, ratio, parent_name, dim_for(node_name)))
    else:
        node_post_index = build_node_post_index(account_name, base_dir)
        for node_name, parent_name in node_to_parent.items():
            if allowed_node_names is not None and node_name not in allowed_node_names:
                continue
            ratio = calc_node_conditional_ratio(
                account_name,
                derived_list,
                node_name,
                base_dir=base_dir,
                node_post_index=node_post_index,
                target_ratio=threshold,
            )
            if ratio >= threshold:
                scored.append((node_name, ratio, parent_name, dim_for(node_name)))

    scored.sort(key=lambda x: x[1], reverse=True)
    top = scored[:top_n]
    return [
        {
            "节点名称": name,
            "条件概率": ratio,
            "父节点名称": parent,
            "所属维度": dim,
        }
        for name, ratio, parent, dim in top
    ]


def _platform_tree_dir() -> Path:
    """平台库人设树目录：../input/xiaohongshu/tree/"""
    return _BASE_INPUT / "xiaohongshu" / "tree"


def _collect_platform_scored_tuples(
    derived_list: list[tuple[str, str]],
    threshold: float,
    max_nodes: int = 12000,
) -> list[tuple[str, float, str, str]]:
    """
    平台库人设树：条件概率 >= threshold 的节点全量收集，按条件概率降序。
    max_nodes 防止极端大树占满内存；截断发生在全局排序之后（保留高分段）。
    """
    tree_dir = _platform_tree_dir()
    if not tree_dir.is_dir():
        return []

    thr = float(threshold)
    scored: list[tuple[str, float, str, str]] = []

    if not derived_list:
        for dim_name, root in load_persona_trees_from_dir(tree_dir):

            def walk(parent_name: str, node_dict: dict) -> None:
                for name, child in (node_dict.get("children") or {}).items():
                    if not isinstance(child, dict):
                        continue
                    ratio = child.get("_ratio")
                    r = 0.0 if ratio is None else float(ratio)
                    if r >= thr:
                        scored.append((name, r, parent_name, dim_name))
                    walk(name, child)

            walk(dim_name, root)
    else:
        node_post_index = build_node_post_index_from_tree_dir(tree_dir)
        node_to_parent_dim: dict[str, tuple[str, str]] = {}
        for dim_name, root in load_persona_trees_from_dir(tree_dir):

            def walk2(parent_name: str, node_dict: dict) -> None:
                for name, child in (node_dict.get("children") or {}).items():
                    if not isinstance(child, dict):
                        continue
                    node_to_parent_dim[name] = (parent_name, dim_name)
                    walk2(name, child)

            walk2(dim_name, root)

        for node_name, (parent_name, dim_name) in node_to_parent_dim.items():
            ratio = calc_node_conditional_ratio(
                "",
                derived_list,
                node_name,
                base_dir=_BASE_INPUT,
                node_post_index=node_post_index,
                target_ratio=thr,
            )
            if ratio >= thr:
                scored.append((node_name, ratio, parent_name, dim_name))

    scored.sort(key=lambda x: x[1], reverse=True)
    if max_nodes > 0 and len(scored) > max_nodes:
        scored = scored[:max_nodes]
    return scored


def get_platform_nodes_by_conditional_ratio(
    derived_list: list[tuple[str, str]],
    threshold: float,
    top_n: int,
) -> list[dict[str, Any]]:
    """
    平台库人设树节点条件概率筛选，计算方式与 get_nodes_by_conditional_ratio 一致
   （同一套 calc_node_conditional_ratio / _post_ids 规则，索引来自 xiaohongshu/tree）。
    derived_list 为空时用节点 _ratio。
    """
    n = max(0, int(top_n))
    scored = _collect_platform_scored_tuples(derived_list, threshold)
    top = scored[:n]
    return [
        {
            "节点名称": name,
            "条件概率": ratio,
            "父节点名称": parent,
            "所属维度": dim,
        }
        for name, ratio, parent, dim in top
    ]


def _parse_derived_list(derived_items: list[dict[str, str]]) -> list[tuple[str, str]]:
    """将 agent 传入的 [{"topic": "x", "source_node": "y"}, ...] 转为 DerivedItem 列表。"""
    out = []
    for item in derived_items:
        if isinstance(item, dict):
            topic = item.get("topic") or item.get("已推导的选题点")
            source = item.get("source_node") or item.get("推导来源人设树节点")
            if topic is not None and source is not None:
                out.append((str(topic).strip(), str(source).strip()))
        elif isinstance(item, (list, tuple)) and len(item) >= 2:
            out.append((str(item[0]).strip(), str(item[1]).strip()))
    return out


# ---------------------------------------------------------------------------
# 3. 平台库人设树辅助节点（基于帖子与平台库人设树匹配结果）
# ---------------------------------------------------------------------------


def _platform_match_topics_by_node(
    post_id: str,
    match_score_threshold: float,
) -> dict[tuple[str, str], dict[str, float]]:
    """
    读取 xiaohongshu/match_data/{post_id}_匹配_all.json，
    返回 (dimension, 人设树节点名) -> {帖子选题点: 最高分}，仅收录 match_score >= match_score_threshold 的对。
    """
    out: dict[tuple[str, str], dict[str, float]] = {}
    if not post_id:
        return out
    path = _BASE_INPUT / "xiaohongshu" / "match_data" / f"{post_id}_匹配_all.json"
    if not path.is_file():
        return out
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except Exception:
        return out
    if not isinstance(data, list):
        return out

    thr = float(match_score_threshold)
    for item in data:
        if not isinstance(item, dict):
            continue
        topic = item.get("name")
        matches = item.get("match_personas")
        if topic is None or not isinstance(matches, list):
            continue
        topic_s = str(topic).strip()
        if not topic_s:
            continue
        for m in matches:
            if not isinstance(m, dict):
                continue
            name = m.get("name")
            dim = m.get("dimension")
            score = m.get("match_score")
            if name is None or dim is None or score is None:
                continue
            try:
                s = float(score)
            except Exception:
                continue
            if s < thr:
                continue
            key = (str(dim).strip(), str(name).strip())
            bucket = out.setdefault(key, {})
            prev = bucket.get(topic_s)
            if prev is None or s > prev:
                bucket[topic_s] = s
    return out


def _platform_node_belonging_dim_from_anchor_nodes(
    anchor_node_names: list[str],
) -> dict[str, str]:
    """
    计算平台库人设树中：节点名 -> 所属最深 derived_dim 锚点节点名。

    逻辑与账号段 _descendant_names_under_tree_nodes 保持一致（但树结构来自 xiaohongshu/tree）。
    """
    if not anchor_node_names:
        return {}
    S = set(anchor_node_names)
    dim_map: dict[str, str] = {}
    tree_dir = _platform_tree_dir()
    if not tree_dir.is_dir():
        return {}

    for dim_root_name, root in load_persona_trees_from_dir(tree_dir):

        def dfs(node_name: str, node_dict: dict, parent_deepest_s: Optional[str]) -> None:
            d_self = node_name if node_name in S else parent_deepest_s
            for cname, cnode in (node_dict.get("children") or {}).items():
                if not isinstance(cnode, dict):
                    continue
                if cname not in S and d_self is not None:
                    dim_map[cname] = d_self
                dfs(cname, cnode, d_self)

        dfs(dim_root_name, root, None)
    return dim_map


def _load_platform_nodes_split(
    post_id: str,
    derived_list: list[tuple[str, str]],
    conditional_ratio_threshold: float,
    match_score_threshold: float,
    top_n: int,
    node_belonging_dim_platform: Optional[dict[str, str]] = None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    """
    平台库人设树：用 _collect_platform_scored_tuples 得到条件概率达标的节点，
    再按 xiaohongshu/match_data 分为「有帖子选题点匹配 / 无匹配」两类，**两类各自按条件概率取 Top 池**（同一全局 TopN 不会挤掉另一类），
    最后分别组装返回：
      - matched：有 match_score >= match_score_threshold 的帖子选题点匹配的节点
      - unmatched：无达标帖子选题点匹配的节点
    两组均要求节点在 node_belonging_dim_platform 中有有效的所属维度（不为「—」）。
    """
    matched: list[dict[str, Any]] = []
    unmatched: list[dict[str, Any]] = []

    topic_map: dict[tuple[str, str], dict[str, float]] = {}
    if post_id:
        topic_map = _platform_match_topics_by_node(post_id, float(match_score_threshold))
    # 维度标签可能与树侧不完全一致：保留一个按节点名聚合的兜底索引，避免误判为“无匹配”。
    topic_map_by_name: dict[str, dict[str, float]] = {}
    for (_dim, n), topics in topic_map.items():
        bucket = topic_map_by_name.setdefault(str(n).strip(), {})
        for t, sc in (topics or {}).items():
            prev = bucket.get(t)
            if prev is None or sc > prev:
                bucket[t] = sc

    # 有 match_data 命中与无命中两类分开按条件概率取 Top，避免混在一个全局 TopN 里挤掉某一类。
    all_scored = _collect_platform_scored_tuples(
        derived_list,
        float(conditional_ratio_threshold),
    )
    if not all_scored:
        return matched, unmatched

    matched_tuples: list[tuple[str, float, str, str]] = []
    unmatched_tuples: list[tuple[str, float, str, str]] = []
    for name, ratio, parent, dim in all_scored:
        lookup_dim = str(dim).strip()
        key = (lookup_dim, str(name).strip())
        topics = topic_map.get(key) or topic_map_by_name.get(str(name).strip()) or {}
        if topics:
            matched_tuples.append((name, ratio, parent, dim))
        else:
            unmatched_tuples.append((name, ratio, parent, dim))

    _pool = max(int(top_n), min(2000, max(500, int(top_n) * 5)))
    matched_tuples = matched_tuples[:_pool]
    unmatched_tuples = unmatched_tuples[:_pool]

    def _emit_tuple_rows(
        tuples: list[tuple[str, float, str, str]],
        *,
        has_topics: bool,
    ) -> None:
        for name, ratio, parent, dim in tuples:
            row = {
                "节点名称": name,
                "条件概率": ratio,
                "父节点名称": parent,
                "所属维度": dim,
            }
            name_s = str(row.get("节点名称") or "").strip()
            out_dim = "—"
            if node_belonging_dim_platform is not None:
                out_dim = node_belonging_dim_platform.get(name_s) or "—"
            if node_belonging_dim_platform is not None and out_dim == "—":
                continue
            row_out = dict(row)
            row_out["所属维度"] = out_dim

            lookup_dim = str(row.get("所属维度") or "").strip()
            key2 = (lookup_dim, name_s)
            topics = topic_map.get(key2) or topic_map_by_name.get(name_s) or {}
            if has_topics:
                if not topics:
                    continue
                topic_items = sorted(topics.items(), key=lambda x: x[1], reverse=True)
                row_out["帖子选题点匹配"] = [{"帖子选题点": t, "匹配分数": sc} for t, sc in topic_items]
                matched.append(row_out)
            else:
                if topics:
                    continue
                row_out["帖子选题点匹配"] = "无"
                unmatched.append(row_out)

    _emit_tuple_rows(matched_tuples, has_topics=True)
    _emit_tuple_rows(unmatched_tuples, has_topics=False)

    return matched, unmatched


def build_platform_tree_section_items_split(
    post_id: str,
    derived_list: list[tuple[str, str]],
    conditional_ratio_threshold: float,
    match_score_threshold: float,
    top_n: int,
    exclude_node_names: set[str],
    node_belonging_dim_platform: Optional[dict[str, str]] = None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    """
    平台库人设树节点：条件概率 + xiaohongshu/match_data 匹配，排除与账号段重复的节点名称，
    返回 (有帖子选题点匹配的节点列表, 无帖子选题点匹配的节点列表)。
    供 find_tree_nodes_by_conditional_ratio 聚合输出使用。
    """
    if not post_id:
        return [], []
    ex = {str(n).strip() for n in exclude_node_names}
    matched, unmatched = _load_platform_nodes_split(
        post_id=post_id,
        derived_list=derived_list,
        conditional_ratio_threshold=float(conditional_ratio_threshold),
        match_score_threshold=float(match_score_threshold),
        top_n=int(top_n),
        node_belonging_dim_platform=node_belonging_dim_platform,
    )
    matched_filtered = [p for p in matched if str(p.get("节点名称", "")).strip() not in ex]
    unmatched_filtered = [p for p in unmatched if str(p.get("节点名称", "")).strip() not in ex]
    return matched_filtered, unmatched_filtered


# ---------------------------------------------------------------------------
# Agent Tools（参考 glob_tool 封装）
# ---------------------------------------------------------------------------

@tool()
async def find_tree_constant_nodes(
    account_name: str,
    post_id: str,
) -> ToolResult:
    """
    获取人设树中的常量节点列表（全局常量与局部常量），并检查每个节点与帖子选题点的匹配情况。

    Args:
    account_name : 账号名，用于定位该账号的人设树数据。
    post_id : 帖子ID，用于加载帖子选题点并与各常量节点做匹配判断。

    Returns:
    ToolResult：
        - title: 结果标题。
        - output: 可读的节点列表文本（每行：节点名称、概率、常量类型、帖子匹配情况）。
        - 出错时 error 为错误信息。
    """
    tree_dir = _tree_dir(account_name)
    if not tree_dir.is_dir():
        return ToolResult(
            title="人设树目录不存在",
            output=f"目录不存在: {tree_dir}",
            error="Directory not found",
        )
    try:
        items = get_constant_nodes(account_name)
        # 批量匹配所有节点与帖子选题点
        if items and post_id:
            node_names = [x["节点名称"] for x in items]
            matched_results = await match_derivation_to_post_points(
                node_names, account_name, post_id, match_threshold=float(DEFAULT_MATCH_THRESHOLD)
            )
            node_match_map: dict[str, list] = {}
            for m in matched_results:
                node_match_map.setdefault(m["推导选题点"], []).append({
                    "帖子选题点": m["帖子选题点"],
                    "匹配分数": m["匹配分数"],
                })
            for item in items:
                matches = node_match_map.get(item["节点名称"], [])
                item["帖子选题点匹配"] = matches if matches else "无"
        if not items:
            output = "未找到常量节点"
        else:
            lines = []
            for x in items:
                match_info = x.get("帖子选题点匹配", "无")
                if isinstance(match_info, list):
                    match_str = "、".join(f"{m['帖子选题点']}({m['匹配分数']})" for m in match_info)
                else:
                    match_str = str(match_info)
                lines.append(f"- {x['节点名称']}\t概率={x['概率']}\t{x['常量类型']}\t帖子选题点匹配={match_str}")
            output = "\n".join(lines)
        return ToolResult(
            title=f"常量节点 ({account_name})",
            output=output,
            metadata={"account_name": account_name, "count": len(items)},
        )
    except Exception as e:
        return ToolResult(
            title="获取常量节点失败",
            output=str(e),
            error=str(e),
        )


@tool()
async def find_tree_nodes_by_conditional_ratio(
    account_name: str,
    post_id: str,
    derived_items: list[dict[str, str]],
    conditional_ratio_threshold: float,
    top_n: int = 100,
    round: int = 1,
    log_id: str = "",
    match_score_threshold: float = DEFAULT_MATCH_THRESHOLD,
) -> ToolResult:
    """
    按条件概率阈值筛选节点：先账号人设树（优先使用），再平台库人设树；两段不合并。
    条件概率计算对两棵树使用同一套规则（calc_node_conditional_ratio / 节点 _post_ids）。
    返回结果按以下配额分配（合计 top_n 条）：
      - 账号人设树节点占 60%，其中有帖子选题点匹配的记录和无帖子选题点匹配的记录各占一半；
      - 平台库人设树节点占 40%，其中有帖子选题点匹配的记录和无帖子选题点匹配的记录各占一半。
    「帖子选题点匹配」仅收录匹配分 >= match_score_threshold 的选题点。

    Args:
    account_name : 账号名，用于定位该账号的人设树数据。
    post_id : 帖子ID，用于加载帖子选题点并与各节点做匹配判断。
    derived_items : 已推导选题点列表，可为空。非空时每项为字典，需含 topic（或「已推导的选题点」）与 source_node（或「推导来源人设树节点」）
    conditional_ratio_threshold : 条件概率阈值，仅返回条件概率 >= 该值的节点。
    top_n : 最终返回总条数上限，按 账号60%/平台40%、有匹配/无匹配各半 分配。
    round : 推导轮次。
    log_id : 推导日志ID
    match_score_threshold : 帖子选题点匹配分阈值，与 point_match 默认一致。

    Returns:
    ToolResult：
        - title: 结果标题。
        - output: 两段文本——先账号人设树，后平台库人设树；
          账号侧匹配来自 input/{账号}/match_data；平台侧条件概率基于 input/xiaohongshu/tree，匹配来自 input/xiaohongshu/match_data。
        - 出错时 error 为错误信息。
    """
    tree_dir = _tree_dir(account_name)
    if not tree_dir.is_dir():
        return ToolResult(
            title="人设树目录不存在",
            output=f"目录不存在: {tree_dir}",
            error="Directory not found",
        )
    try:
        derived_list = _parse_derived_list(derived_items or [])

        allowed: Optional[set[str]] = None
        node_belonging_dim: dict[str, str] = {}
        node_belonging_dim_platform: Optional[dict[str, str]] = None
        dim_source = ""
        derived_dim_names: list[str] = []
        derived_items_len = len(derived_items or [])
        if log_id and str(log_id).strip():
            derived_dim_names = _load_derived_dim_tree_node_names(
                account_name, post_id, str(log_id).strip(), int(round)
            )
            if derived_dim_names:
                allowed, node_belonging_dim = _descendant_names_under_tree_nodes(
                    account_name, derived_dim_names
                )
                node_belonging_dim_platform = _platform_node_belonging_dim_from_anchor_nodes(
                    derived_dim_names
                )
                # 记录实际用到的维度分析文件（与读取逻辑一致）
                log_dir = _dimension_analysis_log_dir(account_name, post_id, str(log_id).strip())
                for r in (int(round), int(round) - 1):
                    if r >= 1 and (log_dir / f"{r}_维度分析.json").is_file():
                        dim_source = f"{r}_维度分析.json (derived_dims -> 全部后代)"
                        break
            else:
                dim_source = "未读到 derived_dims（无对应维度分析文件或为空），未收窄"

        # 当 derived_items 太多时，用 derived_dim_names 作为条件概率计算锚点：
        # 将每个 derived_dim_names 的 name 都映射为 (topic=name, source_node=name)。
        if derived_items_len > 15 and derived_dim_names:
            derived_list = [(n, n) for n in derived_dim_names]

        # 1）账号人设树：按条件概率筛选；帖子选题点匹配仅走账号 match_data（match_derivation_to_post_points）
        items = get_nodes_by_conditional_ratio(
            account_name,
            derived_list,
            conditional_ratio_threshold,
            top_n,
            allowed_node_names=allowed,
            node_belonging_dim=node_belonging_dim if node_belonging_dim else None,
        )
        if items and post_id:
            node_names = [x["节点名称"] for x in items]
            matched_results = await match_derivation_to_post_points(
                node_names, account_name, post_id, match_threshold=float(match_score_threshold)
            )
            node_match_map: dict[str, list] = {}
            for m in matched_results:
                node_match_map.setdefault(m["推导选题点"], []).append({
                    "帖子选题点": m["帖子选题点"],
                    "匹配分数": m["匹配分数"],
                })
            for item in items:
                matches = node_match_map.get(item["节点名称"], [])
                item["帖子选题点匹配"] = matches if matches else "无"

        # 账号配额：占 top_n 的 60%，有/无匹配各一半
        account_quota = int(top_n * 0.6 + 0.5)
        account_with_n = account_quota // 2
        account_without_n = account_quota - account_with_n
        items_with_match = [x for x in items if isinstance(x.get("帖子选题点匹配"), list)]
        items_without_match = [x for x in items if not isinstance(x.get("帖子选题点匹配"), list)]
        items = items_with_match[:account_with_n] + items_without_match[:account_without_n]

        # 2）平台库人设树（条件概率 + xiaohongshu 匹配文件）
        # 平台配额：占 top_n 的 40%，有/无匹配各一半
        platform_quota = top_n - account_quota
        platform_with_n = platform_quota // 2
        platform_without_n = platform_quota - platform_with_n
        # 平台「有匹配」排除账号侧已有帖子选题点匹配的节点名（与账号段去重）。
        # 平台「无匹配」排除已在账号段输出里出现过的节点名（避免重复罗列无新信息的同名节点）。
        account_matched_names = {str(x.get("节点名称", "")).strip() for x in items if isinstance(x.get("帖子选题点匹配"), list)}
        account_all_names = {str(x.get("节点名称", "")).strip() for x in items}
        platform_items: list[dict[str, Any]] = []
        if post_id:
            p_matched_raw, p_unmatched_raw = _load_platform_nodes_split(
                post_id=post_id,
                derived_list=derived_list,
                conditional_ratio_threshold=float(conditional_ratio_threshold),
                match_score_threshold=float(match_score_threshold),
                top_n=top_n,
                node_belonging_dim_platform=node_belonging_dim_platform,
            )
            p_matched = [p for p in p_matched_raw if str(p.get("节点名称", "")).strip() not in account_matched_names]
            p_unmatched = [p for p in p_unmatched_raw if str(p.get("节点名称", "")).strip() not in account_all_names]
            platform_items = p_matched[:platform_with_n] + p_unmatched[:platform_without_n]

        def _format_node_line(x: dict[str, Any]) -> str:
            match_info = x.get("帖子选题点匹配", "无")
            if isinstance(match_info, list):
                match_str = "、".join(f"{m['帖子选题点']}({m['匹配分数']})" for m in match_info)
            else:
                match_str = str(match_info)
            dim_label = x.get("所属维度", "—")
            return (
                f"- {x['节点名称']}\t条件概率={x['条件概率']}\t所属维度={dim_label}"
                f"\t帖子选题点匹配={match_str}"
            )

        lines: list[str] = []
        lines.append(
            "【优先使用】第一节为账号人设树中条件概率达标的节点（占60%配额，有/无帖子匹配各半）；"
            "第二节为平台库人设树中条件概率达标的节点（占40%配额，有/无帖子匹配各半）；"
        )
        lines.append("")
        lines.append("—— 账号人设树节点 ——")
        if not items:
            lines.append(f"（无：未找到条件概率 >= {conditional_ratio_threshold} 的节点）")
        else:
            lines.extend(_format_node_line(x) for x in items)
        lines.append("")
        lines.append("—— 平台库人设树节点 ——")
        if not platform_items:
            lines.append(
                "（无：未找到条件概率达标的节点）"
            )
        else:
            lines.extend(_format_node_line(x) for x in platform_items)

        output = "\n".join(lines)
        return ToolResult(
            title=f"条件概率节点 ({account_name}, 阈值={conditional_ratio_threshold})",
            output=output,
            metadata={
                "account_name": account_name,
                "threshold": conditional_ratio_threshold,
                "match_score_threshold": float(match_score_threshold),
                "top_n": top_n,
                "quota": {
                    "account_quota": account_quota,
                    "account_with_match": len([x for x in items if isinstance(x.get("帖子选题点匹配"), list)]),
                    "account_without_match": len([x for x in items if not isinstance(x.get("帖子选题点匹配"), list)]),
                    "platform_quota": platform_quota,
                    "platform_with_match": len([x for x in platform_items if isinstance(x.get("帖子选题点匹配"), list)]),
                    "platform_without_match": len([x for x in platform_items if not isinstance(x.get("帖子选题点匹配"), list)]),
                },
                "account_tree_count": len(items),
                "platform_tree_count": len(platform_items),
                "count": len(items) + len(platform_items),
                "round": int(round),
                "log_id": str(log_id).strip() if log_id else "",
                "dimension_filter": {
                    "derived_dim_nodes": derived_dim_names,
                    "allowed_descendant_count": len(allowed) if allowed is not None else None,
                    "source": dim_source or ("未提供 log_id，未按维度收窄" if not (log_id and str(log_id).strip()) else ""),
                },
            },
        )
    except Exception as e:
        return ToolResult(
            title="按条件概率查询节点失败",
            output=str(e),
            error=str(e),
        )


def main() -> None:
    """本地测试：用家有大志账号测常量节点与条件概率节点，有 agent 时再跑一遍 tool 接口。"""
    import asyncio

    account_name = "家有大志"
    post_id = "68fb6a5c000000000302e5de"
    log_id = "20260319134630"
    round = 4
    # derived_items = [
    #     {"topic": "分享", "source_node": "分享"},
    #     {"topic": "叙事结构", "source_node": "叙事结构"},
    # ]
    derived_items = [{"topic":"推广","source_node":"推广"},{"topic":"视觉调性","source_node":"视觉调性"}]
    conditional_ratio_threshold = 0.2
    top_n = 200

    # # 1）常量节点（核心函数，无匹配）
    # constant_nodes = get_constant_nodes(account_name)
    # print(f"账号: {account_name} — 常量节点共 {len(constant_nodes)} 个（前 50 个）:")
    # for x in constant_nodes[:50]:
    #     print(f"  - {x['节点名称']}\t概率={x['概率']}\t{x['常量类型']}")
    # print()
    #
    # # 2）条件概率节点（核心函数）
    # derived_list = _parse_derived_list(derived_items)
    # ratio_nodes = get_nodes_by_conditional_ratio(
    #     account_name, derived_list, conditional_ratio_threshold, top_n
    # )
    # print(f"条件概率节点 阈值={conditional_ratio_threshold}, top_n={top_n}, 共 {len(ratio_nodes)} 个:")
    # for x in ratio_nodes:
    #     print(f"  - {x['节点名称']}\t条件概率={x['条件概率']}\t父节点={x['父节点名称']}")
    # print()

    # 3）有 agent 时通过 tool 接口再跑一遍（含帖子选题点匹配）
    if ToolResult is not None:
        async def run_tools():
            r1 = await find_tree_constant_nodes(account_name, post_id=post_id)
            print("--- find_tree_constant_nodes ---")
            print(r1.output[:2000] + "..." if len(r1.output) > 2000 else r1.output)
            r2 = await find_tree_nodes_by_conditional_ratio(
                account_name,
                post_id=post_id,
                derived_items=derived_items,
                conditional_ratio_threshold=conditional_ratio_threshold,
                top_n=top_n,
                round=round,
                log_id=log_id,
            )
            print("\n--- find_tree_nodes_by_conditional_ratio ---")
            print(r2.output)
        asyncio.run(run_tools())


if __name__ == "__main__":
    main()