/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.nutch.util;

import java.io.IOException;
import java.net.URL;
import java.text.SimpleDateFormat;
import java.util.Collection;
import java.util.List;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.map.MultithreadedMapper;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MapFileOutputFormat;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import org.apache.nutch.crawl.CrawlDatum;
import org.apache.nutch.hostdb.HostDatum;
import org.apache.nutch.net.URLFilters;
import org.apache.nutch.net.URLNormalizers;
import org.apache.nutch.protocol.Content;
import org.apache.nutch.protocol.Protocol;
import org.apache.nutch.protocol.ProtocolFactory;
import org.apache.nutch.protocol.ProtocolOutput;
import org.apache.nutch.protocol.ProtocolStatus;
import org.apache.nutch.util.NutchJob;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import crawlercommons.robots.BaseRobotRules;
import crawlercommons.sitemaps.AbstractSiteMap;
import crawlercommons.sitemaps.SiteMap;
import crawlercommons.sitemaps.SiteMapIndex;
import crawlercommons.sitemaps.SiteMapParser;
import crawlercommons.sitemaps.SiteMapURL;

/**
 * <p>Performs Sitemap processing by fetching sitemap links, parsing the content and merging
 * the urls from Sitemap (with the metadata) with the existing crawldb.</p>
 *
 * <p>There are two use cases supported in Nutch's Sitemap processing:</p>
 * <ol>
 *  <li>Sitemaps are considered as "remote seed lists". Crawl administrators can prepare a
 *     list of sitemap links and get only those sitemap pages. This suits well for targeted
 *     crawl of specific hosts.</li>
 *  <li>For open web crawl, it is not possible to track each host and get the sitemap links
 *     manually. Nutch would automatically get the sitemaps for all the hosts seen in the
 *     crawls and inject the urls from sitemap to the crawldb.</li>
 * </ol>
 *
 * <p>For more details see:
 *      https://wiki.apache.org/nutch/SitemapFeature </p>
 */
public class SitemapProcessor extends Configured implements Tool {
  public static final Logger LOG = LoggerFactory.getLogger(SitemapProcessor.class);
  public static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

  public static final String CURRENT_NAME = "current";
  public static final String LOCK_NAME = ".locked";
  public static final String SITEMAP_STRICT_PARSING = "sitemap.strict.parsing";
  public static final String SITEMAP_URL_FILTERING = "sitemap.url.filter";
  public static final String SITEMAP_URL_NORMALIZING = "sitemap.url.normalize";
  public static final String SITEMAP_ALWAYS_TRY_SITEMAPXML_ON_ROOT = "sitemap.url.default.sitemap.xml";
  public static final String SITEMAP_OVERWRITE_EXISTING = "sitemap.url.overwrite.existing";
  public static final String SITEMAP_REDIR_MAX = "sitemap.redir.max";
  public static final String SITEMAP_SIZE_MAX = "sitemap.size.max";

  private static class SitemapMapper extends Mapper<Text, Writable, Text, CrawlDatum> {
    private ProtocolFactory protocolFactory = null;
    private boolean strict = true;
    private boolean filter = true;
    private boolean normalize = true;
    private boolean tryDefaultSitemapXml = true;
    private int maxRedir = 3;
    private URLFilters filters = null;
    private URLNormalizers normalizers = null;
    private CrawlDatum datum = new CrawlDatum();
    private SiteMapParser parser = null;

    public void setup(Context context) {
      Configuration conf = context.getConfiguration();
      int maxSize = conf.getInt(SITEMAP_SIZE_MAX, SiteMapParser.MAX_BYTES_ALLOWED);
      conf.setInt("http.content.limit", maxSize);
      conf.setInt("file.content.limit", maxSize);
      this.protocolFactory = new ProtocolFactory(conf);
      this.filter = conf.getBoolean(SITEMAP_URL_FILTERING, true);
      this.normalize = conf.getBoolean(SITEMAP_URL_NORMALIZING, true);
      this.strict = conf.getBoolean(SITEMAP_STRICT_PARSING, true);
      this.tryDefaultSitemapXml = conf.getBoolean(SITEMAP_ALWAYS_TRY_SITEMAPXML_ON_ROOT, true);
      this.maxRedir = conf.getInt(SITEMAP_REDIR_MAX, 3);
      this.parser = new SiteMapParser(strict);

      if (filter) {
        filters = new URLFilters(conf);
      }
      if (normalize) {
        normalizers = new URLNormalizers(conf, URLNormalizers.SCOPE_DEFAULT);
      }
    }

