nas-tools/dashboard/backend/routers/conversation_tracker.py

import os
import aiosqlite
from fastapi import APIRouter, HTTPException, Query
from typing import Optional

router = APIRouter(tags=["conversations"])

CONVERSATION_TRACKER_DB = os.environ.get("CONVERSATION_TRACKER_DB", "/app/data/claude-code-tracker/conversations.db")


async def _db():
    return aiosqlite.connect(CONVERSATION_TRACKER_DB)


@router.get("/dates")
async def list_dates():
    """List all dates with conversations"""
    async with await _db() as db:
        cursor = await db.execute("""
            SELECT DISTINCT DATE(started_at) as date
            FROM conversations
            ORDER BY date DESC
        """)
        return [r[0] for r in await cursor.fetchall()]


@router.get("/summary/{date}")
async def get_summary(date: str):
    """Get daily summary for a date"""
    async with await _db() as db:
        cursor = await db.execute("""
            SELECT summary_content, conversation_count, total_messages,
                   total_tokens, created_at, project_path
            FROM daily_summaries
            WHERE date = ?
        """, (date,))
        row = await cursor.fetchone()

    if not row:
        raise HTTPException(404, "No summary for this date")

    return {
        "date": date,
        "content": row[0],
        "conversation_count": row[1],
        "total_messages": row[2],
        "total_tokens": row[3],
        "created_at": row[4],
        "projects": row[5]
    }


@router.get("/list")
async def list_conversations(
    date: Optional[str] = None,
    project_path: Optional[str] = None,
    limit: int = Query(50, le=200),
    offset: int = 0
):
    """List conversations with filters"""
    async with await _db() as db:
        query = "SELECT session_id, project_path, slug, started_at, message_count, total_input_tokens, total_output_tokens, model FROM conversations WHERE 1=1"
        params = []

        if date:
            query += " AND DATE(started_at) = ?"
            params.append(date)

        if project_path:
            query += " AND project_path LIKE ?"
            params.append(f"%{project_path}%")

        query += " ORDER BY started_at DESC LIMIT ? OFFSET ?"
        params.extend([limit, offset])

        cursor = await db.execute(query, params)
        rows = await cursor.fetchall()

    return [{
        "session_id": r[0],
        "project_path": r[1],
        "slug": r[2],
        "started_at": r[3],
        "message_count": r[4],
        "total_tokens": r[5] + r[6],
        "model": r[7]
    } for r in rows]


@router.get("/search")
async def search_conversations(
    q: str = Query(..., min_length=2),
    date_from: Optional[str] = None,
    date_to: Optional[str] = None,
    limit: int = Query(50, le=200)
):
    """Search conversations by text"""
    async with await _db() as db:
        query = """
            SELECT DISTINCT m.session_id, c.project_path, c.slug, c.started_at,
                   c.message_count, c.total_input_tokens, c.total_output_tokens
            FROM messages m
            JOIN conversations c ON m.session_id = c.session_id
            WHERE m.content_text LIKE ?
        """
        params = [f"%{q}%"]

        if date_from:
            query += " AND DATE(c.started_at) >= ?"
            params.append(date_from)

        if date_to:
            query += " AND DATE(c.started_at) <= ?"
            params.append(date_to)

        query += " ORDER BY c.started_at DESC LIMIT ?"
        params.append(limit)

        cursor = await db.execute(query, params)
        rows = await cursor.fetchall()

    return [{
        "session_id": r[0],
        "project_path": r[1],
        "slug": r[2],
        "started_at": r[3],
        "message_count": r[4],
        "total_tokens": r[5] + r[6]
    } for r in rows]


