package com.zhiwei.manage.handle;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.stereotype.Component;

import com.zhiwei.manage.bean.Channel;
import com.zhiwei.manage.bean.Data;
import com.zhiwei.manage.bean.FieldIntegerity;
import com.zhiwei.manage.bean.Message;
import com.zhiwei.manage.bean.Messages;
import com.zhiwei.manage.bean.NewsDelayed;
import com.zhiwei.manage.bean.PingUrl;
import com.zhiwei.manage.bean.ServerBean;
import com.zhiwei.manage.bean.Template;
import com.zhiwei.manage.orm.BeanFactory;
import com.zhiwei.manage.service.CrawTemplateServiceImpl;
import com.zhiwei.manage.service.DataServiceImpl;
import com.zhiwei.manage.service.MessageServiceImpl;
import com.zhiwei.manage.service.ServerServiceImpl;
import com.zhiwei.manage.service.TemplateServiceImpl;
import com.zhiwei.manage.util.Config;
import com.zhiwei.manage.util.ESUtil;
import com.zhiwei.manage.util.SendMailUtil;
import com.zhiwei.manage.util.TimeUtil;

/**
 * 主要执行线程
 * 
 * @author admin
 *
 */
@Component
public class MainThread extends Thread {

	private static final Log log = LogFactory.getLog(MainThread.class);
	public static Map<String, Template> mainMap = new ConcurrentHashMap<String, Template>();
	public static Map<String, Template> allTmp = new ConcurrentHashMap<String, Template>();
	public static Map<String, String> map = new HashMap<>();
	static ESUtil es = new ESUtil();
	static Map<String, Integer> counts = new HashMap<>();
	static {
		counts.put("网媒", 1000);
		counts.put("微博", 2000);
		counts.put("微信", 3500);
		counts.put("知乎", 300);
		counts.put("平媒", 10);
		counts.put("今日头条", 600);
		counts.put("贴吧", 50);
		counts.put("论坛", 30);
	}

	public MainThread() {
		this.start();
	}

	/**
	 * 加入执行map
	 * 
	 * @param template
	 */
	public static boolean pushMap(String templateId) {
		Template template = allTmp.get(templateId);
		if (mainMap.get(templateId) != null) {
			return false;
		} else {
			mainMap.put(template.getTemplateId(), template);
			if (template.isNeedDelayed()) {
				NewsDelayed nd = new NewsDelayed();
				nd.setCollection(template.getCollection());
				nd.setNewsDelayed(template.getNewsDelayed());
				nd.setDelayedInterval(template.getDelayedInterval());
				nd.setdBName(template.getdBName());
				nd.setHost(template.getHost());
				nd.setPort(template.getPort());
				nd.setStartNow(false);
				nd.setPt(template.getPt());
				nd.setSavetime(template.getSavetime());
				nd.setTime(template.getTime());
				nd.setSpyderInfoId(template.getSpyderInfoId());
				nd.setTempName(template.getTempName());
				nd.setNextRun(new Date());
				nd.setTestCount(template.getTestCount());
				nd.setTemplateId(templateId);
				DelayedThread.runMap.put(template.getTemplateId(), nd);
			}
			if (template.isNeedPing()) {
				log.info("ping测试添加");
				PingUrl ping = new PingUrl();
				ping.setNextRun(new Date());
				ping.setPingInterval(template.getPingInterval());
				ping.setPingUrlOne(template.getPingUrlOne());
				ping.setPingUrlTwo(template.getPingUrlTwo());
				ping.setPt(template.getPt());
				ping.setTemplateId(templateId);
				ping.setTempName(template.getTempName());
				ping.setStartNow(false);
				PingThread.runMap.put(templateId, ping);
			}
			if (template.isNeedField()) {
				FieldIntegerity fb = new FieldIntegerity();
				fb.setTestCount(template.getTestCount());
				fb.setCollection(template.getCollection());
				fb.setTemplateId(templateId);
				fb.setdBName(template.getdBName());
				fb.setHost(template.getHost());
				fb.setPort(template.getPort());
				fb.setNextRun(new Date());
				fb.setStartNow(false);
				fb.setPt(template.getPt());
				fb.setSpyderInfoId(template.getSpyderInfoId());
				fb.setTempName(template.getTempName());
				fb.setField(template.getField());
				fb.setFieldIntegrityRate(template.getFieldIntegrityRate());
				fb.setFieldInterval(template.getFieldInterval());
				FieldThread.runMap.put(template.getTemplateId(), fb);
			}
			return true;
		}
	}

	/**
	 * 停止该执行
	 * 
	 * @param template
	 */
	public static boolean removeMap(String templateId) {
		try {
			if (DelayedThread.runMap.get(templateId) != null)
				DelayedThread.runMap.remove(templateId);
			if (PingThread.runMap.get(templateId) != null)
				PingThread.runMap.remove(templateId);
			if (FieldThread.runMap.get(templateId) != null)
				FieldThread.runMap.remove(templateId);
			if (mainMap.get(templateId) != null) {
				mainMap.remove(templateId);
			}
			if (allTmp.get(templateId) != null)
				allTmp.remove(templateId);
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		}
	}

	public static void removeAll() {
		allTmp = new ConcurrentHashMap<String, Template>();
		mainMap = new ConcurrentHashMap<String, Template>();
		DelayedThread.runMap = new ConcurrentHashMap<String, NewsDelayed>();
		DelayedThread.queue = new LinkedBlockingQueue<NewsDelayed>(500);
		FieldThread.runMap = new ConcurrentHashMap<String, FieldIntegerity>();
		FieldThread.queue = new LinkedBlockingQueue<FieldIntegerity>(500);
		PingThread.runMap = new ConcurrentHashMap<String, PingUrl>();
		PingThread.queue = new LinkedBlockingQueue<PingUrl>(500);
	}

