package com.uas.search.console.service.impl;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.PrintWriter;
import java.sql.SQLRecoverableException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.lucene.document.Document;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.Term;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import com.alibaba.fastjson.JSONObject;
import com.uas.platform.core.model.PageInfo;
import com.uas.platform.core.model.PageParams;
import com.uas.search.console.LuceneProperties;
import com.uas.search.console.dao.BrandSimpleInfoDao;
import com.uas.search.console.dao.ComponentSimpleInfoDao;
import com.uas.search.console.dao.KindSimpleInfoDao;
import com.uas.search.console.dao.LuceneQueueMessageDao;
import com.uas.search.console.dao.OrderInvoiceSimpleInfoDao;
import com.uas.search.console.dao.OrderSimpleInfoDao;
import com.uas.search.console.dao.PurchaseInvoiceSimpleInfoDao;
import com.uas.search.console.dao.PurchaseSimpleInfoDao;
import com.uas.search.console.jms.AQListener;
import com.uas.search.console.jms.QueueMessageParser;
import com.uas.search.console.model.BrandSimpleInfo;
import com.uas.search.console.model.ComponentSimpleInfo;
import com.uas.search.console.model.KindSimpleInfo;
import com.uas.search.console.model.LuceneQueueMessage;
import com.uas.search.console.model.OrderInvoiceSimpleInfo;
import com.uas.search.console.model.OrderSimpleInfo;
import com.uas.search.console.model.ParsedQueueMessage;
import com.uas.search.console.model.PurchaseInvoiceSimpleInfo;
import com.uas.search.console.model.PurchaseSimpleInfo;
import com.uas.search.console.service.IndexService;
import com.uas.search.console.support.IndexSearcherManager;
import com.uas.search.console.support.IndexWriterManager;
import com.uas.search.console.util.ObjectToDocumentUtils;
import com.uas.search.console.util.SearchConstants;
import com.uas.search.console.util.SearchUtils;
import com.uas.search.exception.SearchException;
import com.uas.search.model.SPage;

/**
 * 创建索引
 * 
 * @author sunyj
 * @since 2016年8月5日 下午2:23:22
 */
@Service
public class IndexServiceImpl implements IndexService {

	@Autowired
	private QueueMessageParser queueMessageParser;

	@Autowired
	private LuceneQueueMessageDao luceneQueueMessageDao;

	@Autowired
	private KindSimpleInfoDao kindDao;

	@Autowired
	private BrandSimpleInfoDao brandDao;

	@Autowired
	private ComponentSimpleInfoDao componentDao;

	@Autowired
	private OrderSimpleInfoDao orderDao;

	@Autowired
	private OrderInvoiceSimpleInfoDao orderInvoiceDao;

	@Autowired
	private PurchaseSimpleInfoDao purchaseDao;

	@Autowired
	private PurchaseInvoiceSimpleInfoDao purchaseInvoiceDao;

	private IndexWriter indexWriter;

	private static IndexWriterManager indexWriterManager = new IndexWriterManager();

	@Autowired
	private AQListener aqListener;

	@Autowired
	private LuceneProperties luceneProperties;

	private IndexSearcherManager indexSearcherManager = IndexSearcherManager.getInstance();

	/**
	 * 是否正在创建索引
	 */
	private boolean creatingIndex = false;

	/**
	 * 从数据库获取数据时的分页大小
	 */
	private static final int PAGE_SIZE = 1000;

	/**
	 * 单个文件存储的最大数据数目，需是PAGE_SIZE的整数倍
	 */
	public static final int SINGLE_FILE_MAX_SIZE = 100000;

	private static Logger logger = LoggerFactory.getLogger(IndexServiceImpl.class);