    public void map(Text key, Writable value, Context context) throws IOException, InterruptedException {
      String url;

      try {
        if (value instanceof CrawlDatum) {
          // If its an entry from CrawlDb, emit it. It will be merged in the reducer
          context.write(key, (CrawlDatum) value);
        }
        else if (value instanceof HostDatum) {
          generateSitemapsFromHostname(key.toString(), context);
        }
        else if (value instanceof Text) {
          // Input can be sitemap URL or hostname
          url = key.toString();
          if (url.startsWith("http://") ||
                url.startsWith("https://") ||
                url.startsWith("ftp://") ||
                url.startsWith("file:/")) {
            // For entry from sitemap urls file, fetch the sitemap, extract urls and emit those
            if((url = filterNormalize(url)) == null) {
              context.getCounter("Sitemap", "filtered_records").increment(1);
              return;
            }

            context.getCounter("Sitemap", "sitemap_seeds").increment(1);
            generateSitemapUrlDatum(protocolFactory.getProtocol(url), url, context); 
          } else {
            LOG.info("generateSitemapsFromHostname: " + key.toString());
            generateSitemapsFromHostname(key.toString(), context);
          }
        }
      } catch (Exception e) {
        LOG.warn("Exception for record {} : {}", key.toString(), StringUtils.stringifyException(e));
      }
    }

    /* Filters and or normalizes the input URL */
    private String filterNormalize(String url) {
      try {
        if (normalizers != null)
          url = normalizers.normalize(url, URLNormalizers.SCOPE_DEFAULT);

        if (filters != null)
          url = filters.filter(url);
      } catch (Exception e) {
        return null;
      }
      return url;
    }
    
    private void generateSitemapsFromHostname(String host, Context context) {
      try {
        // For entry from hostdb, get sitemap url(s) from robots.txt, fetch the sitemap,
        // extract urls and emit those

        // try different combinations of schemes one by one till we get rejection in all cases
        String url;
        if((url = filterNormalize("http://" + host + "/")) == null &&
            (url = filterNormalize("https://" + host + "/")) == null &&
            (url = filterNormalize("ftp://" + host + "/")) == null &&
            (url = filterNormalize("file:/" + host + "/")) == null) {
          context.getCounter("Sitemap", "filtered_records").increment(1);
          return;
        }
        // We may wish to use the robots.txt content as the third parameter for .getRobotRules
        BaseRobotRules rules = protocolFactory.getProtocol(url).getRobotRules(new Text(url), datum, null);
        List<String> sitemaps = rules.getSitemaps();

        if (tryDefaultSitemapXml && sitemaps.size() == 0) {
          sitemaps.add(url + "sitemap.xml");
        }
        for (String sitemap : sitemaps) {
          context.getCounter("Sitemap", "sitemaps_from_hostname").increment(1);
          sitemap = filterNormalize(sitemap);
          if (sitemap == null) {
            context.getCounter("Sitemap", "filtered_sitemaps_from_hostname")
                .increment(1);
          } else {
            generateSitemapUrlDatum(protocolFactory.getProtocol(sitemap),
                sitemap, context);
          }
        }
      } catch (Exception e) {
        LOG.warn("Exception for record {} : {}", host, StringUtils.stringifyException(e));
      }
    }

