{
"tool_response": {
"filePath": "\/opt\/scripts\/pipeline\/pipeline.py",
"oldString": "def run_full_pipeline():\n \"\"\"Run complete pipeline: scan → process → embed.\"\"\"\n print(\"=\" * 50)\n print(\"KI-System Pipeline - Full Run\")\n print(\"=\" * 50)\n\n print(\"\\n[1\/3] Scanning for documents...\")\n files = run_scan()\n\n if files:\n print(f\"\\n[2\/3] Processing {len(files)} documents...\")\n process_queue()\n else:\n print(\"\\n[2\/3] No new documents to process\")\n\n print(\"\\n[3\/3] Embedding remaining chunks...\")\n embedded = embed_pending_chunks()\n print(f\"Embedded {embedded} chunks\")\n\n print(\"\\n\" + \"=\" * 50)\n print(\"Pipeline complete!\")",
"newString": "def run_full_pipeline(run_id=None, pipeline_id=None):\n \"\"\"Run complete pipeline: scan → process → embed.\"\"\"\n progress = PipelineProgress(run_id) if run_id else None\n\n print(\"=\" * 50)\n print(\"KI-System Pipeline - Full Run\")\n if run_id:\n print(f\"Run ID: {run_id}, Pipeline ID: {pipeline_id}\")\n print(\"=\" * 50)\n\n try:\n # Phase 1: Scan\n if progress:\n progress.update_step(\"detect\")\n progress.add_log(\"Scanne nach Dokumenten...\")\n\n print(\"\\n[1\/3] Scanning for documents...\")\n files = scan_directory()\n print(f\"Found {len(files)} files\")\n\n if progress:\n progress.update_progress(total=len(files))\n progress.add_log(f\"{len(files)} Dokumente gefunden\")\n\n if files:\n queued = queue_files(files)\n print(f\"Queued {queued} files\")\n\n # Phase 2: Process each file\n print(f\"\\n[2\/3] Processing {len(files)} documents...\")\n items = db.get_pending_queue_items(limit=100)\n\n total_chunks = 0\n total_embeddings = 0\n processed = 0\n failed = 0\n\n for item in items:\n # Check if cancelled\n if progress and progress.is_cancelled():\n progress.add_log(\"Pipeline abgebrochen durch Benutzer\")\n progress.complete(\"cancelled\")\n print(\"\\nPipeline cancelled by user\")\n return\n\n queue_id = item[\"id\"]\n file_path = item[\"file_path\"]\n file_name = Path(file_path).name\n\n if progress:\n progress.update_document(file_name)\n\n db.update_queue_status(queue_id, \"processing\")\n\n try:\n success, chunks, embedded = process_file(file_path, progress)\n if success:\n db.update_queue_status(queue_id, \"done\")\n processed += 1\n total_chunks += chunks\n total_embeddings += embedded\n else:\n db.update_queue_status(queue_id, \"failed\", \"Processing failed\")\n failed += 1\n except Exception as e:\n db.update_queue_status(queue_id, \"failed\", str(e))\n failed += 1\n if progress:\n progress.add_log(f\"FEHLER bei {file_name}: {str(e)[:50]}\")\n\n if progress:\n progress.update_progress(\n processed=processed,\n failed=failed,\n chunks=total_chunks,\n embeddings=total_embeddings,\n )\n else:\n print(\"\\n[2\/3] No new documents to process\")\n if progress:\n progress.add_log(\"Keine neuen Dokumente gefunden\")\n\n # Phase 3: Embed remaining\n if progress:\n progress.update_step(\"embed\")\n progress.add_log(\"Verarbeite ausstehende Embeddings...\")\n\n print(\"\\n[3\/3] Embedding remaining chunks...\")\n embedded = embed_pending_chunks()\n print(f\"Embedded {embedded} chunks\")\n\n if progress:\n if embedded > 0:\n progress.add_log(f\"{embedded} weitere Embeddings erstellt\")\n\n # Complete\n print(\"\\n\" + \"=\" * 50)\n print(\"Pipeline complete!\")\n\n if progress:\n progress.add_log(\"Pipeline erfolgreich abgeschlossen\")\n progress.complete(\"completed\")\n\n except Exception as e:\n db.log(\"ERROR\", f\"Pipeline error: {e}\")\n print(f\"Error: {e}\")\n if progress:\n progress.add_log(f\"FEHLER: {str(e)}\")\n progress.complete(\"failed\", str(e))\n raise",
"originalFile": "#!\/usr\/bin\/env python3\n\"\"\"\nKI-System Document Pipeline\nMain orchestration script for document processing.\n\nUsage:\n python pipeline.py scan # Scan for new documents\n python pipeline.py process # Process queued documents\n python pipeline.py embed # Embed pending chunks\n python pipeline.py all # Full pipeline run\n python pipeline.py all --pipeline-id=1 --run-id=5 # With tracking\n python pipeline.py file <path> # Process single file\n\"\"\"\n\nimport argparse\nimport json # noqa: I001\nimport os\nimport sys\nimport time\nfrom pathlib import Path\n\nfrom analyze import analyze_document\nfrom chunk import chunk_by_structure\nfrom config import MAX_RETRIES, RETRY_BACKOFF_BASE\nfrom db import db, PipelineProgress\nfrom detect import queue_files, scan_directory\nfrom embed import embed_chunks, embed_pending_chunks\nfrom enrich import run_enrichment_step\nfrom extract import extract, get_full_text\nfrom vision import run_vision_step\n\n\ndef process_file(file_path, progress=None):\n \"\"\"Process a single file through the pipeline.\"\"\"\n db.log(\"INFO\", f\"Processing: {file_path}\")\n file_name = Path(file_path).name\n\n if progress:\n progress.update_document(file_name)\n\n # Extract text\n if progress:\n progress.update_step(\"extract\")\n progress.add_log(f\"Extrahiere Text: {file_name}\")\n\n extraction = extract(file_path)\n if not extraction[\"success\"]:\n db.log(\"ERROR\", f\"Extraction failed: {extraction.get('error')}\")\n if progress:\n progress.add_log(f\"FEHLER: Extraktion fehlgeschlagen\")\n return False, 0, 0\n\n # Get document info\n file_stat = os.stat(file_path)\n\n import hashlib\n\n with open(file_path, \"rb\") as f:\n file_hash = hashlib.sha256(f.read()).hexdigest()\n\n # Insert document\n doc_id = db.insert_document(\n file_path=file_path,\n title=file_name,\n file_type=extraction[\"file_type\"],\n file_size=file_stat.st_size,\n file_hash=file_hash,\n )\n db.log(\"INFO\", f\"Created document: {doc_id}\")\n\n # Vision analysis for PDFs\n if extraction[\"file_type\"] == \".pdf\":\n if progress:\n progress.update_step(\"vision\")\n progress.add_log(\"Vision-Analyse gestartet...\")\n\n db.log(\"INFO\", f\"Running vision analysis for document {doc_id}\")\n vision_config = {\n \"model\": \"minicpm-v:latest\",\n \"store_images\": True,\n \"detect_images\": True,\n \"detect_charts\": True,\n \"detect_tables\": True,\n }\n vision_result = run_vision_step(doc_id, file_path, vision_config)\n if vision_result[\"success\"]:\n db.log(\"INFO\", f\"Vision: {vision_result['pages_analyzed']}\/{vision_result['pages_total']} pages analyzed\")\n if progress:\n progress.add_log(f\"Vision: {vision_result['pages_analyzed']} Seiten analysiert\")\n else:\n db.log(\"WARNING\", f\"Vision analysis failed: {vision_result.get('error')}\")\n\n # Chunk content\n if progress:\n progress.update_step(\"chunk\")\n progress.add_log(\"Erstelle Chunks...\")\n\n chunks = chunk_by_structure(extraction)\n db.log(\"INFO\", f\"Created {len(chunks)} chunks\")\n\n # Store chunks\n for i, chunk in enumerate(chunks):\n chunk_id = db.insert_chunk(\n doc_id=doc_id,\n chunk_index=i,\n content=chunk[\"content\"],\n heading_path=json.dumps(chunk.get(\"heading_path\", [])),\n position_start=chunk.get(\"position_start\", 0),\n position_end=chunk.get(\"position_end\", 0),\n metadata=json.dumps(chunk.get(\"metadata\", {})),\n )\n chunk[\"db_id\"] = chunk_id\n\n if progress:\n progress.add_log(f\"{len(chunks)} Chunks erstellt\")\n\n # Enrich chunks with vision context (for PDFs)\n if extraction[\"file_type\"] == \".pdf\":\n if progress:\n progress.update_step(\"enrich\")\n\n db.log(\"INFO\", f\"Running vision enrichment for document {doc_id}\")\n enrich_result = run_enrichment_step(doc_id)\n if enrich_result[\"success\"]:\n db.log(\"INFO\", f\"Enrichment: {enrich_result['enriched']}\/{enrich_result['total_chunks']} chunks enriched\")\n else:\n db.log(\"WARNING\", f\"Enrichment failed: {enrich_result.get('error')}\")\n\n # Generate embeddings\n if progress:\n progress.update_step(\"embed\")\n progress.add_log(\"Erstelle Embeddings...\")\n\n embedded = embed_chunks(chunks, doc_id, file_name, file_path)\n db.log(\"INFO\", f\"Embedded {embedded}\/{len(chunks)} chunks\")\n\n if progress:\n progress.add_log(f\"{embedded} Embeddings erstellt\")\n\n # Semantic analysis\n if progress:\n progress.update_step(\"analyze\")\n progress.add_log(\"Semantische Analyse...\")\n\n full_text = get_full_text(extraction)\n analysis = analyze_document(doc_id, full_text)\n db.log(\"INFO\", f\"Analysis complete: {analysis}\")\n\n # Update status\n db.update_document_status(doc_id, \"done\")\n\n if progress:\n progress.add_log(f\"Fertig: {file_name}\")\n\n return True, len(chunks), embedded\n\n\ndef process_queue():\n \"\"\"Process items from the queue.\"\"\"\n items = db.get_pending_queue_items(limit=10)\n db.log(\"INFO\", f\"Found {len(items)} items in queue\")\n\n for item in items:\n queue_id = item[\"id\"]\n file_path = item[\"file_path\"]\n retry_count = item[\"retry_count\"]\n\n if retry_count >= MAX_RETRIES:\n db.update_queue_status(queue_id, \"failed\", \"Max retries exceeded\")\n continue\n\n db.update_queue_status(queue_id, \"processing\")\n\n try:\n success = process_file(file_path)\n if success:\n db.update_queue_status(queue_id, \"done\")\n else:\n raise Exception(\"Processing returned False\")\n except Exception as e:\n error_msg = str(e)\n db.update_queue_status(queue_id, \"pending\", error_msg)\n\n # Exponential backoff\n wait_time = RETRY_BACKOFF_BASE ** (retry_count + 1)\n db.log(\"INFO\", f\"Retry {retry_count + 1} in {wait_time}s: {file_path}\")\n time.sleep(wait_time)\n\n\ndef run_scan():\n \"\"\"Scan for new documents.\"\"\"\n files = scan_directory()\n print(f\"Found {len(files)} files\")\n\n if files:\n queued = queue_files(files)\n print(f\"Queued {queued} files\")\n\n return files\n\n\ndef run_full_pipeline():\n \"\"\"Run complete pipeline: scan → process → embed.\"\"\"\n print(\"=\" * 50)\n print(\"KI-System Pipeline - Full Run\")\n print(\"=\" * 50)\n\n print(\"\\n[1\/3] Scanning for documents...\")\n files = run_scan()\n\n if files:\n print(f\"\\n[2\/3] Processing {len(files)} documents...\")\n process_queue()\n else:\n print(\"\\n[2\/3] No new documents to process\")\n\n print(\"\\n[3\/3] Embedding remaining chunks...\")\n embedded = embed_pending_chunks()\n print(f\"Embedded {embedded} chunks\")\n\n print(\"\\n\" + \"=\" * 50)\n print(\"Pipeline complete!\")\n\n\ndef main():\n \"\"\"Main entry point.\"\"\"\n if len(sys.argv) < 2:\n print(__doc__)\n return\n\n command = sys.argv[1].lower()\n\n db.connect()\n\n try:\n if command == \"scan\":\n run_scan()\n\n elif command == \"process\":\n process_queue()\n\n elif command == \"embed\":\n count = embed_pending_chunks()\n print(f\"Embedded {count} chunks\")\n\n elif command == \"all\":\n run_full_pipeline()\n\n elif command == \"file\" and len(sys.argv) > 2:\n file_path = sys.argv[2]\n if os.path.exists(file_path):\n success = process_file(file_path)\n print(f\"Processing {'successful' if success else 'failed'}\")\n else:\n print(f\"File not found: {file_path}\")\n\n elif command == \"status\":\n # Show pipeline status\n cursor = db.execute(\n \"\"\"SELECT status, COUNT(*) as count\n FROM pipeline_queue\n GROUP BY status\"\"\"\n )\n results = cursor.fetchall()\n cursor.close()\n\n print(\"\\nQueue Status:\")\n for r in results:\n print(f\" {r['status']}: {r['count']}\")\n\n cursor = db.execute(\"SELECT COUNT(*) as count FROM documents\")\n doc_count = cursor.fetchone()[\"count\"]\n cursor.close()\n\n cursor = db.execute(\"SELECT COUNT(*) as count FROM chunks\")\n chunk_count = cursor.fetchone()[\"count\"]\n cursor.close()\n\n cursor = db.execute(\"SELECT COUNT(*) as count FROM chunks WHERE qdrant_id IS NOT NULL\")\n embedded_count = cursor.fetchone()[\"count\"]\n cursor.close()\n\n print(f\"\\nDocuments: {doc_count}\")\n print(f\"Chunks: {chunk_count} ({embedded_count} embedded)\")\n\n else:\n print(f\"Unknown command: {command}\")\n print(__doc__)\n\n except Exception as e:\n db.log(\"ERROR\", f\"Pipeline error: {e}\")\n print(f\"Error: {e}\")\n raise\n finally:\n db.disconnect()\n\n\nif __name__ == \"__main__\":\n main()\n",
"structuredPatch": [
{
"oldStart": 199,
"oldLines": 29,
"newStart": 199,
"newLines": 118,
"lines": [
" return files",
" ",
" ",
"-def run_full_pipeline():",
"+def run_full_pipeline(run_id=None, pipeline_id=None):",
" \"\"\"Run complete pipeline: scan → process → embed.\"\"\"",
"+ progress = PipelineProgress(run_id) if run_id else None",
"+",
" print(\"=\" * 50)",
" print(\"KI-System Pipeline - Full Run\")",
"+ if run_id:",
"+ print(f\"Run ID: {run_id}, Pipeline ID: {pipeline_id}\")",
" print(\"=\" * 50)",
" ",
"- print(\"\\n[1\/3] Scanning for documents...\")",
"- files = run_scan()",
"+ try:",
"+ # Phase 1: Scan",
"+ if progress:",
"+ progress.update_step(\"detect\")",
"+ progress.add_log(\"Scanne nach Dokumenten...\")",
" ",
"- if files:",
"- print(f\"\\n[2\/3] Processing {len(files)} documents...\")",
"- process_queue()",
"- else:",
"- print(\"\\n[2\/3] No new documents to process\")",
"+ print(\"\\n[1\/3] Scanning for documents...\")",
"+ files = scan_directory()",
"+ print(f\"Found {len(files)} files\")",
" ",
"- print(\"\\n[3\/3] Embedding remaining chunks...\")",
"- embedded = embed_pending_chunks()",
"- print(f\"Embedded {embedded} chunks\")",
"+ if progress:",
"+ progress.update_progress(total=len(files))",
"+ progress.add_log(f\"{len(files)} Dokumente gefunden\")",
" ",
"- print(\"\\n\" + \"=\" * 50)",
"- print(\"Pipeline complete!\")",
"+ if files:",
"+ queued = queue_files(files)",
"+ print(f\"Queued {queued} files\")",
" ",
"+ # Phase 2: Process each file",
"+ print(f\"\\n[2\/3] Processing {len(files)} documents...\")",
"+ items = db.get_pending_queue_items(limit=100)",
" ",
"+ total_chunks = 0",
"+ total_embeddings = 0",
"+ processed = 0",
"+ failed = 0",
"+",
"+ for item in items:",
"+ # Check if cancelled",
"+ if progress and progress.is_cancelled():",
"+ progress.add_log(\"Pipeline abgebrochen durch Benutzer\")",
"+ progress.complete(\"cancelled\")",
"+ print(\"\\nPipeline cancelled by user\")",
"+ return",
"+",
"+ queue_id = item[\"id\"]",
"+ file_path = item[\"file_path\"]",
"+ file_name = Path(file_path).name",
"+",
"+ if progress:",
"+ progress.update_document(file_name)",
"+",
"+ db.update_queue_status(queue_id, \"processing\")",
"+",
"+ try:",
"+ success, chunks, embedded = process_file(file_path, progress)",
"+ if success:",
"+ db.update_queue_status(queue_id, \"done\")",
"+ processed += 1",
"+ total_chunks += chunks",
"+ total_embeddings += embedded",
"+ else:",
"+ db.update_queue_status(queue_id, \"failed\", \"Processing failed\")",
"+ failed += 1",
"+ except Exception as e:",
"+ db.update_queue_status(queue_id, \"failed\", str(e))",
"+ failed += 1",
"+ if progress:",
"+ progress.add_log(f\"FEHLER bei {file_name}: {str(e)[:50]}\")",
"+",
"+ if progress:",
"+ progress.update_progress(",
"+ processed=processed,",
"+ failed=failed,",
"+ chunks=total_chunks,",
"+ embeddings=total_embeddings,",
"+ )",
"+ else:",
"+ print(\"\\n[2\/3] No new documents to process\")",
"+ if progress:",
"+ progress.add_log(\"Keine neuen Dokumente gefunden\")",
"+",
"+ # Phase 3: Embed remaining",
"+ if progress:",
"+ progress.update_step(\"embed\")",
"+ progress.add_log(\"Verarbeite ausstehende Embeddings...\")",
"+",
"+ print(\"\\n[3\/3] Embedding remaining chunks...\")",
"+ embedded = embed_pending_chunks()",
"+ print(f\"Embedded {embedded} chunks\")",
"+",
"+ if progress:",
"+ if embedded > 0:",
"+ progress.add_log(f\"{embedded} weitere Embeddings erstellt\")",
"+",
"+ # Complete",
"+ print(\"\\n\" + \"=\" * 50)",
"+ print(\"Pipeline complete!\")",
"+",
"+ if progress:",
"+ progress.add_log(\"Pipeline erfolgreich abgeschlossen\")",
"+ progress.complete(\"completed\")",
"+",
"+ except Exception as e:",
"+ db.log(\"ERROR\", f\"Pipeline error: {e}\")",
"+ print(f\"Error: {e}\")",
"+ if progress:",
"+ progress.add_log(f\"FEHLER: {str(e)}\")",
"+ progress.complete(\"failed\", str(e))",
"+ raise",
"+",
"+",
" def main():",
" \"\"\"Main entry point.\"\"\"",
" if len(sys.argv) < 2:"
]
}
],
"userModified": false,
"replaceAll": false
}
}