secont commit

src/main/java/com/homme/demo/service/DocumentStorageService.java

@@ -0,0 +1,162 @@
+package com.homme.demo.service;
+
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Service;
+
+import com.homme.demo.entity.Document;
+import com.homme.demo.entity.DocumentChunck;
+import com.homme.demo.repository.DocumentChunckRepository;
+import com.homme.demo.repository.DocumentRepository;
+
+
+import jakarta.transaction.Transactional;
+
+@Service
+public class DocumentStorageService {
+	
+	@Autowired
+	private DocumentRepository documentRepossitory;
+	
+	@Autowired
+	private DocumentChunckRepository documentChunckRepository;
+	
+	@Autowired
+	private DocumentChunckService documentChunkService;
+	
+	@Autowired
+	private AzureEmbeddingService azureEmbeddingService;
+	
+	@Transactional
+	public ResponseEntity<String> saveDocuemntWithChuncks(
+			String sourceUrl,
+			String rawText,
+			String cleanText) {
+		
+		String message = "";
+		if(documentRepossitory.existsBySourceUrl(sourceUrl)) {
+			
+			message = "Dieses Dokument wurde bereits gespeichert.";
+			return new ResponseEntity<>(message,
+										HttpStatus.CONFLICT);
+		}
+			
+		Document document = Document.builder()
+				.sourceUrl(sourceUrl)
+				.rawText(rawText)
+				.cleanText(cleanText)
+				.createdAt(LocalDate.now())
+				.build();
+		Integer savedDocumentId = documentRepossitory.save(document).getId();
+		
+		
+		
+		ExecutorService pool = Executors.newFixedThreadPool(4);
+		try
+		{
+	
+			List<String> chuncks = documentChunkService.splitTextIntoChunks(cleanText);
+			
+			List<Future<float[]>> futures = new ArrayList<>();
+			
+			for(int i = 0; i < chuncks.size(); i++) {
+				
+				String chunckText = chuncks.get(i);
+				
+				futures.add(pool.submit(() ->
+				      (chunckText == null || chunckText.isBlank())
+				      ?null
+				      : azureEmbeddingService.createEmbeddingFloatArray(chunckText)
+				      ));
+				
+			    }
+				for(int i=0; i < chuncks.size(); i++) {
+					
+					
+					float [] embedding = futures.get(i).get();
+					String chunckText = chuncks.get(i);
+					DocumentChunck documentChunck = DocumentChunck.builder()
+							.chunckIndex(i)
+							.chunckText(chunckText)
+							.documentId(savedDocumentId)
+							.embedding(embedding)
+							.build();
+					documentChunckRepository.save(documentChunck);
+					
+				
+				}
+				message = "Dokument wurde erfolgreich gespeichert.";
+				return new ResponseEntity<>(message,
+						HttpStatus.CREATED);
+		
+		     }catch(InterruptedException | ExecutionException e) {
+			
+					Thread.currentThread().interrupt();
+					message = "Fehler bei parallerer Verarbeitung"+ e.getMessage();
+					return new ResponseEntity<>(message, HttpStatus.BAD_REQUEST);
+			
+		} finally {
+			pool.shutdown();
+		}
+	}
+		
+	@Transactional
+	public int fillMissingEmbedding() {
+		
+		
+		List<DocumentChunck> chuncks = documentChunckRepository.findByEmbeddingIsNull();
+		
+		
+		
+		ExecutorService pool = Executors.newFixedThreadPool(6);
+		
+		List<Future<float[]>> futures = new ArrayList<>();
+		
+		try {
+		
+			for(DocumentChunck chunck: chuncks) {
+				
+				String chunckText = chunck.getChunckText();
+				
+				futures.add(pool.submit(() -> 
+				     (chunckText == null || chunckText.isBlank())
+				     ? null
+				     : azureEmbeddingService.createEmbeddingFloatArray(chunckText)
+				     ));
+				   
+				
+			}
+			int updatedCount = 0;
+			for( int i = 0; i < chuncks.size(); i++) {
+				float[] embedding = futures.get(i).get();
+				chuncks.get(i).setEmbedding(embedding);
+				documentChunckRepository.save(chuncks.get(i));
+				updatedCount++;
+				System.out.println("updatedCount  = "+updatedCount);
+				
+			}
+			return updatedCount;
+			
+		}catch(InterruptedException | ExecutionException e ) {
+			Thread.currentThread().interrupt();
+			throw new RuntimeException("Fehler bei parallerer Verarbeitung: "+e.getMessage());
+			
+		}finally {
+			pool.shutdown();
+		}
+		
+		
+		
+	}
+	
+
+}