D-Net

package eu.dnetlib.validator.service.impls.listeners;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import javax.xml.xpath.XPath;

import javax.xml.xpath.XPathConstants;

import javax.xml.xpath.XPathExpressionException;

import javax.xml.xpath.XPathFactory;

import org.apache.log4j.Logger;

import org.springframework.transaction.annotation.Propagation;

import org.springframework.transaction.annotation.Transactional;

import org.w3c.dom.Node;

import org.w3c.dom.NodeList;

import eu.dnetlib.domain.functionality.validator.Rule;

import eu.dnetlib.validator.commons.dao.jobs.JobsDAO;

import eu.dnetlib.validator.commons.dao.rules.RuleStatus;

import eu.dnetlib.validator.commons.dao.rules.RulesDAO;

import eu.dnetlib.validator.commons.dao.tasks.TaskStored;

import eu.dnetlib.validator.commons.dao.tasks.TasksDAO;

import eu.dnetlib.validator.engine.execution.CompletedTask;

import eu.dnetlib.validator.engine.execution.JobListener;

/**

 * 

 * @author Nikon Gasparis

 *

 */

public class ValidatorJobListener implements JobListener {

	private static Logger logger = Logger.getLogger(ValidatorJobListener.class);

	private Integer jobSubmittedId;

	private String jobSubmittedUser;

	private TasksDAO tasksDao;

	private JobsDAO jobsDao;

	private RulesDAO rulesDao;

	private String validationType;

	private int objsValidated = 0;

	private int score = 0;

	private int internalJobsFinished = 0;

	Map<String,Map<Integer,RuleStatus>> scoreMapPerGroupBy = new HashMap<String,Map<Integer,RuleStatus>>();

	private String valBaseUrl = null;

	private int internalJobsSum = 1;

	private Map<Integer, Rule> ruleCacheMap = new HashMap<Integer, Rule>();

	private String groupBy_xpath = null;

	@Override

	@Transactional(propagation = Propagation.REQUIRED)

	public synchronized void currentResults(List<CompletedTask> tasks, int jobId, Object record, Map<String,Object> recordContext, Throwable t) {

		logger.error("Completed tasks for " + jobId + " (exception). Error: " + t.getMessage());

		List<TaskStored> tasksStored = new ArrayList<TaskStored>();

		List<String> groupBy_values = this.parseGroupByValues(record,groupBy_xpath);

		for (CompletedTask ctask : tasks) {

			TaskStored taskStored = new TaskStored();

			taskStored.setEnded(ctask.finished.toString());

			taskStored.setStarted(ctask.started.toString());

			taskStored.setJobId(jobSubmittedId);

			taskStored.setRecordIdentifier(ctask.valobjId);

			taskStored.setRuleId(ctask.ruleId);

			taskStored.setSuccess(ctask.success);

			taskStored.setStatus("finished");

			tasksStored.add(taskStored);

			logger.debug("JOBID:"+jobSubmittedId+"# Task-failed: rule " + ctask.ruleId + " on " + ctask.valobjId + " with success " + ctask.success+

					" ruleId: "+taskStored.getRuleId() + " error:"+ t.getMessage());

		}

		this.updateScoreMap(groupBy_values, tasksStored);

//		taskStoredDao.saveTasks(tasksStored,groupBy_values);

		if (objsValidated % 100 == 0)

			jobsDao.setStatus(jobSubmittedId, "ongoing", objsValidated, validationType);

		objsValidated++;		

	}

	@Override

	@Transactional(propagation = Propagation.REQUIRED)

