digiops/marketing-parser
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

.

Browse Source
This commit is contained in:
root
2025-11-20 09:27:21 +05:00
parent d6ab1f5681
commit 7af6cf2cd9
3 changed files with 269 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java
+143 -83
View File
@@ -22,6 +22,8 @@ import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.util.retry.Retry;
import reactor.util.retry.RetryBackoffSpec;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import jakarta.annotation.PostConstruct;
@Service
@@ -49,9 +51,14 @@ public class OpenAIAnalyticsService {
@Value("${openai.retry.multiplier:2.0}")
private double retryMultiplier;
@Value("${openai.rateLimit.maxConcurrentRequests:2}")
private int maxConcurrentRequests;
@Value("${openai.api.key}")
private String apiKey;
private Semaphore requestSemaphore;
private String openaiUrl;
public OpenAIAnalyticsService(
@@ -82,7 +89,10 @@ public class OpenAIAnalyticsService {
.defaultHeader("Authorization", "Bearer " + apiKey)
.build();
logger.info("OpenAIAnalyticsService initialized with model: {}", modelName);
this.requestSemaphore = new Semaphore(maxConcurrentRequests, true);
logger.info("OpenAIAnalyticsService initialized with model: {} and max concurrent requests: {}",
modelName, maxConcurrentRequests);
}
public MarketItem.Analytics analyzeText(String text) {
@@ -164,46 +174,61 @@ public class OpenAIAnalyticsService {
requestBody.put("temperature", 0.3);
try {
Map<String, Object> response = this.webClient.post()
.contentType(MediaType.APPLICATION_JSON)
.accept(MediaType.APPLICATION_JSON)
.bodyValue(requestBody)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
})
.retryWhen(createRetrySpec("generate"))
.onErrorResume(err -> {
logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(),
err.getClass().getSimpleName());
if (err instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) err;
if (wcre.getStatusCode().value() == 401) {
logger.error(
"OpenAI API key is invalid or expired. Please check your openai.api.key configuration.");
} else if (wcre.getStatusCode().value() == 429) {
logger.error("OpenAI API rate limit exceeded after all retry attempts.");
} else if (wcre.getStatusCode().value() >= 500) {
logger.error("OpenAI API server error after all retry attempts.");
}
}
return Mono.empty();
})
.block(Duration.ofMillis(timeoutMs));
if (response == null) {
// Acquire semaphore permit before making request
if (!requestSemaphore.tryAcquire(30, TimeUnit.SECONDS)) {
logger.warn("Timeout waiting for OpenAI request slot. Skipping request.");
return null;
}
@SuppressWarnings("unchecked")
List<Map<String, Object>> choices = (List<Map<String, Object>>) response.get("choices");
if (choices != null && !choices.isEmpty()) {
Map<String, Object> firstChoice = choices.get(0);
Map<String, Object> messageObj = (Map<String, Object>) firstChoice.get("message");
if (messageObj != null) {
Object content = messageObj.get("content");
return content == null ? null : String.valueOf(content).trim();
try {
Map<String, Object> response = this.webClient.post()
.contentType(MediaType.APPLICATION_JSON)
.accept(MediaType.APPLICATION_JSON)
.bodyValue(requestBody)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
})
.retryWhen(createRetrySpec("generate"))
.onErrorResume(err -> {
logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(),
err.getClass().getSimpleName());
if (err instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) err;
if (wcre.getStatusCode().value() == 401) {
logger.error(
"OpenAI API key is invalid or expired. Please check your openai.api.key configuration.");
} else if (wcre.getStatusCode().value() == 429) {
logger.error("OpenAI API rate limit exceeded after all retry attempts.");
} else if (wcre.getStatusCode().value() >= 500) {
logger.error("OpenAI API server error after all retry attempts.");
}
}
return Mono.empty();
})
.block(Duration.ofMillis(timeoutMs));
if (response == null) {
return null;
}
@SuppressWarnings("unchecked")
List<Map<String, Object>> choices = (List<Map<String, Object>>) response.get("choices");
if (choices != null && !choices.isEmpty()) {
Map<String, Object> firstChoice = choices.get(0);
Map<String, Object> messageObj = (Map<String, Object>) firstChoice.get("message");
if (messageObj != null) {
Object content = messageObj.get("content");
return content == null ? null : String.valueOf(content).trim();
}
}
return null;
} finally {
// Always release semaphore permit
requestSemaphore.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.warn("Interrupted while waiting for OpenAI request slot: {}", e.getMessage());
return null;
} catch (Exception e) {
logger.warn("Error calling OpenAI generate: {}", e.getMessage());
@@ -229,48 +254,63 @@ public class OpenAIAnalyticsService {
requestBody.put("temperature", 0.3);
try {
Map<String, Object> response = this.webClient.post()
.contentType(MediaType.APPLICATION_JSON)
.accept(MediaType.APPLICATION_JSON)
.bodyValue(requestBody)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
})
.retryWhen(createRetrySpec("generateWithInstruction"))
.onErrorResume(err -> {
logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(),
err.getClass().getSimpleName());
if (err instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) err;
if (wcre.getStatusCode().value() == 401) {
logger.error(
"OpenAI API key is invalid or expired. Please check your openai.api.key configuration.");
} else if (wcre.getStatusCode().value() == 429) {
logger.error("OpenAI API rate limit exceeded after all retry attempts.");
} else if (wcre.getStatusCode().value() >= 500) {
logger.error("OpenAI API server error after all retry attempts.");
}
}
return Mono.empty();
})
.block(Duration.ofMillis(timeoutMs));
if (response == null) {
// Acquire semaphore permit before making request
if (!requestSemaphore.tryAcquire(30, TimeUnit.SECONDS)) {
logger.warn("Timeout waiting for OpenAI request slot. Skipping request.");
return null;
}
@SuppressWarnings("unchecked")
List<Map<String, Object>> choices = (List<Map<String, Object>>) response.get("choices");
if (choices != null && !choices.isEmpty()) {
Map<String, Object> firstChoice = choices.get(0);
Map<String, Object> messageObj = (Map<String, Object>) firstChoice.get("message");
if (messageObj != null) {
Object content = messageObj.get("content");
String result = content == null ? null : String.valueOf(content).trim();
System.out.println("OpenAI response: " + result);
return result;
try {
Map<String, Object> response = this.webClient.post()
.contentType(MediaType.APPLICATION_JSON)
.accept(MediaType.APPLICATION_JSON)
.bodyValue(requestBody)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
})
.retryWhen(createRetrySpec("generateWithInstruction"))
.onErrorResume(err -> {
logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(),
err.getClass().getSimpleName());
if (err instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) err;
if (wcre.getStatusCode().value() == 401) {
logger.error(
"OpenAI API key is invalid or expired. Please check your openai.api.key configuration.");
} else if (wcre.getStatusCode().value() == 429) {
logger.error("OpenAI API rate limit exceeded after all retry attempts.");
} else if (wcre.getStatusCode().value() >= 500) {
logger.error("OpenAI API server error after all retry attempts.");
}
}
return Mono.empty();
})
.block(Duration.ofMillis(timeoutMs));
if (response == null) {
return null;
}
@SuppressWarnings("unchecked")
List<Map<String, Object>> choices = (List<Map<String, Object>>) response.get("choices");
if (choices != null && !choices.isEmpty()) {
Map<String, Object> firstChoice = choices.get(0);
Map<String, Object> messageObj = (Map<String, Object>) firstChoice.get("message");
if (messageObj != null) {
Object content = messageObj.get("content");
String result = content == null ? null : String.valueOf(content).trim();
System.out.println("OpenAI response: " + result);
return result;
}
}
return null;
} finally {
// Always release semaphore permit
requestSemaphore.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.warn("Interrupted while waiting for OpenAI request slot: {}", e.getMessage());
return null;
} catch (Exception e) {
logger.warn("Error calling OpenAI generate (generic): {}", e.getMessage());
@@ -286,13 +326,9 @@ public class OpenAIAnalyticsService {
if (throwable instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) throwable;
int statusCode = wcre.getStatusCode().value();
// Retry on 429 (rate limit) and 5xx (server errors)
boolean shouldRetry = statusCode == 429 || statusCode >= 500;
if (shouldRetry) {
logger.warn("OpenAI API returned {} for operation '{}'. Will retry...", statusCode,
operation);
}
return shouldRetry;
// Only retry on 429 (rate limit) and 5xx (server errors)
// Don't retry on 401, 403, etc.
return statusCode == 429 || statusCode >= 500;
}
// Retry on network errors (timeouts, connection issues)
return throwable instanceof java.util.concurrent.TimeoutException
@@ -301,11 +337,35 @@ public class OpenAIAnalyticsService {
})
.doBeforeRetry(retrySignal -> {
long attempt = retrySignal.totalRetries() + 1;
logger.info("Retrying OpenAI request for operation '{}' (attempt {}/{})",
operation, attempt, maxRetryAttempts);
Throwable failure = retrySignal.failure();
Duration retryAfter = null;
if (failure instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) failure;
if (wcre.getStatusCode().value() == 429) {
// Try to extract Retry-After header
String retryAfterHeader = wcre.getHeaders().getFirst("Retry-After");
if (retryAfterHeader != null) {
try {
int seconds = Integer.parseInt(retryAfterHeader);
retryAfter = Duration.ofSeconds(seconds);
logger.warn(
"OpenAI API returned 429 for operation '{}'. Retry-After: {} seconds. Will retry in {}ms (attempt {}/{})",
operation, seconds, retryAfter.toMillis(), attempt, maxRetryAttempts);
} catch (NumberFormatException e) {
// Ignore if header is not a number
}
}
}
}
if (retryAfter == null) {
logger.warn("OpenAI API returned error for operation '{}'. Will retry (attempt {}/{})",
operation, attempt, maxRetryAttempts);
}
})
.onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> {
logger.error("OpenAI request for operation '{}' exhausted all {} retry attempts",
logger.error("OpenAI request for operation '{}' exhausted all {} retry attempts. Giving up.",
operation, maxRetryAttempts);
return retrySignal.failure();
});
src/main/java/kz/konturai/parser/service/OpenAiChartService.java
+123 -63
View File
@@ -19,6 +19,9 @@ import org.slf4j.LoggerFactory;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.util.retry.Retry;
import reactor.util.retry.RetryBackoffSpec;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import jakarta.annotation.PostConstruct;
@Service
public class OpenAiChartService {
@@ -45,6 +48,11 @@ public class OpenAiChartService {
@Value("${openai.retry.multiplier:2.0}")
private double retryMultiplier;
@Value("${openai.rateLimit.maxConcurrentRequests:2}")
private int maxConcurrentRequests;
private Semaphore requestSemaphore;
public OpenAiChartService(
@Value("${openai.api.url:https://api.openai.com/v1/chat/completions}") String apiUrl,
@Value("${openai.api.key:}") String apiKey) {
@@ -66,6 +74,12 @@ public class OpenAiChartService {
.build();
}
@PostConstruct
public void initialize() {
this.requestSemaphore = new Semaphore(maxConcurrentRequests, true);
logger.info("OpenAiChartService initialized with max concurrent requests: {}", maxConcurrentRequests);
}
public Mono<String> getChartDataJson(String aggregatedLearnings) {
return getChartDataJson(aggregatedLearnings, "ru");
}
@@ -99,60 +113,86 @@ public class OpenAiChartService {
logger.info("Calling OpenAI API for task '{}' with model '{}'.", taskName, modelName);
logger.info("OpenAI Prompt: {}", prompt);
return this.webClient.post()
.contentType(MediaType.APPLICATION_JSON)
.accept(MediaType.APPLICATION_JSON)
.bodyValue(body)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
})
.timeout(Duration.ofMillis(timeoutMs))
.retryWhen(createRetrySpec(taskName))
.doOnSuccess(response -> {
// 4. Логируем успешный, но еще не обработанный ответ
logger.info("Raw OpenAI response for task '{}': {}", taskName, response);
})
.map(resp -> {
try {
List<Map<String, Object>> choices = (List<Map<String, Object>>) resp.get("choices");
if (choices == null || choices.isEmpty()) {
// 5. Логируем аномальные, но не ошибочные ситуации
logger.warn("OpenAI response for task '{}' contained no 'choices'.", taskName);
return null;
}
Map<String, Object> message = (Map<String, Object>) choices.get(0).get("message");
if (message == null) {
logger.warn("OpenAI choice for task '{}' contained no 'message'.", taskName);
return null;
}
Object content = message.get("content");
if (content == null) {
logger.warn("OpenAI message for task '{}' contained no 'content'.", taskName);
return null;
}
String responseContent = String.valueOf(content);
logger.info("Successfully extracted content from OpenAI for task '{}'. Content length: {}",
taskName, responseContent.length());
return responseContent;
} catch (Exception e) {
// 6. Логируем ошибку парсинга, если структура ответа неожиданная
logger.error("Failed to parse OpenAI response for task '{}'. Response body: {}", taskName, resp,
e);
return null;
// Acquire semaphore permit before making request
return Mono.fromCallable(() -> {
try {
if (!requestSemaphore.tryAcquire(30, TimeUnit.SECONDS)) {
logger.warn("Timeout waiting for OpenAI request slot for task '{}'. Skipping request.", taskName);
return null;
}
return requestSemaphore;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.warn("Interrupted while waiting for OpenAI request slot for task '{}'", taskName);
return null;
}
})
.flatMap(semaphore -> {
if (semaphore == null) {
return Mono.empty();
}
})
.doOnError(e -> {
// 7. Логируем ошибки сети или HTTP-статусов (4xx, 5xx)
if (e instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) e;
logger.error("Error from OpenAI API for task '{}'. Status: {}, Body: {}",
taskName, wcre.getStatusCode(), wcre.getResponseBodyAsString(), e);
} else {
logger.error("Generic WebClient error for task '{}'.", taskName, e);
}
})
.onErrorResume(e -> Mono.empty()); // В случае любой ошибки возвращаем пустой результат, чтобы не
// прерывать цепочку вызовов
return this.webClient.post()
.contentType(MediaType.APPLICATION_JSON)
.accept(MediaType.APPLICATION_JSON)
.bodyValue(body)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
})
.timeout(Duration.ofMillis(timeoutMs))
.retryWhen(createRetrySpec(taskName))
.doFinally(signalType -> {
// Always release semaphore permit
semaphore.release();
})
.doOnSuccess(response -> {
// 4. Логируем успешный, но еще не обработанный ответ
logger.info("Raw OpenAI response for task '{}': {}", taskName, response);
})
.map(resp -> {
try {
List<Map<String, Object>> choices = (List<Map<String, Object>>) resp.get("choices");
if (choices == null || choices.isEmpty()) {
// 5. Логируем аномальные, но не ошибочные ситуации
logger.warn("OpenAI response for task '{}' contained no 'choices'.", taskName);
return null;
}
Map<String, Object> message = (Map<String, Object>) choices.get(0).get("message");
if (message == null) {
logger.warn("OpenAI choice for task '{}' contained no 'message'.", taskName);
return null;
}
Object content = message.get("content");
if (content == null) {
logger.warn("OpenAI message for task '{}' contained no 'content'.", taskName);
return null;
}
String responseContent = String.valueOf(content);
logger.info(
"Successfully extracted content from OpenAI for task '{}'. Content length: {}",
taskName, responseContent.length());
return responseContent;
} catch (Exception e) {
// 6. Логируем ошибку парсинга, если структура ответа неожиданная
logger.error("Failed to parse OpenAI response for task '{}'. Response body: {}",
taskName, resp,
e);
return null;
}
})
.doOnError(e -> {
// 7. Логируем ошибки сети или HTTP-статусов (4xx, 5xx)
if (e instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) e;
logger.error("Error from OpenAI API for task '{}'. Status: {}, Body: {}",
taskName, wcre.getStatusCode(), wcre.getResponseBodyAsString(), e);
} else {
logger.error("Generic WebClient error for task '{}'.", taskName, e);
}
})
.onErrorResume(e -> Mono.empty()); // В случае любой ошибки возвращаем пустой результат,
// чтобы не
// прерывать цепочку вызовов
});
}
private String buildChartDataPrompt(String learningsAsString, String language) {
@@ -265,13 +305,9 @@ public class OpenAiChartService {
if (throwable instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) throwable;
int statusCode = wcre.getStatusCode().value();
// Retry on 429 (rate limit) and 5xx (server errors)
boolean shouldRetry = statusCode == 429 || statusCode >= 500;
if (shouldRetry) {
logger.warn("OpenAI API returned {} for operation '{}'. Will retry...", statusCode,
operation);
}
return shouldRetry;
// Only retry on 429 (rate limit) and 5xx (server errors)
// Don't retry on 401, 403, etc.
return statusCode == 429 || statusCode >= 500;
}
// Retry on network errors (timeouts, connection issues)
return throwable instanceof java.util.concurrent.TimeoutException
@@ -280,11 +316,35 @@ public class OpenAiChartService {
})
.doBeforeRetry(retrySignal -> {
long attempt = retrySignal.totalRetries() + 1;
logger.info("Retrying OpenAI request for operation '{}' (attempt {}/{})",
operation, attempt, maxRetryAttempts);
Throwable failure = retrySignal.failure();
Duration retryAfter = null;
if (failure instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) failure;
if (wcre.getStatusCode().value() == 429) {
// Try to extract Retry-After header
String retryAfterHeader = wcre.getHeaders().getFirst("Retry-After");
if (retryAfterHeader != null) {
try {
int seconds = Integer.parseInt(retryAfterHeader);
retryAfter = Duration.ofSeconds(seconds);
logger.warn(
"OpenAI API returned 429 for operation '{}'. Retry-After: {} seconds. Will retry in {}ms (attempt {}/{})",
operation, seconds, retryAfter.toMillis(), attempt, maxRetryAttempts);
} catch (NumberFormatException e) {
// Ignore if header is not a number
}
}
}
}
if (retryAfter == null) {
logger.warn("OpenAI API returned error for operation '{}'. Will retry (attempt {}/{})",
operation, attempt, maxRetryAttempts);
}
})
.onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> {
logger.error("OpenAI request for operation '{}' exhausted all {} retry attempts",
logger.error("OpenAI request for operation '{}' exhausted all {} retry attempts. Giving up.",
operation, maxRetryAttempts);
return retrySignal.failure();
});
src/main/resources/application.properties
+3 -2
View File
@@ -58,10 +58,11 @@ openai.api.key=sk-proj-zEsggv3MCgvZ2aQbRhpCfyZPKn-isyylNhO16dekXGpM3yzi9H4g0zSjp
openai.api.url=https://api.openai.com/v1/chat/completions
openai.model.name=gpt-4o-mini
openai.timeoutMs=90000
openai.retry.maxAttempts=5
openai.retry.initialDelayMs=1000
openai.retry.maxAttempts=3
openai.retry.initialDelayMs=2000
openai.retry.maxDelayMs=60000
openai.retry.multiplier=2.0
openai.rateLimit.maxConcurrentRequests=2
# Email Configuration
spring.mail.host=smtp.gmail.com