Agent/examples/piaoquan_demand/pattern_service.py

"""Pattern Mining 核心服务 - 挖掘+存储+查询"""
import json
import time
import traceback
from datetime import datetime

from sqlalchemy import insert, select, bindparam

from db_manager import DatabaseManager
from models import (
    Base, Post, TopicPatternExecution, TopicPatternMiningConfig,
    TopicPatternItemset, TopicPatternItemsetItem,
    TopicPatternCategory, TopicPatternElement,
)

db = DatabaseManager()


def _rebuild_source_data_from_db(session, execution_id: int) -> dict:
    """从 topic_pattern_element 表重建 source_data 格式

    Returns:
        {post_id: {点类型: [{点: str, 实质: [{名称, 分类路径}], 形式: [...], 意图: [...]}]}}
    """
    rows = session.query(TopicPatternElement).filter(
        TopicPatternElement.execution_id == execution_id
    ).all()

    # 先按 (post_id, point_type, point_text) 分组
    from collections import defaultdict
    post_point_map = defaultdict(lambda: defaultdict(lambda: defaultdict(
        lambda: {'实质': [], '形式': [], '意图': []}
    )))

    for r in rows:
        point_key = r.point_text or ''
        dim_name = r.element_type  # 实质/形式/意图
        if dim_name not in ('实质', '形式', '意图'):
            continue
        path_list = r.category_path.split('>') if r.category_path else []
        post_point_map[r.post_id][r.point_type][point_key][dim_name].append({
            '名称': r.name,
            '详细描述': r.description or '',
            '分类路径': path_list,
        })

    # 转换为 source_data 格式
    source_data = {}
    for post_id, point_types in post_point_map.items():
        post_data = {}
        for point_type, points in point_types.items():
            post_data[point_type] = [
                {'点': point_text, **dims}
                for point_text, dims in points.items()
            ]
        source_data[post_id] = post_data

    return source_data


def _upsert_post_metadata(session, post_metadata: dict):
    """将 API 返回的 post_metadata upsert 到 Post 表

    Args:
        post_metadata: {post_id: {"account_name": ..., "merge_leve2": ..., "platform": ...}}
    """
    if not post_metadata:
        return

    # 查出已存在的 post_id
    existing_post_ids = set()
    all_pids = list(post_metadata.keys())
    BATCH = 500
    for i in range(0, len(all_pids), BATCH):
        batch = all_pids[i:i + BATCH]
        rows = session.query(Post.post_id).filter(Post.post_id.in_(batch)).all()
        existing_post_ids.update(r[0] for r in rows)

    # 新增的直接 insert
    new_dicts = []
    update_dicts = []
    for pid, meta in post_metadata.items():
        if pid in existing_post_ids:
            update_dicts.append((pid, meta))
        else:
            new_dicts.append({
                'post_id': pid,
                'account_name': meta.get('account_name'),
                'merge_leve2': meta.get('merge_leve2'),
                'platform': meta.get('platform'),
            })

    if new_dicts:
        batch_size = 1000
        for i in range(0, len(new_dicts), batch_size):
            session.execute(insert(Post), new_dicts[i:i + batch_size])

    # 已存在的更新
    for pid, meta in update_dicts:
        session.query(Post).filter(Post.post_id == pid).update({
            'account_name': meta.get('account_name'),
            'merge_leve2': meta.get('merge_leve2'),
            'platform': meta.get('platform'),
        })

    session.flush()
    print(f"[Post metadata] upsert 完成: 新增 {len(new_dicts)}, 更新 {len(update_dicts)}")


def ensure_tables():
    """确保表存在"""
    Base.metadata.create_all(db.engine)


def _parse_target_depth(target_depth):
    """解析 target_depth（纯数字转为 int）"""
    try:
        return int(target_depth)
    except (ValueError, TypeError):
        return target_depth


def _parse_item_string(item_str: str, dimension_mode: str) -> dict:
    """解析 item 字符串，提取点类型、维度、分类路径、元素名称

    item 格式（根据 dimension_mode）：
      full:                 点类型_维度_路径  或  点类型_维度_路径||名称
      point_type_only:      点类型_路径      或  点类型_路径||名称
      substance_form_only:  维度_路径        或  维度_路径||名称

    Returns:
        {'point_type', 'dimension', 'category_path', 'element_name'}
    """
    point_type = None
    dimension = None
    element_name = None

    if dimension_mode == 'full':
        parts = item_str.split('_', 2)
        if len(parts) >= 3:
            point_type, dimension, path_part = parts
        else:
            path_part = item_str
    elif dimension_mode == 'point_type_only':
        parts = item_str.split('_', 1)
        if len(parts) >= 2:
            point_type, path_part = parts
        else:
            path_part = item_str
    elif dimension_mode == 'substance_form_only':
        parts = item_str.split('_', 1)
        if len(parts) >= 2:
            dimension, path_part = parts
        else:
            path_part = item_str
    else:
        path_part = item_str

    # 分离路径和元素名称
    if '||' in path_part:
        category_path, element_name = path_part.split('||', 1)
    else:
        category_path = path_part

    return {
        'point_type': point_type,
        'dimension': dimension,
        'category_path': category_path or None,
        'element_name': element_name,
    }


