{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/scripts\/pipeline\/knowledge\/storage.py",
"content": "\"\"\"Knowledge Storage für Wissensextraktion.\"\"\"\nimport json\n\nimport sys\nsys.path.insert(0, \"\/var\/www\/scripts\/pipeline\")\n\nfrom db import db\n\nfrom .models import KnowledgeLevel, KnowledgeType\n\n\ndef store_knowledge(\n level: KnowledgeLevel,\n source_id: int,\n knowledge_type: KnowledgeType,\n data: dict,\n model_name: str = \"unknown\",\n):\n \"\"\"Speichere Wissen in der entsprechenden Tabelle.\"\"\"\n table_map = {\n KnowledgeLevel.PAGE: \"page_knowledge\",\n KnowledgeLevel.SECTION: \"section_knowledge\",\n KnowledgeLevel.DOCUMENT: \"document_knowledge\",\n }\n\n id_field_map = {\n KnowledgeLevel.PAGE: \"page_id\",\n KnowledgeLevel.SECTION: \"section_id\",\n KnowledgeLevel.DOCUMENT: \"document_id\",\n }\n\n table = table_map[level]\n id_field = id_field_map[level]\n\n try:\n cursor = db.execute(\n f\"\"\"INSERT INTO {table} ({id_field}, knowledge_type, data, model_used, created_at)\n VALUES (%s, %s, %s, %s, NOW())\"\"\", # noqa: S608 - table name is controlled\n (source_id, knowledge_type.value, json.dumps(data), model_name),\n )\n db.commit()\n cursor.close()\n except Exception as e:\n db.log(\"ERROR\", f\"Fehler beim Speichern in {table}: {e}\")\n\n\nclass KnowledgeStorage:\n \"\"\"Wrapper für Knowledge Storage mit Modell-Kontext.\"\"\"\n\n def __init__(self, model_name: str):\n \"\"\"Initialisiere mit Modellname.\"\"\"\n self.model_name = model_name\n\n def store(\n self, level: KnowledgeLevel, source_id: int, knowledge_type: KnowledgeType, data: dict\n ):\n \"\"\"Speichere Wissen mit Modell-Kontext.\"\"\"\n store_knowledge(level, source_id, knowledge_type, data, self.model_name)\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/scripts\/pipeline\/knowledge\/storage.py",
"content": "\"\"\"Knowledge Storage für Wissensextraktion.\"\"\"\nimport json\n\nimport sys\nsys.path.insert(0, \"\/var\/www\/scripts\/pipeline\")\n\nfrom db import db\n\nfrom .models import KnowledgeLevel, KnowledgeType\n\n\ndef store_knowledge(\n level: KnowledgeLevel,\n source_id: int,\n knowledge_type: KnowledgeType,\n data: dict,\n model_name: str = \"unknown\",\n):\n \"\"\"Speichere Wissen in der entsprechenden Tabelle.\"\"\"\n table_map = {\n KnowledgeLevel.PAGE: \"page_knowledge\",\n KnowledgeLevel.SECTION: \"section_knowledge\",\n KnowledgeLevel.DOCUMENT: \"document_knowledge\",\n }\n\n id_field_map = {\n KnowledgeLevel.PAGE: \"page_id\",\n KnowledgeLevel.SECTION: \"section_id\",\n KnowledgeLevel.DOCUMENT: \"document_id\",\n }\n\n table = table_map[level]\n id_field = id_field_map[level]\n\n try:\n cursor = db.execute(\n f\"\"\"INSERT INTO {table} ({id_field}, knowledge_type, data, model_used, created_at)\n VALUES (%s, %s, %s, %s, NOW())\"\"\", # noqa: S608 - table name is controlled\n (source_id, knowledge_type.value, json.dumps(data), model_name),\n )\n db.commit()\n cursor.close()\n except Exception as e:\n db.log(\"ERROR\", f\"Fehler beim Speichern in {table}: {e}\")\n\n\nclass KnowledgeStorage:\n \"\"\"Wrapper für Knowledge Storage mit Modell-Kontext.\"\"\"\n\n def __init__(self, model_name: str):\n \"\"\"Initialisiere mit Modellname.\"\"\"\n self.model_name = model_name\n\n def store(\n self, level: KnowledgeLevel, source_id: int, knowledge_type: KnowledgeType, data: dict\n ):\n \"\"\"Speichere Wissen mit Modell-Kontext.\"\"\"\n store_knowledge(level, source_id, knowledge_type, data, self.model_name)\n",
"structuredPatch": [],
"originalFile": null
}
}