task_hook.py

Code Hygiene Score: 80

Keine Issues gefunden.

Dependencies 13

Funktionen 6

Code

#!/usr/bin/env python3
"""
Task Hook für Claude Code
Erstellt automatisch Tasks aus bestimmten Patterns in User-Prompts

Erkannte Patterns:
- TODO: <text>
- TASK: <text>
- @task <text>
- #task <text>
"""

import json
import os
import sys
import re
import uuid
import pymysql
from pathlib import Path
from datetime import datetime
from typing import Dict, Any, Optional, List

# .env aus Hook-Verzeichnis laden
from dotenv import load_dotenv
load_dotenv(Path(__file__).parent / '.env')

# Konfiguration aus Environment-Variablen
DB_CONFIG = {
    'host': os.environ.get('CLAUDE_DB_HOST', 'localhost'),
    'port': int(os.environ.get('CLAUDE_DB_PORT', '3306')),
    'user': os.environ.get('CLAUDE_DB_USER', 'root'),
    'password': os.environ.get('CLAUDE_DB_PASSWORD', ''),
    'database': os.environ.get('CLAUDE_DB_NAME', 'ki_dev'),
    'charset': 'utf8mb4'
}

# Task-Erkennungsmuster
TASK_PATTERNS = [
    (re.compile(r'TODO:\s*(.+?)(?:\n|$)', re.IGNORECASE), 'todo'),
    (re.compile(r'TASK:\s*(.+?)(?:\n|$)', re.IGNORECASE), 'task'),
    (re.compile(r'@task\s+(.+?)(?:\n|$)', re.IGNORECASE), 'mention'),
    (re.compile(r'#task\s+(.+?)(?:\n|$)', re.IGNORECASE), 'hashtag'),
]


def get_connection():
    """Erstellt Datenbankverbindung"""
    return pymysql.connect(**DB_CONFIG)


def generate_uuid() -> str:
    """Generiert eine UUID"""
    return str(uuid.uuid4())


def extract_tasks(text: str) -> List[Dict[str, str]]:
    """Extrahiert Tasks aus Text"""
    tasks = []

    for pattern, source in TASK_PATTERNS:
        matches = pattern.findall(text)
        for match in matches:
            title = match.strip()
            if title and len(title) >= 3:  # Mindestlänge
                tasks.append({
                    'title': title[:255],  # Max 255 Zeichen
                    'source': source
                })

    return tasks


def create_task(title: str, description: str, created_by: str) -> Optional[int]:
    """Erstellt einen neuen Task in der Datenbank"""
    try:
        connection = get_connection()

        with connection.cursor() as cursor:
            task_uuid = generate_uuid()
            now = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')

            sql = """
                INSERT INTO tasks (
                    uuid, title, description, type, priority, status,
                    created_by, created_by_type, created_at, updated_at
                ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            """

            cursor.execute(sql, (
                task_uuid,
                title,
                description,
                'ai_task',
                'medium',
                'pending',
                created_by,
                'ai',
                now,
                now
            ))

            task_id = cursor.lastrowid

            # Kommentar hinzufügen
            comment_sql = """
                INSERT INTO task_comments (
                    task_id, author, author_type, comment_type, content, metadata, created_at
                ) VALUES (%s, %s, %s, %s, %s, %s, %s)
            """

            cursor.execute(comment_sql, (
                task_id,
                'claude-code-hook',
                'system',
                'note',
                'Task automatisch aus User-Prompt erstellt',
                json.dumps({'source': 'task_hook', 'pattern': 'auto-detect'}),
                now
            ))

            connection.commit()
            return task_id

    except Exception as e:
        print(f"Task creation error: {e}", file=sys.stderr)
        return None
    finally:
        if 'connection' in locals():
            connection.close()


def process_event(data: Dict[str, Any]) -> None:
    """Verarbeitet ein Claude Code Event"""
    event_name = data.get('hook_event_name', '')

    # Nur UserPromptSubmit verarbeiten
    if event_name != 'UserPromptSubmit':
        return

    # Prompt aus verschiedenen Stellen extrahieren
    prompt = ''

    if 'prompt' in data:
        prompt = data['prompt']
    elif 'tool_input' in data and isinstance(data['tool_input'], dict):
        prompt = data['tool_input'].get('prompt', '')
        prompt += ' ' + data['tool_input'].get('message', '')

    if not prompt:
        return

    # Tasks extrahieren
    tasks = extract_tasks(prompt)

    if not tasks:
        return

    # User ermitteln
    created_by = os.environ.get('USER', 'claude-code-hook')

    # Tasks erstellen
    created_count = 0
    for task in tasks:
        task_id = create_task(
            title=task['title'],
            description=f"Auto-created from {task['source']} pattern in Claude Code prompt",
            created_by=created_by
        )
        if task_id:
            created_count += 1
            print(f"Task #{task_id} created: {task['title'][:50]}...", file=sys.stderr)

    if created_count > 0:
        print(f"Created {created_count} task(s) from prompt", file=sys.stderr)


def main():
    """Hauptfunktion"""
    try:
        # Input von stdin lesen
        input_data = json.load(sys.stdin)

        # Event verarbeiten
        process_event(input_data)

        # Immer continue zurückgeben (non-blocking)
        print(json.dumps({"continue": True}))
        sys.exit(0)

    except json.JSONDecodeError as e:
        print(f"JSON decode error: {e}", file=sys.stderr)
        print(json.dumps({"continue": True}))
        sys.exit(0)
    except Exception as e:
        print(f"Hook error: {e}", file=sys.stderr)
        print(json.dumps({"continue": True}))
        sys.exit(0)


if __name__ == '__main__':
    main()
← Übersicht