def _store_category_tree_snapshot(session, execution_id: int, categories: list, source_data: dict):
    """存储分类树快照 + 帖子级元素记录（代替 data_cache JSON）

    Args:
        session: DB session
        execution_id: 执行 ID
        categories: API 返回的分类列表 [{stable_id, name, description, ...}, ...]
        source_data: 帖子元素数据 {post_id: {灵感点: [{点:..., 实质:[{名称, 详细描述, 分类路径}], ...}], ...}}
    """
    t_snapshot_start = time.time()

    # ── 1. 插入分类树节点 ──
    t0 = time.time()
    cat_dicts = []
    if categories:
        for c in categories:
            cat_dicts.append({
                'execution_id': execution_id,
                'source_stable_id': c['stable_id'],
                'source_type': c['source_type'],
                'name': c['name'],
                'description': c.get('description') or None,
                'category_nature': c.get('category_nature'),
                'path': c.get('path'),
                'level': c.get('level'),
                'parent_id': None,
                'parent_source_stable_id': c.get('parent_stable_id'),
                'element_count': 0,
            })

        # 批量 INSERT（利用 insertmanyvalues 合并为多行 VALUES）
        batch_size = 1000
        for i in range(0, len(cat_dicts), batch_size):
            session.execute(insert(TopicPatternCategory), cat_dicts[i:i + batch_size])
        session.flush()

        # 查回所有刚插入的行，建立 stable_id → (id, row_data) 映射
        inserted_rows = session.execute(
            select(TopicPatternCategory)
            .where(TopicPatternCategory.execution_id == execution_id)
        ).scalars().all()

        stable_id_to_row = {row.source_stable_id: row for row in inserted_rows}

        # 批量回填 parent_id
        updates = []
        for row in inserted_rows:
            if row.parent_source_stable_id and row.parent_source_stable_id in stable_id_to_row:
                parent = stable_id_to_row[row.parent_source_stable_id]
                updates.append({'_id': row.id, '_parent_id': parent.id})
                row.parent_id = parent.id  # 同步更新内存对象

        if updates:
            session.connection().execute(
                TopicPatternCategory.__table__.update()
                .where(TopicPatternCategory.__table__.c.id == bindparam('_id'))
                .values(parent_id=bindparam('_parent_id')),
                updates
            )

    print(f"[Execution {execution_id}] 写入分类树节点: {len(cat_dicts)} 条, 耗时 {time.time() - t0:.2f}s")

    # 构建 path → TopicPatternCategory 映射（path 格式: "/食品/水果" → 用于匹配分类路径）
    path_to_cat = {}
    if categories:
        for row in inserted_rows:
            if row.path:
                path_to_cat[row.path] = row

    # ── 2. 从 source_data 逐帖子逐元素写入 TopicPatternElement ──
    t0 = time.time()
    elem_dicts = []
    cat_elem_counts = {}  # category_id → count

    for post_id, post_data in source_data.items():
        for point_type in ['灵感点', '目的点', '关键点']:
            for point in post_data.get(point_type, []):
                point_text = point.get('点', '')
                for elem_type in ['实质', '形式', '意图']:
                    for elem in point.get(elem_type, []):
                        path_list = elem.get('分类路径', [])
                        path_str = '/' + '/'.join(path_list) if path_list else None
                        cat_row = path_to_cat.get(path_str) if path_str else None
                        path_label = '>'.join(path_list) if path_list else None

                        elem_dicts.append({
                            'execution_id': execution_id,
                            'post_id': post_id,
                            'point_type': point_type,
                            'point_text': point_text,
                            'element_type': elem_type,
                            'name': elem.get('名称', ''),
                            'description': elem.get('详细描述') or None,
                            'category_id': cat_row.id if cat_row else None,
                            'category_path': path_label,
                        })

                        if cat_row:
                            cat_elem_counts[cat_row.id] = cat_elem_counts.get(cat_row.id, 0) + 1

    t_build = time.time() - t0
    print(f"[Execution {execution_id}] 构建元素字典: {len(elem_dicts)} 条, 耗时 {t_build:.2f}s")

    # 批量写入元素（Core insert 利用 insertmanyvalues 合并多行）
    t0 = time.time()
    if elem_dicts:
        batch_size = 5000
        for i in range(0, len(elem_dicts), batch_size):
            session.execute(insert(TopicPatternElement), elem_dicts[i:i + batch_size])

    # 回填分类节点的 element_count（批量 UPDATE）
    if cat_elem_counts:
        elem_count_updates = [{'_id': cat_id, '_count': count} for cat_id, count in cat_elem_counts.items()]
        session.connection().execute(
            TopicPatternCategory.__table__.update()
            .where(TopicPatternCategory.__table__.c.id == bindparam('_id'))
            .values(element_count=bindparam('_count')),
            elem_count_updates
        )

    session.commit()
    t_write = time.time() - t0
    print(f"[Execution {execution_id}] 写入元素到DB: {len(elem_dicts)} 条, "
          f"batch_size=5000, 批次={len(elem_dicts) // 5000 + (1 if len(elem_dicts) % 5000 else 0)}, "
          f"耗时 {t_write:.2f}s")
    print(f"[Execution {execution_id}] 分类树快照总计: {len(cat_dicts)} 个分类, {len(elem_dicts)} 个元素, "
          f"总耗时 {time.time() - t_snapshot_start:.2f}s")

    # 返回 path → category row 映射，供后续 itemset item 关联使用
    return path_to_cat



# ==================== 删除/重建 ====================

def delete_execution_results(execution_id: int):
    """删除频繁项集结果（保留 execution 配置记录）"""
    session = db.get_session()
    try:
        # 删除 itemset items（通过 itemset_id 关联）
        itemset_ids = [r.id for r in session.query(TopicPatternItemset.id).filter(
            TopicPatternItemset.execution_id == execution_id
        ).all()]
        if itemset_ids:
            session.query(TopicPatternItemsetItem).filter(
                TopicPatternItemsetItem.itemset_id.in_(itemset_ids)
            ).delete(synchronize_session=False)

        # 删除 itemsets
        session.query(TopicPatternItemset).filter(
            TopicPatternItemset.execution_id == execution_id
        ).delete(synchronize_session=False)

        # 删除 mining configs
        session.query(TopicPatternMiningConfig).filter(
            TopicPatternMiningConfig.execution_id == execution_id
        ).delete(synchronize_session=False)

        # 删除 elements
        session.query(TopicPatternElement).filter(
            TopicPatternElement.execution_id == execution_id
        ).delete(synchronize_session=False)

        # 删除 categories
        session.query(TopicPatternCategory).filter(
            TopicPatternCategory.execution_id == execution_id
        ).delete(synchronize_session=False)

        # 更新 execution 状态
        exe = session.query(TopicPatternExecution).filter(
            TopicPatternExecution.id == execution_id
        ).first()
        if exe:
            exe.status = 'deleted'
            exe.itemset_count = 0
            exe.post_count = None
            exe.end_time = None

        session.commit()

        invalidate_graph_cache(execution_id)

        return True
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()




# ==================== 查询接口 ====================

def get_executions(page: int = 1, page_size: int = 20):
    """获取执行列表"""
    session = db.get_session()
    try:
        total = session.query(TopicPatternExecution).count()
        rows = session.query(TopicPatternExecution).order_by(
            TopicPatternExecution.id.desc()
        ).offset((page - 1) * page_size).limit(page_size).all()

        return {
            'total': total,
            'page': page,
            'page_size': page_size,
            'executions': [_execution_to_dict(e) for e in rows],
        }
    finally:
        session.close()


def get_execution_detail(execution_id: int):
    """获取执行详情"""
    session = db.get_session()
    try:
        exe = session.query(TopicPatternExecution).filter(
            TopicPatternExecution.id == execution_id
        ).first()
        if not exe:
            return None
        return _execution_to_dict(exe)
    finally:
        session.close()


def get_itemsets(execution_id: int, combination_type: str = None,
                 min_support: int = None, page: int = 1, page_size: int = 50,
                 sort_by: str = 'absolute_support', mining_config_id: int = None,
                 itemset_id: int = None, dimension_mode: str = None):
    """查询项集"""
    session = db.get_session()
    try:
        query = session.query(TopicPatternItemset).filter(
            TopicPatternItemset.execution_id == execution_id
        )
        if itemset_id:
            query = query.filter(TopicPatternItemset.id == itemset_id)
        if mining_config_id:
            query = query.filter(TopicPatternItemset.mining_config_id == mining_config_id)
        elif dimension_mode:
            # 按维度模式筛选：找到该模式下所有 config_id
            config_ids = [c[0] for c in session.query(TopicPatternMiningConfig.id).filter(
                TopicPatternMiningConfig.execution_id == execution_id,
                TopicPatternMiningConfig.dimension_mode == dimension_mode,
            ).all()]
            if config_ids:
                query = query.filter(TopicPatternItemset.mining_config_id.in_(config_ids))
            else:
                return {'total': 0, 'page': page, 'page_size': page_size, 'itemsets': []}
        if combination_type:
            query = query.filter(TopicPatternItemset.combination_type == combination_type)
        if min_support is not None:
            query = query.filter(TopicPatternItemset.absolute_support >= min_support)

        total = query.count()

        if sort_by == 'support':
            query = query.order_by(TopicPatternItemset.support.desc())
        elif sort_by == 'item_count':
            query = query.order_by(TopicPatternItemset.item_count.desc(), TopicPatternItemset.absolute_support.desc())
        else:
            query = query.order_by(TopicPatternItemset.absolute_support.desc())

        rows = query.offset((page - 1) * page_size).limit(page_size).all()

        # 批量加载 items
        itemset_ids = [r.id for r in rows]
        all_items = session.query(TopicPatternItemsetItem).filter(
            TopicPatternItemsetItem.itemset_id.in_(itemset_ids)
        ).all() if itemset_ids else []

        items_by_itemset = {}
        for it in all_items:
            items_by_itemset.setdefault(it.itemset_id, []).append(_itemset_item_to_dict(it))

        return {
            'total': total,
            'page': page,
            'page_size': page_size,
            'itemsets': [_itemset_to_dict(r, items=items_by_itemset.get(r.id, [])) for r in rows],
        }
    finally:
        session.close()


