{
"tool_response": {
"filePath": "\/var\/www\/scripts\/pipeline\/db.py",
"oldString": " result = cursor.fetchone()\n cursor.close()\n return result\n\n # Protokoll Logging (LLM calls)",
"newString": " 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 = cursor.fetchall()\n cursor.close()\n return list(results) if results else []\n\n def get_entity_type_codes(self) -> set[str]:\n \"\"\"Get set of valid entity type codes.\"\"\"\n cursor = self.execute(\n \"SELECT code FROM entity_types WHERE is_active = 1\"\n )\n results = cursor.fetchall()\n cursor.close()\n return {r[\"code\"] for r in results} if results else set()\n\n def build_entity_prompt_categories(self) -> str:\n \"\"\"Build categories section for entity extraction prompt from DB.\"\"\"\n types = self.get_entity_types()\n lines = []\n for t in types:\n lines.append(f\" {t['code']}: {t['criteria']}\")\n return \"\\n\".join(lines)\n\n # Protokoll Logging (LLM calls)",
"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 # Protokoll Logging (LLM calls)\n def log_to_protokoll(\n self,\n client_name,\n request,\n response=None,\n model_name=None,\n tokens_input=0,\n tokens_output=0,\n duration_ms=0,\n status=\"completed\",\n error_message=None,\n ):\n \"\"\"\n Log LLM call to ki_dev.protokoll table.\n\n Args:\n client_name: Caller identifier (e.g., 'content-studio', 'pipeline')\n request: The prompt sent to the LLM\n response: The LLM response\n ... [TRUNCATED-64fe93ad22c0b05d]",
"structuredPatch": [
{
"oldStart": 263,
"oldLines": 6,
"newStart": 263,
"newLines": 41,
"lines": [
" cursor.close()",
" return result",
" ",
"+ # Entity Types",
"+ def get_entity_types(self, active_only: bool = True) -> list[dict]:",
"+ \"\"\"Get all entity types from database.",
"+",
"+ Returns:",
"+ List of dicts with code, name, description, criteria, indicators, examples",
"+ \"\"\"",
"+ query = \"\"\"SELECT code, name, description, criteria, indicators, examples",
"+ FROM entity_types\"\"\"",
"+ if active_only:",
"+ query += \" WHERE is_active = 1\"",
"+ query += \" ORDER BY sort_order\"",
"+",
"+ cursor = self.execute(query)",
"+ results = cursor.fetchall()",
"+ cursor.close()",
"+ return list(results) if results else []",
"+",
"+ def get_entity_type_codes(self) -> set[str]:",
"+ \"\"\"Get set of valid entity type codes.\"\"\"",
"+ cursor = self.execute(",
"+ \"SELECT code FROM entity_types WHERE is_active = 1\"",
"+ )",
"+ results = cursor.fetchall()",
"+ cursor.close()",
"+ return {r[\"code\"] for r in results} if results else set()",
"+",
"+ def build_entity_prompt_categories(self) -> str:",
"+ \"\"\"Build categories section for entity extraction prompt from DB.\"\"\"",
"+ types = self.get_entity_types()",
"+ lines = []",
"+ for t in types:",
"+ lines.append(f\" {t['code']}: {t['criteria']}\")",
"+ return \"\\n\".join(lines)",
"+",
" # Protokoll Logging (LLM calls)",
" def log_to_protokoll(",
" self,"
]
}
],
"userModified": false,
"replaceAll": false
}
}