	public synchronized void currentResults(List<CompletedTask> tasks, int jobId, Object record, Map<String,Object> recordContext) {

		List<Map<String, String>> veloList = new ArrayList<Map<String, String>>();

		logger.debug("JOBID:"+jobSubmittedId+"# Updating Completed Tasks");

		Map<Integer,Boolean> ruleIdsSuccess = new HashMap<Integer,Boolean>();

		Map<String, String> veloMap = null;

		List<TaskStored> tasksStored = new ArrayList<TaskStored>();

		List<String> groupBy_values = this.parseGroupByValues(record,groupBy_xpath);

		for (CompletedTask ctask : tasks) {

			veloMap = new HashMap<String, String>();

			TaskStored taskStored = new TaskStored();

			taskStored.setEnded(ctask.finished.toString());

			taskStored.setStarted(ctask.started.toString());

			taskStored.setJobId(jobSubmittedId);

			taskStored.setRecordIdentifier(ctask.valobjId);

			taskStored.setRuleId(ctask.ruleId);

			taskStored.setSuccess(ctask.success);

			taskStored.setStatus("finished");

			tasksStored.add(taskStored);

//			logger.debug("JOBID:"+jobSubmittedId+"# Task: rule " + ctask.ruleId + " on " + ctask.valobjId + " with success " + ctask.success+

//					" ruleId: "+taskStored.getRuleId());

			ruleIdsSuccess.put(taskStored.getRuleId(),ctask.success);

//			logger.debug("inserted on map: "+ "id: " + taskStored.getRuleId());

			if (!ctask.success) {

				veloMap.put("id", Integer.toString(taskStored.getRuleId()));

				veloMap.put("error", "...");

			    veloList.add(veloMap);

			}

		}

		this.updateScoreMap(groupBy_values, tasksStored);

		// WORKFLOWS BEGIN

		int score = this.calculateRecordScore(ruleIdsSuccess);

		Map<String,String> recordValidationResult = new HashMap<String,String>();

		recordValidationResult.put("score", Integer.toString(score));

		recordValidationResult.put("id", tasks.get(0).valobjId);

		if (score > 0)

			recordValidationResult.put("status", "pass");

		else

			recordValidationResult.put("status", "fail");

		recordContext.put("veloList", (List<Map<String, String>>) veloList);

		recordContext.put("recordValidationResult", (Map<String,String>) recordValidationResult);

		recordContext.put("score", (int) score);

		// WORKFLOWS END

		if (objsValidated % 10 == 0)

			jobsDao.setStatus(jobSubmittedId, "ongoing", objsValidated, validationType);

		objsValidated++;

	}

	@Override

	@Transactional(propagation = Propagation.REQUIRED)

	public synchronized void finished(int jobId, Map<String,Object> jobContext) {

		internalJobsFinished++;

		if(internalJobsFinished == internalJobsSum) { 

			logger.debug("JOBID:"+ this.jobSubmittedId +"# Set job finished");

			tasksDao.saveTasks(this.scoreMapPerGroupBy.get("all"));

			jobContext.put("jobSubmittedId", (int) jobSubmittedId);

			jobContext.put("jobSubmittedUser", (String) jobSubmittedUser);

			score = jobsDao.setJobFinished(jobSubmittedId, this.scoreMapPerGroupBy, null, false, objsValidated, validationType);

			logger.debug("score_"+validationType +": " + score);

			jobContext.put("score_"+validationType, (int) score);

			if (jobSubmittedUser.contains("Workflow")) {

				jobsDao.setTotalJobFinished(jobId, null, false);

			}

		} else {

			logger.debug("JOBID:"+jobSubmittedId+"#Job finished. Waiting "+ (internalJobsSum-internalJobsFinished) + " job(s) to finish" );

		}

	}

	@Override

	@Transactional(propagation = Propagation.REQUIRED)

	public synchronized void failed(int jobId, Map<String,Object> jobContext, Throwable t) {

		internalJobsFinished++;

		if(internalJobsFinished == internalJobsSum) { 		

			logger.debug("JOBID:"+jobSubmittedId+"# Set job finished-failed");

			jobContext.put("jobSubmittedId", (int) jobSubmittedId);

			jobContext.put("jobSubmittedUser",(String) jobSubmittedUser);

			score = jobsDao.setJobFinished(jobSubmittedId, this.scoreMapPerGroupBy,t.getMessage(), true, objsValidated, validationType);

			jobContext.put("score_"+validationType, (int) 0);

			if (jobSubmittedUser.contains("Workflow")) {

				jobsDao.setTotalJobFinished(jobId, t.getMessage(), true);

			}

		} else {

			logger.debug("JOBID:"+jobSubmittedId+"#Job finished. Waiting "+ (internalJobsSum-internalJobsFinished) + " job(s) to finish" );

		}

	}

	private synchronized void updateScoreMap(List<String> groupBy_values, List<TaskStored> tasksStored) {

		for (String groupBy : groupBy_values) {

			logger.debug("updating map for groupBy value: " + groupBy);

			Map<Integer,RuleStatus> scoreMapPerRule = null;

			if((scoreMapPerRule=this.scoreMapPerGroupBy.get(groupBy)) == null) {

				logger.debug("map for groupBy value: " + groupBy + " doesn't exist");

				scoreMapPerRule = new HashMap<Integer, RuleStatus>();

			} else {

				logger.debug("map for groupBy value: " + groupBy + " exists");

			}

			for (TaskStored task : tasksStored) {

				RuleStatus ruleSt = null;

				if((ruleSt=scoreMapPerRule.get(task.getRuleId())) == null) {

					logger.debug("ruleStatus for rule with id : " + task.getRuleId() + " doesn't exist");

					ruleSt = new RuleStatus();

				} 

				Rule rule = null;

				if((rule=ruleCacheMap.get(task.getRuleId())) == null) {

					rule = rulesDao.get(task.getRuleId());

					ruleCacheMap.put(task.getRuleId(), rule);

				}

				ruleSt.setMandatory(rule.isMandatory());

				ruleSt.setWeight(rule.getWeight());

				ruleSt.total++;

				if (task.getSuccess())

					ruleSt.success++;

				else {

					if (groupBy.equals("all") && ruleSt.getFailedTasks().size() < 10 )

						ruleSt.getFailedTasks().add(task);

				}

				scoreMapPerRule.put(task.getRuleId(), ruleSt);

			}

			this.scoreMapPerGroupBy.put(groupBy, scoreMapPerRule);

		}

	} 