def get_itemset_posts(itemset_ids):
    """获取一个或多个项集的匹配帖子和结构化 items

    Args:
        itemset_ids: 单个 int 或 int 列表

    Returns:
        列表，每项含 id, dimension_mode, target_depth, items, post_ids, absolute_support
    """
    if isinstance(itemset_ids, int):
        itemset_ids = [itemset_ids]

    session = db.get_session()
    try:
        itemsets = session.query(TopicPatternItemset).filter(
            TopicPatternItemset.id.in_(itemset_ids)
        ).all()
        if not itemsets:
            return []

        # 批量加载 mining_config 信息
        config_ids = set(r.mining_config_id for r in itemsets)
        configs = session.query(TopicPatternMiningConfig).filter(
            TopicPatternMiningConfig.id.in_(config_ids)
        ).all() if config_ids else []
        config_map = {c.id: c for c in configs}

        # 批量加载所有 items
        all_items = session.query(TopicPatternItemsetItem).filter(
            TopicPatternItemsetItem.itemset_id.in_(itemset_ids)
        ).all()
        items_by_itemset = {}
        for it in all_items:
            items_by_itemset.setdefault(it.itemset_id, []).append(it)

        # 按传入顺序组装结果
        id_to_itemset = {r.id: r for r in itemsets}
        results = []
        for iid in itemset_ids:
            r = id_to_itemset.get(iid)
            if not r:
                continue
            cfg = config_map.get(r.mining_config_id)
            results.append({
                'id': r.id,
                'dimension_mode': cfg.dimension_mode if cfg else None,
                'target_depth': cfg.target_depth if cfg else None,
                'item_count': r.item_count,
                'absolute_support': r.absolute_support,
                'support': r.support,
                'items': [_itemset_item_to_dict(it) for it in items_by_itemset.get(iid, [])],
                'post_ids': r.matched_post_ids or [],
            })
        return results
    finally:
        session.close()


def get_combination_types(execution_id: int, mining_config_id: int = None):
    """获取某执行下的 combination_type 列表及计数"""
    session = db.get_session()
    try:
        from sqlalchemy import func
        query = session.query(
            TopicPatternItemset.combination_type,
            func.count(TopicPatternItemset.id).label('count'),
        ).filter(
            TopicPatternItemset.execution_id == execution_id
        )
        if mining_config_id:
            query = query.filter(TopicPatternItemset.mining_config_id == mining_config_id)
        rows = query.group_by(
            TopicPatternItemset.combination_type
        ).order_by(
            func.count(TopicPatternItemset.id).desc()
        ).all()

        return [{'combination_type': r[0], 'count': r[1]} for r in rows]
    finally:
        session.close()


def get_category_tree(execution_id: int, source_type: str = None):
    """获取某次执行的分类树快照

    Returns:
        {
            "categories": [...],  # 平铺的分类节点列表
            "tree": [...],        # 树状结构（嵌套 children）
            "element_count": N,   # 元素总数
        }
    """
    session = db.get_session()
    try:
        # 查询分类节点
        cat_query = session.query(TopicPatternCategory).filter(
            TopicPatternCategory.execution_id == execution_id
        )
        if source_type:
            cat_query = cat_query.filter(TopicPatternCategory.source_type == source_type)
        categories = cat_query.all()

        # 按 category_id + name 统计元素（含 post_ids）
        from collections import defaultdict
        from sqlalchemy import func
        elem_query = session.query(
            TopicPatternElement.category_id,
            TopicPatternElement.name,
            TopicPatternElement.post_id,
        ).filter(
            TopicPatternElement.execution_id == execution_id,
        )
        if source_type:
            elem_query = elem_query.filter(TopicPatternElement.element_type == source_type)
        elem_rows = elem_query.all()

        # 聚合: (category_id, name) → {count, post_ids}
        elem_agg = defaultdict(lambda: defaultdict(lambda: {'count': 0, 'post_ids': set()}))
        for cat_id, name, post_id in elem_rows:
            elem_agg[cat_id][name]['count'] += 1
            elem_agg[cat_id][name]['post_ids'].add(post_id)

        cat_elements = defaultdict(list)
        for cat_id, names in elem_agg.items():
            for name, data in names.items():
                cat_elements[cat_id].append({
                    'name': name,
                    'count': data['count'],
                    'post_ids': sorted(data['post_ids']),
                })

        # 构建平铺列表 + 树
        cat_list = []
        by_id = {}
        for c in categories:
            node = {
                'id': c.id,
                'source_stable_id': c.source_stable_id,
                'source_type': c.source_type,
                'name': c.name,
                'description': c.description,
                'category_nature': c.category_nature,
                'path': c.path,
                'level': c.level,
                'parent_id': c.parent_id,
                'element_count': c.element_count,
                'elements': cat_elements.get(c.id, []),
                'children': [],
            }
            cat_list.append(node)
            by_id[c.id] = node

        # 建树
        roots = []
        for node in cat_list:
            if node['parent_id'] and node['parent_id'] in by_id:
                by_id[node['parent_id']]['children'].append(node)
            else:
                roots.append(node)

        # 递归计算子树元素总数
        def sum_elements(node):
            total = node['element_count']
            for child in node['children']:
                total += sum_elements(child)
            node['total_element_count'] = total
            return total

        for root in roots:
            sum_elements(root)

        total_elements = sum(len(v) for v in cat_elements.values())
        return {
            'categories': [{k: v for k, v in n.items() if k != 'children'} for n in cat_list],
            'tree': roots,
            'category_count': len(cat_list),
            'element_count': total_elements,
        }
    finally:
        session.close()


def get_category_tree_compact(execution_id: int, source_type: str = None) -> str:
    """构建紧凑文本格式的分类树（节省token）

    格式示例:
      [12] 食品 [实质] (5个元素) — 各类食品相关内容
        [13] 水果 [实质] (3个元素) — 水果类
        [14] 蔬菜 [实质] (2个元素) — 蔬菜类
    """
    tree_data = get_category_tree(execution_id, source_type=source_type)
    roots = tree_data.get('tree', [])

    lines = []
    lines.append(f"分类数: {tree_data['category_count']}  元素数: {tree_data['element_count']}")
    lines.append("")

    def _render(nodes, indent=0):
        for node in nodes:
            prefix = "  " * indent
            desc_preview = ""
            if node.get("description"):
                desc = node["description"]
                desc_preview = f" — {desc[:30]}..." if len(desc) > 30 else f" — {desc}"
            nature_tag = f"[{node['category_nature']}]" if node.get("category_nature") else ""
            elem_count = node.get('element_count', 0)
            total_count = node.get('total_element_count', elem_count)
            count_info = f"({elem_count}个元素)" if elem_count == total_count else f"({elem_count}个元素, 含子树共{total_count})"
            # 只列出元素名称，不含 post_ids
            elem_names = [e['name'] for e in node.get('elements', [])]
            elem_str = ""
            if elem_names:
                if len(elem_names) <= 5:
                    elem_str = f" 元素: {', '.join(elem_names)}"
                else:
                    elem_str = f" 元素: {', '.join(elem_names[:5])}...等{len(elem_names)}个"
            lines.append(f"{prefix}[{node['id']}] {node['name']} {nature_tag} {count_info}{desc_preview}{elem_str}")
            if node.get('children'):
                _render(node['children'], indent + 1)

    _render(roots)
    return "\n".join(lines) if lines else "(空树)"