	@Override
	public Long createIndexes(List<String> tableNames, Boolean componentFromFiles) {
		if (creatingIndex) {
			throw new SearchException("已存在线程在创建索引，不可重复请求");
		}
		creatingIndex = true;
		// 如果索引实时更新处于开启状态，需要关闭（以免两者同时操作索引出现问题）
		if (aqListener.isRunning()) {
			logger.info("索引实时更新服务正在运行，尝试关闭索引实时更新服务...");
			aqListener.stop();
		}

		if (CollectionUtils.isEmpty(tableNames)) {
			tableNames = SearchUtils.getTableNames();
		}

		try {
			Long startTime = new Date().getTime();

			for (String tableName : tableNames) {
				Long start = new Date().getTime();
				Long size = 0L;
				deleteIndexes(tableName);
				if (tableName.equals(SearchConstants.KIND_TABLE_NAME)) {
					size = createKindIndexes();
				} else if (tableName.equals(SearchConstants.BRAND_TABLE_NAME)) {
					size = createBrandIndexes();
				} else if (tableName.equals(SearchConstants.COMPONENT_TABLE_NAME)) {
					// 只有明确指定不从本地文件读取数据时，才从数据库获取数据建立索引
					if (componentFromFiles != null && !componentFromFiles) {
						size = createComponentIndexes();
					} else {
						size = createComponentIndexesWithFiles();
					}
				} else if (tableName.equals(SearchConstants.ORDER_TABLE_NAME)) {
					size = createOrderIndexes();
				} else if (tableName.equals(SearchConstants.ORDER_INVOICE_TABLE_NAME)) {
					size = createOrderInvoiceIndexes();
				} else if (tableName.equals(SearchConstants.PURCHASE_TABLE_NAME)) {
					size = createPurchaseIndexes();
				} else if (tableName.equals(SearchConstants.PURCHASE_INVOICE_TABLE_NAME)) {
					size = createPurchaseInvoiceIndexes();
				} else {
					logger.error("不支持创建该表索引：" + tableName);
					continue;
				}
				indexWriterManager.release(tableName);
				Long end = new Date().getTime();
				logger.info(String.format("创建%s索引： %s条，耗时%.2fs\n ", tableName, size, (end - start) / 1000.0));
			}

			Long endTime = new Date().getTime();
			logger.info(String.format("索引创建成功， 共用时间%.2fs\n", (endTime - startTime) / 1000.0));
			return endTime - startTime;
		} catch (Exception e) {
			// 防止SQLRecoverableException导致应用终止
			throw new SearchException(e).setDetailedMessage(e);
		} finally {
			creatingIndex = false;
		}
	}

	/**
	 * 删除指定表的索引
	 * 
	 * @param tableName
	 *            表名
	 * @throws InterruptedException
	 * @throws IOException
	 */
	private void deleteIndexes(String tableName) throws InterruptedException, IOException {
		indexWriter = indexWriterManager.get(tableName);
		logger.info("正在清理旧索引..." + tableName);
		indexWriter.deleteAll();
		indexWriter.commit();
		logger.info("旧索引清理完毕..." + tableName);
	}

	/**
	 * 创建类目索引
	 * 
	 * @return 写入的类目索引数
	 * @throws IOException
	 */
	private Long createKindIndexes() throws IOException {
		logger.info("正在创建类目索引...");
		List<KindSimpleInfo> kinds = kindDao.findAll();
		logger.info("发现数据：" + kinds.size() + "条");
		return createIndexesWithObjects(kinds.toArray());
	}

	/**
	 * 创建品牌索引
	 * 
	 * @return 写入的品牌索引数
	 * @throws IOException
	 */
	private Long createBrandIndexes() throws IOException {
		logger.info("正在创建品牌索引...");
		List<BrandSimpleInfo> brands = brandDao.findAll();
		logger.info("发现数据：" + brands.size() + "条");
		return createIndexesWithObjects(brands.toArray());
	}