	private int calculateRecordScore(Map<Integer,Boolean> ruleIdsSuccess) {

		float score = 0;

		float sum = 0;

		float weights = 0 ;

		for (Map.Entry<Integer,Boolean> entry : ruleIdsSuccess.entrySet()) {

//			RuleStored rule = ruleStoredDao.get(entry.getKey().toString());

			Rule rule = ruleCacheMap.get(entry.getKey());

			if (rule.isMandatory()) {

				weights += rule.getWeight();

				if (entry.getValue()) {

					sum += rule.getWeight();

				}

			}

			score = (sum/weights)*100;

		}

		return (int) score;

	}

	private synchronized List<String> parseGroupByValues(Object record, String xpath) {

		List<String> groupBy_values = null;

		logger.debug("groupBy field: "+xpath);

		try{

			XPathFactory factory = XPathFactory.newInstance();

	        XPath xPath = factory.newXPath();

	        NodeList rtl = (NodeList) xPath.evaluate(xpath+"/text()",record ,XPathConstants.NODESET);

	        groupBy_values = new ArrayList<String>();

	        if (rtl.getLength() > 0) {

	        	for(int i=0; i<rtl.getLength(); i++) {

	        		Node childNode = rtl.item(i);

	        		groupBy_values.add(childNode.getNodeValue());

	        		logger.debug("value: "+childNode.getNodeValue());

	        	}

	        	groupBy_values.add("all");

	        } else {

	        	groupBy_values.add("all");

	        }

	 	} catch (XPathExpressionException e) {

			logger.error("error getting object"+ e);

	 	}

		return groupBy_values;

	}

	public Integer getJobSubmittedId() {

		return jobSubmittedId;

	}

	public void setJobSubmittedId(Integer jobSubmittedId) {

		this.jobSubmittedId = jobSubmittedId;

	}

	public String getJobSubmittedUser() {

		return jobSubmittedUser;

	}

	public void setJobSubmittedUser(String jobSubmittedUser) {

		this.jobSubmittedUser = jobSubmittedUser;

	}

	public String getGroupBy_xpath() {

		return groupBy_xpath;

	}

	public void setGroupBy_xpath(String groupBy_xpath) {

		this.groupBy_xpath = groupBy_xpath;

	}

	public String getValBaseUrl() {

		return valBaseUrl;

	}

	public void setValBaseUrl(String valBaseUrl) {

		this.valBaseUrl = valBaseUrl;

	}

	public int getInternalJobsSum() {

		return internalJobsSum;

	}

	public void setInternalJobsSum(int internalJobsSum) {

		this.internalJobsSum = internalJobsSum;

	}

	public Map<String, Map<Integer, RuleStatus>> getScoreMapPerGroupBy() {

		return scoreMapPerGroupBy;

	}

	public void setScoreMapPerGroupBy(

			Map<String, Map<Integer, RuleStatus>> scoreMapPerGroupBy) {

		this.scoreMapPerGroupBy = scoreMapPerGroupBy;

	}

	public TasksDAO getTasksDao() {

		return tasksDao;

	}

	public void setTasksDao(TasksDAO tasksDao) {

		this.tasksDao = tasksDao;

	}

	public JobsDAO getJobsDao() {

		return jobsDao;

	}

	public void setJobsDao(JobsDAO jobsDao) {

		this.jobsDao = jobsDao;

	}

	public RulesDAO getRulesDao() {

		return rulesDao;

	}

	public void setRulesDao(RulesDAO rulesDao) {

		this.rulesDao = rulesDao;

	}

	public Map<Integer, Rule> getRuleCacheMap() {

		return ruleCacheMap;

	}

	public void setRuleCacheMap(Map<Integer, Rule> ruleCacheMap) {

		this.ruleCacheMap = ruleCacheMap;

	}

	public String getValidationType() {

		return validationType;

	}

	public void setValidationType(String validationType) {

		this.validationType = validationType;

	}

}