def get_category_elements(category_id: int, execution_id: int = None,
                          account_name: str = None, merge_leve2: str = None):
    """获取某个分类节点下的元素列表（按名称去重聚合，附带来源帖子）"""
    session = db.get_session()
    try:
        from sqlalchemy import func

        # 按名称聚合，统计出现次数 + 去重帖子数
        from sqlalchemy import func
        query = session.query(
            TopicPatternElement.name,
            TopicPatternElement.element_type,
            TopicPatternElement.category_path,
            func.count(TopicPatternElement.id).label('occurrence_count'),
            func.count(func.distinct(TopicPatternElement.post_id)).label('post_count'),
            func.group_concat(func.distinct(TopicPatternElement.point_type)).label('point_types'),
        ).filter(
            TopicPatternElement.category_id == category_id
        )
        # JOIN Post 表做 DB 侧过滤
        query = _apply_post_filter(query, session, account_name, merge_leve2)

        rows = query.group_by(
            TopicPatternElement.name,
            TopicPatternElement.element_type,
            TopicPatternElement.category_path,
        ).order_by(
            func.count(TopicPatternElement.id).desc()
        ).all()

        return [{
            'name': r.name,
            'element_type': r.element_type,
            'point_types': sorted(r.point_types.split(',')) if r.point_types else [],
            'category_path': r.category_path,
            'occurrence_count': r.occurrence_count,
            'post_count': r.post_count,
        } for r in rows]
    finally:
        session.close()


def get_execution_post_ids(execution_id: int, search: str = None):
    """获取某执行下的所有去重帖子ID列表，支持按ID搜索"""
    session = db.get_session()
    try:
        query = session.query(TopicPatternElement.post_id).filter(
            TopicPatternElement.execution_id == execution_id,
        ).distinct()

        if search:
            query = query.filter(TopicPatternElement.post_id.like(f'%{search}%'))

        post_ids = sorted([r[0] for r in query.all()])
        return {'post_ids': post_ids, 'total': len(post_ids)}
    finally:
        session.close()


def get_post_elements(execution_id: int, post_ids: list):
    """获取指定帖子的元素数据，按帖子ID分组，每个帖子按点类型→元素类型组织

    Returns:
        {post_id: {point_type: [{point_text, elements: {实质: [...], 形式: [...], 意图: [...]}}]}}
    """
    session = db.get_session()
    try:
        from collections import defaultdict

        rows = session.query(TopicPatternElement).filter(
            TopicPatternElement.execution_id == execution_id,
            TopicPatternElement.post_id.in_(post_ids),
        ).all()

        # 组织: post_id → point_type → point_text → element_type → [elements]
        raw = defaultdict(lambda: defaultdict(lambda: defaultdict(lambda: defaultdict(list))))
        for r in rows:
            raw[r.post_id][r.point_type][r.point_text or ''][r.element_type].append({
                'name': r.name,
                'category_path': r.category_path,
                'description': r.description,
            })

        # 转换为列表结构
        result = {}
        for post_id, point_types in raw.items():
            post_points = {}
            for pt, points in point_types.items():
                pt_list = []
                for point_text, elem_types in points.items():
                    pt_list.append({
                        'point_text': point_text,
                        'elements': {
                            '实质': elem_types.get('实质', []),
                            '形式': elem_types.get('形式', []),
                            '意图': elem_types.get('意图', []),
                        }
                    })
                post_points[pt] = pt_list
            result[post_id] = post_points

        return result
    finally:
        session.close()


# ==================== Item Graph 缓存 + 渐进式查询 ====================

_graph_cache = {}  # key: (execution_id, mining_config_id) → graph dict




def invalidate_graph_cache(execution_id: int = None):
    """清除 graph 缓存"""
    if execution_id is None:
        _graph_cache.clear()
    else:
        keys_to_remove = [k for k in _graph_cache if k[0] == execution_id]
        for k in keys_to_remove:
            del _graph_cache[k]


def compute_item_graph_nodes(execution_id: int, mining_config_id: int = None):
    """返回所有节点（meta + edge_summary），不含边详情"""
    graph, config, error = _get_or_compute_graph(execution_id, mining_config_id)
    if error:
        return {'error': error}
    if graph is None:
        return None

    nodes = {}
    for item_key, item_data in graph.items():
        edge_summary = {'co_in_post': 0, 'hierarchy': 0}
        for target, edge_types in item_data.get('edges', {}).items():
            if 'co_in_post' in edge_types:
                edge_summary['co_in_post'] += 1
            if 'hierarchy' in edge_types:
                edge_summary['hierarchy'] += 1
        nodes[item_key] = {
            'meta': item_data['meta'],
            'edge_summary': edge_summary,
        }

    return {'nodes': nodes}


def compute_item_graph_edges(execution_id: int, item_key: str, mining_config_id: int = None):
    """返回指定节点的所有边"""
    graph, config, error = _get_or_compute_graph(execution_id, mining_config_id)
    if error:
        return {'error': error}
    if graph is None:
        return None

    item_data = graph.get(item_key)
    if not item_data:
        return {'error': f'未找到节点: {item_key}'}

    return {'item': item_key, 'edges': item_data.get('edges', {})}


# ==================== 序列化 ====================

def get_mining_configs(execution_id: int):
    """获取某执行下的 mining config 列表"""
    session = db.get_session()
    try:
        rows = session.query(TopicPatternMiningConfig).filter(
            TopicPatternMiningConfig.execution_id == execution_id
        ).all()
        return [_mining_config_to_dict(r) for r in rows]
    finally:
        session.close()


def _execution_to_dict(e):
    return {
        'id': e.id,
        'merge_leve2': e.merge_leve2,
        'platform': e.platform,
        'account_name': e.account_name,
        'post_limit': e.post_limit,
        'min_absolute_support': e.min_absolute_support,
        'classify_execution_id': e.classify_execution_id,
        'mining_configs': e.mining_configs,
        'post_count': e.post_count,
        'itemset_count': e.itemset_count,
        'status': e.status,
        'error_message': e.error_message,
        'start_time': e.start_time.isoformat() if e.start_time else None,
        'end_time': e.end_time.isoformat() if e.end_time else None,
    }


def _mining_config_to_dict(c):
    return {
        'id': c.id,
        'execution_id': c.execution_id,
        'dimension_mode': c.dimension_mode,
        'target_depth': c.target_depth,
        'transaction_count': c.transaction_count,
        'itemset_count': c.itemset_count,
    }


def _itemset_to_dict(r, items=None):
    d = {
        'id': r.id,
        'execution_id': r.execution_id,
        'combination_type': r.combination_type,
        'item_count': r.item_count,
        'support': r.support,
        'absolute_support': r.absolute_support,
        'dimensions': r.dimensions,
        'is_cross_point': r.is_cross_point,
        'matched_post_ids': r.matched_post_ids,
    }
    if items is not None:
        d['items'] = items
    return d


def _itemset_item_to_dict(it):
    return {
        'id': it.id,
        'point_type': it.point_type,
        'dimension': it.dimension,
        'category_id': it.category_id,
        'category_path': it.category_path,
        'element_name': it.element_name,
    }


