Protokoll #19165

{
    "event": "PreToolUse",
    "tool_name": "Edit",
    "tool_input": {
        "file_path": "\/var\/www\/scripts\/pipeline\/db.py",
        "old_string": "        cursor.close()\n\n    # Chunk Operations\n    def insert_chunk(",
        "new_string": "        cursor.close()\n\n    # Page Operations\n    def insert_page(self, doc_id, page_number, text_content, token_count=None):\n        \"\"\"Insert a document page.\"\"\"\n        if token_count is None:\n            token_count = len(text_content.split()) if text_content else 0\n        cursor = self.execute(\n            \"\"\"INSERT INTO document_pages\n               (document_id, page_number, text_content, token_count, created_at)\n               VALUES (%s, %s, %s, %s, NOW())\n               ON DUPLICATE KEY UPDATE\n               text_content = VALUES(text_content),\n               token_count = VALUES(token_count)\"\"\",\n            (doc_id, page_number, text_content, token_count),\n        )\n        self.commit()\n        page_id = cursor.lastrowid\n        if page_id == 0:\n            cursor_select = self.execute(\n                \"SELECT id FROM document_pages WHERE document_id = %s AND page_number = %s\",\n                (doc_id, page_number),\n            )\n            result = cursor_select.fetchone()\n            cursor_select.close()\n            page_id = result[\"id\"] if result else None\n        cursor.close()\n        return page_id\n\n    def get_page_id(self, doc_id, page_number):\n        \"\"\"Get page ID by document and page number.\"\"\"\n        cursor = self.execute(\n            \"SELECT id FROM document_pages WHERE document_id = %s AND page_number = %s\",\n            (doc_id, page_number),\n        )\n        result = cursor.fetchone()\n        cursor.close()\n        return result[\"id\"] if result else None\n\n    # Chunk Operations\n    def insert_chunk("
    }
}

