sanguo_moziplus_v2/src/blackboard/queries.py

"""黑板读操作（查询封装）"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any, Dict, List, Optional

from .db import get_connection
from .models import Task


class Queries:
    """黑板查询（只读）"""

    def __init__(self, db_path: Path):
        self.db_path = db_path

    def _conn(self):
        return get_connection(self.db_path)

    def task_summary(self) -> Dict[str, int]:
        """任务状态汇总"""
        conn = self._conn()
        try:
            rows = conn.execute(
                "SELECT status, COUNT(*) as cnt FROM tasks GROUP BY status"
            ).fetchall()
            return {r["status"]: r["cnt"] for r in rows}
        finally:
            conn.close()

    def tasks_by_assignee(self, assignee: str) -> List[Task]:
        """查询某 Agent 的任务"""
        conn = self._conn()
        try:
            rows = conn.execute(
                "SELECT * FROM tasks WHERE assignee=? ORDER BY priority ASC",
                (assignee,),
            ).fetchall()
            return [Task.from_row(r) for r in rows]
        finally:
            conn.close()

    def blocked_tasks_with_deps(self) -> List[Dict[str, Any]]:
        """查询 blocked 任务及其依赖"""
        conn = self._conn()
        try:
            rows = conn.execute(
                "SELECT * FROM tasks WHERE status='blocked'"
            ).fetchall()
            result = []
            for r in rows:
                deps = json.loads(r["depends_on"] or "[]")
                if deps:
                    dep_rows = conn.execute(
                        f"SELECT id, status FROM tasks WHERE id IN ({','.join('?' * len(deps))})",
                        deps,
                    ).fetchall()
                    dep_info = {dr["id"]: dr["status"] for dr in dep_rows}
                    all_done = all(s == "done" for s in dep_info.values())
                else:
                    dep_info = {}
                    all_done = True
                result.append({
                    "task_id": r["id"],
                    "title": r["title"],
                    "depends_on": deps,
                    "dep_status": dep_info,
                    "all_deps_done": all_done,
                })
            return result
        finally:
            conn.close()

    def tasks_by_status(self, status: str) -> List[Task]:
        """查询指定状态的所有任务"""
        conn = self._conn()
        try:
            rows = conn.execute(
                "SELECT * FROM tasks WHERE status=? ORDER BY priority ASC",
                (status,),
            ).fetchall()
            return [Task.from_row(r) for r in rows]
        finally:
            conn.close()

    def pending_dispatchable(self) -> List[Task]:
        """查询可调度的 pending 任务（依赖已满足）"""
        conn = self._conn()
        try:
            rows = conn.execute(
                "SELECT * FROM tasks WHERE status='pending' ORDER BY priority ASC"
            ).fetchall()
            result = []
            for r in rows:
                deps = json.loads(r["depends_on"] or "[]")
                if not deps:
                    result.append(Task.from_row(r))
                    continue
                # 检查依赖是否全部完成
                placeholders = ",".join("?" * len(deps))
                dep_rows = conn.execute(
                    f"SELECT id FROM tasks WHERE id IN ({placeholders}) AND status='done'",
                    deps,
                ).fetchall()
                if len(dep_rows) == len(deps):
                    result.append(Task.from_row(r))
            return result
        finally:
            conn.close()

    def recent_events(self, limit: int = 20) -> List[Dict[str, Any]]:
        """最近事件"""
        conn = self._conn()
        try:
            rows = conn.execute(
                "SELECT * FROM events ORDER BY created_at DESC LIMIT ?",
                (limit,),
            ).fetchall()
            return [dict(r) for r in rows]
        finally:
            conn.close()

    def task_detail(self, task_id: str) -> Optional[Dict[str, Any]]:
        """任务详情聚合（含关联数据）"""
        conn = self._conn()
        try:
            row = conn.execute("SELECT * FROM tasks WHERE id=?", (task_id,)).fetchone()
            if not row:
                return None
            task = dict(row)
            # 关联评论数 + 产出数
            task["comments_count"] = conn.execute(
                "SELECT COUNT(*) FROM comments WHERE task_id=?", (task_id,)
            ).fetchone()[0]
            task["outputs_count"] = conn.execute(
                "SELECT COUNT(*) FROM outputs WHERE task_id=?", (task_id,)
            ).fetchone()[0]
            # 最新审查状态
            rev_row = conn.execute(
                "SELECT verdict FROM reviews WHERE task_id=? ORDER BY created_at DESC LIMIT 1",
                (task_id,),
            ).fetchone()
            task["review_status"] = rev_row["verdict"] if rev_row else None
            # 最新事件
            evt_row = conn.execute(
                "SELECT detail FROM events WHERE task_id=? ORDER BY created_at DESC LIMIT 1",
                (task_id,),
            ).fetchone()
            task["latest_event_detail"] = evt_row["detail"] if evt_row else None
            return task
        finally:
            conn.close()

    def task_events(self, task_id: str, limit: int = 50) -> List[Dict[str, Any]]:
        """任务事件列表"""
        conn = self._conn()
        try:
            rows = conn.execute(
                "SELECT * FROM events WHERE task_id=? ORDER BY created_at DESC LIMIT ?",
                (task_id, limit),
            ).fetchall()
            return [dict(r) for r in rows]
        finally:
            conn.close()

    def task_experiences(self, task_id: str) -> List[Dict[str, Any]]:
        """任务关联经验"""
        conn = self._conn()
        try:
            rows = conn.execute(
                """SELECT e.*, GROUP_CONCAT(et.tag) as tags
                FROM experiences e
                LEFT JOIN experience_tags et ON e.experience_id = et.experience_id
                WHERE e.task_id=?
                GROUP BY e.experience_id
                ORDER BY e.created_at DESC""",
                (task_id,),
            ).fetchall()
            return [dict(r) for r in rows]
        finally:
            conn.close()

    def db_size_bytes(self) -> int:
        return self.db_path.stat().st_size if self.db_path.exists() else 0