task_creator.py

Code Hygiene Score: 93

Keine Issues gefunden.

Dependencies 8

Funktionen 4

Code

#!/usr/bin/env python3
"""
Task Creator - Erstellt Tasks für Quality Violations.

Schreibt direkt in ki_dev.tasks via MariaDB.
"""

import json
import os
import uuid
from datetime import datetime
from pathlib import Path
from typing import Optional

import mysql.connector


def get_db_connection():
    """Erstellt DB-Verbindung zu ki_dev."""
    return mysql.connector.connect(
        host=os.environ.get("CLAUDE_DB_HOST", "localhost"),
        user=os.environ.get("CLAUDE_DB_USER", "root"),
        password=os.environ.get("CLAUDE_DB_PASSWORD", ""),
        database=os.environ.get("CLAUDE_DB_NAME", "ki_dev")
    )


def create_violation_task(
    file_path: str,
    rule_id: str,
    message: str,
    severity: str = "block"  # "block" oder "warn"
) -> Optional[int]:
    """
    Erstellt einen Task für eine Quality Violation.

    Args:
        file_path: Pfad zur betroffenen Datei
        rule_id: Regel-ID (z.B. "P1.1", "W1.2")
        message: Vollständige Fehlermeldung
        severity: "block" (Pre-Hook) oder "warn" (Post-Hook)

    Returns:
        Task-ID oder None bei Fehler
    """
    filename = Path(file_path).name

    # Task-Titel
    title = f"Quality [{rule_id}]: {filename}"

    # Task-Beschreibung mit Struktur
    description = f"""## Violation

**Datei:** `{file_path}`
**Regel:** {rule_id}
**Severity:** {severity.upper()}

**Meldung:**
{message}

---

## Analyse

*Zu analysieren:*
- Was ist die Ursache der Violation?
- Welche Abhängigkeiten sind betroffen?
- Gibt es ähnliche Violations in anderen Dateien?

---

## Plan

*Zu planen:*
1. [ ] Violation verstehen
2. [ ] Lösung entwerfen
3. [ ] Änderungen implementieren
4. [ ] Testen

---

## Umsetzung

*Nach Abschluss dokumentieren:*
- Was wurde geändert?
- Welche Dateien wurden angepasst?
- Sind weitere Änderungen nötig?
"""

    # Metadata
    metadata = {
        "source": "quality-hook",
        "file_path": file_path,
        "rule_id": rule_id,
        "severity": severity,
        "message": message,
        "detected_at": datetime.now().isoformat()
    }

    try:
        conn = get_db_connection()
        cursor = conn.cursor()

        # Prüfe ob bereits ein offener Task für diese Datei+Regel existiert
        cursor.execute("""
            SELECT id FROM tasks
            WHERE status IN ('pending', 'in_progress')
            AND JSON_EXTRACT(metadata, '$.file_path') = %s
            AND JSON_EXTRACT(metadata, '$.rule_id') = %s
            LIMIT 1
        """, (file_path, rule_id))

        existing = cursor.fetchone()
        if existing:
            # Task existiert bereits, nicht duplizieren
            cursor.close()
            conn.close()
            return existing[0]

        # Neuen Task erstellen
        cursor.execute("""
            INSERT INTO tasks (uuid, title, description, type, status, created_by, created_by_type, metadata)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
        """, (
            str(uuid.uuid4()),
            title,
            description,
            "ai_task",
            "pending",
            "quality-hook",
            "ai",
            json.dumps(metadata)
        ))

        task_id = cursor.lastrowid
        conn.commit()
        cursor.close()
        conn.close()

        return task_id

    except Exception as e:
        # Bei DB-Fehler nicht crashen, nur loggen
        import sys
        print(f"[QUALITY] Task creation failed: {e}", file=sys.stderr)
        return None


def create_warning_task(file_path: str, rule_id: str, message: str) -> Optional[int]:
    """
    Erstellt einen Task für eine einzelne Warnung.

    Args:
        file_path: Pfad zur betroffenen Datei
        rule_id: Regel-ID (z.B. "W1.1")
        message: Vollständige Warnmeldung

    Returns:
        Task-ID oder None bei Fehler
    """
    return create_violation_task(file_path, rule_id, message, "warn")


def create_tasks_for_warnings(file_path: str, warnings: list) -> list:
    """
    Erstellt Tasks für mehrere Warnungen.

    Returns:
        Liste der erstellten Task-IDs
    """
    task_ids = []

    for warning in warnings:
        # Parse Rule-ID aus Warnung (Format: "W1.2: ...")
        parts = warning.split(":", 1)
        if len(parts) >= 2:
            rule_id = parts[0].strip()
            message = parts[1].strip()
        else:
            rule_id = "UNKNOWN"
            message = warning

        task_id = create_violation_task(file_path, rule_id, message, "warn")
        if task_id:
            task_ids.append(task_id)

    return task_ids
← Übersicht