	@Override
	public void run() {
		log.info("=======日常入库启动======");
		CrawTemplateServiceImpl crawService = BeanFactory.getBean(CrawTemplateServiceImpl.class);
		DataServiceImpl dataService = BeanFactory.getBean(DataServiceImpl.class);
		ServerServiceImpl serverService = BeanFactory.getBean(ServerServiceImpl.class);
		MessageServiceImpl messageService = BeanFactory.getBean(MessageServiceImpl.class);
		TemplateServiceImpl templateService = BeanFactory.getBean(TemplateServiceImpl.class);
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH");
		List<Channel> channel = serverService.findDisChannel();
		/**初始化模版列表***/
//		if(allTmp==null) {
//			templateService.findAll();
//		}
		
		Calendar loop = Calendar.getInstance();
		while (true) {
			Calendar start = Calendar.getInstance();
			if (start.get(Calendar.HOUR_OF_DAY) < 1) {
				start.add(Calendar.DATE, -1);
			}
			Calendar end = Calendar.getInstance();
			Calendar inTime = Calendar.getInstance();
			inTime.setTime(end.getTime());
			inTime.add(Calendar.HOUR_OF_DAY, -1);
			try {
				start.set(Calendar.HOUR_OF_DAY, 0);
				start.setTime(sdf.parse(sdf.format(start.getTime())));
				end.setTime(sdf.parse(sdf.format(end.getTime())));
				int count = 0;
				for (Channel chan : channel) {
					map.put(chan.getPt(), chan.getValue());
					try {
						ServerBean ser = serverService.findServer(chan.getPt());
						count = es.getCountByPt(start.getTime(), end.getTime(), chan.getPt(), chan.getValue(), ser);// 当日的数据量
						System.out.println(count);
						Data data = new Data();
						data.setCount(count);
						data.setPt(chan.getPt());
						data.setTime(start.getTime());
						dataService.insert(data);
						log.info(sdf.format(start.getTime()) + "|" + chan.getPt() + "|入库");
						count = es.getCountByPt(inTime.getTime(), end.getTime(), chan.getPt(), chan.getValue(), ser);// 小时的数据量
						if (inTime.get(Calendar.HOUR_OF_DAY) > 6)
							if (counts.get(chan.getPt()) != null)
								if (count < counts.get(chan.getPt())) {
									Messages m = new Messages();
									m.setTitle("渠道低数据量预警{" + chan.getPt() + "}");
									m.setContent("1小时内的数据量为：" + count);
									m.setDate(sdf.format(start.getTime()));
									for (String mail : Config.getVal("count_mail").split(",")) {
										SendMailUtil.sendMessage(mail, m);
									}
								}
					} catch (Exception e) {
						log.error(chan.getPt() + "入库出错，错误信息{}", e);
					}
				}
				Thread.sleep(1000 * 120);
				
				Set<String> keys = allTmp.keySet();
				System.out.println("keys size is :::"+ keys.size());
				List<String> needRemove = new ArrayList<>();
				for (Iterator<String> it = keys.iterator(); it.hasNext();) {
					String key = it.next();
					Template tmp = allTmp.get(key);
					if (tmp.getSpyderInfoId() != null) {
						int cwCount = 0;
						String pt = crawService.findPt(tmp.getSpyderInfoId());
						cwCount = es.getCounts(start.getTime(), end.getTime(), pt, map.get(pt), tmp.getSpyderInfoId(),
								serverService.findServer(pt));
						log.info(tmp.getTempName()+"====="+start.getTime()+"======"+end.getTime()+"======"+pt+"====="+cwCount);
						try {
							Data data = new Data();
							data.setCount(cwCount);
							data.setTime(start.getTime());
							data.setTempName(tmp.getTempName());
							dataService.insert(data);
							log.info(TimeUtil.yearToDay(start.getTime()) + "|" + tmp.getTempName() + "|入库");
							if (cwCount == 0) {
								loop.setTime(start.getTime());
								loop.add(Calendar.DATE, -2);
								cwCount = es.getCounts(loop.getTime(), end.getTime(), pt, map.get(pt),
										tmp.getSpyderInfoId(), serverService.findServer(pt));
								log.info(tmp.getTempName()+"====="+loop.getTime()+"======"+end.getTime()+"======"+pt+"====="+cwCount);
								if (cwCount < 3) {
									if (StringUtils.isBlank(tmp.getTempName())) {
										continue;
									}
									Message msg = new Message();
									msg.setTempName(tmp.getTempName());
									msg.setCreateDate(new Date());
									msg.setErrorMsg("连续三天数据不足3条，请确认采集是否有问题");
									msg.setPt(pt);
									msg.setTemplateLv(1);
									messageService.insert(msg);
								} else {
									Message msg = messageService.findMsg(tmp.getTempName(), null);
									if (msg != null) {
										needRemove.add(msg.getMessageId());
									}
								}
							} else {
								Message msg = messageService.findMsg(tmp.getTempName(), null);
								if (msg != null) {
									needRemove.add(msg.getMessageId());
								}
							}
						} catch (Exception e) {
							log.error(tmp.getTempName() + "入库出错，错误信息{}", e);
						}
					}
					sleep(1000);
				}
				// 有数据了，移除错误信息
				messageService.delete(needRemove);
			} catch (Exception e) {
				e.printStackTrace();
			}
			try {
				Thread.sleep(1000 * 3600);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}