	private Long createComponentIndexesWithFiles() {
		logger.info("正在创建器件索引...");
		Long size = 0L;
		try {
			// 从本地路径读取器件数据
			File[] files = new File(luceneProperties.getComponentsDir()).listFiles();
			if (files == null || files.length == 0) {
				logger.info("创建器件索引失败，原因：器件数据文件不存在！");
				return 0L;
			}
			// 将要创建的索引总数目约为：文件数目*单个文件的行数
			long totalSize = files.length * SINGLE_FILE_MAX_SIZE;
			for (File file : files) {
				logger.info("读取文件: " + file.getName());
				BufferedReader bufferedReader = new BufferedReader(new FileReader(file));
				String line = null;
				while (!StringUtils.isEmpty(line = bufferedReader.readLine())) {
					ComponentSimpleInfo component = JSONObject.parseObject(line, ComponentSimpleInfo.class);
					Document document = ObjectToDocumentUtils.toDocument(component);
					if (document != null) {
						size++;
						// 每创建10000条，打印一次进度
						if (size % 10000 == 0) {
							logger.info(String.format("Component indexed...................%.2f%%",
									size * 100.0 / totalSize));
						}
						indexWriter.addDocument(document);
					}
				}
				indexWriter.commit();
				bufferedReader.close();
			}
		} catch (FileNotFoundException e) {
			logger.error("创建器件索引失败，原因：器件数据文件不存在！");
			return 0L;
		} catch (IOException e) {
			logger.error("", e);
		}
		return size;
	}

	/**
	 * 创建器件索引，从数据库取数据
	 * 
	 * @return 写入的器件索引数
	 * @throws IOException
	 */
	public Long createComponentIndexes() throws IOException {
		logger.info("正在创建器件索引...");
		Long size = 0L;
		PageParams params = new PageParams();
		int page = 1;
		params.setCount(PAGE_SIZE);
		params.setPage(page);

		PageInfo info = new PageInfo(params);
		Page<ComponentSimpleInfo> pageResult = componentDao.findAll(info);
		long totalElements = pageResult.getTotalElements();
		logger.info("Number of components: " + totalElements);
		// 用于记录上次提交索引时的创建进度
		double recordProgress = 0;
		while (totalElements > size) {
			List<ComponentSimpleInfo> components = pageResult.getContent();
			for (ComponentSimpleInfo component : components) {
				Document document = ObjectToDocumentUtils.toDocument(component);
				if (document != null) {
					indexWriter.addDocument(document);
				}
			}
			size += components.size();

			// 器件索引的创建进度（百分比）
			double indexProgress = size * 100.0 / totalElements;
			logger.info(String.format("Component indexed...................%.2f%%", indexProgress));
			// 每创建5%，提交一次，避免内存耗尽，发生OutOfMemoryError
			if (indexProgress - recordProgress > 5) {
				indexWriter.commit();
				recordProgress = indexProgress;
			}

			page++;
			params.setPage(page);
			info = new PageInfo(params);
			pageResult = componentDao.findAll(info);
		}
		indexWriter.commit();
		return size;
	}

	private Long createOrderIndexes() {
		logger.info("正在创建销售单索引...");
		List<OrderSimpleInfo> orders = orderDao.findAll();
		logger.info("发现数据：" + orders.size() + "条");
		return createIndexesWithObjects(orders.toArray());
	}

	private Long createOrderInvoiceIndexes() {
		logger.info("正在创建销售发货单索引...");
		List<OrderInvoiceSimpleInfo> orderInvoices = orderInvoiceDao.findAll();
		logger.info("发现数据：" + orderInvoices.size() + "条");
		return createIndexesWithObjects(orderInvoices.toArray());
	}

	private Long createPurchaseIndexes() {
		logger.info("正在创建采购单索引...");
		List<PurchaseSimpleInfo> purchases = purchaseDao.findAll();
		logger.info("发现数据：" + purchases.size() + "条");
		return createIndexesWithObjects(purchases.toArray());
	}

	private Long createPurchaseInvoiceIndexes() {
		logger.info("正在创建采购发货单索引...");
		List<PurchaseInvoiceSimpleInfo> purchaseInvoices = purchaseInvoiceDao.findAll();
		logger.info("发现数据：" + purchaseInvoices.size() + "条");
		return createIndexesWithObjects(purchaseInvoices.toArray());
	}

