promotion project增加字段重试

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/entity/BytedanceCdpProject.java

@@ -169,6 +169,7 @@ public class BytedanceCdpProject {
     private String status;
     private String statusFirst;
     private String statusSecond;
+    private String projectList;
 
 
     /**
@@ -186,6 +187,7 @@ public class BytedanceCdpProject {
     }
 
     public BytedanceCdpProject(JSONObject o) {
+        this.projectList = o.toString();
         this.advertiserId = o.getLong("advertiser_id");
         this.projectId = o.getLong("project_id");
         this.projectName = o.getString("name");

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris/sink/SimpleHttpDorisSinkClient.java

@@ -1,10 +1,12 @@
 package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import cn.com.ctop.job.bytedance.data.utils.JsonUtils;
+import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.http.HttpEntity;
 import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpPut;
 import org.apache.http.entity.StringEntity;
 import org.apache.http.impl.client.CloseableHttpClient;
@@ -75,6 +77,7 @@ public class SimpleHttpDorisSinkClient implements DorisSinkClient {
 
         put.setEntity(new StringEntity(jsonBody, StandardCharsets.UTF_8));
 
+
         try (CloseableHttpClient httpClient = HttpClients.custom()
                 .setRedirectStrategy(new StrictRedirectStrategy())  // 支持 307/301 重定向
                 .build();
@@ -82,14 +85,37 @@ public class SimpleHttpDorisSinkClient implements DorisSinkClient {
             int code = response.getStatusLine().getStatusCode();
             HttpEntity entity = response.getEntity();
             String responseBody = EntityUtils.toString(entity, "UTF-8");
+            String fsWebhookUrl = "https://open.feishu.cn/open-apis/bot/v2/hook/371e7945-7258-4574-a2eb-d7cd0490f86c";
             if (code != 200 && code != 201) {
+                sendFsMessage(fsWebhookUrl, "stream load 异常 " + label, "Doris stream load failed. Status: " + code + ", response: " + responseBody);
                 throw new RuntimeException("Doris stream load failed. Status: " + code + ", response: " + responseBody);
             }
             // 可选: 解析返回内容判断 success
-            if (!responseBody.contains("\"Status\": \"Success\"")) {
+            if (!responseBody.contains("\"Status\": \"Success\"") || responseBody.contains("ErrorURL")) {
+                sendFsMessage(fsWebhookUrl, "stream load 异常 " + label, responseBody);
                 throw new RuntimeException("Stream load returned non-success: " + responseBody);
             }
             log.info("Doris stream load success: {}", responseBody);
         }
     }
+
+    //发送飞书消息
+    private static void sendFsMessage(String url, String head, String message) {
+        JSONObject body = new JSONObject();
+        body.put("msg_type", "interactive");
+        JSONArray elements = new JSONArray();
+        elements.add(new JSONObject()
+                .fluentPut("tag", "div")
+                .fluentPut("text", new JSONObject().fluentPut("content", message).fluentPut("tag", "lark_md")));
+        JSONObject header = new JSONObject().fluentPut("title", new JSONObject().fluentPut("content", head).fluentPut("tag", "plain_text"));
+        body.put("card", new JSONObject().fluentPut("elements", elements).fluentPut("header", header));
+        HttpPost httpPost = new HttpPost(url);
+        httpPost.setEntity(new StringEntity(body.toString(), StandardCharsets.UTF_8));
+        try (CloseableHttpClient httpClient = HttpClients.custom().build();
+             CloseableHttpResponse response = httpClient.execute(httpPost)) {
+            response.getStatusLine().getStatusCode();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
 }