def _to_list(value):
    """将 str 或 list 统一转为 list，None 保持 None"""
    if value is None:
        return None
    if isinstance(value, str):
        return [value]
    return list(value)


def _apply_post_filter(query, session, account_name=None, merge_leve2=None):
    """对 TopicPatternElement 查询追加 Post 表 JOIN 过滤（DB 侧完成，不加载到内存）。

    account_name / merge_leve2 支持 str（单个）或 list（多个，OR 逻辑）。
    如果无需筛选，原样返回 query。
    """
    names = _to_list(account_name)
    leve2s = _to_list(merge_leve2)
    if not names and not leve2s:
        return query

    query = query.join(Post, TopicPatternElement.post_id == Post.post_id)
    if names:
        query = query.filter(Post.account_name.in_(names))
    if leve2s:
        query = query.filter(Post.merge_leve2.in_(leve2s))

    return query


def _get_filtered_post_ids_set(session, account_name=None, merge_leve2=None):
    """从 Post 表筛选帖子ID，返回 Python set。

    account_name / merge_leve2 支持 str（单个）或 list（多个，OR 逻辑）。
    仅用于需要在内存中做集合运算的场景（如 JSON matched_post_ids 交集、co-occurrence 交集）。
    """
    names = _to_list(account_name)
    leve2s = _to_list(merge_leve2)
    if not names and not leve2s:
        return None

    query = session.query(Post.post_id)
    if names:
        query = query.filter(Post.account_name.in_(names))
    if leve2s:
        query = query.filter(Post.merge_leve2.in_(leve2s))

    return set(r[0] for r in query.all())


# ==================== TopicBuild Agent 专用查询 ====================

def get_top_itemsets(
    execution_id: int,
    top_n: int = 20,
    mining_config_id: int = None,
    combination_type: str = None,
    min_support: int = None,
    is_cross_point: bool = None,
    min_item_count: int = None,
    max_item_count: int = None,
    sort_by: str = 'absolute_support',
):
    """获取 Top N 项集（直接返回前 N 条，不分页）

    比 get_itemsets 更适合 Agent 场景：直接拿到最有价值的 Top N，不需要翻页。
    """
    session = db.get_session()
    try:
        query = session.query(TopicPatternItemset).filter(
            TopicPatternItemset.execution_id == execution_id
        )
        if mining_config_id:
            query = query.filter(TopicPatternItemset.mining_config_id == mining_config_id)
        if combination_type:
            query = query.filter(TopicPatternItemset.combination_type == combination_type)
        if min_support is not None:
            query = query.filter(TopicPatternItemset.absolute_support >= min_support)
        if is_cross_point is not None:
            query = query.filter(TopicPatternItemset.is_cross_point == is_cross_point)
        if min_item_count is not None:
            query = query.filter(TopicPatternItemset.item_count >= min_item_count)
        if max_item_count is not None:
            query = query.filter(TopicPatternItemset.item_count <= max_item_count)

        # 排序
        if sort_by == 'support':
            query = query.order_by(TopicPatternItemset.support.desc())
        elif sort_by == 'item_count':
            query = query.order_by(TopicPatternItemset.item_count.desc(),
                                   TopicPatternItemset.absolute_support.desc())
        else:
            query = query.order_by(TopicPatternItemset.absolute_support.desc())

        total = query.count()
        rows = query.limit(top_n).all()

        # 批量加载 items
        itemset_ids = [r.id for r in rows]
        all_items = session.query(TopicPatternItemsetItem).filter(
            TopicPatternItemsetItem.itemset_id.in_(itemset_ids)
        ).all() if itemset_ids else []

        items_by_itemset = {}
        for it in all_items:
            items_by_itemset.setdefault(it.itemset_id, []).append(_itemset_item_to_dict(it))

        return {
            'total': total,
            'showing': len(rows),
            'itemsets': [_itemset_to_dict(r, items=items_by_itemset.get(r.id, [])) for r in rows],
        }
    finally:
        session.close()


def search_top_itemsets(
    execution_id: int,
    category_ids: list = None,
    dimension_mode: str = None,
    top_n: int = 20,
    min_support: int = None,
    min_item_count: int = None,
    max_item_count: int = None,
    sort_by: str = 'absolute_support',
    account_name: str = None,
    merge_leve2: str = None,
):
    """搜索频繁项集（Agent 专用，精简返回）

    - category_ids 为空：返回所有 depth 下的 Top N
    - category_ids 非空：返回同时包含所有指定分类的项集（跨 depth 汇总）
    - dimension_mode：按挖掘维度模式筛选（full/substance_form_only/point_type_only）
    返回精简字段，不含 matched_post_ids。
    支持按 account_name / merge_leve2 筛选帖子范围，过滤后重算 support。
    """
    from sqlalchemy import distinct, func

    session = db.get_session()
    try:
        # 帖子筛选集合（需要 set 做 JSON matched_post_ids 交集）
        filtered_post_ids = _get_filtered_post_ids_set(session, account_name, merge_leve2)

        # 获取适用的 mining_config 列表
        config_query = session.query(TopicPatternMiningConfig).filter(
            TopicPatternMiningConfig.execution_id == execution_id,
        )
        if dimension_mode:
            config_query = config_query.filter(TopicPatternMiningConfig.dimension_mode == dimension_mode)
        all_configs = config_query.all()
        if not all_configs:
            return {'total': 0, 'showing': 0, 'groups': {}}

        config_map = {c.id: c for c in all_configs}

        # 按 mining_config_id 分别查询，每组各取 top_n
        groups = {}
        total = 0
        filtered_supports = {}  # itemset_id -> filtered_absolute_support（跨组共享）

        for cfg in all_configs:
            dm = cfg.dimension_mode
            td = cfg.target_depth
            group_key = f"{dm}/{td}"

            # 构建该 config 的基础查询
            if category_ids:
                subq = session.query(TopicPatternItemsetItem.itemset_id).join(
                    TopicPatternItemset,
                    TopicPatternItemsetItem.itemset_id == TopicPatternItemset.id,
                ).filter(
                    TopicPatternItemset.mining_config_id == cfg.id,
                    TopicPatternItemsetItem.category_id.in_(category_ids),
                ).group_by(
                    TopicPatternItemsetItem.itemset_id
                ).having(
                    func.count(distinct(TopicPatternItemsetItem.category_id)) >= len(category_ids)
                ).subquery()

                query = session.query(TopicPatternItemset).filter(
                    TopicPatternItemset.id.in_(session.query(subq.c.itemset_id))
                )
            else:
                query = session.query(TopicPatternItemset).filter(
                    TopicPatternItemset.mining_config_id == cfg.id,
                )

            if min_support is not None:
                query = query.filter(TopicPatternItemset.absolute_support >= min_support)
            if min_item_count is not None:
                query = query.filter(TopicPatternItemset.item_count >= min_item_count)
            if max_item_count is not None:
                query = query.filter(TopicPatternItemset.item_count <= max_item_count)

            # 排序
            if sort_by == 'support':
                order_clauses = [TopicPatternItemset.support.desc()]
            elif sort_by == 'item_count':
                order_clauses = [TopicPatternItemset.item_count.desc(),
                                 TopicPatternItemset.absolute_support.desc()]
            else:
                order_clauses = [TopicPatternItemset.absolute_support.desc()]
            query = query.order_by(*order_clauses)

            if filtered_post_ids is not None:
                # 流式扫描：只加载 id + matched_post_ids
                scan_query = session.query(
                    TopicPatternItemset.id,
                    TopicPatternItemset.matched_post_ids,
                ).filter(
                    TopicPatternItemset.id.in_(query.with_entities(TopicPatternItemset.id).subquery())
                ).order_by(*order_clauses)

                SCAN_BATCH = 200
                matched_ids = []
                scan_offset = 0
                while True:
                    batch = scan_query.offset(scan_offset).limit(SCAN_BATCH).all()
                    if not batch:
                        break
                    for item_id, mpids in batch:
                        matched = set(mpids or []) & filtered_post_ids
                        if matched:
                            filtered_supports[item_id] = len(matched)
                            matched_ids.append(item_id)
                    scan_offset += SCAN_BATCH
                    if len(matched_ids) >= top_n * 3:
                        break

                # 按筛选后的 support 重新排序
                if sort_by == 'support':
                    matched_ids.sort(key=lambda i: filtered_supports[i] / max(len(filtered_post_ids), 1), reverse=True)
                elif sort_by != 'item_count':
                    matched_ids.sort(key=lambda i: filtered_supports[i], reverse=True)

                group_total = len(matched_ids)
                selected_ids = matched_ids[:top_n]
                group_rows = session.query(TopicPatternItemset).filter(
                    TopicPatternItemset.id.in_(selected_ids)
                ).all() if selected_ids else []
                row_map = {r.id: r for r in group_rows}
                group_rows = [row_map[i] for i in selected_ids if i in row_map]
            else:
                group_total = query.count()
                group_rows = query.limit(top_n).all()

            total += group_total

            if not group_rows:
                continue

            # 批量加载该组的 items
            group_itemset_ids = [r.id for r in group_rows]
            group_items = session.query(
                TopicPatternItemsetItem.itemset_id,
                TopicPatternItemsetItem.point_type,
                TopicPatternItemsetItem.dimension,
                TopicPatternItemsetItem.category_id,
                TopicPatternItemsetItem.category_path,
                TopicPatternItemsetItem.element_name,
            ).filter(
                TopicPatternItemsetItem.itemset_id.in_(group_itemset_ids)
            ).all()

            items_by_itemset = {}
            for it in group_items:
                slim = {
                    'point_type': it.point_type,
                    'dimension': it.dimension,
                    'category_id': it.category_id,
                    'category_path': it.category_path,
                }
                if it.element_name:
                    slim['element_name'] = it.element_name
                items_by_itemset.setdefault(it.itemset_id, []).append(slim)

            itemsets_out = []
            for r in group_rows:
                itemset_out = {
                    'id': r.id,
                    'item_count': r.item_count,
                    'absolute_support': filtered_supports[r.id] if filtered_post_ids is not None and r.id in filtered_supports else r.absolute_support,
                    'support': r.support,
                    'items': items_by_itemset.get(r.id, []),
                }
                if filtered_post_ids is not None and r.id in filtered_supports:
                    itemset_out['original_absolute_support'] = r.absolute_support
                itemsets_out.append(itemset_out)

            groups[group_key] = {
                'dimension_mode': dm,
                'target_depth': td,
                'total': group_total,
                'itemsets': itemsets_out,
            }

        showing = sum(len(g['itemsets']) for g in groups.values())
        return {'total': total, 'showing': showing, 'groups': groups}
    finally:
        session.close()