	/**
	 * 利用对象数组创建索引
	 * 
	 * @param objects
	 *            对象数组，可为KindSimpleInfo、BrandSimpleInfo、OrderSimpleInfo、
	 *            OrderInvoiceSimpleInfo、PurchaseSimpleInfo、
	 *            PurchaseInvoiceSimpleInfo
	 * @return 对象数组的数量
	 */
	private Long createIndexesWithObjects(Object[] objects) {
		if (objects == null || objects.length < 1)
			return 0L;

		long count = 0;
		try {
			for (Object object : objects) {
				Document document = ObjectToDocumentUtils.toDocument(object);
				if (document != null) {
					indexWriter.addDocument(document);
					count++;
				}
			}
			indexWriter.commit();
		} catch (IOException e) {
			logger.error("", e);
		}
		return count;
	}

	@Override
	public Long downloadComponentDataFromDatabase(Integer startFileIndex) {
		int fileIndex = 1;
		if (startFileIndex != null) {
			if (startFileIndex < 1) {
				throw new SearchException("startFileIndex不小于1");
			} else {
				fileIndex = startFileIndex;
			}
		}

		Long startTime = new Date().getTime();
		logger.info("下载器件... ");

		// 分页获取数据
		PageParams pageParams = new PageParams();
		pageParams.setPage((fileIndex - 1) * SINGLE_FILE_MAX_SIZE / PAGE_SIZE + 1);
		pageParams.setCount(PAGE_SIZE);
		PageInfo pageInfo = new PageInfo(pageParams);
		Page<ComponentSimpleInfo> pageResult = componentDao.findAll(pageInfo);

		// 数据库中数据的总数目
		if (pageResult.getTotalElements() < (fileIndex - 1) * SINGLE_FILE_MAX_SIZE) {
			throw new SearchException("startFileIndex不超过"
					+ ((int) Math.ceil(pageResult.getTotalElements() / (1.0 * SINGLE_FILE_MAX_SIZE))));
		}
		long totalElements = pageResult.getTotalElements() - (fileIndex - 1) * SINGLE_FILE_MAX_SIZE;
		logger.info("发现数据：" + totalElements + "条");
		// 已翻页的数据数目
		Long size = 0L;
		PrintWriter printWriter = null;
		int count = 0;
		try {
			File file = new File(luceneProperties.getComponentsDir());
			if (!file.exists()) {
				file.mkdirs();
			}
			printWriter = new PrintWriter(luceneProperties.getComponentsDir() + "/" + fileIndex + ".txt");
			while (totalElements > size) {
				// 一个文件存放100000条数据，一旦超过，写入新的文件
				if (count == SINGLE_FILE_MAX_SIZE) {
					count = 0;
					printWriter.flush();
					printWriter.close();
					fileIndex++;
					printWriter = new PrintWriter(luceneProperties.getComponentsDir() + "/" + fileIndex + ".txt");
				}
				List<ComponentSimpleInfo> content = pageResult.getContent();
				for (ComponentSimpleInfo element : content) {
					printWriter.println(JSONObject.toJSONString(element));
					count++;
				}
				size += content.size();
				logger.info(String.format("Downloaded...................%.2f%%", size * 100.0 / totalElements));

				pageParams.setPage(pageParams.getPage() + 1);
				pageInfo = new PageInfo(pageParams);
				pageResult = componentDao.findAll(pageInfo);
			}
			printWriter.flush();
			printWriter.close();
		} catch (FileNotFoundException e) {
			logger.error("", e);
		}

		Long endTime = new Date().getTime();
		logger.info(String.format("下载数据%s条，耗时%.2fs\n ", totalElements, (endTime - startTime) / 1000.0));
		return endTime - startTime;
	}

