nas-tools/claude-code-tracker/summarizer.py

import os
import logging
from datetime import datetime, timedelta
import asyncio
import aiosqlite
from anthropic import AsyncAnthropic

import db

logger = logging.getLogger(__name__)

ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY")
ANTHROPIC_BASE_URL = os.environ.get("ANTHROPIC_BASE_URL", "https://www.bytecatcode.org")
SUMMARY_HOUR = int(os.environ.get("SUMMARY_HOUR", "23"))
TRIGGER_PATH = os.environ.get("TRIGGER_PATH", "/app/data/trigger")


async def generate_summary_for_date(date_str: str):
    """Generate summary for a specific date"""
    try:
        logger.info(f"Generating summary for {date_str}")

        # Query conversations for this date
        async with aiosqlite.connect(db.DB_PATH) as conn:
            # Get conversations
            cursor = await conn.execute("""
                SELECT session_id, project_path, slug, message_count,
                       total_input_tokens, total_output_tokens
                FROM conversations
                WHERE DATE(started_at) = ? OR DATE(last_updated_at) = ?
                ORDER BY started_at
            """, (date_str, date_str))
            conversations = await cursor.fetchall()

            if not conversations:
                logger.info(f"No conversations found for {date_str}")
                return

            # Get sample messages from each conversation
            conversation_summaries = []
            total_messages = 0
            total_tokens = 0
            projects = set()

            for conv in conversations:
                session_id, project_path, slug, msg_count, input_tokens, output_tokens = conv
                projects.add(project_path)
                total_messages += msg_count
                total_tokens += (input_tokens + output_tokens)

                # Get first few user messages to understand what was worked on
                cursor = await conn.execute("""
                    SELECT content_text FROM messages
                    WHERE session_id = ? AND role = 'user' AND content_text IS NOT NULL
                    ORDER BY timestamp
                    LIMIT 3
                """, (session_id,))
                user_messages = await cursor.fetchall()

                # Get tool usage
                cursor = await conn.execute("""
                    SELECT tool_name, COUNT(*) as count
                    FROM tool_calls
                    WHERE message_uuid IN (
                        SELECT message_uuid FROM messages WHERE session_id = ?
                    )
                    GROUP BY tool_name
                    ORDER BY count DESC
                    LIMIT 5
                """, (session_id,))
                tools = await cursor.fetchall()

                conversation_summaries.append({
                    'slug': slug or 'Untitled',
                    'project': project_path.split('/')[-1] if project_path else 'unknown',
                    'messages': msg_count,
                    'tokens': input_tokens + output_tokens,
                    'user_messages': [msg[0][:200] for msg in user_messages if msg[0]],
                    'tools': [f"{tool[0]} ({tool[1]}x)" for tool in tools]
                })

        # Format for Claude
        formatted_convs = []
        for i, conv in enumerate(conversation_summaries, 1):
            formatted_convs.append(f"""
**Conversation {i}: {conv['slug']}**
- Project: {conv['project']}
- Messages: {conv['messages']} ({conv['tokens']} tokens)
- Initial requests: {', '.join(conv['user_messages'][:2]) if conv['user_messages'] else 'N/A'}
- Tools used: {', '.join(conv['tools']) if conv['tools'] else 'None'}
""")

        prompt = f"""You are analyzing a developer's Claude Code conversations from {date_str}.

**Overview:**
- Conversations: {len(conversations)}
- Total messages: {total_messages}
- Total tokens: {total_tokens:,}
- Projects: {', '.join(sorted(projects))}

**Conversations:**
{''.join(formatted_convs)}

Generate a concise daily summary covering:
1. Main tasks and goals worked on
2. Key decisions and solutions implemented
3. Files and components modified (if evident from tool usage)
4. Challenges encountered and how they were resolved
5. Tools and patterns used
6. Overall progress and outcomes

Use markdown formatting. Be specific and technical. Focus on what was accomplished. Keep it under 500 words."""

        # Call Claude API
        client = AsyncAnthropic(api_key=ANTHROPIC_API_KEY, base_url=ANTHROPIC_BASE_URL)
        response = await client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=4096,
            messages=[{"role": "user", "content": prompt}]
        )

        summary_content = response.content[0].text

        # Store summary
        await db.upsert_summary(date_str, {
            'project_path': ', '.join(sorted(projects)),
            'conversation_count': len(conversations),
            'total_messages': total_messages,
            'total_tokens': total_tokens,
            'summary_content': summary_content
        })

        logger.info(f"Summary generated for {date_str}")

    except Exception as e:
        logger.error(f"Error generating summary for {date_str}: {e}", exc_info=True)


async def check_trigger():
    """Check for manual trigger file"""
    if os.path.exists(TRIGGER_PATH):
        try:
            os.remove(TRIGGER_PATH)
            logger.info("Manual trigger detected")
            # Generate summary for yesterday
            yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
            await generate_summary_for_date(yesterday)
        except Exception as e:
            logger.error(f"Error processing trigger: {e}")


async def summarize_daily():
    """Daily summarization loop"""
    while True:
        try:
            now = datetime.now()

            # Check for manual trigger
            await check_trigger()

            # Check if it's time for daily summary
            if now.hour == SUMMARY_HOUR and now.minute < 5:
                yesterday = (now - timedelta(days=1)).strftime("%Y-%m-%d")
                await generate_summary_for_date(yesterday)

                # Sleep until next hour to avoid duplicate runs
                await asyncio.sleep(3600)
            else:
                # Check every 5 minutes
                await asyncio.sleep(300)

        except Exception as e:
            logger.error(f"Error in summarization loop: {e}", exc_info=True)
            await asyncio.sleep(300)