def get_category_co_occurrences(
    execution_id: int,
    category_ids: list,
    top_n: int = 30,
    account_name: str = None,
    merge_leve2: str = None,
):
    """查询多个分类的共现关系

    找到同时包含所有指定分类下元素的帖子，统计这些帖子中其他分类的出现频率。
    支持叠加多分类，结果为同时满足所有分类共现的交集。

    Returns:
        {"matched_post_count", "input_categories": [...], "co_categories": [...]}
    """
    from sqlalchemy import distinct

    session = db.get_session()
    try:
        # 帖子筛选（需要 set 做交集运算）
        filtered_post_ids = _get_filtered_post_ids_set(session, account_name, merge_leve2)

        # 1. 对每个分类ID找到包含该分类元素的帖子集合，取交集
        post_sets = []
        input_cat_infos = []
        for cat_id in category_ids:
            # 获取分类信息
            cat = session.query(TopicPatternCategory).filter(
                TopicPatternCategory.id == cat_id,
            ).first()
            if cat:
                input_cat_infos.append({
                    'category_id': cat.id, 'name': cat.name, 'path': cat.path,
                })

            rows = session.query(distinct(TopicPatternElement.post_id)).filter(
                TopicPatternElement.execution_id == execution_id,
                TopicPatternElement.category_id == cat_id,
            ).all()
            post_sets.append(set(r[0] for r in rows))

        if not post_sets:
            return {'matched_post_count': 0, 'input_categories': input_cat_infos, 'co_categories': []}

        common_post_ids = post_sets[0]
        for s in post_sets[1:]:
            common_post_ids &= s

        # 应用帖子筛选
        if filtered_post_ids is not None:
            common_post_ids &= filtered_post_ids

        if not common_post_ids:
            return {'matched_post_count': 0, 'input_categories': input_cat_infos, 'co_categories': []}

        # 2. 在 DB 侧按分类聚合统计（排除输入分类自身）
        from sqlalchemy import func

        common_post_list = list(common_post_ids)

        # 分批 UNION 查询避免 IN 列表过长，DB 侧 GROUP BY 聚合
        BATCH = 500
        cat_stats = {}  # category_id -> {category_id, category_path, element_type, count, post_count}
        for i in range(0, len(common_post_list), BATCH):
            batch = common_post_list[i:i + BATCH]
            rows = session.query(
                TopicPatternElement.category_id,
                TopicPatternElement.category_path,
                TopicPatternElement.element_type,
                func.count(TopicPatternElement.id).label('cnt'),
                func.count(func.distinct(TopicPatternElement.post_id)).label('post_count'),
            ).filter(
                TopicPatternElement.execution_id == execution_id,
                TopicPatternElement.post_id.in_(batch),
                TopicPatternElement.category_id.isnot(None),
                ~TopicPatternElement.category_id.in_(category_ids),
            ).group_by(
                TopicPatternElement.category_id,
                TopicPatternElement.category_path,
                TopicPatternElement.element_type,
            ).all()

            for r in rows:
                key = r.category_id
                if key not in cat_stats:
                    cat_stats[key] = {
                        'category_id': r.category_id,
                        'category_path': r.category_path,
                        'element_type': r.element_type,
                        'count': 0,
                        'post_count': 0,
                    }
                cat_stats[key]['count'] += r.cnt
                cat_stats[key]['post_count'] += r.post_count  # 跨批次近似值，足够排序用

        # 3. 补充分类名称（只查需要的列）
        cat_ids_to_lookup = list(cat_stats.keys())
        if cat_ids_to_lookup:
            cats = session.query(
                TopicPatternCategory.id, TopicPatternCategory.name,
            ).filter(
                TopicPatternCategory.id.in_(cat_ids_to_lookup),
            ).all()
            cat_name_map = {c.id: c.name for c in cats}
            for cs in cat_stats.values():
                cs['name'] = cat_name_map.get(cs['category_id'], '')

        # 4. 按出现帖子数排序
        co_categories = sorted(cat_stats.values(), key=lambda x: x['post_count'], reverse=True)[:top_n]

        return {
            'matched_post_count': len(common_post_ids),
            'input_categories': input_cat_infos,
            'co_categories': co_categories,
        }
    finally:
        session.close()


