/*
    * 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.scutter;
  
  import java.io.BufferedReader;
  import java.io.BufferedWriter;
  import java.io.File;
  import java.io.FileReader;
  import java.io.FileWriter;
  import java.io.IOException;
  import java.io.PrintWriter;
  import java.net.MalformedURLException;
  import java.net.URL;
  import java.util.ArrayList;
  import java.util.HashSet;
  import java.util.Hashtable;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Set;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.ThreadPoolExecutor;
  import java.util.concurrent.TimeUnit;
  import java.util.regex.Pattern;
  
  import org.openrdf.elmo.scutter.Util;
  import org.openrdf.model.Resource;
  import org.openrdf.model.Value;
  import org.openrdf.query.BindingSet;
  import org.openrdf.query.MalformedQueryException;
  import org.openrdf.query.QueryEvaluationException;
  import org.openrdf.query.TupleQueryResult;
  import org.openrdf.query.QueryLanguage;
  
  import org.openrdf.repository.Repository;
  import org.openrdf.repository.RepositoryConnection;
  import org.openrdf.repository.RepositoryException;
  import org.openrdf.repository.RepositoryResult;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Scutter is the main class of the RDF crawler (scutter). It can be invoked
   * from the command line or built into applications. Scutter implements Runnable
   * so that it can be run as a background thread.
   * 
   * @author Peter Mika (original version for Jena by Matt Biddulph.)
   * @version $Revision: 1.20 $
   */
  
  public class Scutter implements Runnable {
  
  	List<URL> _urls = new ArrayList<URL>(); // the queue of URLs to be visited
  
  	HashSet<URL> _visited = new HashSet<URL>(); // the URLs visited so far
  
  	private RetrieverFactory _factory = null; // the repository where the
  
  	// collected data should be
  	// added
  
  	private boolean _stop = false; // flag to indicate that the crawler should
  
  	// be stopped
  
  	private final static String VISITED_QUERY = "SELECT DISTINCT url from {x} socionet:provenance {url} "
  			+ "using namespace "
  			+ "socionet = <http://www.cs.vu.nl/~pmika/socionet#>";
  
  	/**
  	 * @see EDU.oswego.cs.dl.util.concurrent.BoundedLinkedQueue
  	 */
  	public final static int LINKEDQUEUE_SIZE = 1000; // see
 
 	/**
 	 * @see EDU.oswego.cs.dl.util.concurrent.PooledExecutor
 	 */
 	public final static int DEFAULT_MAXTHREADS = 20; // see
 
 	/**
 	 * Time to wait for Scutter to finish after stopped (in sec). before saving
 	 * queue to disk. Time to wait is dependent on the size of the thread pool.
 	 */
 	public final static int STOP_TIME = 1 * 60;
 
 	/**
 	 * Threshold for the number of URLs seen from a domain before it becomes
 	 * blacklisted
 	 */
 	private final static int BLACKLIST_THRESHOLD = 500; // if there are more
 
 	/**
 	 * Stores how many URLs we have visited so far from a domain
 	 * 
 	 */
 	private Hashtable<String, Integer> _domains = new Hashtable<String, Integer>();
 
 	private final Logger _logger = LoggerFactory.getLogger(Scutter.class);
 
 	/**
 	 * Defines the whitelist (domains that are allowed even when on blacklist)
 	 */
 	private Pattern _domainPattern = null;
 
 	/**
 	 * Defines whether we store metadata with statements
 	 */
 	private boolean _metadata = false; //
 
 	/**
 	 * Enables automatic blacklisting
 	 */
 	private boolean _autoBlackList = true;
 
 	/**
 	 * Size limit (in KB) for files to be parsed. Files larger will not be
 	 * loaded.
 	 */
 	public final static int DEFAULT_SIZE_LIMIT = 1000; // in kilobytes
 
 	/**
 	 * @see EDU.oswego.cs.dl.util.concurrent.PooledExecutor
 	 */
 	private int _sizelimit = DEFAULT_SIZE_LIMIT;
 
 	/**
 	 * @see EDU.oswego.cs.dl.util.concurrent.PooledExecutor
 	 */
 	private int _maxThreads = DEFAULT_MAXTHREADS;
 
 	private File _queueFile;
 
 	private File _blacklistFile;
 	
 	private Resource[] _context;
 
 	/**
 	 * Scutter command-line tool.
 	 * 
 	 * @param args
 	 *            First argument is the URL of Sesame server, second argument is
 	 *            the repository name. Remaining arguments are interpreted as
 	 *            URLs and are used to initialize the queue.
 	 */
 	public final static void main(String[] args) throws Exception {
 		// Log4j configuration
 
 		// BasicConfigurator.configure();
 		// Logger.getRootLogger().setLevel(Level.INFO);
 
 		Scutter s = null;
 		try {
 			if (args.length < 3) {
 				throw new Exception("Usage: java -jar elmo-scutter.jar <server> <repository> <starturl1> <starturl2> ...");
 			}
 			String[] urls = new String[args.length - 2];
 			for (int i = 2; i < args.length; i++) {
 				urls[i - 2] = args[i];
 			}
 			Repository repository = Util.initRepository(args[0], args[1]);
 
 			// Create factory for simple retrievers
 			RetrieverFactory factory = new RetrieverFactory(repository, false);
 			s = new Scutter(factory);
 			s.initQueue(urls);
 		} catch (Exception e) {
 			e.printStackTrace();
 			System.exit(1);
 		}
 		s.run();
 	}
 
 	/**
 	 * Add a list of URLs to the queue
 	 * 
 	 * @param urls
 	 *            Array of strings (URLs)
 	 */
 
 	public void initQueue(String[] urls) {
 		for (int i = 0; i < urls.length; i++) {
 			try {
 				System.err.println("considering: " + urls[i]);
 				if (urls[i] != null) {
 					addURL(new URL(urls[i].trim()));
 				}
 			} catch (MalformedURLException e) {
 				_logger.error(e.toString());
 			}
 		}
 	}
 
 	/**
 	 * Get the queue. Beware: may mutate while the scutter is running.
 	 * 
 	 * @return List of URL objects
 	 */
 	public List getQueue() {
 		return _urls;
 	}
 
 	/**
 	 * Get the set of URLs visited so far. Beware: may mutate while the scutter
 	 * is running.
 	 * 
 	 * @return Set of URL objects
 	 */
 	public Set getVisited() {
 		return _visited;
 	}
 
 	/**
 	 * Clear the queue and visited lists.
 	 * 
 	 * @return false if clear fails
 	 */
 	public boolean clear() {
 		try {
 			_urls.clear();
 			_visited.clear();
 			_domains.clear();
 			_logger.info("Queue and visited lists are cleared");
 			return true;
 		} catch (Exception e) {
 			_logger.error(e.toString());
 			return false;
 		}
 	}
 
 	protected void addVisited(URL url) {
 		_visited.add(url);
 
 		String host = Scutter.getHost(url);
 		synchronized (_domains) {
 			if (_domains.containsKey(host)) {
 				int currentCount = ((Integer) _domains.get(host)).intValue();
 				_domains.put(host, new Integer(currentCount + 1));
 			} else {
 				_domains.put(host, new Integer(1));
 			}
 		}
 	}
 
 	/**
 	 * Load the URLs of sources visited so far from the repository.
 	 * 
 	 * @throws RepositoryException
 	 * @throws RepositoryException
 	 * @throws QueryEvaluationException
 	 * @throws QueryEvaluationException
 	 * @throws MalformedQueryException
 	 * 
 	 */
 	protected void loadVisited() {
 		//Visited URLs are simply the set of the contexts in which statements are placed
 		try {
 			RepositoryConnection con = _factory.getTarget().getConnection();
 			RepositoryResult<Resource> results = null; 
 			 
 			try {
 				results = con.getContextIDs(); 
 			   
 				while (results.hasNext()) {
 					Resource nextContext = results.next();
 					try {
 						addVisited(new URL(nextContext.toString()));
 					} catch (MalformedURLException mue) {
 						// ignore
 					}
 					_logger.info("Visited URLs reloaded.");
 				}
 			} finally {
 				if (results != null) results.close();
 				con.close();
 			}
 		} catch (Exception e) {
 			_logger.error("Reloading visited URLs failed");
 			_logger.error(e.getMessage(), e);
 		}
 	}
 
 	/**
 	 * Add URLs to the queue from a file on the disk. Format is one URL per
 	 * line.
 	 * 
 	 * @return number of URLs loaded
 	 */
 	public int loadQueue() {
 		int result = 0;
 
 		BufferedReader in = null;
 		try {
 			in = new BufferedReader(new FileReader(_queueFile));
 			String currentLine = in.readLine();
 
 			while (currentLine != null) {
 				try {
 					addURL(new URL(currentLine.trim()));
 					result++;
 				} catch (MalformedURLException mue) {
 					// error in the queue file
 				}
 
 				currentLine = in.readLine();
 			}
 
 			_logger.info("Queue reloaded from file.");
 		} catch (IOException ioe) {
 			_logger.error("Loading queue failed.");
 		} finally {
 			if (in != null)
 				try {
 					in.close();
 				} catch (IOException e) {
 				}
 		}
 
 		return result;
 	}
 
 	/**
 	 * Save the status of the queue to the disk. Format is one URL per line.
 	 * 
 	 */
 	public void saveQueue() {
 		PrintWriter out = null;
 		try {
 			if (_queueFile != null) {
 				_queueFile.delete();
 				// returns false if the file already exists
 				_queueFile.createNewFile();
 			}
 			out = new PrintWriter(
 					new BufferedWriter(new FileWriter(_queueFile)));
 			synchronized (_urls) {
 				for (int i = 0; i < _urls.size(); i++) {
 					out.println(_urls.get(i));
 				}
 			}
 
 			out.flush();
 
 			_logger.info("Queue saved to file.");
 		} catch (IOException ioe) {
 			_logger
 					.error("Creating or writing queue file failed, check servlet configuration.");
 			// temp file is null
 			_queueFile = null;
 		} finally {
 			if (out != null)
 				out.close();
 		}
 	}
 
 	/**
 	 * Add prefixes to the blacklist from a file on the disk. Format is one
 	 * prefix per line.
 	 * 
 	 * @return number of URLs loaded
 	 */
 	public int loadBlacklist() {
 		int result = 0;
 
 		BufferedReader in = null;
 		try {
 			in = new BufferedReader(new FileReader(_blacklistFile));
 			String currentLine = in.readLine();
 
 			while (currentLine != null) {
 				synchronized (_domains) {
 					_domains.put(currentLine.trim(), new Integer(
 							BLACKLIST_THRESHOLD + 1));
 				}
 				result++;
 				currentLine = in.readLine();
 			}
 
 			_logger.info("Blacklist reloaded from file.");
 		} catch (IOException ioe) {
 			_logger.error("Loading blacklist failed.");
 		} finally {
 			if (in != null)
 				try {
 					in.close();
 				} catch (IOException e) {
 
 				}
 		}
 
 		return result;
 	}
 
 	/**
 	 * Save the status of the blacklist to the disk. Format is one prefix per
 	 * line.
 	 * 
 	 * @throws IOException
 	 */
 	public void saveBlacklist() throws IOException {
 		PrintWriter out = null;
 		try {
 			if (_blacklistFile != null) {
 				_blacklistFile.delete();
 				// returns false if the file already exists
 				_blacklistFile.createNewFile();
 			}
 			out = new PrintWriter(new BufferedWriter(new FileWriter(
 					_blacklistFile)));
 			Iterator it = _domains.keySet().iterator();
 			while (it.hasNext()) {
 				String host = (String) it.next();
 				if (((Integer) _domains.get(host)).intValue() > BLACKLIST_THRESHOLD) {
 					out.println(host);
 				}
 			}
 
 			out.flush();
 
 			_logger.info("Blacklist saved to file.");
 		} catch (IOException ioe) {
 			_logger
 					.error("Creating or writing blacklist file failed, check servlet configuration.");
 			// temp file is null
 			_blacklistFile = null;
 		} finally {
 			if (out != null)
 				out.close();
 		}
 	}
 
 	/**
 	 * Stop the scutter. The thread will sleep for a specified time to allow the
 	 * scutter to finish.
 	 * 
 	 * @throws IOException
 	 */
 	public void stop() throws IOException {
 		_stop = true;
 		// wait a bit for it to finish
 		try {
 			Thread.sleep(STOP_TIME);
 		} catch (InterruptedException ie) {
 		}
 		if (_queueFile != null)
 			saveQueue();
 		if (_blacklistFile != null)
 			saveBlacklist();
 
 		_logger.info("Scutter stopped.");
 	}
 
 	/**
 	 * Sets a pattern (regular expression) for limiting crawling to those URLs
 	 * that match this pattern. In other words, this pattern defines the
 	 * whitelist.
 	 * 
 	 * @param p
 	 *            Pattern to use
 	 */
 	public void setDomainPattern(Pattern p) {
 		if (p != null) {
 			_domainPattern = p;
 		}
 	}
 
 	/**
 	 * Return the pattern used for the whitelist.
 	 * 
 	 * @return Pattern
 	 */
 	public Pattern getDomainPattern() {
 		return _domainPattern;
 	}
 
 	/**
 	 * Set whether the scutter should produce metadata.
 	 * 
 	 * @param metadata
 	 */
 	public void setStoreMetadata(boolean metadata) {
 		_metadata = metadata;
 	}
 
 	/**
 	 * Determine whether the scutter is set to produce metadata.
 	 * 
 	 * @return Flag indicating if the metadata feature is turned on
 	 */
 	public boolean getStoreMetadata() {
 		return _metadata;
 	}
 
 	/**
 	 * Set whether the scutter should automatically put sites on blacklist after
 	 * a number of profiles has been collected from that site.
 	 * 
 	 * @param metadata
 	 */
 	public void setAutoBlackList(boolean blacklist) {
 		_autoBlackList = blacklist;
 	}
 
 	/**
 	 * Determine whether automatic blacklisting is enabled.
 	 * 
 	 * @return Flag indicating if the metadata feature is turned on
 	 */
 	public boolean getAutoBlackList() {
 		return _autoBlackList;
 	}
 
 	/**
 	 * Set the size limit for files to be loaded
 	 * 
 	 * @param sizelimit
 	 */
 	public void setSizeLimit(int sizelimit) {
 		_sizelimit = sizelimit;
 	}
 
 	/**
 	 * 
 	 * 
 	 * @return Size limit for files to be loaded
 	 */
 	public int getSizeLimit() {
 		return _sizelimit;
 	}
 
 	/**
 	 * @return Returns the maxThreads.
 	 */
 	public int getMaxThreads() {
 		return _maxThreads;
 	}
 
 	/**
 	 * @param threads
 	 *            The maxThreads to set.
 	 */
 	public void setMaxThreads(int threads) {
 		_maxThreads = threads;
 	}
 
 	/**
 	 * Create a new scutter.
 	 * 
 	 * @param repository
 	 *            Sesame repository to be used for storing the data
 	 * 
 	 * @throws Exception
 	 */
 	public Scutter(RetrieverFactory factory) throws Exception {
 		_factory = factory;
 
 		loadVisited();
 	}
 
 	private void shutdownAndAwaitTermination(ExecutorService pool) {
 		pool.shutdown(); // Disable new tasks from being submitted
 		try {
 			// Wait a while for existing tasks to terminate
 			if (!pool.awaitTermination(STOP_TIME, TimeUnit.SECONDS)) {
 				pool.shutdownNow(); // Cancel currently executing tasks
 				// Wait a while for tasks to respond to being cancelled
 				if (!pool.awaitTermination(STOP_TIME, TimeUnit.SECONDS))
 					System.err.println("Pool did not terminate");
 			}
 		} catch (InterruptedException ie) {
 			// (Re-)Cancel if current thread also interrupted
 			pool.shutdownNow();
 			// Preserve interrupt status
 			Thread.currentThread().interrupt();
 		}
 	}
 
 	/**
 	 * Main method that loops infinitely or until the scutter is stopped.
 	 * 
 	 */
 	public void run() {
 		_logger.info("Scutter started.");
 
 		if (_queueFile != null)
 			loadQueue();
 		if (_blacklistFile != null)
 			loadBlacklist();
 
 		_stop = false;
 
 		ExecutorService executor = Executors.newFixedThreadPool(_maxThreads);
 
 		while (!_stop) {
 
 			if (_urls.size() > 0) {
 
 				Retriever r = null;
 				URL url = null;
 				synchronized (_urls) {
 					// Need to instantiate new object in case this thread
 					// modifies the value of url
 					try {
 						url = new URL(((URL) _urls.remove(0)).toString());
 					} catch (MalformedURLException mue) {
 						_logger.warn("Malformed URL: " + url.toString());
 					}
 					/*
 					 * //Check Content-Length try { if
 					 * (Util.getContentLength(url.toString()) > _sizelimit *
 					 * 1000) { _logger.info("File is too large: " + url);
 					 * continue; } } catch (HttpException e) {
 					 * _logger.warn("Checking content length failed for " +
 					 * url); } catch (IOException ioe) { _logger.warn("Checking
 					 * content length failed for " + url); } catch
 					 * (RuntimeException rte) { _logger.warn("Checking content
 					 * length failed for " + url); }
 					 */
 
 				}
 				_logger.info("Putting " + url
 						+ " into the execute queue, queue size now "
 						+ _urls.size());
 
 				r = _factory.getRetriever(url, this);
 
 				executor.execute(r);
 				_logger.debug("Pool size is "
 						+ ((ThreadPoolExecutor) executor).getPoolSize()
 						+ ", queue size is "
 						+ ((ThreadPoolExecutor) executor).getQueue().size());
 
 			}
 		}
 
 		shutdownAndAwaitTermination(executor);
 		_logger.info("Scutter task exiting.");
 	}
 
 	/**
 	 * Add a single URL to the queue.
 	 * 
 	 * @param url
 	 */
 	public void addURL(URL url) {
 		synchronized (_urls) {
 			if (_visited.contains(url)) {
 				_logger.debug("Already visited : " + url);
 			} else if (inBlacklist(url)) {
 				_logger.info("Blacklisted domain : " + url);
 			} else if (_domainPattern != null
 					&& !_domainPattern.matcher(url.toString()).matches()
 					&& !url.getHost().equalsIgnoreCase("localhost")) {
 				_logger.info("Not on whitelist : " + url);
 			} else {
 				_urls.add(url);
 				addVisited(url);
 			}
 		}
 	}
 
 	/**
 	 * Check if a URL is on the blacklist.
 	 * 
 	 * @param url
 	 * @return
 	 */
 	protected boolean inBlacklist(URL url) {
 
 		String host = Scutter.getHost(url);
 		if (_autoBlackList
 				&& _domains.containsKey(host)
 				&& ((Integer) _domains.get(host)).intValue() > BLACKLIST_THRESHOLD) {
 			return true;
 		} else {
 			return false;
 		}
 
 	}
 
 	private static String getHost(URL url) {
 		String host = url.getHost();
 		/*
 		 * Doesn't work for addresses such as example.co.uk //Get upper two
 		 * levels of domain name String[] parts = host.split("\\."); if
 		 * (parts.length > 1) { return parts[parts.length-2] + '.' +
 		 * parts[parts.length-1]; } else { return host; }
 		 */
 		return host;
 
 	}
 
 	/**
 	 * @return Returns the blacklistFile.
 	 */
 	public File getBlacklistFile() {
 		return _blacklistFile;
 	}
 
 	/**
 	 * @param blacklistFile
 	 *            The blacklistFile to set.
 	 */
 	public void setBlacklistFile(File blacklistFile) {
 		_blacklistFile = blacklistFile;
 	}
 
 	/**
 	 * @return Returns the queueFile.
 	 */
 	public File getQueueFile() {
 		return _queueFile;
 	}
 
 	/**
 	 * @param queueFile
 	 *            The queueFile to set.
 	 */
 	public void setQueueFile(File queueFile) {
 		_queueFile = queueFile;
 	}
 
 	public void setContext(Resource[] context) {
 		// TODO Auto-generated method stub
 		_context = context;
 	}
 	public Resource[] getContext() {
 		// TODO Auto-generated method stub
 		return _context;
 	}
 	
 	
 }