开发接口自定义入湖接口推送

src/main/java/com/skyversation/poiaddr/addquery/AddressQueryEngine.java

@@ -294,6 +294,14 @@ public class AddressQueryEngine {
         }
     }
 
+    public Object putDataToSJ_Big_Data_kafka_server(JSONObject params) {
+        ResponseEntity response = AddressNetTools.getInstance().requestGetOrPost(HttpMethod.POST, Constant.INPUT_SJ_BIGDATA_SERVER_URL, params, null, 0);
+        if (response != null && response.hasBody()) {
+            return response.getBody() + "";
+        }
+        return null;
+    }
+
     public Object putDataToSJ_Big_Data(JSONObject datas) {
         JSONObject params = new JSONObject();
         params.put("catalogName", "松江区规范化地名地址库");

src/main/java/com/skyversation/poiaddr/addquery/Constant.java

@@ -32,6 +32,8 @@ public class Constant {
 
     public static String SJ_BIGDATA_KAFKA_URL = "http://116.246.2.84:2222/data/kafka-api";
 
+    public static String INPUT_SJ_BIGDATA_SERVER_URL = "http://121.43.55.7:10015/poiApi/corporate_library/saveDatasToSJBigDataKafka";
+
     public static void setTowns(String[] town) {
         getInstance().towns = town;
     }

src/main/java/com/skyversation/poiaddr/controller/CorporateLibraryController.java

@@ -165,6 +165,32 @@ public class CorporateLibraryController {
         return "处理完成！用时" + (endTime - startTime) / 1000 + "秒！" + ScheduledTasks.OID_TAG;
     }
 
+    /**
+     * 121服务器开放接口，入湖
+     * @param Datas TyyszAddress
+     * @return
+     */
+    @PostMapping(value = "/saveDatasToSJBigDataKafka")
+    public Object saveDatasToSJBigDataKafka(@RequestBody JSONObject Datas){
+        return yyskAddressStandardizationService.saveDatasToSJBigDataKafka(Datas);
+    }
+
+    /**
+     * 本地调用121入湖接口
+     * @param startOidIndex 按照批次来讲的话，现在是第二批数据也就是2kw，同时也是固定值
+     * @param page 从0开始
+     * @param pageSize 一次5000条
+     * @return
+     */
+    @PostMapping(value = "/tyaToServer")
+    public Object tyaToServer(@RequestParam(name = "startOidIndex") Integer startOidIndex,@RequestParam(name = "page") Integer page,@RequestParam(name = "pageSize") Integer pageSize){
+        long startTime = System.currentTimeMillis();
+        startOidIndex = startOidIndex + (page * pageSize);
+        yyskAddressStandardizationService.tyaToServer(startOidIndex,page,pageSize);
+        long endTime = System.currentTimeMillis();
+        return "处理完成！用时" + (endTime - startTime) / 1000 + "秒！";
+    }
+
     /**
      * fileRootPath：单个文件目录
      * pageSize: 批次请求数据条数

src/main/java/com/skyversation/poiaddr/entity/TYyszAddress.java

@@ -0,0 +1,230 @@
+package com.skyversation.poiaddr.entity;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Table;
+import javax.persistence.Id;
+import java.io.Serializable;
+import java.util.Date;
+import java.math.BigDecimal;
+
+/**
+ * @Description  
+ * @Author  LiuMengxiang
+ * @Date 2025-05-30 15:10:42 
+ */
+
+@AllArgsConstructor
+@NoArgsConstructor
+@Data
+@Entity
+@Table ( name ="t_yysz_address" , schema = "")
+public class TYyszAddress  implements Serializable {
+
+	private static final long serialVersionUID =  3773205680338123194L;
+
+	@Id
+   	@Column(name = "oid" )
+	private Long oid;
+
+   	@Column(name = "code" )
+	private String code;
+
+   	@Column(name = "sourceaddress" )
+	private String sourceaddress;
+
+   	@Column(name = "city" )
+	private String city;
+
+   	@Column(name = "county" )
+	private String county;
+
+   	@Column(name = "town" )
+	private String town;
+
+   	@Column(name = "community" )
+	private String community;
+
+   	@Column(name = "village" )
+	private String village;
+
+   	@Column(name = "squad" )
+	private String squad;
+
+   	@Column(name = "szone" )
+	private String szone;
+
+   	@Column(name = "street" )
+	private String street;
+
+   	@Column(name = "door" )
+	private String door;
+
+   	@Column(name = "resregion" )
+	private String resregion;
+
+   	@Column(name = "building" )
+	private String building;
+
+   	@Column(name = "building_num" )
+	private String buildingNum;
+
+   	@Column(name = "unit" )
+	private String unit;
+
+   	@Column(name = "floor" )
+	private String floor;
+
+   	@Column(name = "room" )
+	private String room;
+
+   	@Column(name = "unique_code" )
+	private String uniqueCode;
+
+   	@Column(name = "room_of_floor" )
+	private Long roomOfFloor;
+
+   	@Column(name = "source" )
+	private String source;
+
+   	@Column(name = "lon" )
+	private BigDecimal lon;
+
+   	@Column(name = "lat" )
+	private BigDecimal lat;
+
+   	@Column(name = "shape" )
+	private String shape;
+
+   	@Column(name = "belong_building" )
+	private String belongBuilding;
+
+   	@Column(name = "address_type" )
+	private Long addressType;
+
+   	@Column(name = "lv" )
+	private Long lv;
+
+   	@Column(name = "is_delete" )
+	private Long isDelete;
+
+   	@Column(name = "is_multi" )
+	private Long isMulti;
+
+   	@Column(name = "createtime" )
+	private Date createtime;
+
+   	@Column(name = "updatetime" )
+	private Date updatetime;
+
+   	@Column(name = "alt" )
+	private BigDecimal alt;
+
+   	@Column(name = "model_type" )
+	private Long modelType;
+
+   	@Column(name = "full_place" )
+	private String fullPlace;
+
+   	@Column(name = "security_grade" )
+	private Long securityGrade;
+
+   	@Column(name = "city_code" )
+	private String cityCode;
+
+   	@Column(name = "county_code" )
+	private String countyCode;
+
+   	@Column(name = "town_code" )
+	private String townCode;
+
+   	@Column(name = "community_code" )
+	private String communityCode;
+
+   	@Column(name = "grid_name" )
+	private String gridName;
+
+   	@Column(name = "grid_code" )
+	private String gridCode;
+
+   	@Column(name = "data_type" )
+	private String dataType;
+
+   	@Column(name = "mphid" )
+	private String mphid;
+
+   	@Column(name = "address_code" )
+	private String addressCode;
+
+   	@Column(name = "systemid" )
+	private String systemid;
+
+   	@Column(name = "type" )
+	private String type;
+
+   	@Column(name = "x" )
+	private String X;
+
+   	@Column(name = "y" )
+	private String Y;
+
+   	@Column(name = "label" )
+	private String label;
+
+   	@Column(name = "addr_tag" )
+	private String addrTag;
+
+   	@Column(name = "multiple_unique" )
+	private String multipleUnique;
+
+   	@Column(name = "standardize_task_id" )
+	private String standardizeTaskId;
+
+   	@Column(name = "dlhh" )
+	private String dlhh;
+
+   	@Column(name = "signature_data" )
+	private String signatureData;
+
+   	@Column(name = "signature_check_state" )
+	private Long signatureCheckState;
+
+   	@Column(name = "signature_time" )
+	private Date signatureTime;
+
+   	@Column(name = "signature_check_time" )
+	private Date signatureCheckTime;
+
+   	@Column(name = "is_history" )
+	private Long isHistory;
+
+   	@Column(name = "ylmc" )
+	private String ylmc;
+
+   	@Column(name = "ylbm" )
+	private String ylbm;
+
+   	@Column(name = "address" )
+	private String address;
+
+   	@Column(name = "ybdd" )
+	private String ybdd;
+
+   	@Column(name = "location" )
+	private String location;
+
+   	@Column(name = "tydz_jc" )
+	private String tydzJc;
+
+   	@Column(name = "match_level" )
+	private String matchLevel;
+
+   	@Column(name = "no1" )
+	private String no1;
+
+   	@Column(name = "no2" )
+	private String no2;
+}

src/main/java/com/skyversation/poiaddr/service/AreaService.java

@@ -40,6 +40,9 @@ public class AreaService {
     @Autowired
     private AmapAddressV3Service amapAddressV3Service;
 
+    @Autowired
+    private TyyszAddressV3Service tyyszAddressV3Service;
+
     @Autowired
     private YyskAddressStandardizationSerice yyskAddressStandardizationSerice;
 
@@ -129,6 +132,10 @@ public class AreaService {
         return amapAddressV3Service.getDataByPages(page,pageSize);
     }
 
+    public Page<TYyszAddress> getTyaDataByPage(Integer page, Integer pageSize){
+        return tyyszAddressV3Service.getDataByPages(page,pageSize);
+    }
+
     public List<YyszAddressQp> saveYyszAddressQp(List<YyszAddressQp> amapAddressV3List){
         return yyszAddressQpService.saveYyszAddressQp(amapAddressV3List);
     }

src/main/java/com/skyversation/poiaddr/service/TyyszAddressRepository.java

@@ -0,0 +1,10 @@
+package com.skyversation.poiaddr.service;
+
+import com.skyversation.poiaddr.entity.TYyszAddress;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+import javax.annotation.Resource;
+
+@Resource
+public interface TyyszAddressRepository extends JpaRepository<TYyszAddress, String> {
+}

src/main/java/com/skyversation/poiaddr/service/impl/TyyszAddressV3Service.java

@@ -0,0 +1,23 @@
+package com.skyversation.poiaddr.service.impl;
+
+import com.skyversation.poiaddr.entity.TYyszAddress;
+import com.skyversation.poiaddr.service.TyyszAddressRepository;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
+import org.springframework.stereotype.Service;
+
+
+@Service
+public class TyyszAddressV3Service {
+
+    @Autowired
+    private TyyszAddressRepository tyyszAddressRepository;
+
+    public Page<TYyszAddress> getDataByPages(Integer page, Integer pageSize){
+        Pageable pageable = PageRequest.of(page,pageSize);
+        return tyyszAddressRepository.findAll(pageable);
+    }
+
+}

src/main/java/com/skyversation/poiaddr/service/impl/YyskAddressStandardizationServiceImpl.java

@@ -241,15 +241,6 @@ public class YyskAddressStandardizationServiceImpl {
         }
     }
 
-    /**
-     * 需要迭代获取数据库中的数据
-     * 然后首先判断是否存在街镇数据，存在的话根据经纬度落点判断点位是否在这个镇里面
-     * 不存在街镇数据的话直接重新请求
-     * 最后把新的数据更新到数据库里面
-     *
-     * @param page     当前页码
-     * @param pageSize 每页数量
-     */
     public void dataCleaning(Integer page, Integer pageSize) {
         System.out.println("page：" + page + "pageSize:" + pageSize);
         long startTime = System.currentTimeMillis();
@@ -319,7 +310,6 @@ public class YyskAddressStandardizationServiceImpl {
         }
     }
 
-
     /**
      * 需要迭代获取数据库中的数据
      * 然后首先判断是否存在街镇数据，存在的话根据经纬度落点判断点位是否在这个镇里面
@@ -563,6 +553,70 @@ public class YyskAddressStandardizationServiceImpl {
         return now.format(formatter);
     }
 
+    public Object saveDatasToSJBigDataKafka(JSONObject dataList) {
+        long startTime = System.currentTimeMillis();
+        if (dataList != null && dataList.size() > 0) {
+            try {
+                return AddressQueryEngine.getInstance().putDataToSJ_Big_Data(dataList);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+            long endTime = System.currentTimeMillis();
+            dataList.clear();
+            System.out.println("处理单批次用时" + (endTime - startTime) / 1000 + "秒！");
+        } else {
+            System.out.println("该文件无有效数据！");
+        }
+        return null;
+    }
+
+
+    public void tyaToServer(Integer startOidIndex, Integer page, Integer pageSize) {
+        System.out.println("page：" + page + "pageSize:" + pageSize);
+        long startTime = System.currentTimeMillis();
+        Page<TYyszAddress> sqlDatas = AreaService.getInstance().getTyaDataByPage(page, pageSize);
+        List<TYyszAddress> tYyszAddressList = sqlDatas.getContent();
+        JSONObject dataList = new JSONObject();
+//          操作类型 1:新增,2:更新3:删除
+        dataList.put("optFlag", 1);
+        JSONArray datas = new JSONArray();
+        for (TYyszAddress tYyszAddress : tYyszAddressList) {
+            tYyszAddress.setOid(Long.valueOf(startOidIndex));
+            startOidIndex++;
+            JSONObject dataItem = new JSONObject();
+            dataItem.put("oid", Long.valueOf(startOidIndex));
+            startOidIndex++;
+            dataItem.put("address", initAddrMap.outAddrMapInAddr(tYyszAddress.getSourceaddress()).getAddress());
+            dataItem.put("sourceaddress", tYyszAddress.getSourceaddress());
+            dataItem.put("city", tYyszAddress.getCity());
+            dataItem.put("county", tYyszAddress.getCounty());
+            dataItem.put("town", tYyszAddress.getTown());
+            dataItem.put("community", tYyszAddress.getCommunity());
+            dataItem.put("lon", tYyszAddress.getX());
+            dataItem.put("lat", tYyszAddress.getY());
+            dataItem.put("address_type", tYyszAddress.getAddressType());
+            dataItem.put("createtime", getCurrentDateTime());
+            dataItem.put("updatetime", getCurrentDateTime());
+            dataItem.put("match_level", tYyszAddress.getMatchLevel());
+            datas.add(dataItem);
+        }
+        dataList.put("data", datas);
+        tYyszAddressList.clear();
+//      TODO 发送远程请求
+        try {
+            System.out.println(AddressQueryEngine.getInstance().putDataToSJ_Big_Data_kafka_server(dataList));
+        } catch (Exception e) {
+            e.printStackTrace();
+            System.err.println(page + "页数据入湖异常" + e);
+        }
+        long endTime = System.currentTimeMillis();
+        System.out.println("单批数据处理完成,用时：" + (endTime - startTime) / 1000 + "秒！");
+        if (sqlDatas.getTotalPages() > page) {
+            page++;
+            tyaToServer(startOidIndex, page, pageSize);
+        }
+    }
+
     public void readFileToPg(String fileRootPath, Integer startFileIndex, Integer endFileIndex, Integer sort, Integer pageSize) {
         System.out.println("<<<<<<<<------readFileToPg{fileIndex:" + startFileIndex + "}");
         long startTime = System.currentTimeMillis();