	@Override
	public Object save(Object obj) {
		if (obj != null) {
			String tableName = SearchUtils.getTableName(obj.getClass());
			Document document = ObjectToDocumentUtils.toDocument(obj);
			if (document != null) {
				try {
					indexWriter = indexWriterManager.get(tableName);
					indexWriter.addDocument(document);
					indexWriter.commit();
					indexSearcherManager.flushCache(tableName, null);
					return obj;
				} catch (IOException | InterruptedException e) {
					logger.error("", e);
				} finally {
					indexWriterManager.release(tableName);
				}
			} else {
				logger.info("对象转为Document时为null：" + obj);
			}
		}
		return null;
	}

	@Override
	public Object update(Object obj) {
		if (obj != null) {
			String tableName = SearchUtils.getTableName(obj.getClass());
			Document document = ObjectToDocumentUtils.toDocument(obj);
			if (document != null) {
				try {
					indexWriter = indexWriterManager.get(tableName);
					if (obj instanceof KindSimpleInfo) {
						indexWriter.updateDocument(
								new Term(SearchConstants.KIND_ID_FIELD, String.valueOf(((KindSimpleInfo) obj).getId())),
								document);
					} else if (obj instanceof BrandSimpleInfo) {
						indexWriter.updateDocument(new Term(SearchConstants.BRAND_ID_FIELD,
								String.valueOf(((BrandSimpleInfo) obj).getId())), document);
					} else if (obj instanceof ComponentSimpleInfo) {
						indexWriter.updateDocument(new Term(SearchConstants.COMPONENT_ID_FIELD,
								String.valueOf(((ComponentSimpleInfo) obj).getId())), document);
					} else if (obj instanceof OrderSimpleInfo) {
						indexWriter.updateDocument(new Term(SearchConstants.ORDER_ID_FIELD,
								String.valueOf(((OrderSimpleInfo) obj).getId())), document);
					} else if (obj instanceof OrderInvoiceSimpleInfo) {
						indexWriter.updateDocument(new Term(SearchConstants.ORDER_INVOICE_ID_FIELD,
								String.valueOf(((OrderInvoiceSimpleInfo) obj).getId())), document);
					} else if (obj instanceof PurchaseSimpleInfo) {
						indexWriter.updateDocument(new Term(SearchConstants.PURCHASE_ID_FIELD,
								String.valueOf(((PurchaseSimpleInfo) obj).getId())), document);
					} else if (obj instanceof PurchaseInvoiceSimpleInfo) {
						indexWriter.updateDocument(new Term(SearchConstants.PURCHASE_INVOICE_ID_FIELD,
								String.valueOf(((PurchaseInvoiceSimpleInfo) obj).getId())), document);
					} else {
						throw new SearchException("Message parsing failed!");
					}
					indexWriter.commit();
					indexSearcherManager.flushCache(tableName, null);
					return obj;
				} catch (IOException | InterruptedException e) {
					logger.error("", e);
				} finally {
					indexWriterManager.release(tableName);
				}
			} else {
				logger.info("对象转为Document时为null：" + obj);
			}
		}
		return null;
	}

