agent-foundation/src/main/java/com/company/agentfoundation/integration/ngp/NgpApiClient.java

package com.company.agentfoundation.integration.ngp;

import com.company.agentfoundation.config.FoundationProperties;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.channel.ChannelOption;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientRequestException;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.netty.http.client.HttpClient;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HexFormat;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicLong;

@Component
public class NgpApiClient {
  static final String SALES_POST_BILL_LIST = "ngp.bill.salepostbill.list";
  static final String ESHOP_DELIVERY_BILL_LIST = "ngp.eshopsaleorder.bills";
  static final String STORE_LIST = "ngp.otype.list";
  static final int DELIVERY_TIME = 3;
  static final int DELIVERED = 100;

  private final FoundationProperties.Ngp config;
  private final NgpTokenProvider tokenProvider;
  private final ObjectMapper objectMapper;
  private final WebClient webClient;
  private final Duration readTimeout;
  private final AtomicLong lastTimestamp = new AtomicLong();

  public NgpApiClient(
      FoundationProperties properties,
      NgpTokenProvider tokenProvider,
      ObjectMapper objectMapper,
      WebClient.Builder builder
  ) {
    this.config = properties.getNgp();
    this.tokenProvider = tokenProvider;
    this.objectMapper = objectMapper;
    HttpClient httpClient = HttpClient.create()
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Math.max(config.getConnectTimeoutMs(), 1))
        .responseTimeout(Duration.ofMillis(Math.max(config.getReadTimeoutMs(), 1)));
    this.webClient = builder.clone()
        .baseUrl(config.getBaseUrl())
        .clientConnector(new ReactorClientHttpConnector(httpClient))
        .codecs(codecs -> codecs.defaultCodecs()
            .maxInMemorySize(Math.max(config.getMaxResponseBytes(), 256 * 1024)))
        .build();
    this.readTimeout = Duration.ofMillis(Math.max(config.getReadTimeoutMs(), 1));
  }

  public JsonNode salesPostBillPage(String beginTime, String endTime, int pageIndex) {
    return call(SALES_POST_BILL_LIST, salesPostBillRequestBody(beginTime, endTime, pageIndex));
  }

  public JsonNode eshopDeliveryBillPage(String beginTime, String endTime, int pageIndex) {
    return call(ESHOP_DELIVERY_BILL_LIST,
        eshopDeliveryBillRequestBody(beginTime, endTime, pageIndex));
  }

  public JsonNode storePage(int pageIndex) {
    return call(STORE_LIST, storeRequestBody(pageIndex));
  }

  Map<String, Object> salesPostBillRequestBody(String beginTime, String endTime, int pageIndex) {
    Map<String, Object> body = new LinkedHashMap<>();
    body.put("pageSize", Math.min(Math.max(config.getPageSize(), 1), 200));
    body.put("pageIndex", pageIndex);
    body.put("vchtypes", List.of(2000));
    body.put("beginTime", beginTime);
    body.put("endTime", endTime);
    return body;
  }

  Map<String, Object> eshopDeliveryBillRequestBody(String beginTime, String endTime, int pageIndex) {
    Map<String, Object> body = new LinkedHashMap<>();
    body.put("pageSize", Math.min(Math.max(config.getPageSize(), 1), 200));
    body.put("pageIndex", pageIndex);
    body.put("beginTime", beginTime);
    body.put("endTime", endTime);
    body.put("timeType", DELIVERY_TIME);
    body.put("deliverStatusPackage", DELIVERED);
    body.put("showAllEshopType", true);
    return body;
  }

  Map<String, Object> storeRequestBody(int pageIndex) {
    Map<String, Object> body = new LinkedHashMap<>();
    body.put("pageSize", Math.min(Math.max(config.getPageSize(), 1), 200));
    body.put("pageIndex", pageIndex);
    body.put("showAll", true);
    body.put("showAllEshopType", true);
    return body;
  }

  private JsonNode call(String method, Object requestBody) {
    int attempts = Math.max(config.getMaxRetries(), 0) + 1;
    boolean tokenRefreshed = false;
    String token = tokenProvider.getAccessToken();
    RuntimeException lastFailure = null;
    for (int attempt = 1; attempt <= attempts; attempt++) {
      try {
        JsonNode response = execute(method, requestBody, token);
        if (!tokenRefreshed && isExpiredToken(response)) {
          tokenRefreshed = true;
          token = tokenProvider.refreshAccessToken(token);
          attempt--;
          continue;
        }
        validate(response, method);
        return response;
      } catch (WebClientResponseException exception) {
        lastFailure = exception;
        if (!transientStatus(exception.getStatusCode().value()) || attempt == attempts) break;
        pause(attempt);
      } catch (WebClientRequestException exception) {
        lastFailure = exception;
        if (attempt == attempts) break;
        pause(attempt);
      }
    }
    throw new NgpApiException("NGP 接口调用失败：" + method + "；" + rootMessage(lastFailure), lastFailure);
  }

  private JsonNode execute(String method, Object requestBody, String token) {
    String body;
    try {
      body = objectMapper.writeValueAsString(requestBody);
    } catch (JsonProcessingException exception) {
      throw new NgpApiException("NGP 请求序列化失败", exception);
    }
    Map<String, String> systemParameters = new LinkedHashMap<>();
    systemParameters.put("app_key", config.getAppKey());
    systemParameters.put("method", method);
    systemParameters.put("timestamp", nextTimestamp());
    systemParameters.put("token", token);
    String signature = sign(systemParameters, body, config.getAppSecret());

    return webClient.post()
        .uri(builder -> {
          builder.path(config.getRestPath());
          systemParameters.forEach(builder::queryParam);
          builder.queryParam("sign", signature);
          return builder.build();
        })
        .contentType(MediaType.APPLICATION_JSON)
        .accept(MediaType.APPLICATION_JSON)
        .bodyValue(body)
        .retrieve()
        .bodyToMono(JsonNode.class)
        .block(readTimeout);
  }

  static String sign(Map<String, String> systemParameters, String body, String appSecret) {
    if (appSecret == null || appSecret.isBlank()) throw new NgpApiException("NGP AppSecret 未配置");
    StringBuilder source = new StringBuilder();
    new TreeMap<>(systemParameters).forEach((name, value) -> source.append(name).append(value));
    source.append(body);
    try {
      Mac mac = Mac.getInstance("HmacMD5");
      mac.init(new SecretKeySpec(appSecret.getBytes(StandardCharsets.UTF_8), "HmacMD5"));
      return HexFormat.of().formatHex(mac.doFinal(source.toString().getBytes(StandardCharsets.UTF_8)));
    } catch (Exception exception) {
      throw new NgpApiException("NGP 签名生成失败", exception);
    }
  }

  private String nextTimestamp() {
    return String.valueOf(lastTimestamp.updateAndGet(previous -> Math.max(System.currentTimeMillis(), previous + 1)));
  }

  private void validate(JsonNode response, String method) {
    if (response == null || response.isNull()) throw new NgpApiException("NGP 接口返回空响应：" + method);
    int code = response.path("code").asInt(-1);
    if (code != 0 && code != 200) {
      throw new NgpApiException("NGP 接口业务失败：" + method + "，code=" + code
          + "，message=" + response.path("message").asText("未知错误"));
    }
  }

  static boolean isExpiredToken(JsonNode response) {
    if (response == null) return false;
    String message = response.path("message").asText(response.path("msg").asText("")).toLowerCase();
    return message.contains("token") && (message.contains("expired") || message.contains("invalid")
        || message.contains("error") || message.contains("失效") || message.contains("过期")
        || message.contains("错误"));
  }

  private boolean transientStatus(int status) {
    return status == 429 || status >= 500;
  }

  private void pause(int attempt) {
    try {
      Thread.sleep(Math.min(1000L, 200L * attempt));
    } catch (InterruptedException exception) {
      Thread.currentThread().interrupt();
      throw new NgpApiException("NGP 接口重试被中断", exception);
    }
  }

  private String rootMessage(Throwable throwable) {
    if (throwable == null) return "未知错误";
    Throwable current = throwable;
    while (current.getCause() != null && current.getCause() != current) current = current.getCause();
    return current.getMessage() == null ? current.getClass().getSimpleName() : current.getMessage();
  }
}