去掉不使用的功能

入库MySQL

dsp/src/main/java/com/jsc/dsp/config/Configuration.java

@@ -16,18 +16,6 @@ import java.util.concurrent.Executors;
 @Component
 public class Configuration {
 
-    @Value("${es.ip}")
-    String esIp;
-
-    @Value("${es.port}")
-    Integer esPort;
-
-    @Value("${es.username}")
-    String esUsername;
-
-    @Value("${es.password}")
-    String esPassword;
-
     @Bean
     public JacksonJsonParser getJacksonParser() {
         return new JacksonJsonParser();
@@ -48,8 +36,4 @@ public class Configuration {
         return Executors.newFixedThreadPool(4);
     }
 
-    @Bean
-    public RestHighLevelClient esClient() {
-        return EsUtils.getElasticsearchClient(esIp, esPort, esUsername, esPassword);
-    }
 }

dsp/src/main/java/com/jsc/dsp/service/FileDlService.java

@@ -6,6 +6,7 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.json.JacksonJsonParser;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.annotation.StreamListener;
@@ -21,6 +22,7 @@ import java.util.concurrent.Executors;
 
 @Component
 @EnableBinding(FileDlBinding.class)
+@ConditionalOnProperty(name = "switch.enable-file-dl-service", havingValue = "true", matchIfMissing = true)
 public class FileDlService extends StreamService {
 
     @Autowired
@@ -78,31 +80,14 @@ public class FileDlService extends StreamService {
             }
             int dlResult = fileUtils.downloadFromUrl(fileURL, protoSavePath);
             if (dlResult == 1) {
-                File transferPath = new File(transferBackupPath);
+                File transferPath = new File(protoSavePath);
                 File[] files = transferPath.listFiles();
                 if (files != null && files.length > 0) {
                     for (File transferFile : files) {
                         if (transferFile.getName().endsWith(".tar.gz")) {
-                            if (transferFile.getName().startsWith("attach")) {
-                                try {
-                                    fileUtils.UnzipTarGzip(transferFile.getAbsolutePath(), nginxPath);
-                                    logger.info("Unzip attachments " + transferFile.getName());
-                                } catch (Exception e) {
-                                    logger.error("Unzip error!");
-                                }
-                            } else {
-                                try {
-                                    fileUtils.UnzipTarGzip(transferFile.getAbsolutePath(), fileUnzipPath);
-                                    logger.info("Unzip " + transferFile.getName());
-                                } catch (Exception e) {
-                                    logger.error("Unzip error!");
-                                }
-                            }
-                        }
                             fileUtils.moveFileToBackupFolder(transferFile, keepBackupFile);
                         }
-//                    Runnable upload2Ceph = () -> fileUtils.uploadToCeph(fileUnzipPath);
+                    }
-//                    pool.execute(upload2Ceph);
                 }
             } else if (dlResult == 0) {
                 logger.error("File " + fileName + " download failure");

dsp/src/main/java/com/jsc/dsp/service/ProtobufService.java

@@ -12,12 +12,14 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.json.JacksonJsonParser;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.annotation.StreamListener;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.Resource;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.OutputStream;
@@ -30,6 +32,7 @@ import java.util.Map;
 
 @Component
 @EnableBinding(ProtobufBinding.class)
+@ConditionalOnProperty(name = "switch.enable-protobuf-service", havingValue = "true", matchIfMissing = true)
 public class ProtobufService extends StreamService {
 
     @Autowired
@@ -41,9 +44,6 @@ public class ProtobufService extends StreamService {
     @Value("${custom.proto_save_path}")
     String protoSavePath;
 
-    @Value("${custom.transfer_backup_path}")
-    String transferBackupPath;
-
     @Value("${custom.keep_backup_file}")
     String keepBackupFile;
 
@@ -55,7 +55,7 @@ public class ProtobufService extends StreamService {
 
     private final Logger logger = LogManager.getLogger(ProtobufService.class.getName());
 
-    @Autowired
+    @Resource
     private ProtobufBinding source;
 
     @Override
@@ -169,7 +169,7 @@ public class ProtobufService extends StreamService {
             }
             logger.debug("protobuf done");
             // 转移备份目录的todist文件
-            File transferPath = new File(transferBackupPath);
+            File transferPath = new File(protoSavePath);
             File[] files = transferPath.listFiles();
             if (files != null && files.length > 0) {
                 for (File transferFile : files) {

dsp/src/main/java/com/jsc/dsp/service/StorageService.java

@@ -20,6 +20,7 @@ import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.json.JacksonJsonParser;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.annotation.StreamListener;
@@ -33,6 +34,7 @@ import java.util.Base64.Decoder;
 
 @Component
 @EnableBinding(StorageBinding.class)
+@ConditionalOnProperty(name = "switch.enable-storage-service", havingValue = "true", matchIfMissing = true)
 public class StorageService extends StreamService {
 
     @Autowired
@@ -44,27 +46,15 @@ public class StorageService extends StreamService {
     @Autowired
     JacksonJsonParser jsonParser;
 
-    @Value("${es.ip}")
-    String esIp;
-
-    @Value("${es.port}")
-    Integer esPort;
-
-    @Value("${es.username}")
-    String esUsername;
-
-    @Value("${es.password}")
-    String esPassword;
-
-    @Value("${es.index}")
-    String esIndex;
-
     @Value("${custom.dev-mode}")
     boolean devMode;
 
     @Value("${custom.local-file-storage-path}")
     String localFileStoragePath;
 
+    @Value("${custom.websiteWhiteList}")
+    String websiteWhiteListString;
+
     @Value("${db.driver}")
     String dbDriver;
 
@@ -80,6 +70,7 @@ public class StorageService extends StreamService {
 
     private final Logger logger = LogManager.getLogger(StorageService.class.getName());
 
+
     @Override
     public void sendMessage(byte[] msg) {
         source.StorageOutput().send(MessageBuilder.withPayload(msg).build());
@@ -91,8 +82,8 @@ public class StorageService extends StreamService {
     @Override
     @StreamListener(StorageBinding.STORAGE_PIPELINE_IN)
     public void receiveMessage(Object payload) {
+        List<String> websiteWhiteList = Arrays.asList(websiteWhiteListString.split(";"));
         String tempString;
-        ObjectMapper objectMapper = new ObjectMapper();
         try {
             tempString = new String(base64.decode(payload.toString()), StandardCharsets.UTF_8);
             Map<String, Object> data = jsonParser.parseMap(tempString);
@@ -101,7 +92,6 @@ public class StorageService extends StreamService {
             if ("public_info_data_".equals(protoName)) {
                 EsSets.Builder esSetsBuilder = EsSets.newBuilder();
                 EsSets esSets = EsSets.parseFrom(data.get("content").toString().getBytes(StandardCharsets.ISO_8859_1));
-                List<Object> localStorageItems = new ArrayList<>();
                 List<Indeximos> dbStorageItems = new ArrayList<>();
                 BulkRequest bulkRequest = new BulkRequest();
                 bulkRequest.timeout("5s");
@@ -138,6 +128,9 @@ public class StorageService extends StreamService {
                             }
                         }
                     }
+                    // 只导出目标站点的数据
+                    if (websiteWhiteList.contains(indeximos.getEs_sitename())) {
+                        logger.info("开始处理站点【" + indeximos.getEs_sitename() + "】的数据入库流程");
                         String uuid = UUID.randomUUID().toString().replaceAll("-", "");
                         String es_urlname = indeximos.getEs_urlname();
                         if (!es_urlname.isEmpty()) {
@@ -145,6 +138,7 @@ public class StorageService extends StreamService {
                             UUID _uuid = UUID.nameUUIDFromBytes(es_urlname.getBytes());
                             uuid = _uuid.toString().replaceAll("-", "");
                         }
+                        indeximos.setEs_urltitle(indeximos.getEs_urltitle().trim());
                         indeximos.setEs_sid(uuid);
                         indeximos.setEs_links(indeximos.getEs_links());
                         indeximos.setEs_loadtime(StringUtils.TimestampToStringDate(System.currentTimeMillis()));
@@ -163,36 +157,8 @@ public class StorageService extends StreamService {
                                 }
                             }
                         }
-                    IndexRequest indexRequest = new IndexRequest(esIndex);
-                    indexRequest.id(indeximos.getEs_sid());
-                    indexRequest.source(objectMapper.writeValueAsString(indeximos), XContentType.JSON);
-                    bulkRequest.add(indexRequest);
-                    Es es_temp = builder.build();
-                    esSetsBuilder.addEs(es_temp);
-                    List<String> localizedOption = JSON.parseArray(indeximos.getEs_urltopic(), String.class);
-                    if (indeximos.getEs_carriertype().equals("wechat")) {
                         dbStorageItems.add(indeximos);
                     }
-                    if (localizedOption != null && localizedOption.size() > 0) {
-                        //本地存储用
-                        if (localizedOption.contains("json")) {
-                            localStorageItems.add(indeximos);
-                        }
-                        //入库MySQL
-                        if (localizedOption.contains("mysql")) {
-                            dbStorageItems.add(indeximos);
-                        }
-                    }
-                }
-                EsUtils.EsSaveBulkRequest(esIp, esPort, esUsername, esPassword, bulkRequest);
-                if (localStorageItems.size() > 0) {
-                    String entityItemsString = JSON.toJSONString(localStorageItems);
-                    String entityFileFullPath = localFileStoragePath + esIndex + "_" + System.currentTimeMillis() + ".json";
-                    if (FileUtils.saveStringToFile(entityItemsString, entityFileFullPath)) {
-                        logger.info("Local file store to " + entityFileFullPath);
-                    } else {
-                        logger.error("Local file store error!");
-                    }
                 }
                 if (dbStorageItems.size() > 0) {
                     if (DBUtils.insertIntoDB(dbDriver, dbUrl, dbUser, dbPassword, dbStorageItems)) {

dsp/src/main/java/com/jsc/dsp/utils/AutoPatroller.java

@@ -37,99 +37,19 @@ public class AutoPatroller {
     @Value("${custom.socialUpdateAPI}")
     String socialUpdateAPI;
 
-    @Value("${es.ip}")
+//    @Scheduled(cron = "0 45 0/3 * * *")
-    String esIp;
+//    public void checkNewsSite() {
-
+//        checkWebsite("es_sitename", "es_carriertype", "news");
-    @Value("${es.port}")
+//    }
-    Integer esPort;
+//
-
+//    @Scheduled(cron = "0 15 1/3 * * *")
-    @Value("${es.username}")
+//    public void checkWechat() {
-    String esUsername;
+//        checkSocial("es_authors", "es_carriertype", "wechat", "5");
-
+//    }
-    @Value("${es.password}")
+//
-    String esPassword;
+//    @Scheduled(cron = "0 0 2/4 * * *")
-
+//    public void checkArticleSite() {
-    @Scheduled(cron = "0 45 0/3 * * *")
+//        checkWebsite("es_sitename", "es_carriertype", "article");
-    public void checkNewsSite() {
+//    }
-        checkWebsite("es_sitename", "es_carriertype", "news");
-    }
-
-    @Scheduled(cron = "0 15 1/3 * * *")
-    public void checkWechat() {
-        checkSocial("es_authors", "es_carriertype", "wechat", "5");
-    }
-
-    @Scheduled(cron = "0 0 2/4 * * *")
-    public void checkArticleSite() {
-        checkWebsite("es_sitename", "es_carriertype", "article");
-    }
-
-    public void checkWebsite(String aggFieldName, String queryFieldName, String queryFieldValue) {
-        try {
-            Map<String, SearchAggregation> searchAggregationMap = performAggregationSearch(
-                    esIp, esPort, esUsername, esPassword, aggFieldName, queryFieldName, queryFieldValue);
-            JSONObject dataObject = new JSONObject();
-            dataObject.put("carrierType", queryFieldValue);
-            String rsp = HttpUtils.post(websiteQueryAPI, dataObject);
-            JSONObject rspObj = JSON.parseObject(rsp);
-            if (rspObj.getIntValue("code") == 200) {
-                JSONArray rspArr = rspObj.getJSONArray("content");
-                for (Object obj : rspArr) {
-                    TargetWebsite targetWebsite = JSONObject.parseObject(obj.toString(), TargetWebsite.class);
-                    String siteName = targetWebsite.getSiteName();
-                    if (searchAggregationMap.containsKey(siteName)) {
-                        SearchAggregation checkInfo = searchAggregationMap.get(siteName);
-                        targetWebsite.setCheckTotalNum(checkInfo.getCount());
-                        targetWebsite.setCheckLastTime(checkInfo.getLastTime());
-                        targetWebsite.setCheckUpdateTime(new Date());
-                        String updateRsp = HttpUtils.post(websiteUpdateAPI, targetWebsite);
-                        JSONObject updateRspObj = JSONObject.parseObject(updateRsp);
-                        if (updateRspObj.getIntValue("code") != 200) {
-                            logger.warning("更新站点【" + siteName + "】巡检信息失败");
-                        }
-                        Thread.sleep(updateInterval);
-                    }
-                }
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
-        logger.info("站点巡检完毕");
-    }
-
-    public void checkSocial(String aggFieldName, String queryFieldName, String queryFieldValue, String socialTypeCode) {
-        try {
-            Map<String, SearchAggregation> searchAggregationMap = performAggregationSearch(
-                    esIp, esPort, esUsername, esPassword, aggFieldName, queryFieldName, queryFieldValue);
-            TargetSocial postData = new TargetSocial();
-            postData.setUserFlag("0");
-            postData.setUserType(socialTypeCode);
-            String rsp = HttpUtils.post(socialQueryAPI, postData);
-            JSONObject rspObj = JSON.parseObject(rsp);
-            if (rspObj.getIntValue("code") == 200) {
-                JSONArray rspArr = rspObj.getJSONArray("content");
-                for (Object obj : rspArr) {
-                    TargetSocial targetSocial = JSONObject.parseObject(obj.toString(), TargetSocial.class);
-                    String userName = targetSocial.getUserName();
-                    if (searchAggregationMap.containsKey(userName)) {
-                        SearchAggregation checkInfo = searchAggregationMap.get(userName);
-                        targetSocial.setCheckTotalNum(checkInfo.getCount());
-                        targetSocial.setCheckLastTime(checkInfo.getLastTime());
-                        targetSocial.setCheckUpdateTime(new Date());
-                        String updateRsp = HttpUtils.post(socialUpdateAPI, targetSocial);
-                        JSONObject updateRspObj = JSONObject.parseObject(updateRsp);
-                        if (updateRspObj.getIntValue("code") != 200) {
-                            logger.warning("更新账号【" + userName + "】巡检信息失败");
-                        }
-                        Thread.sleep(updateInterval);
-                    }
-                }
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
-        logger.info("社交帐号巡检完毕");
-
-    }
 
 }

dsp/src/main/resources/application.yml

@@ -6,13 +6,12 @@ spring:
       kafka:
         binder:
           brokers: 47.113.231.200:9092
-          zkNodes: 47.113.231.200:2181
           auto-create-topics: true
           healthTimeout: 600
       bindings:
         file_dl_pipeline_in:
           destination: stream-file-dl
-          group: file-dl
+          group: file-dl-test
           consumer:
             pollTimeout: 60
         file_dl_pipeline_out:
@@ -20,7 +19,7 @@ spring:
           content-type: text/plain
         protobuf_pipeline_in:
           destination: stream-protobuf
-          group: protobuf
+          group: protobuf-test
           consumer:
             pollTimeout: 60
         protobuf_pipeline_out:
@@ -28,7 +27,7 @@ spring:
           content-type: text/plain
         storage_pipeline_in:
           destination: stream-db
-          group: db
+          group: db-test
           consumer:
             pollTimeout: 60
         storage_pipeline_out:
@@ -48,33 +47,28 @@ topics:
   stream-db: com.jsc.dsp.service.StorageService
   stream-file-dl: com.jsc.dsp.service.FileDlService
 
-es:
+switch:
-  ip: 8.130.95.27
+  enable-storage-service: true
-  port: 28087
+  enable-file-dl-service: false
-  username: elastic
+  enable-protobuf-service: false
-  password: passok123A
+
-  index: indeximos
-  type: default
-ceph:
-  aws-access-key: JH8OF0D9ZJYYXBFYB5OD
-  aws-secret-key: FuptELjiPQOQNR6tPOVL777n3dGe3bZCDJphyiz0
-  endpoint: 192.168.1.16:28090
 db:
   driver: com.mysql.cj.jdbc.Driver
-  url: jdbc:mysql://8.130.95.27:28089/dsp
+  url: jdbc:mysql://47.113.231.200:28089/dsp
   user: root
   password: passok123A
 custom:
   dev-mode: false
   filter-words-query-api: http://47.115.228.133:28081/api/open/wordBank/queryAll
   filter-words-update-interval-ms: 3600000
-  local-file-storage-path: E:/data/local-storage/
+  local-file-storage-path: D:/data/local-storage/
   proto_save_path: D:/data/spider_data/proto/
-  transfer_backup_path: E:/data/transfer_backup/
+  transfer_backup_path: D:/data/transfer_backup/
-  file_unzip_path: E:/html-full/
+  file_unzip_path: D:/html-full/
-  keep_backup_file: E:/data/dbzq_backup/
+  keep_backup_file: D:/data/dbzq_backup/
-  nginx_path: E:/OSC-3.0/app/osdp_board/html/
+  nginx_path: D:/OSC-3.0/app/osdp_board/html/
   websiteQueryAPI: http://47.115.228.133:28081/api/open/target/website/queryAllInfo
   websiteUpdateAPI: http://47.115.228.133:28081/api/open/target/website/update
   socialQueryAPI: http://47.115.228.133:28081/api/open/target/social/queryAll?sortBy=id&shuffleResult=false
   socialUpdateAPI: http://47.115.228.133:28081/api/open/target/social/update
+  websiteWhiteList: 能源界（国内信息）;能源界（国际信息）;中国能源新闻网;新华能源网;中国能源网（能源战略）;中国农网（三农要闻）;中国经济网（三农经济）;中华粮网（粮食安全）;美国之音（中国版面）;美国之音（中美关系）;美国之音（台海两岸版面）;美国之音（港澳版面）;看中国（看大陆版面）;看中国（重点新闻）;德国之声（中国报道）;纽约时报中文网（中国版面）;大纪元（一周大陆新闻）;EnergyNow;联合国粮农组织;路透社（中国版面）

dsp/src/main/resources/logback-spring.xml

@@ -8,7 +8,7 @@
     <contextName>logback</contextName>
  
     <!-- name的值是变量的名称，value的值时变量定义的值。通过定义的值会被插入到logger上下文中。定义后，可以使“${}”来使用变量。 -->
-    <property name="log.path" value="E:/dsp-logs" />
+    <property name="log.path" value="D:/dsp-logs" />
  
     <!--0. 日志格式和颜色渲染 -->
     <!-- 彩色日志依赖的渲染类 -->
@@ -27,7 +27,7 @@
         <encoder>
             <Pattern>${CONSOLE_LOG_PATTERN}</Pattern>
             <!-- 设置字符集 -->
-            <charset>GBK</charset>
+            <charset>UTF-8</charset>
         </encoder>
     </appender>