    private void generateSitemapUrlDatum(Protocol protocol, String url, Context context) throws Exception {
      ProtocolOutput output = protocol.getProtocolOutput(new Text(url), datum);
      ProtocolStatus status = output.getStatus();
      Content content = output.getContent();

      // Following redirects http > https and what else
      int maxRedir = this.maxRedir;
      while (!output.getStatus().isSuccess() && output.getStatus().isRedirect() && maxRedir > 0) {
        String[] stuff = output.getStatus().getArgs();
        url = filterNormalize(stuff[0]);
        
        // get out!
        if (url == null) {
          break;
        }
        output = protocol.getProtocolOutput(new Text(url), datum);
        status = output.getStatus();
        content = output.getContent();
        
        maxRedir--;
      }

      if(status.getCode() != ProtocolStatus.SUCCESS) {
        // If there were any problems fetching the sitemap, log the error and let it go. Not sure how often
        // sitemaps are redirected. In future we might have to handle redirects.
        context.getCounter("Sitemap", "failed_fetches").increment(1);
        LOG.error("Error while fetching the sitemap. Status code: {} for {}", status.getCode(), url);
        return;
      }

      AbstractSiteMap asm = parser.parseSiteMap(content.getContentType(), content.getContent(), new URL(url));

      if(asm instanceof SiteMap) {
        LOG.info("Parsing sitemap file: {}", asm.getUrl().toString());
        SiteMap sm = (SiteMap) asm;
        Collection<SiteMapURL> sitemapUrls = sm.getSiteMapUrls();
        for(SiteMapURL sitemapUrl: sitemapUrls) {
          // If 'strict' is ON, only allow valid urls. Else allow all urls
          if(!strict || sitemapUrl.isValid()) {
            String key = filterNormalize(sitemapUrl.getUrl().toString());

            if (key != null) {
              CrawlDatum sitemapUrlDatum = new CrawlDatum();
              sitemapUrlDatum.setStatus(CrawlDatum.STATUS_INJECTED);
              sitemapUrlDatum.setScore((float) sitemapUrl.getPriority());

              if(sitemapUrl.getChangeFrequency() != null) {
                int fetchInterval = -1;
                switch(sitemapUrl.getChangeFrequency()) {
                  case ALWAYS:  fetchInterval = 1;        break;
                  case HOURLY:  fetchInterval = 3600;     break; // 60*60
                  case DAILY:   fetchInterval = 86400;    break; // 60*60*24
                  case WEEKLY:  fetchInterval = 604800;   break; // 60*60*24*7
                  case MONTHLY: fetchInterval = 2592000;  break; // 60*60*24*30
                  case YEARLY:  fetchInterval = 31536000; break; // 60*60*24*365
                  case NEVER:   fetchInterval = Integer.MAX_VALUE; break; // Loose "NEVER" contract
                }
                sitemapUrlDatum.setFetchInterval(fetchInterval);
              }

              if(sitemapUrl.getLastModified() != null) {
                sitemapUrlDatum.setModifiedTime(sitemapUrl.getLastModified().getTime());
              }

              context.write(new Text(key), sitemapUrlDatum);
            }
          }
        }
      }
      else if (asm instanceof SiteMapIndex) {
        SiteMapIndex index = (SiteMapIndex) asm;
        Collection<AbstractSiteMap> sitemapUrls = index.getSitemaps();

        if (sitemapUrls.isEmpty()) {
          return;
        }

        LOG.info("Parsing sitemap index file: {}", index.getUrl().toString());
        for (AbstractSiteMap sitemap : sitemapUrls) {
          String sitemapUrl = filterNormalize(sitemap.getUrl().toString());
          if (sitemapUrl != null) {
            generateSitemapUrlDatum(protocol, sitemapUrl, context);
          }
        }
      }
    }
  }

  private static class SitemapReducer extends Reducer<Text, CrawlDatum, Text, CrawlDatum> {
    CrawlDatum sitemapDatum  = null;
    CrawlDatum originalDatum = null;

    private boolean overwriteExisting = false; // DO NOT ENABLE!!

    public void setup(Context context) {
      Configuration conf = context.getConfiguration();
      this.overwriteExisting = conf.getBoolean(SITEMAP_OVERWRITE_EXISTING, false);
    }

