{
"tool_response": {
"filePath": "\/var\/www\/scripts\/pipeline\/db.py",
"oldString": "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)",
"newString": "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 # Stopwords\n def get_stopwords(self, active_only: bool = True) -> list[str]:\n \"\"\"Get list of stopword canonical forms for entity filtering.\n\n Returns:\n List of canonical stopword strings (lowercase, normalized)\n \"\"\"\n query = \"SELECT canonical_form FROM stopwords\"\n if active_only:\n query += \" WHERE is_active = 1\"\n\n cursor = self.execute(query)\n results = cursor.fetchall()\n cursor.close()\n return [r[\"canonical_form\"] for r in results] if results else []\n\n def is_stopword(self, word: str) -> bool:\n \"\"\"Check if a word is in the stopword list.\"\"\"\n canonical = self._normalize_stopword(word)\n stopwords = self.get_stopwords()\n return canonical in stopwords\n\n def _normalize_stopword(self, word: str) -> str:\n \"\"\"Normalize word to canonical form for stopword matching.\"\"\"\n import re\n import unicodedata\n\n result = word.lower().strip()\n # German umlauts\n replacements = {\"ä\": \"ae\", \"ö\": \"oe\", \"ü\": \"ue\", \"ß\": \"ss\"}\n for old, new in replacements.items():\n result = result.replace(old, new)\n # Normalize unicode\n result = unicodedata.normalize(\"NFKD\", result)\n result = result.encode(\"ascii\", \"ignore\").decode(\"ascii\")\n # Keep only alphanumeric\n result = re.sub(r\"[^a-z0-9]\", \"\", result)\n return result\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 # 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(\n self,\n doc_id,\n chunk_index,\n content,\n heading_path,\n position_start=None,\n position_end=None,\n metadata=None,\n page_id=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, page_id, chunk_index, content, token_count, heading_path, metadata)\n VALUES (%s, %s, %s, %s, %s, %s, %s)\"\"\",\n (doc_id, page_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, ve... [TRUNCATED-019761556a23261b]",
"structuredPatch": [
{
"oldStart": 341,
"oldLines": 6,
"newStart": 341,
"newLines": 45,
"lines": [
" lines.append(f\" {t['code']}: {t['criteria']}\")",
" return \"\\n\".join(lines)",
" ",
"+ # Stopwords",
"+ def get_stopwords(self, active_only: bool = True) -> list[str]:",
"+ \"\"\"Get list of stopword canonical forms for entity filtering.",
"+",
"+ Returns:",
"+ List of canonical stopword strings (lowercase, normalized)",
"+ \"\"\"",
"+ query = \"SELECT canonical_form FROM stopwords\"",
"+ if active_only:",
"+ query += \" WHERE is_active = 1\"",
"+",
"+ cursor = self.execute(query)",
"+ results = cursor.fetchall()",
"+ cursor.close()",
"+ return [r[\"canonical_form\"] for r in results] if results else []",
"+",
"+ def is_stopword(self, word: str) -> bool:",
"+ \"\"\"Check if a word is in the stopword list.\"\"\"",
"+ canonical = self._normalize_stopword(word)",
"+ stopwords = self.get_stopwords()",
"+ return canonical in stopwords",
"+",
"+ def _normalize_stopword(self, word: str) -> str:",
"+ \"\"\"Normalize word to canonical form for stopword matching.\"\"\"",
"+ import re",
"+ import unicodedata",
"+",
"+ result = word.lower().strip()",
"+ # German umlauts",
"+ replacements = {\"ä\": \"ae\", \"ö\": \"oe\", \"ü\": \"ue\", \"ß\": \"ss\"}",
"+ for old, new in replacements.items():",
"+ result = result.replace(old, new)",
"+ # Normalize unicode",
"+ result = unicodedata.normalize(\"NFKD\", result)",
"+ result = result.encode(\"ascii\", \"ignore\").decode(\"ascii\")",
"+ # Keep only alphanumeric",
"+ result = re.sub(r\"[^a-z0-9]\", \"\", result)",
"+ return result",
"+",
" # Protokoll Logging (LLM calls)",
" def log_to_protokoll(",
" self,"
]
}
],
"userModified": false,
"replaceAll": false
}
}