	@Override
	public Object delete(Object obj) {
		if (obj != null) {
			String tableName = SearchUtils.getTableName(obj.getClass());
			try {
				indexWriter = indexWriterManager.get(tableName);
				if (obj instanceof KindSimpleInfo) {
					indexWriter.deleteDocuments(
							new Term(SearchConstants.KIND_ID_FIELD, String.valueOf(((KindSimpleInfo) obj).getId())));
				} else if (obj instanceof BrandSimpleInfo) {
					indexWriter.deleteDocuments(
							new Term(SearchConstants.BRAND_ID_FIELD, String.valueOf(((BrandSimpleInfo) obj).getId())));
				} else if (obj instanceof ComponentSimpleInfo) {
					indexWriter.deleteDocuments(new Term(SearchConstants.COMPONENT_ID_FIELD,
							String.valueOf(((ComponentSimpleInfo) obj).getId())));
				} else if (obj instanceof OrderSimpleInfo) {
					indexWriter.deleteDocuments(
							new Term(SearchConstants.ORDER_ID_FIELD, String.valueOf(((OrderSimpleInfo) obj).getId())));
				} else if (obj instanceof OrderInvoiceSimpleInfo) {
					indexWriter.deleteDocuments(new Term(SearchConstants.ORDER_INVOICE_ID_FIELD,
							String.valueOf(((OrderInvoiceSimpleInfo) obj).getId())));
				} else if (obj instanceof PurchaseSimpleInfo) {
					indexWriter.deleteDocuments(new Term(SearchConstants.PURCHASE_ID_FIELD,
							String.valueOf(((PurchaseSimpleInfo) obj).getId())));
				} else if (obj instanceof PurchaseInvoiceSimpleInfo) {
					indexWriter.deleteDocuments(new Term(SearchConstants.PURCHASE_INVOICE_ID_FIELD,
							String.valueOf(((PurchaseInvoiceSimpleInfo) obj).getId())));
				} else {
					throw new SearchException("Message parsing failed!");
				}
				indexWriter.commit();
				indexSearcherManager.flushCache(tableName, null);
				return obj;
			} catch (IOException e) {
				logger.error("", e);
			} catch (InterruptedException e) {
				logger.error("", e);
			} finally {
				indexWriterManager.release(tableName);
			}
		}
		return null;
	}

	@Override
	public List<Object> maintainIndexes(ParsedQueueMessage parsedQueueMessage) {
		if (parsedQueueMessage == null) {
			throw new SearchException("参数不能为空：parsedQueueMessage");
		}

		Object[] objects = parsedQueueMessage.getObjects();
		if (objects == null) {
			return null;
		}
		List<Object> maintainedObjects = new ArrayList<>();
		try {
			// 新增、更新索引
			if (parsedQueueMessage.isInsert() || parsedQueueMessage.isUpdate()) {
				for (Object object : objects) {
					Object maintainedObject = update(object);
					if (maintainedObject != null) {
						maintainedObjects.add(maintainedObject);
					}
				}
			}
			// 删除索引
			else if (parsedQueueMessage.isDelete()) {
				for (Object object : objects) {
					Object maintainedObject = delete(object);
					if (maintainedObject != null) {
						maintainedObjects.add(maintainedObject);
					}
				}
			} else {
				throw new SearchException("message parsing failed!");
			}
		} catch (Throwable e) {
			throw new SearchException(e).setDetailedMessage(e);
		}
		return maintainedObjects;
	}

	@Override
	public List<Object> maintainIndexes(String tableName, String ids, String method) {
		if (StringUtils.isEmpty(tableName) || StringUtils.isEmpty(ids) || StringUtils.isEmpty(method)) {
			throw new SearchException("参数不能为空：tableName，ids，method");
		}
		Map<String, Object> map = new HashMap<>();
		map.put("table", tableName);
		String[] strs = ids.split(",");
		List<Long> idList = new ArrayList<>();
		for (String str : strs) {
			idList.add(Long.parseLong(str));
		}
		map.put("ids", idList);
		map.put("method", method);
		ParsedQueueMessage parsedQueueMessage = queueMessageParser.parse(JSONObject.toJSONString(map));

		if (parsedQueueMessage == null) {
			logger.error("message parsing failed!");
			return null;
		}
		return maintainIndexes(parsedQueueMessage);
	}

	@Override
	public SPage<LuceneQueueMessage> getListenDetails(Integer page, Integer size, String searchContent) {
		try {
			return luceneQueueMessageDao.findAll(page, size, searchContent);
		} catch (SQLRecoverableException e) {
			throw new SearchException(e).setDetailedMessage(e);
		}
	}

	@Override
	public boolean dequeueLuceneQueueMessage(String messageId) {
		try {
			return luceneQueueMessageDao.dequeueLuceneQueueMessage(messageId);
		} catch (SQLRecoverableException e) {
			throw new SearchException(e).setDetailedMessage(e);
		}
	}

}