    public void reduce(Text key, Iterable<CrawlDatum> values, Context context)
        throws IOException, InterruptedException {
      sitemapDatum  = null;
      originalDatum = null;

      for (CrawlDatum curr: values) {
        if(curr.getStatus() == CrawlDatum.STATUS_INJECTED) {
          sitemapDatum = new CrawlDatum();
          sitemapDatum.set(curr);
        }
        else {
          originalDatum = new CrawlDatum();
          originalDatum.set(curr);
        }
      }

      if(originalDatum != null) {
        // The url was already present in crawldb. If we got the same url from sitemap too, save
        // the information from sitemap to the original datum. Emit the original crawl datum
        if(sitemapDatum != null && overwriteExisting) {
          originalDatum.setScore(sitemapDatum.getScore());
          originalDatum.setFetchInterval(sitemapDatum.getFetchInterval());
          originalDatum.setModifiedTime(sitemapDatum.getModifiedTime());
        }

        context.getCounter("Sitemap", "existing_sitemap_entries").increment(1);
        context.write(key, originalDatum);
      }
      else if(sitemapDatum != null) {
        // For the newly discovered links via sitemap, set the status as unfetched and emit
        context.getCounter("Sitemap", "new_sitemap_entries").increment(1);
        sitemapDatum.setStatus(CrawlDatum.STATUS_DB_UNFETCHED);
        context.write(key, sitemapDatum);
      }
    }
  }

