package jp.hasc.hasctool.core.runtime.filter.file.count;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import jp.hasc.hasctool.core.runtime.filter.AbstractFilter;
import jp.hasc.hasctool.core.data.EnumCommand;
import jp.hasc.hasctool.core.runtime.RuntimeContext;
import jp.hasc.hasctool.core.util.CoreUtil;

/**
 * メタ情報別にファイル数をカウントします
 * mergeInf_:　","区切りでキーワードを入力（優先順）
 * @author hiro
 */
public class MetaFileCounter extends AbstractFilter{

	private String metaElement_;
	private HashMap<String, Integer> hmap_;
	private String mergeInf_;
	private String key_ = null;
	private ArrayList<String> mergeList_;
	private int total_;

	@Override
	public void setup(RuntimeContext context) {
		super.setup(context);
		hmap_ = new HashMap<String, Integer>();
		total_ = 0;
	}

	@Override
	public void processMessage(Object message) throws InterruptedException {
		if (message instanceof String) {
			String s =(String)message;
			count(s);
			total_++;
		} else if (message == EnumCommand.END) {
			Iterator<String> it = hmap_.keySet().iterator();
			while (it.hasNext()) {
				String key = (String) it.next();
				int count =  hmap_.get(key);
				String row = key + "," + count;
				outputMessage(row);
			}
			outputMessage("total is :"+total_);
			System.out.println("countEnd");
			outputMessage(EnumCommand.END);
		} else {
			//BEGIN message
			outputMessage(message);
		}
	}

	private void count(String filePath){
		try{
			InputStream inps = getRuntimeContext().getFileStreamProvider().openInputStream(filePath);
			BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inps, RuntimeContext.DEFAULT_CHARSET));
			while(true){
				String line = bufferedReader.readLine();
				if (line == null) {
					break;
				} else if (0 <= line.indexOf(metaElement_)){
					String[] row = line.split(":");
					String key = row[row.length-1];
					key = key.trim();
					if(mergeInf_ != null){
						key = merge(key);
					}
					if(!(key.equals(key_))){
						System.out.println(key);
						key_ = key;
					}
					if((hmap_.containsKey(key))){
						int count = hmap_.get(key);
						count++;
						hmap_.put(key,count);
					} else {
						hmap_.put(key,1);
					}
					break;
				}
			}
			bufferedReader.close();
			inps.close();
		} catch (IOException ex) {
			CoreUtil.throwAsRuntimeException(ex);
		} catch (RuntimeException ex) {
		}
	}

	private String merge(String metaInf){
		mergeList_ = new ArrayList<String>();
		String[] val = mergeInf_.split(",");
		for(int i=0; i<val.length; i++){
			mergeList_.add(val[i]);
		}
		for(int i=0; i<mergeList_.size(); i++){
			String mergeInf = mergeList_.get(i);
			if (0 <= metaInf.indexOf(mergeInf)){
				metaInf = mergeInf;
				break;
			}
		}
		return metaInf;
	}

	public void setMetaElement(String metaEle) {
		metaElement_ = metaEle;
	}

	public void setMergeMetaInf(String metaInf) {
		mergeInf_ = metaInf;
	}
}
