Add monitoring with another system

docker-compose.yaml

@@ -88,7 +88,7 @@ services:
       #      FILE_CHUNK_SIZE: 1048576
       FILE_CHUNK_SIZE: 1000000
       POSTGRES_DB: no_copy_
-
+#      SPRING_KAFKA_BOOTSTRAP_SERVERS: kafka:9092
       POSTGRES_USER: ${POSTGRES_USER}
       POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
       POSTGRES_PORT: 5432

src/main/java/ru/soune/nocopy/configuration/security/SecurityConfig.java

@@ -25,7 +25,7 @@ public class SecurityConfig {
                 .sessionManagement(session ->
                         session.sessionCreationPolicy(SessionCreationPolicy.STATELESS))
                 .authorizeHttpRequests(auth -> auth
-                        .requestMatchers("api/v{version}/files/public-download/**").permitAll()
+                        .requestMatchers("/api/v{version}/files/public-download/**").permitAll()
                         .requestMatchers("/api/files/public/**").permitAll()
                         .requestMatchers("/api/auth/**").permitAll()
                         .requestMatchers("/api/file/link/**").permitAll()

src/main/java/ru/soune/nocopy/controller/HealtCheckController.java

@@ -1,16 +1,40 @@
 package ru.soune.nocopy.controller;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.web.bind.annotation.*;
+import ru.soune.nocopy.dto.monitoring.MonitoringDTO;
+import ru.soune.nocopy.entity.monitoring.FileMonitoringEntity;
+import ru.soune.nocopy.repository.FileMonitoringRepository;
 
 import java.util.HashMap;
 import java.util.Map;
 
 @RestController
 @RequestMapping("check/api")
+@Slf4j
 public class HealtCheckController {
 
+    private final KafkaTemplate<String, Object> kafkaTemplate;
+
+    private final FileMonitoringRepository monitoringRepository;
+
+    private final ObjectMapper objectMapper;
+
+    @Value("${server.baseurl}")
+    private String baseUrl;
+
+    public HealtCheckController(KafkaTemplate<String, Object> kafkaTemplate, FileMonitoringRepository monitoringRepository, ObjectMapper objectMapper) {
+        this.kafkaTemplate = kafkaTemplate;
+        this.monitoringRepository = monitoringRepository;
+        this.objectMapper = objectMapper;
+    }
+
     @GetMapping("/healt")
     public HttpStatus healtCheck() {
         return HttpStatus.OK;

src/main/java/ru/soune/nocopy/dto/file/YandexSearchResponse.java

@@ -32,6 +32,9 @@ public class YandexSearchResponse {
 
         @JsonProperty("host")
         private String host;
+
+        @JsonProperty("file_id")
+        private String fileId;
     }
 
     private int page;

src/main/java/ru/soune/nocopy/dto/monitoring/MonitoringDTO.java

@@ -0,0 +1,13 @@
+package ru.soune.nocopy.dto.monitoring;
+
+import lombok.Builder;
+import lombok.Data;
+
+@Data
+@Builder
+public class MonitoringDTO {
+    String fileId;
+    String baseUrl;
+    String engine;
+    String searchType;
+}

src/main/java/ru/soune/nocopy/kafka/MonitoringResultListener.java

@@ -1,21 +1,22 @@
 package ru.soune.nocopy.kafka;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 import ru.soune.nocopy.dto.file.YandexSearchResponse;
+import ru.soune.nocopy.service.violation.ViolationService;
 
 @Component
 @Slf4j
 @RequiredArgsConstructor
 public class MonitoringResultListener {
 
-    private final ObjectMapper objectMapper;
+    private final ViolationService violationService;
 
-//    @KafkaListener(topics = "monitoring-results", groupId = "dashboard")
+    @KafkaListener(topics = "monitoring-results", groupId = "dashboard")
-    public void handleCommand(String message) {
+    public void handleCommand(YandexSearchResponse.ImageResult imageResult) {
-        YandexSearchResponse yandexSearchResponse = objectMapper.convertValue(message, YandexSearchResponse.class);
+        violationService.processViolation(imageResult.getUrl(), imageResult.getHost(), imageResult.getPageUrl(),
+                imageResult.getFileId(), imageResult.getPageTitle());
     }
 }

src/main/java/ru/soune/nocopy/scheduler/MonitoringScheduler.java

@@ -4,11 +4,13 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
+import ru.soune.nocopy.dto.monitoring.MonitoringDTO;
 import ru.soune.nocopy.entity.file.FileStatus;
 import ru.soune.nocopy.entity.monitoring.FileMonitoringEntity;
 import ru.soune.nocopy.entity.notification.NotificationType;
@@ -33,7 +35,12 @@ public class MonitoringScheduler {
 
     private final NotificationService notificationService;
 
-    private final KafkaTemplate<String, FileMonitoringEntity> kafkaTemplate;
+    private final KafkaTemplate<Object, String> kafkaTemplate;
+
+    private final ObjectMapper objectMapper;
+
+    @Value("${server.baseurl}")
+    private String baseUrl;
 
     @Scheduled(cron = "0 1 2 * * *", zone = "Europe/Moscow")
     @Transactional

src/main/java/ru/soune/nocopy/service/monitoring/MonitoringSearchService.java

@@ -5,11 +5,14 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
 import jakarta.mail.MessagingException;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.MessageSource;
+import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import ru.soune.nocopy.configuration.search.SearchProperties;
 import ru.soune.nocopy.dto.file.YandexSearchResponse;
+import ru.soune.nocopy.dto.monitoring.MonitoringDTO;
 import ru.soune.nocopy.dto.tarriff.TariffDTO;
 import ru.soune.nocopy.entity.monitoring.FileMonitoringEntity;
 import ru.soune.nocopy.entity.monitoring.MonitoringType;
@@ -58,7 +61,28 @@ public class MonitoringSearchService {
 
     private final NotificationService notificationService;
 
-    private final MessageSource messageSource;
+    private final ObjectMapper objectMapper;
+
+    private final KafkaTemplate<Object, String> kafkaTemplate;
+
+    @Value("${server.baseurl}")
+    private String baseUrl;
+
+    //                MonitoringDTO monitoringGoogleDTO = MonitoringDTO.builder().baseUrl(baseUrl).
+    //                        fileId(monitoring.getFile().getId())
+    //                        .engine("google_lens")
+    //                        .searchType("exact_matches")
+    //                        .build();
+    //
+    //                kafkaTemplate.send("monitoring-commands", objectMapper.writeValueAsString(monitoringGoogleDTO));
+    //
+    //                MonitoringDTO monitoringYandexDTO = MonitoringDTO.builder().baseUrl(baseUrl).
+    //                        fileId(monitoring.getFile().getId())
+    //                        .engine("yandex_reverse_image")
+    //                        .searchType("visual_matches")
+    //                        .build();
+    //
+    //                kafkaTemplate.send("monitoring-commands", objectMapper.writeValueAsString(monitoringYandexDTO));
 
     @Transactional(noRollbackFor = TariffNotFoundException.class)
     public void processFileSearch(FileMonitoringEntity monitoring) throws MessagingException, IOException {
@@ -81,20 +105,26 @@ public class MonitoringSearchService {
 
             log.info("Monitoring search settings: useYandex={}, useGoogle={}", useYandex, useGoogle);
 
-            List<YandexSearchResponse.ImageResult> allResults = new ArrayList<>();
+//            List<YandexSearchResponse.ImageResult> allResults = new ArrayList<>();
 
             if (useYandex) {
                 try {
-                    String yandexResponse = searchImageService.searchReverseByPublicUrl(
+//                    String yandexResponse = searchImageService.searchReverseByPublicUrl(
-                            monitoring.getFile(), "yandex_reverse_image", "visual_matches");
+//                            monitoring.getFile(), "yandex_reverse_image", "visual_matches");
+//
+//                    List<YandexSearchResponse.ImageResult> yandexImages =
+//                            searchImageService.getAllImagesWithoutPagination(yandexResponse, "visual_matches");
+//
+//                    allResults.addAll(yandexImages);
+//                    log.info("Yandex search found {} results", yandexImages.size());
+                    MonitoringDTO monitoringYandexDTO = MonitoringDTO.builder().baseUrl(baseUrl).
+                            fileId(monitoring.getFile().getId())
+                            .engine("yandex_reverse_image")
+                            .searchType("visual_matches")
+                            .build();
 
-                    List<YandexSearchResponse.ImageResult> yandexImages =
+                    kafkaTemplate.send("monitoring-commands", objectMapper.writeValueAsString(monitoringYandexDTO));
-                            searchImageService.getAllImagesWithoutPagination(yandexResponse, "visual_matches");
+                } catch (IOException e) {
-
-                    allResults.addAll(yandexImages);
-                    log.info("Yandex search found {} results", yandexImages.size());
-
-                } catch (TimeoutException | IOException e) {
                     log.error("Error processing monitoring search", e);
                     monitoring.setLastRunStatus("ERROR: " + e.getMessage());
                     updateNextRun(monitoring);
@@ -115,17 +145,23 @@ public class MonitoringSearchService {
 
             if (useGoogle) {
                 try {
-                    String googleResponse = searchImageService.searchReverseByPublicUrl(
+//                    String googleResponse = searchImageService.searchReverseByPublicUrl(
-                            monitoring.getFile(), "google_lens", "exact_matches");
+//                            monitoring.getFile(), "google_lens", "exact_matches");
+//
+//                    List<YandexSearchResponse.ImageResult> googleImages =
+//                            searchImageService.getAllImagesWithoutPagination(googleResponse,
+//                                    "exact_matches");
+//
+//                    allResults.addAll(googleImages);
+//                    log.info("Google search found {} results", googleImages.size());
+                    MonitoringDTO monitoringGoogleDTO = MonitoringDTO.builder().baseUrl(baseUrl).
+                            fileId(monitoring.getFile().getId())
+                            .engine("google_lens")
+                            .searchType("exact_matches")
+                            .build();
 
-                    List<YandexSearchResponse.ImageResult> googleImages =
+                    kafkaTemplate.send("monitoring-commands", objectMapper.writeValueAsString(monitoringGoogleDTO));
-                            searchImageService.getAllImagesWithoutPagination(googleResponse,
+                } catch (IOException e) {
-                                    "exact_matches");
-
-                    allResults.addAll(googleImages);
-                    log.info("Google search found {} results", googleImages.size());
-
-                } catch (TimeoutException | IOException e) {
                     log.error("Error processing monitoring search", e);
                     monitoring.setLastRunStatus("ERROR: " + e.getMessage());
                     updateNextRun(monitoring);
@@ -146,25 +182,25 @@ public class MonitoringSearchService {
                 log.info("Google search is disabled by settings");
             }
 
-            List<YandexSearchResponse.ImageResult> uniqueResults = allResults;
+//            List<YandexSearchResponse.ImageResult> uniqueResults = allResults;
-            if (useYandex && useGoogle) {
+//            if (useYandex && useGoogle) {
-                uniqueResults = searchImageService.removeDuplicateUrls(
+//                uniqueResults = searchImageService.removeDuplicateUrls(
-                        useYandex ? allResults : new ArrayList<>(),
+//                        useYandex ? allResults : new ArrayList<>(),
-                        useGoogle ? allResults : new ArrayList<>()
+//                        useGoogle ? allResults : new ArrayList<>()
-                );
+//                );
-                log.info("After deduplication: {} unique results", uniqueResults.size());
+//                log.info("After deduplication: {} unique results", uniqueResults.size());
-            }
+//            }
 
-            for (YandexSearchResponse.ImageResult imageResult : uniqueResults) {
+//            for (YandexSearchResponse.ImageResult imageResult : uniqueResults) {
-                violationService.processViolation(imageResult, monitoring.getFile(), null);
+//                violationService.processViolation(imageResult, monitoring.getFile(), null);
-            }
+//            }
 
-            monitoring.setLastRunStatus("SUCCESS");
+//            monitoring.setLastRunStatus("SUCCESS");
             updateNextRun(monitoring);
             monitoringRepository.save(monitoring);
-            if (uniqueResults.isEmpty()) {
+//            if (uniqueResults.isEmpty()) {
-                log.info("No results found for monitoring file {}", monitoring.getFile().getId());
+//                log.info("No results found for monitoring file {}", monitoring.getFile().getId());
-            }
+//            }
 
         } catch (TariffNotFoundException e) {
             TariffInfo activeTariffInfo = user.getActiveTariffInfo();

src/main/java/ru/soune/nocopy/service/violation/ViolationService.java

@@ -13,7 +13,6 @@ import ru.soune.nocopy.dto.violation.ViolationStatisticsResponse;
 import ru.soune.nocopy.entity.file.FileEntity;
 import ru.soune.nocopy.entity.file.FileStatus;
 import ru.soune.nocopy.entity.search.GlobalSearchResult;
-import ru.soune.nocopy.entity.user.User;
 import ru.soune.nocopy.entity.violation.Violation;
 import ru.soune.nocopy.exception.UserNotHavePermission;
 import ru.soune.nocopy.repository.ComplaintEntityRepository;
@@ -72,6 +71,26 @@ public class ViolationService {
         }
     }
 
+    public void processViolation(String url, String host, String pageUrl, String fileId, String pageTitle) {
+        String urlHash = DigestUtils.sha256Hex(url);
+        FileEntity file = fileEntityRepository.findByFileId(fileId);
+
+        if (!violationRepository.existsByUrlHash(urlHash)) {
+            Violation violation = new Violation();
+
+            violation.setHost(host);
+            violation.setUrl(url);
+            violation.setUrlHash(urlHash);
+            violation.setPageUrl(pageUrl);
+            violation.setPageTitle(pageTitle);
+            violation.setFileEntity(file);
+            violation.setCreatedDate(LocalDateTime.now());
+            violation.setStatus(ViolationStatus.NEW.name());
+
+            violationRepository.save(violation);
+        }
+    }
+
     public Page<Violation> getViolationsByFilesAndFilters(
             List<FileEntity> targetFiles,
             LocalDateTime startDate,

src/main/resources/application.yaml

@@ -1,13 +1,15 @@
 spring:
-#  kafka:
+  kafka:
-#    bootstrap-servers: kafka:9092
+    bootstrap-servers: kafka:9092
-#    consumer:
+    consumer:
-#      group-id: dashboard
+      group-id: dashboard
-#      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
-#      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
-#    producer:
+      properties:
-#      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+        spring.json.trusted.packages: "ru.soune.nocopy.dto.monitoring.*,ru.no_copy.monitoring.dto.*"
-#      value-serializer: org.apache.kafka.common.serialization.StringSerializer
+    producer:
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.apache.kafka.common.serialization.StringSerializer
   cloud:
     compatibility-verifier:
       enabled: false