/*
    * Copyright (c) 2007, Peter Mika All rights reserved.
    * 
    * Redistribution and use in source and binary forms, with or without
    * modification, are permitted provided that the following conditions are met:
    * 
    * - Redistributions of source code must retain the above copyright notice, this
    *   list of conditions and the following disclaimer.
    * - Redistributions in binary form must reproduce the above copyright notice,
   *   this list of conditions and the following disclaimer in the documentation
   *   and/or other materials provided with the distribution. 
   * - Neither the name of the openrdf.org nor the names of its contributors may
   *   be used to endorse or promote products derived from this software without
   *   specific prior written permission.
   * 
   * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
   * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
   * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
   * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
   * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
   * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
   * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
   * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
   * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
   * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
   * POSSIBILITY OF SUCH DAMAGE.
   * 
   */
  package org.openrdf.elmo.smusher;
  
  import java.io.UnsupportedEncodingException;
  import java.security.NoSuchAlgorithmException;
  import java.util.Iterator;
  
  import org.openrdf.OpenRDFException;
  import org.openrdf.elmo.sesame.SesameManager;
  import org.openrdf.elmo.smusher.listener.SmusherListener;
  import org.openrdf.elmo.smusher.Util;
  import org.openrdf.model.Resource;
  import org.openrdf.model.URI;
  import org.openrdf.model.Value;
  import org.openrdf.query.BindingSet;
  import org.openrdf.query.TupleQueryResult;
  import org.openrdf.repository.Repository;
  import org.openrdf.repository.RepositoryConnection;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public class EmailSmusher extends AbstractSmusher {
       
  	protected final static Logger _logger = LoggerFactory.getLogger(EmailSmusher.class);
  	   
      private final static String EMAIL_AND_CHECKSUM_QUERY = 
          "SELECT DISTINCT person, mail FROM {person} foaf:mbox {mail} " + " UNION " +
          "SELECT DISTINCT person, mail FROM {person} owl:sameAs {other} foaf:mbox {mail} " + " UNION " +
          //"SELECT DISTINCT person, mail FROM {other} owl:sameAs {person}, {other} foaf:mbox {mail} " + " UNION " +   
          "SELECT DISTINCT person, checksum FROM {person} foaf:mbox_sha1sum {checksum} " + " UNION " +
          "SELECT DISTINCT person, checksum FROM {person} owl:sameAs {other} foaf:mbox_sha1sum {checksum} " +     
          //"SELECT DISTINCT person, checksum FROM {other} owl:sameAs {person}, {other} foaf:mbox_sha1sum {checksum} " +   
          "USING NAMESPACE foaf=<" + org.openrdf.model.vocabulary.FOAF.NAMESPACE + ">," +
          "owl=<" + org.openrdf.model.vocabulary.OWL.NAMESPACE + ">";
     
      private static String normalizeEmail(Value email) {
          String value = email.toString();
          if (email instanceof URI) {
              //add mailto: if required and calculate checksum
              if (!value.startsWith("mailto:")) {
                  value = value + "mailto:";
              } 
              try {
                  value = Util.createSHA1(value);
              } catch (UnsupportedEncodingException e) {
                  //ignore
              } catch (NoSuchAlgorithmException e) {
                  //ignore
              }
          }
          return value;
      }
      
      	 
      public void smush(SesameManager firstRepository, SesameManager secondRepository) throws OpenRDFException {
          //Start all the registered listeners
          Iterator it = _listeners.iterator();
          while (it.hasNext()) {
              SmusherListener nextListener = (SmusherListener) it.next();
              nextListener.start();
          }
          
      	//Select all emails and checksums and compare them pairwise
     
      	RepositoryConnection firstConnection = firstRepository.getConnection();
   		RepositoryConnection secondConnection = null;
   		   		
  		TupleQueryResult firstResult = null;
  		TupleQueryResult secondResult = null;
  		
  		try {
  	     		
  		    firstResult = firstConnection.prepareTupleQuery(org.openrdf.query.QueryLanguage.SERQL,
  						EMAIL_AND_CHECKSUM_QUERY).evaluate();
  		    
  		    if (!firstRepository.equals(secondRepository)) {
             	secondConnection = secondRepository.getConnection();
      		    secondResult = secondConnection.prepareTupleQuery(org.openrdf.query.QueryLanguage.SERQL,
      						EMAIL_AND_CHECKSUM_QUERY).evaluate();
      		} 
         	while (firstResult.hasNext()) {
         		while (secondResult.hasNext()) {
         			BindingSet firstSet = firstResult.next();
         			BindingSet secondSet = secondResult.next();
         			Value firstValue = firstSet.getValue("mail");
                     Value secondValue = secondSet.getValue("mail");
                     if (normalizeEmail(firstValue)
                             .equalsIgnoreCase(normalizeEmail(secondValue))) {
                         Iterator subit = _listeners.iterator();
                         while (subit.hasNext()) {
                             SmusherListener nextListener = (SmusherListener) subit.next();
                             nextListener.matchFound(
                                     (Resource) firstSet.getValue("person"), 
                                     (Resource) secondSet.getValue("person"));    
                         }
                     }
                    
         		
         		}        		
         	}       	
 		} finally {
 			if (firstResult != null) firstResult.close();
 			if (firstConnection != null) firstConnection.close();
 			if (secondResult != null) secondResult.close();
 			if (secondConnection != null) secondConnection.close();			
     	}
         		
          
         //Finish all the registered listeners
         it = _listeners.iterator();
         while (it.hasNext()) {
             SmusherListener nextListener = (SmusherListener) it.next();
             nextListener.finish();
         }
         
     }
 }