{
    "tool_response": {
        "filePath": "\/var\/www\/scripts\/pipeline\/db.py",
        "oldString": "        cursor.close()\n\n    # Chunk Operations\n    def insert_chunk(",
        "newString": "        cursor.close()\n\n    # Page Operations\n    def insert_page(self, doc_id, page_number, text_content, token_count=None):\n        \"\"\"Insert a document page.\"\"\"\n        if token_count is None:\n            token_count = len(text_content.split()) if text_content else 0\n        cursor = self.execute(\n            \"\"\"INSERT INTO document_pages\n               (document_id, page_number, text_content, token_count, created_at)\n               VALUES (%s, %s, %s, %s, NOW())\n               ON DUPLICATE KEY UPDATE\n               text_content = VALUES(text_content),\n               token_count = VALUES(token_count)\"\"\",\n            (doc_id, page_number, text_content, token_count),\n        )\n        self.commit()\n        page_id = cursor.lastrowid\n        if page_id == 0:\n            cursor_select = self.execute(\n                \"SELECT id FROM document_pages WHERE document_id = %s AND page_number = %s\",\n                (doc_id, page_number),\n            )\n            result = cursor_select.fetchone()\n            cursor_select.close()\n            page_id = result[\"id\"] if result else None\n        cursor.close()\n        return page_id\n\n    def get_page_id(self, doc_id, page_number):\n        \"\"\"Get page ID by document and page number.\"\"\"\n        cursor = self.execute(\n            \"SELECT id FROM document_pages WHERE document_id = %s AND page_number = %s\",\n            (doc_id, page_number),\n        )\n        result = cursor.fetchone()\n        cursor.close()\n        return result[\"id\"] if result else None\n\n    # Chunk Operations\n    def insert_chunk(",
        "originalFile": "\"\"\"\nDatabase operations for KI-System Pipeline\n\"\"\"\n\nimport mysql.connector\nfrom mysql.connector import Error\n\nfrom config import DB_CONFIG\n\n\nclass Database:\n    \"\"\"MariaDB connection wrapper with document and chunk operations.\"\"\"\n\n    def __init__(self):\n        self.connection = None\n\n    def connect(self):\n        \"\"\"Establish database connection.\"\"\"\n        try:\n            self.connection = mysql.connector.connect(**DB_CONFIG)\n            return True\n        except Error as e:\n            print(f\"Database connection error: {e}\")\n            return False\n\n    def disconnect(self):\n        \"\"\"Close database connection.\"\"\"\n        if self.connection and self.connection.is_connected():\n            self.connection.close()\n\n    def execute(self, query, params=None):\n        \"\"\"Execute a query and return the cursor.\"\"\"\n        cursor = self.connection.cursor(dictionary=True)\n        cursor.execute(query, params or ())\n        return cursor\n\n    def commit(self):\n        \"\"\"Commit the current transaction.\"\"\"\n        self.connection.commit()\n\n    # Document Operations\n    def document_exists(self, file_path):\n        \"\"\"Check if document already exists.\"\"\"\n        cursor = self.execute(\"SELECT id FROM documents WHERE source_path = %s\", (file_path,))\n        result = cursor.fetchone()\n        cursor.close()\n        return result[\"id\"] if result else None\n\n    def document_is_done(self, file_path):\n        \"\"\"Check if document is already fully processed (status='done').\"\"\"\n        cursor = self.execute(\"SELECT id, status FROM documents WHERE source_path = %s\", (file_path,))\n        result = cursor.fetchone()\n        cursor.close()\n        if result and result[\"status\"] == \"done\":\n            return result[\"id\"]\n        return None\n\n    def insert_document(self, file_path, title, file_type, file_size, file_hash):\n        \"\"\"Insert a new document or update existing one.\"\"\"\n        import os\n\n        folder_path = os.path.dirname(file_path)\n        cursor = self.execute(\n            \"\"\"INSERT INTO documents\n               (source_path, folder_path, filename, mime_type, file_size, file_hash, status)\n               VALUES (%s, %s, %s, %s, %s, %s, 'processing')\n               ON DUPLICATE KEY UPDATE\n               file_hash = VALUES(file_hash),\n               file_size = VALUES(file_size),\n               status = 'processing',\n               processed_at = NULL,\n               error_message = NULL\"\"\",\n            (file_path, folder_path, title, file_type, file_size, file_hash),\n        )\n        self.commit()\n        doc_id = cursor.lastrowid\n        # If ON DUPLICATE KEY UPDATE was triggered, lastrowid is 0\n        if doc_id == 0:\n            cursor_select = self.execute(\"SELECT id FROM documents WHERE source_path = %s\", (file_path,))\n            result = cursor_select.fetchone()\n            cursor_select.close()\n            doc_id = result[\"id\"] if result else None\n        cursor.close()\n        return doc_id\n\n    def update_document_status(self, doc_id, status, error_message=None):\n        \"\"\"Update document processing status.\"\"\"\n        if error_message:\n            cursor = self.execute(\n                \"\"\"UPDATE documents\n                   SET status = %s, error_message = %s, processed_at = NOW()\n                   WHERE id = %s\"\"\",\n                (status, error_message, doc_id),\n            )\n        else:\n            cursor = self.execute(\n                \"UPDATE documents SET status = %s, processed_at = NOW() WHERE id = %s\", (status, doc_id)\n            )\n        self.commit()\n        cursor.close()\n\n    # Chunk Operations\n    def insert_chunk(\n        self, doc_id, chunk_index, content, heading_path, position_start=None, position_end=None, metadata=None\n    ):\n        \"\"\"Insert a text chunk.\"\"\"\n        # Calculate token count (rough estimate: 4 chars per token)\n        token_count = len(content) \/\/ 4\n        cursor = self.execute(\n            \"\"\"INSERT INTO chunks\n               (document_id, chunk_index, content, token_count, heading_path, metadata)\n               VALUES (%s, %s, %s, %s, %s, %s)\"\"\",\n            (doc_id, chunk_index, content, token_count, heading_path, metadata),\n        )\n        self.commit()\n        chunk_id = cursor.lastrowid\n        cursor.close()\n        return chunk_id\n\n    def get_chunks_for_embedding(self, limit=100):\n        \"\"\"Get chunks that need embeddings.\"\"\"\n        cursor = self.execute(\n            \"\"\"SELECT c.id, c.content, c.document_id\n               FROM chunks c\n               WHERE c.qdrant_id IS NULL\n               ORDER BY c.created_at\n               LIMIT %s\"\"\",\n            (limit,),\n        )\n        results = cursor.fetchall()\n        cursor.close()\n        return results\n\n    def update_chunk_qdrant_id(self, chunk_id, qdrant_id):\n        \"\"\"Update chunk with Qdrant point ID.\"\"\"\n        cursor = self.execute(\"UPDATE chunks SET qdrant_id = %s WHERE id = %s\", (qdrant_id, chunk_id))\n        self.commit()\n        cursor.close()\n\n    # Queue Operations\n    def add_to_queue(self, file_path, action=\"process\"):\n        \"\"\"Add item to pipeline queue.\"\"\"\n        cursor = self.execute(\n            \"\"\"INSERT INTO pipeline_queue\n               (file_path, action, status, retry_count, created_at)\n               VALUES (%s, %s, 'pending', 0, NOW())\"\"\",\n            (file_path, action),\n        )\n        self.commit()\n        queue_id = cursor.lastrowid\n        cursor.close()\n        return queue_id\n\n    def get_pending_queue_items(self, limit=10):\n        \"\"\"Get pending items from queue.\"\"\"\n        cursor = self.execute(\n            \"\"\"SELECT id, file_path, action, retry_count\n               FROM pipeline_queue\n               WHERE status = 'pending'\n               ORDER BY created_at\n               LIMIT %s\"\"\",\n            (limit,),\n        )\n        results = cursor.fetchall()\n        cursor.close()\n        return results\n\n    def update_queue_status(self, queue_id, status, error_message=None):\n        \"\"\"Update queue item status.\"\"\"\n        if status == \"error\" and error_message:\n            cursor = self.execute(\n                \"\"\"UPDATE pipeline_queue\n                   SET status = %s, error_message = %s,\n                       retry_count = retry_count + 1, updated_at = NOW()\n                   WHERE id = %s\"\"\",\n                (status, error_message, queue_id),\n            )\n        else:\n            cursor = self.execute(\n                \"UPDATE pipeline_queue SET status = %s, updated_at = NOW() WHERE id = %s\", (status, queue_id)\n            )\n        self.commit()\n        cursor.close()\n\n    # Logging\n    def log(self, level, message, context=None):\n        \"\"\"Write to pipeline log (ki_dev database).\"\"\"\n        import json\n\n        import mysql.connector\n\n        from config import DB_LOG_CONFIG\n\n        try:\n            # Context must be valid JSON\n            if context is not None:\n                if isinstance(context, str):\n                    context = json.dumps({\"info\": context})\n                elif isinstance(context, dict):\n                    context = json.dumps(context)\n                else:\n                    context = json.dumps({\"data\": str(context)})\n\n            # Use separate connection to ki_dev for logging\n            log_conn = mysql.connector.connect(**DB_LOG_CONFIG)\n            cursor = log_conn.cursor()\n            cursor.execute(\n                \"\"\"INSERT INTO pipeline_log\n                   (level, message, context, created_at)\n                   VALUES (%s, %s, %s, NOW())\"\"\",\n                (level, message, context),\n            )\n            log_conn.commit()\n            cursor.close()\n            log_conn.close()\n        except Exception:  # noqa: S110\n            # Logging should never break the pipeline\n            pass\n\n    # Prompts\n    def get_prompt(self, name, version=None):\n        \"\"\"Get a prompt by name (optionally specific version).\"\"\"\n        if version:\n            cursor = self.execute(\"SELECT content FROM prompts WHERE name = %s AND version = %s\", (name, version))\n        else:\n            cursor = self.execute(\n                \"\"\"SELECT content FROM prompts\n                   WHERE name = %s AND is_active = 1\n                   ORDER BY version DESC LIMIT 1\"\"\",\n                (name,),\n            )\n        result = cursor.fetchone()\n        cursor.close()\n        return result[\"content\"] if result else None\n\n    def get_prompt_by_use_case(self, use_case: str, version: str = None) -> dict | None:\n        \"\"\"\n        Get prompt by use_case with full metadata for provenance tracking.\n\n        Args:\n            use_case: The use case (entity_extraction, semantic_analysis, statement_extraction, etc.)\n            version: Optional specific version (otherwise latest active)\n\n        Returns:\n            Dict with id, name, version, content, use_case or None\n        \"\"\"\n        if version:\n            cursor = self.execute(\n                \"\"\"SELECT id, name, version, content, use_case\n                   FROM prompts\n                   WHERE use_case = %s AND version = %s\"\"\",\n                (use_case, version),\n            )\n        else:\n            cursor = self.execute(\n                \"\"\"SELECT id, name, version, content, use_case\n                   FROM prompts\n                   WHERE use_case = %s AND is_active = 1\n                   ORDER BY version DESC LIMIT 1\"\"\",\n                (use_case,),\n            )\n        result = cursor.fetchone()\n        cursor.close()\n        return result\n\n    # Entity Types\n    def get_entity_types(self, active_only: bool = True) -> list[dict]:\n        \"\"\"Get all entity types from database.\n\n        Returns:\n            List of dicts with code, name, description, criteria, indicators, examples\n        \"\"\"\n        query = \"\"\"SELECT code, name, description, criteria, indicators, examples\n                   FROM entity_types\"\"\"\n        if active_only:\n            query += \" WHERE is_active = 1\"\n        query += \" ORDER BY sort_order\"\n\n        cursor = self.execute(query)\n        results = cu... [TRUNCATED-b7dc19d3d25faec4]",
        "structuredPatch": [
            {
                "oldStart": 99,
                "oldLines": 6,
                "newStart": 99,
                "newLines": 43,
                "lines": [
                    "         self.commit()",
                    "         cursor.close()",
                    " ",
                    "+    # Page Operations",
                    "+    def insert_page(self, doc_id, page_number, text_content, token_count=None):",
                    "+        \"\"\"Insert a document page.\"\"\"",
                    "+        if token_count is None:",
                    "+            token_count = len(text_content.split()) if text_content else 0",
                    "+        cursor = self.execute(",
                    "+            \"\"\"INSERT INTO document_pages",
                    "+               (document_id, page_number, text_content, token_count, created_at)",
                    "+               VALUES (%s, %s, %s, %s, NOW())",
                    "+               ON DUPLICATE KEY UPDATE",
                    "+               text_content = VALUES(text_content),",
                    "+               token_count = VALUES(token_count)\"\"\",",
                    "+            (doc_id, page_number, text_content, token_count),",
                    "+        )",
                    "+        self.commit()",
                    "+        page_id = cursor.lastrowid",
                    "+        if page_id == 0:",
                    "+            cursor_select = self.execute(",
                    "+                \"SELECT id FROM document_pages WHERE document_id = %s AND page_number = %s\",",
                    "+                (doc_id, page_number),",
                    "+            )",
                    "+            result = cursor_select.fetchone()",
                    "+            cursor_select.close()",
                    "+            page_id = result[\"id\"] if result else None",
                    "+        cursor.close()",
                    "+        return page_id",
                    "+",
                    "+    def get_page_id(self, doc_id, page_number):",
                    "+        \"\"\"Get page ID by document and page number.\"\"\"",
                    "+        cursor = self.execute(",
                    "+            \"SELECT id FROM document_pages WHERE document_id = %s AND page_number = %s\",",
                    "+            (doc_id, page_number),",
                    "+        )",
                    "+        result = cursor.fetchone()",
                    "+        cursor.close()",
                    "+        return result[\"id\"] if result else None",
                    "+",
                    "     # Chunk Operations",
                    "     def insert_chunk(",
                    "         self, doc_id, chunk_index, content, heading_path, position_start=None, position_end=None, metadata=None"
                ]
            }
        ],
        "userModified": false,
        "replaceAll": false
    }
}