def get_element_co_occurrences(
    execution_id: int,
    element_names: list,
    top_n: int = 30,
    account_name: str = None,
    merge_leve2: str = None,
):
    """查询多个元素的共现关系

    找到同时包含所有指定元素的帖子，统计这些帖子中其他元素的出现频率。
    支持叠加多元素，结果为同时满足所有元素共现的交集。

    Returns:
        {"matched_post_count", "co_elements": [{"name", "element_type", "category_path", "count", "post_ids"}, ...]}
    """
    from sqlalchemy import distinct, func

    session = db.get_session()
    try:
        # 帖子筛选（需要 set 做交集运算）
        filtered_post_ids = _get_filtered_post_ids_set(session, account_name, merge_leve2)

        # 1. 对每个元素名找到包含它的帖子集合，取交集
        post_sets = []
        for name in element_names:
            rows = session.query(distinct(TopicPatternElement.post_id)).filter(
                TopicPatternElement.execution_id == execution_id,
                TopicPatternElement.name == name,
            ).all()
            post_sets.append(set(r[0] for r in rows))

        if not post_sets:
            return {'matched_post_count': 0, 'co_elements': []}

        common_post_ids = post_sets[0]
        for s in post_sets[1:]:
            common_post_ids &= s

        # 应用帖子筛选
        if filtered_post_ids is not None:
            common_post_ids &= filtered_post_ids

        if not common_post_ids:
            return {'matched_post_count': 0, 'co_elements': []}

        # 2. 在 DB 侧按元素聚合统计（排除输入元素自身）
        from sqlalchemy import func

        common_post_list = list(common_post_ids)

        # 分批查询 + DB 侧 GROUP BY 聚合，避免加载全量行到内存
        BATCH = 500
        element_stats = {}  # (name, element_type) -> {...}
        for i in range(0, len(common_post_list), BATCH):
            batch = common_post_list[i:i + BATCH]
            rows = session.query(
                TopicPatternElement.name,
                TopicPatternElement.element_type,
                TopicPatternElement.category_path,
                TopicPatternElement.category_id,
                func.count(TopicPatternElement.id).label('cnt'),
                func.count(func.distinct(TopicPatternElement.post_id)).label('post_count'),
                func.group_concat(func.distinct(TopicPatternElement.point_type)).label('point_types'),
            ).filter(
                TopicPatternElement.execution_id == execution_id,
                TopicPatternElement.post_id.in_(batch),
                ~TopicPatternElement.name.in_(element_names),
            ).group_by(
                TopicPatternElement.name,
                TopicPatternElement.element_type,
                TopicPatternElement.category_path,
                TopicPatternElement.category_id,
            ).all()

            for r in rows:
                key = (r.name, r.element_type)
                if key not in element_stats:
                    element_stats[key] = {
                        'name': r.name,
                        'element_type': r.element_type,
                        'category_path': r.category_path,
                        'category_id': r.category_id,
                        'count': 0,
                        'post_count': 0,
                        '_point_types': set(),
                    }
                element_stats[key]['count'] += r.cnt
                element_stats[key]['post_count'] += r.post_count
                if r.point_types:
                    element_stats[key]['_point_types'].update(r.point_types.split(','))

        # 3. 转换 point_types set 为 sorted list，按出现帖子数排序
        for es in element_stats.values():
            es['point_types'] = sorted(es.pop('_point_types'))
        co_elements = sorted(element_stats.values(), key=lambda x: x['post_count'], reverse=True)[:top_n]

        return {
            'matched_post_count': len(common_post_ids),
            'input_elements': element_names,
            'co_elements': co_elements,
        }
    finally:
        session.close()


def search_itemsets_by_category(
    execution_id: int,
    category_id: int = None,
    category_path: str = None,
    include_subtree: bool = False,
    dimension: str = None,
    point_type: str = None,
    top_n: int = 20,
    sort_by: str = 'absolute_support',
):
    """查找包含某个特定分类的项集

    通过 JOIN TopicPatternItemsetItem 筛选包含指定分类的项集。
    支持按 category_id 精确匹配，也支持按 category_path 前缀匹配（include_subtree=True 时匹配子树）。

    Returns:
        {"total", "showing", "itemsets": [...]}
    """
    session = db.get_session()
    try:
        from sqlalchemy import distinct

        # 先找出符合条件的 itemset_id
        item_query = session.query(distinct(TopicPatternItemsetItem.itemset_id)).join(
            TopicPatternItemset,
            TopicPatternItemsetItem.itemset_id == TopicPatternItemset.id,
        ).filter(
            TopicPatternItemset.execution_id == execution_id
        )

        if category_id is not None:
            item_query = item_query.filter(TopicPatternItemsetItem.category_id == category_id)
        elif category_path is not None:
            if include_subtree:
                # 前缀匹配: "食品" 匹配 "食品>水果", "食品>水果>苹果" 等
                item_query = item_query.filter(
                    TopicPatternItemsetItem.category_path.like(f"{category_path}%")
                )
            else:
                item_query = item_query.filter(
                    TopicPatternItemsetItem.category_path == category_path
                )

        if dimension:
            item_query = item_query.filter(TopicPatternItemsetItem.dimension == dimension)
        if point_type:
            item_query = item_query.filter(TopicPatternItemsetItem.point_type == point_type)

        matched_ids = [r[0] for r in item_query.all()]

        if not matched_ids:
            return {'total': 0, 'showing': 0, 'itemsets': []}

        # 查询这些 itemset 的完整信息
        query = session.query(TopicPatternItemset).filter(
            TopicPatternItemset.id.in_(matched_ids)
        )
        if sort_by == 'support':
            query = query.order_by(TopicPatternItemset.support.desc())
        elif sort_by == 'item_count':
            query = query.order_by(TopicPatternItemset.item_count.desc(),
                                   TopicPatternItemset.absolute_support.desc())
        else:
            query = query.order_by(TopicPatternItemset.absolute_support.desc())

        total = len(matched_ids)
        rows = query.limit(top_n).all()

        # 批量加载 items
        itemset_ids = [r.id for r in rows]
        all_items = session.query(TopicPatternItemsetItem).filter(
            TopicPatternItemsetItem.itemset_id.in_(itemset_ids)
        ).all() if itemset_ids else []

        items_by_itemset = {}
        for it in all_items:
            items_by_itemset.setdefault(it.itemset_id, []).append(_itemset_item_to_dict(it))

        return {
            'total': total,
            'showing': len(rows),
            'itemsets': [_itemset_to_dict(r, items=items_by_itemset.get(r.id, [])) for r in rows],
        }
    finally:
        session.close()


def search_elements(execution_id: int, keyword: str, element_type: str = None, limit: int = 50,
                    account_name: str = None, merge_leve2: str = None):
    """按名称关键词搜索元素，返回去重聚合结果（附带分类信息）"""
    session = db.get_session()
    try:
        from sqlalchemy import func

        query = session.query(
            TopicPatternElement.name,
            TopicPatternElement.element_type,
            TopicPatternElement.category_id,
            TopicPatternElement.category_path,
            func.count(TopicPatternElement.id).label('occurrence_count'),
            func.count(func.distinct(TopicPatternElement.post_id)).label('post_count'),
            func.group_concat(func.distinct(TopicPatternElement.point_type)).label('point_types'),
        ).filter(
            TopicPatternElement.execution_id == execution_id,
            TopicPatternElement.name.like(f"%{keyword}%"),
        )
        if element_type:
            query = query.filter(TopicPatternElement.element_type == element_type)
        # JOIN Post 表做 DB 侧过滤
        query = _apply_post_filter(query, session, account_name, merge_leve2)

        rows = query.group_by(
            TopicPatternElement.name,
            TopicPatternElement.element_type,
            TopicPatternElement.category_id,
            TopicPatternElement.category_path,
        ).order_by(
            func.count(TopicPatternElement.id).desc()
        ).limit(limit).all()

        return [{
            'name': r.name,
            'element_type': r.element_type,
            'point_types': sorted(r.point_types.split(',')) if r.point_types else [],
            'category_id': r.category_id,
            'category_path': r.category_path,
            'occurrence_count': r.occurrence_count,
            'post_count': r.post_count,
        } for r in rows]
    finally:
        session.close()


