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 java.util.Set;

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.DaoException;

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.ValidatorException;

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 Set<Integer> blacklistRuleIds;

	private int objsValidated = 0;

	private int score = 0;

	private int internalJobsFinished = 0;

	private int internalJobsSum = 1;

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

	private String valBaseUrl = null;

	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) throws ValidatorException {

		try {

			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);

			if (objsValidated % 100 == 0)

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

			objsValidated++;		

		} catch (Exception e) {

			logger.error("Error while proccessing results");

			throw new ValidatorException("Error while proccessing results", e);

		}

	}

	@Override

	@Transactional(propagation = Propagation.REQUIRED)

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

		try {

			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);

			int recordScore = -1;

			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);

//				    if (scoreTriggerRules.contains(ctask.ruleId)) {

//				    	recordScore=0;

//				    }

				    Rule rule = this.getRule(taskStored.getRuleId());

				    String xpath = null;

				    veloMap.put("name", rule.getName());

				    if (! rule.getType().contains("Chain"))

				    	xpath = rule.getConfiguration().getProperty("xpath");

				    else {

				    	xpath = this.getRule(Integer.parseInt(rule.getConfiguration().getProperty("rule_2"))).getConfiguration().getProperty("xpath");

				    }

				    veloMap.put("xpath", xpath);

				    veloMap.put("mandatory", Boolean.toString(rule.isMandatory()));

				}

			}

			this.updateScoreMap(groupBy_values, tasksStored);

			// WORKFLOWS BEGIN

			int recordBlacklistScore = -1;

			if (recordScore == -1) {

				recordScore = this.calculateRecordScore(ruleIdsSuccess);

				if (blacklistRuleIds != null )

					recordBlacklistScore = this.calculateBlacklistRecordScore(ruleIdsSuccess, blacklistRuleIds);

			}

			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);

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

			// WORKFLOWS END

			if (objsValidated % 100 == 0)

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

			objsValidated++;

		} catch (Exception e) {

			logger.error("Error while proccessing results");

			throw new ValidatorException("Error while proccessing results", e);

		}

	}

	@Override

	@Transactional(propagation = Propagation.REQUIRED)

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

		try {

			internalJobsFinished++;

			if(internalJobsFinished == internalJobsSum) { 

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

				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);

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

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

					jobsDao.setTotalJobFinished(jobId, null, false);

				}

			} else {

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

			}

		} catch (Exception e) {

			logger.error("Error while finalizing successfull job");

		}

	}

	@Override

	@Transactional(propagation = Propagation.REQUIRED)

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

		try {

			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" );

			}

		} catch (Exception e) {

			logger.error("Error while finalizing failed job");

		}

	}

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

		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) throws Exception {

		float score = 0;

		float sum = 0;

		float weights = 0 ;

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

			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 int calculateBlacklistRecordScore(Map<Integer,Boolean> ruleIdsSuccess, Set<Integer> blacklistRuleIds) throws Exception {

		float score = 0;

		float sum = 0;

		float weights = 0 ;

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

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

			if (rule.isMandatory() && blacklistRuleIds.contains(entry.getKey())) {

				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;

	}

	private Rule getRule(int id) throws DaoException {

		Rule rule = null;

		if((rule=ruleCacheMap.get(id)) == null) {

			rule = rulesDao.get(id);

			ruleCacheMap.put(id, rule);

		}

		return rule;

	}

	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;

	}

	public Set<Integer> getBlacklistRuleIds() {

		return blacklistRuleIds;

	}

	public void setBlacklistRuleIds(Set<Integer> blacklistRuleIds) {

		this.blacklistRuleIds = blacklistRuleIds;

	}

}