package com.zzsn;

import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.zzsn.configuration.SpringContextUtil;
import com.zzsn.crawler.WeixinSiteThread;
import com.zzsn.entity.SiteMsgTemple;
import com.zzsn.job.KafkaConsumerJob;
import com.zzsn.job.SouGouSearch;
import com.zzsn.util.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.servlet.support.SpringBootServletInitializer;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

//@SpringBootApplication
@Slf4j
@SpringBootApplication(scanBasePackages = "com.zzsn")
public class WeixinCrawlerApplication extends SpringBootServletInitializer  implements CommandLineRunner {
    @Override
    protected SpringApplicationBuilder configure(SpringApplicationBuilder builder) {
        return builder.sources(WeixinCrawlerApplication.class);
    }
    public static void main(String[] args) {
        SpringApplication.run(WeixinCrawlerApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
//            try {
//                loadSiteMsg();
//            }catch (Exception e){
//                loadSiteMsg();
//            } finally {
//            }

    }
    public void loadSiteMsg(){
        try{
            KafkaConsumerJob kafkaConsumerJob= SpringContextUtil.getBean(KafkaConsumerJob.class);
            System.out.println("进入定时获取mq消息");
            //1.创建消费者
            KafkaConsumer<String, String> consumer = kafkaConsumerJob.createConsumer();
            // 消费某个主题的某个分区数据
            ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
            String kafkaConsumerPartition = Constants.KAFKA_CONSUMER_PARTITION;
            String[] partitions = kafkaConsumerPartition.split(",");
            for (int i = 0; i < partitions.length; i++) {
                topicPartitions.add(new TopicPartition(Constants.KAFKA_CONSUMER_TOPIC, Integer.parseInt(partitions[i])));
            }
            consumer.assign(topicPartitions);
            while(true){
                //消费者是一个长期运行的程序，通过持续轮询向Kafka请求数据。在其他线程中调用consumer.wakeup()可以退出循环
                //在0ms内等待Kafka的broker返回数据.超时参数指定poll在多久之后可以返回，不管有没有可用的数据都要返回
                ConsumerRecords<String, String> records = consumer.poll(300);
                if (records != null && records.count() > 0) {
                    for (ConsumerRecord record : records) {
                        try {
                        System.out.println("kafka消息："+record.value().toString());
                        SiteMsgTemple siteMsgTemple = new Gson().fromJson(record.value().toString(), SiteMsgTemple.class);
                        String siteName = siteMsgTemple.getSiteName();
                        int msgTime=0;
                        try {
                            msgTime= SouGouSearch.getMsgTime(siteName);//获取最新的资讯时间差
                        }catch (Exception e){
                            msgTime=0;
                        }
                        if(msgTime>7){//微信的资讯时间超过4小时的不采集
                            log.info("资讯发布时间超过5小时的公众号："+siteName);
                            System.out.println("资讯发布时间超过5小时的公众号："+siteName);
                            try {
                                Thread.sleep(1000*10);
                            }catch (Exception e){

                            }
                            continue;
                        }else{
                            log.info(siteName+"————资讯最新发布时间："+msgTime);
                        }
                        WeixinSiteThread siteThread = new WeixinSiteThread();
                        siteThread.siteMsgTemple = siteMsgTemple;
                        siteThread.crawler();
                        Thread.sleep(1000*30);
                        }catch (Exception e){
                            continue;
                        }
                    }
                }
            }
        }catch (Exception e){
//            e.printStackTrace();
            System.out.println(e.getMessage());
            System.out.println("程序异常+++++");
            try {
                Thread.sleep(30000);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
        }
    }

    public void loadSiteMsgpy(){
        try{
            KafkaConsumerJob kafkaConsumerJob= SpringContextUtil.getBean(KafkaConsumerJob.class);
            System.out.println("进入定时获取mq消息");
            //1.创建消费者
            KafkaConsumer<String, String> consumer = kafkaConsumerJob.createConsumer();
            // 消费某个主题的某个分区数据
            ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
            String kafkaConsumerPartition = Constants.KAFKA_CONSUMER_PARTITION;
            String[] partitions = kafkaConsumerPartition.split(",");
            for (int i = 0; i < partitions.length; i++) {
                topicPartitions.add(new TopicPartition(Constants.KAFKA_CONSUMER_TOPIC, Integer.parseInt(partitions[i])));
            }
            consumer.assign(topicPartitions);
            while(true){
                //消费者是一个长期运行的程序，通过持续轮询向Kafka请求数据。在其他线程中调用consumer.wakeup()可以退出循环
                //在0ms内等待Kafka的broker返回数据.超时参数指定poll在多久之后可以返回，不管有没有可用的数据都要返回
                ConsumerRecords<String, String> records = consumer.poll(300);
                if (records != null && records.count() > 0) {
                    for (ConsumerRecord record : records) {
                        System.out.println("kafka消息："+record.value().toString());
                        SiteMsgTemple siteMsgTemple = new Gson().fromJson(record.value().toString(), SiteMsgTemple.class);
                        String siteName = siteMsgTemple.getSiteName();
                        String url="http://49.4.24.191:8005/sougou_wechart/?query="+siteName;
                        //这个属于发起请求并获取json数据（得引入hutool，即上面引入的第二个依赖）
                        String json= HttpUtil.createGet(url).execute().body();
                        //转化请求的 json 数据
                        JSONObject jsonObject = JSONObject.parseObject(json);
                        //获取 error 返回状态码
                        String latestNewsTime = jsonObject.getString("latestNewsTime");
                        if(!latestNewsTime.contains("None")){
                            SimpleDateFormat simpleFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm");
                            String sd = simpleFormat.format(new Date(Long.parseLong(latestNewsTime)*1000));
                            log.info(siteName+"————资讯最新发布时间："+sd);
                            int hours = SouGouSearch.getBeforeTime(latestNewsTime);
                            if(hours>5){
                                log.info("资讯发布时间超过5小时的公众号："+siteName);
                            System.out.println("资讯发布时间超过5小时的公众号："+siteName);
                                try {
                                    Thread.sleep(1000*30);
                                }catch (Exception e){
                                }
                                continue;
                            }
                        }else{
                            //这个属于发起请求并获取json数据（得引入hutool，即上面引入的第二个依赖）
                             json= HttpUtil.createGet(url).execute().body();
                            //转化请求的 json 数据
                             jsonObject = JSONObject.parseObject(json);
                            //获取 error 返回状态码
                             latestNewsTime = jsonObject.getString("latestNewsTime");
                            if(!latestNewsTime.contains("None")){
                                SimpleDateFormat simpleFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm");
                                String sd = simpleFormat.format(new Date(Long.parseLong(latestNewsTime)*1000));
                                log.info(siteName+"————资讯最新发布时间："+sd);
                                int hours = SouGouSearch.getBeforeTime(latestNewsTime);
                                if(hours>5){
                                    log.info("资讯发布时间超过5小时的公众号："+siteName);
                                    System.out.println("资讯发布时间超过5小时的公众号："+siteName);
                                    try {
                                        Thread.sleep(1000*30);
                                    }catch (Exception e){
                                    }
                                    continue;
                                }
                            }else{
                                log.info(siteName+"————资讯最新发布时间："+latestNewsTime);
                            }
                        }
                        WeixinSiteThread siteThread = new WeixinSiteThread();
                        siteThread.siteMsgTemple = siteMsgTemple;
                        siteThread.crawler();
                        try {
                            Thread.sleep(1000*60);
                        }catch (Exception e){

                        }
                    }
                }
            }
        }catch (Exception e){
            System.out.println(e.getMessage());
            System.out.println("程序异常+++++");
            try {
                Thread.sleep(30000);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
        }
    }

    public  static KafkaConsumer<String, String> createConsumer() {
        Properties properties = new Properties();
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, Constants.KAFKA_CONSUMER_SERVERS);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, Constants.KAFKA_CONSUMER_GROUP_ID);
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        //kafka数据的读取方式
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,Constants.KAFKA_CONSUMER_AUTO_OFFSET_RESET);
//         latest earliest
        //时间间隔设置为1h
        properties.put("max.poll.interval.ms", 60*60*1000);
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
        return new KafkaConsumer<>(properties);
    }

}