def get_category_by_id(category_id: int):
    """获取单个分类节点详情"""
    session = db.get_session()
    try:
        cat = session.query(TopicPatternCategory).filter(
            TopicPatternCategory.id == category_id
        ).first()
        if not cat:
            return None
        return {
            'id': cat.id,
            'source_stable_id': cat.source_stable_id,
            'source_type': cat.source_type,
            'name': cat.name,
            'description': cat.description,
            'category_nature': cat.category_nature,
            'path': cat.path,
            'level': cat.level,
            'parent_id': cat.parent_id,
            'element_count': cat.element_count,
        }
    finally:
        session.close()


def get_category_detail_with_context(execution_id: int, category_id: int):
    """获取分类节点的完整上下文: 自身信息 + 祖先链 + 子节点 + 元素列表"""
    session = db.get_session()
    try:
        from sqlalchemy import func

        cat = session.query(TopicPatternCategory).filter(
            TopicPatternCategory.id == category_id
        ).first()
        if not cat:
            return None

        # 祖先链（向上回溯到根）
        ancestors = []
        current = cat
        while current.parent_id:
            parent = session.query(TopicPatternCategory).filter(
                TopicPatternCategory.id == current.parent_id
            ).first()
            if not parent:
                break
            ancestors.insert(0, {
                'id': parent.id, 'name': parent.name,
                'path': parent.path, 'level': parent.level,
            })
            current = parent

        # 直接子节点
        children = session.query(TopicPatternCategory).filter(
            TopicPatternCategory.parent_id == category_id,
            TopicPatternCategory.execution_id == execution_id,
        ).all()
        children_list = [{
            'id': c.id, 'name': c.name, 'path': c.path,
            'level': c.level, 'element_count': c.element_count,
        } for c in children]

        # 元素列表（去重聚合）
        elem_rows = session.query(
            TopicPatternElement.name,
            TopicPatternElement.element_type,
            func.count(TopicPatternElement.id).label('occurrence_count'),
            func.count(func.distinct(TopicPatternElement.post_id)).label('post_count'),
            func.group_concat(func.distinct(TopicPatternElement.point_type)).label('point_types'),
        ).filter(
            TopicPatternElement.category_id == category_id,
        ).group_by(
            TopicPatternElement.name,
            TopicPatternElement.element_type,
        ).order_by(
            func.count(TopicPatternElement.id).desc()
        ).limit(100).all()

        elements = [{
            'name': r.name, 'element_type': r.element_type,
            'point_types': sorted(r.point_types.split(',')) if r.point_types else [],
            'occurrence_count': r.occurrence_count, 'post_count': r.post_count,
        } for r in elem_rows]

        # 同级兄弟节点（同 parent_id）
        siblings = []
        if cat.parent_id:
            sibling_rows = session.query(TopicPatternCategory).filter(
                TopicPatternCategory.parent_id == cat.parent_id,
                TopicPatternCategory.execution_id == execution_id,
                TopicPatternCategory.id != category_id,
            ).all()
            siblings = [{
                'id': s.id, 'name': s.name, 'path': s.path,
                'element_count': s.element_count,
            } for s in sibling_rows]

        return {
            'category': {
                'id': cat.id, 'name': cat.name, 'description': cat.description,
                'source_type': cat.source_type, 'category_nature': cat.category_nature,
                'path': cat.path, 'level': cat.level, 'element_count': cat.element_count,
            },
            'ancestors': ancestors,
            'children': children_list,
            'siblings': siblings,
            'elements': elements,
        }
    finally:
        session.close()


def search_categories(execution_id: int, keyword: str, source_type: str = None, limit: int = 30):
    """按名称关键词搜索分类节点，附带该分类涉及的 point_type 列表"""
    session = db.get_session()
    try:
        query = session.query(TopicPatternCategory).filter(
            TopicPatternCategory.execution_id == execution_id,
            TopicPatternCategory.name.like(f"%{keyword}%"),
        )
        if source_type:
            query = query.filter(TopicPatternCategory.source_type == source_type)

        rows = query.limit(limit).all()
        if not rows:
            return []

        # 批量查询每个分类涉及的 point_type
        cat_ids = [c.id for c in rows]
        pt_rows = session.query(
            TopicPatternElement.category_id,
            TopicPatternElement.point_type,
        ).filter(
            TopicPatternElement.category_id.in_(cat_ids),
        ).group_by(
            TopicPatternElement.category_id,
            TopicPatternElement.point_type,
        ).all()
        pt_by_cat = {}
        for cat_id, pt in pt_rows:
            pt_by_cat.setdefault(cat_id, set()).add(pt)

        return [{
            'id': c.id, 'name': c.name, 'description': c.description,
            'source_type': c.source_type, 'category_nature': c.category_nature,
            'path': c.path, 'level': c.level, 'element_count': c.element_count,
            'parent_id': c.parent_id,
            'point_types': sorted(pt_by_cat.get(c.id, [])),
        } for c in rows]
    finally:
        session.close()


def get_element_category_chain(execution_id: int, element_name: str, element_type: str = None):
    """从元素名称反查其所属分类链

    返回该元素出现在哪些分类下，以及每个分类的完整祖先路径。
    """
    session = db.get_session()
    try:
        from sqlalchemy import func

        query = session.query(
            TopicPatternElement.category_id,
            TopicPatternElement.category_path,
            TopicPatternElement.element_type,
            func.count(TopicPatternElement.id).label('occurrence_count'),
            func.count(func.distinct(TopicPatternElement.post_id)).label('post_count'),
        ).filter(
            TopicPatternElement.execution_id == execution_id,
            TopicPatternElement.name == element_name,
        )
        if element_type:
            query = query.filter(TopicPatternElement.element_type == element_type)

        rows = query.group_by(
            TopicPatternElement.category_id,
            TopicPatternElement.category_path,
            TopicPatternElement.element_type,
        ).all()

        results = []
        for r in rows:
            # 获取分类节点详情
            cat_info = None
            ancestors = []
            if r.category_id:
                cat = session.query(TopicPatternCategory).filter(
                    TopicPatternCategory.id == r.category_id
                ).first()
                if cat:
                    cat_info = {
                        'id': cat.id, 'name': cat.name,
                        'path': cat.path, 'level': cat.level,
                        'source_type': cat.source_type,
                    }
                    # 回溯祖先
                    current = cat
                    while current.parent_id:
                        parent = session.query(TopicPatternCategory).filter(
                            TopicPatternCategory.id == current.parent_id
                        ).first()
                        if not parent:
                            break
                        ancestors.insert(0, {
                            'id': parent.id, 'name': parent.name,
                            'path': parent.path, 'level': parent.level,
                        })
                        current = parent

            results.append({
                'category_id': r.category_id,
                'category_path': r.category_path,
                'element_type': r.element_type,
                'occurrence_count': r.occurrence_count,
                'post_count': r.post_count,
                'category': cat_info,
                'ancestors': ancestors,
            })

        return results
    finally:
        session.close()