  public void sitemap(Path crawldb, Path hostdb, Path sitemapUrlDir, boolean strict, boolean filter,
                      boolean normalize, int threads) throws Exception {
    long start = System.currentTimeMillis();
    if (LOG.isInfoEnabled()) {
      LOG.info("SitemapProcessor: Starting at {}", sdf.format(start));
    }

    FileSystem fs = crawldb.getFileSystem(getConf());
    Path old = new Path(crawldb, "old");
    Path current = new Path(crawldb, "current");
    Path tempCrawlDb = new Path(crawldb, "crawldb-" + Integer.toString(new Random().nextInt(Integer.MAX_VALUE)));

    // lock an existing crawldb to prevent multiple simultaneous updates
    Path lock = new Path(crawldb, LOCK_NAME);
    if (!fs.exists(current))
      fs.mkdirs(current);

    LockUtil.createLockFile(fs, lock, false);

    Configuration conf = getConf();
    conf.setBoolean(SITEMAP_STRICT_PARSING, strict);
    conf.setBoolean(SITEMAP_URL_FILTERING, filter);
    conf.setBoolean(SITEMAP_URL_NORMALIZING, normalize);
    conf.setBoolean("mapreduce.fileoutputcommitter.marksuccessfuljobs", false);

    Job job = Job.getInstance(conf, "SitemapProcessor_" + crawldb.toString());
    job.setJarByClass(SitemapProcessor.class);

    // add crawlDb, sitemap url directory and hostDb to input paths
    MultipleInputs.addInputPath(job, current, SequenceFileInputFormat.class);

    if (sitemapUrlDir != null)
      MultipleInputs.addInputPath(job, sitemapUrlDir, KeyValueTextInputFormat.class);

    if (hostdb != null)
      MultipleInputs.addInputPath(job, new Path(hostdb, CURRENT_NAME), SequenceFileInputFormat.class);

    FileOutputFormat.setOutputPath(job, tempCrawlDb);

    job.setOutputFormatClass(MapFileOutputFormat.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(CrawlDatum.class);

    job.setMapperClass(MultithreadedMapper.class);
    MultithreadedMapper.setMapperClass(job, SitemapMapper.class);
    MultithreadedMapper.setNumberOfThreads(job, threads);
    job.setReducerClass(SitemapReducer.class);

    try {
      boolean success = job.waitForCompletion(true);
      if (!success) {
        String message = "SitemapProcessor_" + crawldb.toString()
            + " job did not succeed, job status: " + job.getStatus().getState()
            + ", reason: " + job.getStatus().getFailureInfo();
        LOG.error(message);
        NutchJob.cleanupAfterFailure(tempCrawlDb, lock, fs);
        // throw exception so that calling routine can exit with error
        throw new RuntimeException(message);
      }

      boolean preserveBackup = conf.getBoolean("db.preserve.backup", true);
      if (!preserveBackup && fs.exists(old))
        fs.delete(old, true);
      else
        FSUtils.replace(fs, old, current, true);

      FSUtils.replace(fs, current, tempCrawlDb, true);
      LockUtil.removeLockFile(fs, lock);

      if (LOG.isInfoEnabled()) {
        long filteredRecords = job.getCounters().findCounter("Sitemap", "filtered_records").getValue();
        long fromHostname = job.getCounters().findCounter("Sitemap", "sitemaps_from_hostname").getValue();
        long fromSeeds = job.getCounters().findCounter("Sitemap", "sitemap_seeds").getValue();
        long failedFetches = job.getCounters().findCounter("Sitemap", "failed_fetches").getValue();
        long newSitemapEntries = job.getCounters().findCounter("Sitemap", "new_sitemap_entries").getValue();

        LOG.info("SitemapProcessor: Total records rejected by filters: {}", filteredRecords);
        LOG.info("SitemapProcessor: Total sitemaps from host name: {}", fromHostname);
        LOG.info("SitemapProcessor: Total sitemaps from seed urls: {}", fromSeeds);
        LOG.info("SitemapProcessor: Total failed sitemap fetches: {}", failedFetches);
        LOG.info("SitemapProcessor: Total new sitemap entries added: {}", newSitemapEntries);

        long end = System.currentTimeMillis();
        LOG.info("SitemapProcessor: Finished at {}, elapsed: {}", sdf.format(end), TimingUtil.elapsedTime(start, end));
      }
    } catch (IOException | InterruptedException | ClassNotFoundException e) {
      LOG.error("SitemapProcessor_" + crawldb.toString(), e);
      NutchJob.cleanupAfterFailure(tempCrawlDb, lock, fs);
      throw e;
    }
  }

  public static void main(String[] args) throws Exception {
    int res = ToolRunner.run(NutchConfiguration.create(), new SitemapProcessor(), args);
    System.exit(res);
  }

  public static void usage() {
    System.err.println("Usage:\n SitemapProcessor <crawldb> [-hostdb <hostdb>] [-sitemapUrls <url_dir>] " +
        "[-threads <threads>] [-force] [-noStrict] [-noFilter] [-noNormalize]\n");

    System.err.println("\t<crawldb>\t\tpath to crawldb where the sitemap urls would be injected");
    System.err.println("\t-hostdb <hostdb>\tpath of a hostdb. Sitemap(s) from these hosts would be downloaded");
    System.err.println("\t-sitemapUrls <url_dir>\tpath to directory with sitemap urls or hostnames");
    System.err.println("\t-threads <threads>\tNumber of threads created per mapper to fetch sitemap urls (default: 8)");
    System.err.println("\t-force\t\t\tforce update even if CrawlDb appears to be locked (CAUTION advised)");
    System.err.println("\t-noStrict\t\tBy default Sitemap parser rejects invalid urls. '-noStrict' disables that.");
    System.err.println("\t-noFilter\t\tturn off URLFilters on urls (optional)");
    System.err.println("\t-noNormalize\t\tturn off URLNormalizer on urls (optional)");
  }

  public int run(String[] args) throws Exception {
    if (args.length < 3) {
      usage();
      return -1;
    }

    Path crawlDb = new Path(args[0]);
    Path hostDb = null;
    Path urlDir = null;
    boolean strict = true;
    boolean filter = true;
    boolean normalize = true;
    int threads = 8;

    for (int i = 1; i < args.length; i++) {
      if (args[i].equals("-hostdb")) {
        hostDb = new Path(args[++i]);
        LOG.info("SitemapProcessor: hostdb: {}", hostDb);
      }
      else if (args[i].equals("-sitemapUrls")) {
        urlDir = new Path(args[++i]);
        LOG.info("SitemapProcessor: sitemap urls dir: {}", urlDir);
      }
      else if (args[i].equals("-threads")) {
        threads = Integer.valueOf(args[++i]);
        LOG.info("SitemapProcessor: threads: {}", threads);
      }
      else if (args[i].equals("-noStrict")) {
        LOG.info("SitemapProcessor: 'strict' parsing disabled");
        strict = false;
      }
      else if (args[i].equals("-noFilter")) {
        LOG.info("SitemapProcessor: filtering disabled");
        filter = false;
      }
      else if (args[i].equals("-noNormalize")) {
        LOG.info("SitemapProcessor: normalizing disabled");
        normalize = false;
      }
      else {
        LOG.info("SitemapProcessor: Found invalid argument \"{}\"\n", args[i]);
        usage();
        return -1;
      }
    }

    try {
      sitemap(crawlDb, hostDb, urlDir, strict, filter, normalize, threads);
      return 0;
    } catch (Exception e) {
      LOG.error("SitemapProcessor: {}", StringUtils.stringifyException(e));
      return -1;
    }
  }
}