/*
    * 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 info.aduna.iteration.Iterations;
  
  import java.io.IOException;
  import java.net.MalformedURLException;
  import java.net.URL;
  import java.util.Collection;
  import java.util.Date;
  import java.util.List;
  
  import org.apache.commons.httpclient.HttpException;
  import org.openrdf.OpenRDFException;
  import org.openrdf.model.Resource;
  import org.openrdf.model.Statement;
  import org.openrdf.model.URI;
  import org.openrdf.model.Value;
  import org.openrdf.model.vocabulary.RDFS;
  import org.openrdf.repository.Repository;
  import org.openrdf.repository.RepositoryConnection;
  import org.openrdf.repository.RepositoryException;
  import org.openrdf.repository.RepositoryResult;
  import org.openrdf.repository.sail.SailRepository;
  import org.openrdf.rio.ParseErrorListener;
  import org.openrdf.rio.RDFHandlerException;
  import org.openrdf.rio.RDFParseException;
  import org.openrdf.rio.RDFParser;
  import org.openrdf.rio.helpers.StatementCollector;
  import org.openrdf.rio.rdfxml.RDFXMLParser;
  import org.openrdf.sail.memory.MemoryStore;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * A SimpleRetriever is instantiated by the Scutter class for every URL to be
   * visited. The Retriver fetches that document and adds the allowed statements
   * to an in-memory repository. Statements can be optionally filtered by specying
   * a StatementFilter. The seeAlso statements found in the document are added to
   * the scutter queue.
   * 
   * @author Peter Mika (original version for Jena by Matt Biddulph.)
   * @version $Revision: 1.8 $
   */
  
  public class SimpleRetriever implements Retriever {
  
  	protected URL _url;
  
  	protected Scutter _scutter;
  
  	protected Repository _repository = null;
  
  	protected boolean _ok = true;
  
  	protected final static Logger _logger = LoggerFactory
  			.getLogger(SimpleRetriever.class);
  
  	// NOTE: we can make parser static, but only if the parser itself is
  	// threadsafe
  	// and that is doubtful: there is only one StatementHandler inside and that
  	// should be different for every URL (content goes into separate
  	// repositories)
  	protected RDFParser _parser = null;
  
  	protected StatementFilter _filter = null;
  
  	protected DocumentHandler _handler = null;
  
  	public SimpleRetriever(final URL url, final Repository repository,
  			final Scutter scutter) {
  		setUrl(url);
 		setRepository(repository);
 		setScutter(scutter);
 		setFilter(new DefaultStatementFilter());
 		// Use the SAX2-compliant Xerces parser:
 		System.setProperty("org.xml.sax.driver",
 				"org.apache.xerces.parsers.SAXParser");
 		_parser = new RDFXMLParser(repository.getValueFactory());
 		_parser.setVerifyData(true);
 		_parser.setStopAtFirstError(false);
 
 		_handler = new SimpleDocumentHandler(_filter);
 	}
 
 	static class DefaultStatementFilter implements StatementFilter {
 		public boolean allowStatement(Resource subject, URI predicate,
 				Value object) {
 			/*
 			 * Example: store just RDF, RDF-S, FOAF and DC triples. Ignore
 			 * statements about resources in the SOCIONET namespace.
 			 * 
 			 * if
 			 * ((predicate.toString().startsWith(flink.util.Constants.FOAF_NS) ||
 			 * predicate.toString().startsWith(flink.util.Constants.WGS84_POS_NS) ||
 			 * predicate.toString().startsWith(flink.util.Constants.RDF_NS) ||
 			 * predicate.toString().startsWith(flink.util.Constants.RDFS_NS)) &&
 			 * !subject.toString().startsWith(flink.util.Constants.SOCIONET_NS)) {
 			 * 
 			 * return true; } else { return false; }
 			 * 
 			 */
 			return true;
 		}
 	}
 
 	class SimpleDocumentHandler extends StatementCollector implements
 			DocumentHandler {
 
 		public final static int DEFAULT_MAXSIZE = 10000;
 
 		// Interrupt execution after this many statements have been processed
 		protected final int _maxSize = DEFAULT_MAXSIZE;
 
 		protected int _stmtCount = 0;
 
 		protected StatementFilter _filter = new DefaultStatementFilter();
 
 		public SimpleDocumentHandler(StatementFilter filter) {
 			if (filter != null) {
 				_filter = filter;
 			}
 		}
 
 		public void parse(URL url) {
 
 			// Parse the data from inputStream,
 			// resolving any relative URIs against the URL of the source
 			// Set statement and error handler
 			_parser.setParseErrorListener(new RetrieverErrorListener());
 			_parser.setRDFHandler(this);
 			try {
 				_parser.parse(Util.getDocumentAsInputStream(url.toString()),
 						url.toString());
 			} catch (RDFParseException e) {
 				_logger.error(e.getMessage(), e);
 			} catch (RDFHandlerException e) {
 				_logger.error(e.getMessage(), e);
 			} catch (HttpException e) {
 				_logger.error(e.getMessage(), e);
 			} catch (IOException e) {
 				_logger.error(e.getMessage(), e);
 			}
 
 		}
 
 		public void handleStatement(Statement stmt) {
 			// Stop parsing in a brutal way if we exceeded the limit
 			if (_stmtCount++ > _maxSize) {
 				throw new RuntimeException("Document longer than " + _maxSize
 						+ " statements, interrupting parsing.");
 			}
 			if (_filter.allowStatement(stmt.getSubject(), stmt.getPredicate(),
 					stmt.getObject())) {
 
 				super.handleStatement(stmt);
 
 			}
 		}
 
 		public boolean followLinks() {
 			return true;
 		}
 
 		public boolean aggregateContent() {
 			return true;
 		}
 
 		public Collection<Statement> getCollectedStatements() {
 			return getStatements();
 
 		}
 
 	}
 
 	class RetrieverErrorListener implements ParseErrorListener {
 
 		public void error(java.lang.String msg, int lineNo, int colNo) {
 			_logger.error(msg);
 			_ok = false;
 		}
 
 		public void fatalError(java.lang.String msg, int lineNo, int colNo) {
 			_logger.error(msg);
 			_ok = false;
 		}
 
 		public void warning(java.lang.String msg, int lineNo, int colNo) {
 
 		}
 	}
 
 	public void setRepository(Repository r) {
 		_repository = r;
 	}
 
 	public void setScutter(Scutter s) {
 		_scutter = s;
 	}
 
 	public void setUrl(URL u) {
 		_url = u;
 	}
 
 	public void setFilter(StatementFilter f) {
 		_filter = f;
 	}
 
 	public void run() {
 		Repository temp = null;
 		try {
 			_logger.debug("Parsing " + _url);
 
 			_handler.parse(_url);
 
 			// Create new non-inferencing in-memory repository
 			temp = new SailRepository(new MemoryStore());
 			temp.initialize();
 
 			// Add the collected statements to the temporary repository
 			RepositoryConnection tempCon = null;
 			try {
 				tempCon = temp.getConnection();
 				tempCon.add(_handler.getCollectedStatements());
 			} finally {
 				tempCon.close();
 			}
 
 			if (_ok) {
 				_logger.info("Finished retrieving and parsing.");
 
 				if (!Thread.currentThread().isInterrupted()) {
 					if (_handler.followLinks()) {
 						// Add seeAlso links to the scutter queue
 						addSeeAlsos(temp);
 					} else {
 						_logger.debug("Ignoring links in " + _url);
 					}
 					if (_handler.aggregateContent()) {
 						// Add the statements to the repository
 						aggregate(temp, _url.toString());
 					} else {
 						_logger.debug("Ignoring content in " + _url);
 					}
 				}
 			} else {
 				_logger.warn("Errors during parsing, not adding to the model.");
 			}
 		} catch (Throwable t) {
 			_logger.warn("Errors loading " + _url, t);
 		} finally {
 			temp = null;
 		}
 	}
 
 	protected void aggregate(Repository incoming, String url)
 			throws RepositoryException {
 		Date date = new Date();
 		int count = 0;
 
 		// There is only a single aggregator and we synchronize on that
 		// so we shouldn't need transactions, right?
 		RepositoryConnection con = null;
 		try {
 			con = incoming.getConnection();
 			RepositoryResult<Statement> stmts = con.getStatements(null, null,
 					null, true);
 
 			List<Statement> stmtList = Iterations.asList(stmts);
 
 			try {
 
 				for (Statement nextStatement : stmtList) {
 					
 					if (_logger.isDebugEnabled()) {
 						if (nextStatement.getPredicate().getLocalName().equals("name")) {
 							_logger
 									.info("Found name: "
 											+ nextStatement.getObject());
 						}
 					}
 					// TODO: store metadata in context
 					// if (_scutter.getMetadata()) {
 					// Util.writeReificationStatement(rdfxml, nextStatement,
 					// url,
 					// date);
 					// rdfxml.writeStatement(nextStatement.getSubject(),
 					// nextStatement.getPredicate(), nextStatement
 					// .getObject());
 					// count += 7;
 					// }
 					count++;
 
 					// System.err.println("*** Document from " + url + "
 					// starts here.");
 					// System.err.println(writer.toString());
 					// System.err.println("*** Document from " + url + "
 					// ends here");
 					// Upload RDF document to Sesame
 				}
 			} finally {
 				// TODO: do I need to close this?
 				stmts.close(); // make sure the result object is closed
 				// properly
 			}
 
 			if (count > 0) {
 				con = null;
 				try {
 					con = _repository.getConnection();
 					if (_scutter.getStoreMetadata()) {
 						con.add(stmtList, _scutter.getContext());
 					} else {
 						con.add(stmtList);		
 					}
 					_logger.debug("Added " + count + " triples to model.");
 				} catch (OpenRDFException e) {
 					_logger.error("Uploading document from " + url
 							+ " produced errors.");
 				} finally {
 					if (con != null)
 						con.close();
 				}
 
 			} else {
 				_logger.debug("No statements in document " + url);
 			}
 		} catch (OpenRDFException e) {
 			_logger.error(e.getMessage(), e);
 		} finally {
 			if (con != null)
 				con.close();
 		}
 	}
 
 	/**
 	 * Add all rdfs:seeAlso statements to the queue from the specified
 	 * repository.
 	 * 
 	 * @param incoming
 	 * @throws RepositoryException
 	 */
 	public void addSeeAlsos(Repository incoming) throws RepositoryException {
 		RepositoryConnection con = null;
 
 		try {
 			con = incoming.getConnection();
 			// Get statements with the rdfs:seeAlso predicate
 			RepositoryResult<Statement> stmts = con.getStatements(null,
 					RDFS.SEEALSO, null, true);
 
 			try {
 				while (stmts.hasNext()) {
 					Statement nextStatement = stmts.next();
 					// should be a safe cast...
 					URI object = (URI) nextStatement.getObject();
 					try {
 						_scutter.addURL(new URL(object.toString()));
 					} catch (MalformedURLException e) {
 						// shouldn't really happen...
 						_logger.warn("Bad URL: " + object.toString());
 					}
 				}
 			} finally {
 				if (stmts != null)
 					stmts.close();
 			}
 		} catch (OpenRDFException e) {
 			_logger.error(e.getMessage(), e);
 		} catch (ClassCastException e) {
 			_logger.error(e.getMessage(), e);
 		} finally {
 			if (con != null)
 				con.close();
 		}
 
 	}
 
 }