hzya
/
nifi-hzyadev-bundle
2
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: hzya:d34165962c5d5fbfe00f74067576b7fad8a67f38
Branches Tags
hzya:master
hzya:zanhuo
..
compare: hzya:e512bddad2c5ce4ad2f206a0f33b50261ac80311
Branches Tags
hzya:master
hzya:zanhuo

No commits in common. "d34165962c5d5fbfe00f74067576b7fad8a67f38" and "e512bddad2c5ce4ad2f206a0f33b50261ac80311" have entirely different histories.
d34165962c ... e512bddad2

1 changed files with 7 additions and 28 deletions
Show Stats Expand all files Collapse all files

35
nifi-hzyadev-bundle/hzya-nifi-GetJackyunOpenData-processors/src/main/java/com/hzya/frame/GetJackyunOpenDataProcessor.java
View File

@ -1,7 +1,6 @@
package com.hzya.frame;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.hzya.frame.dto.ResultDataDto;
import com.hzya.frame.dto.ResultDto;
@ -244,17 +243,8 @@ public class GetJackyunOpenDataProcessor extends AbstractProcessor {
}
ResultDto result = resultDataDto.getResult();
//2025年9月12日11:39:07 补充判断逻辑
//吉客云返回接口的类型，可能出现两种情况：
//{"code":200,"msg":"","result":{"data":{}}}
//{"code":200,"msg":"","result":{"data":{"warehouseInfo":[]}}}
if (result != null && result.getData() != null && result.getData().size() > 0 && result.getData().get(0) instanceof JSONObject) {
JSONObject jsonObject = (JSONObject) result.getData().get(0);
if (!checkIsNull(jsonObject)) {
hasNextPageInWindow = false;
}
}
if (hasNextPageInWindow && result != null && result.getData() != null && result.getData().size() > 0 && !result.getData().isEmpty()) {
if (result != null && result.getData() != null && !result.getData().isEmpty()) {
hasFetchedDataInThisRun = true;
getLogger().info("任务 '{}' 在时间窗 [{} -> {}] 的第 {} 页获取到 {} 条数据。", stateKey, windowStartTimeStr, windowEndTimeStr, currentPage, result.getData().size());
FlowFile ff = session.create();
@ -278,7 +268,11 @@ public class GetJackyunOpenDataProcessor extends AbstractProcessor {
// 成功处理完一个时间窗口（包括所有页），立即更新MongoDB中的状态
String newWatermarkStr = sdf.format(new Date(newWatermarkMillis));
stateCollection.updateOne(Filters.eq("_id", stateKey), Updates.set("value", newWatermarkStr), new UpdateOptions().upsert(true));
stateCollection.updateOne(
Filters.eq("_id", stateKey),
Updates.set("value", newWatermarkStr),
new UpdateOptions().upsert(true)
);
getLogger().info("任务 '{}' 的时间窗口 [{} -> {}] 数据同步完成，状态水位线更新至: {}", stateKey, windowStartTimeStr, windowEndTimeStr, newWatermarkStr);
// 更新循环变量，为下一个时间窗口做准备
@ -359,19 +353,4 @@ public class GetJackyunOpenDataProcessor extends AbstractProcessor {
}
return hexValue.toString();
}
private boolean checkIsNull(JSONObject jsonObject) {
if (jsonObject != null) {
for (Map.Entry<String, Object> entry : jsonObject.entrySet()) {
String key = entry.getKey();
Object value = entry.getValue();
if (value instanceof JSONArray) {
if (((JSONArray) value).size() > 0) {
return true;
}
}
}
}
return false;
}
}