@router.get("/stats")
async def get_stats(
    date_from: Optional[str] = None,
    date_to: Optional[str] = None
):
    """Get overall statistics"""
    async with await _db() as db:
        query = "SELECT COUNT(*), SUM(message_count), SUM(total_input_tokens + total_output_tokens) FROM conversations WHERE 1=1"
        params = []

        if date_from:
            query += " AND DATE(started_at) >= ?"
            params.append(date_from)

        if date_to:
            query += " AND DATE(started_at) <= ?"
            params.append(date_to)

        cursor = await db.execute(query, params)
        stats = await cursor.fetchone()

        # Get top projects
        cursor = await db.execute("""
            SELECT project_path, COUNT(*) as count
            FROM conversations
            GROUP BY project_path
            ORDER BY count DESC
            LIMIT 10
        """)
        projects = await cursor.fetchall()

        # Get tool usage
        cursor = await db.execute("""
            SELECT tool_name, COUNT(*) as count
            FROM tool_calls
            GROUP BY tool_name
            ORDER BY count DESC
            LIMIT 10
        """)
        tools = await cursor.fetchall()

    return {
        "total_conversations": stats[0] or 0,
        "total_messages": stats[1] or 0,
        "total_tokens": stats[2] or 0,
        "top_projects": [{"path": p[0], "count": p[1]} for p in projects],
        "top_tools": [{"name": t[0], "count": t[1]} for t in tools]
    }


@router.post("/trigger")
async def trigger_summary():
    """Trigger manual summary generation"""
    trigger_path = os.environ.get("TRIGGER_PATH", "/app/data/claude-code-tracker/trigger")

    # Validate path
    trigger_path = os.path.abspath(trigger_path)
    forbidden_prefixes = ("/etc/", "/root/", "/sys/", "/proc/", "/boot/")
    if any(trigger_path.startswith(prefix) for prefix in forbidden_prefixes):
        raise HTTPException(status_code=400, detail="Invalid trigger path")

    os.makedirs(os.path.dirname(trigger_path), exist_ok=True)
    with open(trigger_path, "w") as f:
        f.write("1")

    return {"status": "triggered"}


@router.get("/{session_id}")
async def get_conversation(session_id: str):
    """Get conversation details"""
    async with await _db() as db:
        # Get conversation metadata
        cursor = await db.execute("""
            SELECT project_path, git_branch, slug, started_at, last_updated_at,
                   message_count, total_input_tokens, total_output_tokens, model
            FROM conversations
            WHERE session_id = ?
        """, (session_id,))
        conv = await cursor.fetchone()

        if not conv:
            raise HTTPException(404, "Conversation not found")

        # Get message count
        cursor = await db.execute("""
            SELECT COUNT(*) FROM messages WHERE session_id = ?
        """, (session_id,))
        msg_count = (await cursor.fetchone())[0]

    return {
        "session_id": session_id,
        "project_path": conv[0],
        "git_branch": conv[1],
        "slug": conv[2],
        "started_at": conv[3],
        "last_updated_at": conv[4],
        "message_count": msg_count,
        "total_input_tokens": conv[6],
        "total_output_tokens": conv[7],
        "model": conv[8]
    }


@router.get("/{session_id}/messages")
async def get_messages(
    session_id: str,
    limit: int = Query(100, le=500),
    offset: int = 0
):
    """Get messages for a conversation"""
    async with await _db() as db:
        cursor = await db.execute("""
            SELECT message_uuid, role, content_text, timestamp, model,
                   input_tokens, output_tokens, has_tool_use, has_thinking
            FROM messages
            WHERE session_id = ?
            ORDER BY timestamp
            LIMIT ? OFFSET ?
        """, (session_id, limit, offset))
        rows = await cursor.fetchall()

        messages = []
        for r in rows:
            msg = {
                "uuid": r[0],
                "role": r[1],
                "content": r[2],
                "timestamp": r[3],
                "model": r[4],
                "input_tokens": r[5],
                "output_tokens": r[6],
                "has_tool_use": bool(r[7]),
                "has_thinking": bool(r[8]),
                "tool_calls": []
            }

            # Get tool calls for this message
            if msg["has_tool_use"]:
                cursor2 = await db.execute("""
                    SELECT tool_name, tool_input, tool_result, is_error
                    FROM tool_calls
                    WHERE message_uuid = ?
                """, (r[0],))
                tool_rows = await cursor2.fetchall()
                msg["tool_calls"] = [{
                    "name": tr[0],
                    "input": tr[1],
                    "result": tr[2],
                    "is_error": bool(tr[3])
                } for tr in tool_rows]

            messages.append(msg)

    return messages