/*
 * 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.hostdb;

import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.net.MalformedURLException;
import java.net.URL;

import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.conf.Configuration;

import org.apache.nutch.crawl.CrawlDatum;
import org.apache.nutch.crawl.NutchWritable;
import org.apache.nutch.metadata.Nutch;
import org.apache.nutch.net.URLFilters;
import org.apache.nutch.net.URLNormalizers;
import org.apache.nutch.protocol.ProtocolStatus;
import org.apache.nutch.util.URLUtil;

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

/**
 * Mapper ingesting HostDB and CrawlDB entries. Additionally it can also read
 * host score info from a plain text key/value file generated by the
 * Webgraph's NodeDumper tool.
 */
public class UpdateHostDbMapper
  extends Mapper<Text, Writable, Text, NutchWritable> {

  private static final Logger LOG = LoggerFactory
      .getLogger(MethodHandles.lookup().lookupClass());
  protected Text host = new Text();
  protected HostDatum hostDatum = null;
  protected CrawlDatum crawlDatum = null;
  protected String reprUrl = null;
  protected String buffer = null;
  protected String[] args = null;
  protected boolean filter = false;
  protected boolean normalize = false;
  protected boolean readingCrawlDb = false;
  protected URLFilters filters = null;
  protected URLNormalizers normalizers = null;

  @Override
  public void setup(Mapper<Text, Writable, Text, NutchWritable>.Context context) {
    Configuration conf = context.getConfiguration();
    readingCrawlDb = conf.getBoolean("hostdb.reading.crawldb", false);
    filter = conf.getBoolean(UpdateHostDb.HOSTDB_URL_FILTERING, false);
    normalize = conf.getBoolean(UpdateHostDb.HOSTDB_URL_NORMALIZING, false);

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

  /**
   * Filters and or normalizes the input hostname by applying the configured URL
   * filters and normalizers the URL &quot;http://hostname/&quot;.
   *
   * @param hostName
   *          the input hostname
   * @return the normalized hostname, or null if the URL is excluded by URL
   *         filters or failed to be normalized converted
   */
  protected String filterNormalize(String hostName) {

    if (!filter && !normalize) {
      // nothing to do
      return hostName;
    }

    String url = "http://" + hostName + "/";
    String normalizedUrl = url;

    try {
      if (normalize)
        normalizedUrl = normalizers.normalize(url, URLNormalizers.SCOPE_DEFAULT);
      if (filter)
        normalizedUrl = filters.filter(normalizedUrl);
      if (normalizedUrl == null)
        return null;
    } catch (Exception e) {
      return null;
    }

    if (normalizedUrl.equals(url)) {
      // URL did not change during normalization
      return hostName;
    }

    // Turn back to host
    return URLUtil.getHost(url);
  }

  /**
    * Mapper ingesting records from the HostDB, CrawlDB and plaintext host
    * scores file. Statistics and scores are passed on.
    *
    * @param key record {@link org.apache.hadoop.io.Text} key
    * @param value associated {@link org.apache.hadoop.io.Writable} object
    * @param context {@link org.apache.hadoop.mapreduce.Reducer.Context} for
    * writing custom counters and output.
    */
  @Override
  public void map(Text key, Writable value,
    Context context)
    throws IOException, InterruptedException {

    // Get the key!
    String keyStr = key.toString();

    // Check if we process records from the CrawlDB
    if (value instanceof CrawlDatum) {

      URL url;
      try {
        url = new URL(keyStr);
      } catch (MalformedURLException e) {
        context.getCounter("UpdateHostDb", "malformed_url").increment(1);
        return;
      }
      String hostName = URLUtil.getHost(url);

      // Get the normalized and filtered host of this URL
      buffer = filterNormalize(hostName);

      // Filtered out?
      if (buffer == null) {
        context.getCounter("UpdateHostDb", "filtered_records").increment(1);
        LOG.debug("UpdateHostDb: {} crawldatum has been filtered", hostName);
        return;
      }

      // Set the host of this URL
      host.set(buffer);
      crawlDatum = (CrawlDatum) value;
      hostDatum = new HostDatum();

      /**
        * TODO: fix multi redirects: host_a => host_b/page => host_c/page/whatever
        * http://www.ferienwohnung-armbruster.de/
        * http://www.ferienwohnung-armbruster.de/website/
        * http://www.ferienwohnung-armbruster.de/website/willkommen.php
        *
        * We cannot reresolve redirects for host objects as CrawlDatum metadata is
        * not available. We also cannot reliably use the reducer in all cases
        * since redirects may be across hosts or even domains. The example
        * above has redirects that will end up in the same reducer. During that
        * phase, however, we do not know which URL redirects to the next URL.
        */
      // Do not resolve homepages when the root URL is unfetched
      if (crawlDatum.getStatus() != CrawlDatum.STATUS_DB_UNFETCHED) {
        // Get the protocol
        String protocol = URLUtil.getProtocol(url);
        
        // Check if the current key is equals the host
        if (URLUtil.isHomePageOf(url, buffer)) {
          // Check if this is a redirect to the real home page
          if (crawlDatum.getStatus() == CrawlDatum.STATUS_DB_REDIR_PERM ||
            crawlDatum.getStatus() == CrawlDatum.STATUS_DB_REDIR_TEMP) {

            // Obtain the repr url for this redirect via protocolstatus from the metadata
            ProtocolStatus z = (ProtocolStatus) crawlDatum.getMetaData()
                .get(Nutch.WRITABLE_PROTO_STATUS_KEY);

            // Get the protocol status' arguments
            args = z.getArgs();

            // ..and the possible redirect URL
            reprUrl = args[0];

            // Am i a redirect?
            if (reprUrl != null) {
              LOG.debug("UpdateHostDb: homepage: {} redirects to: {}", keyStr,
                  args[0]);
              context.write(host, new NutchWritable(hostDatum));
              hostDatum.setHomepageUrl(reprUrl);
            } else {
              LOG.debug(
                  "UpdateHostDb: homepage: {} redirects to: {} but has been filtered out",
                  keyStr, args[0]);
            }
          } else {
            // need to construct the URL anew as the hostname may be change by normalization
            String homepage = protocol + "://" + buffer + "/";
            hostDatum.setHomepageUrl(homepage);
            context.write(host, new NutchWritable(hostDatum));
            LOG.debug("UpdateHostDb: homepage: {}", homepage);
          }
        }
      }

      // Always emit crawl datum
      context.write(host, new NutchWritable(crawlDatum));

    } else if (value instanceof HostDatum) {
      // we got a record from the hostdb
      buffer = filterNormalize(keyStr);

      // Filtered out?
      if (buffer == null) {
        context.getCounter("UpdateHostDb", "filtered_records").increment(1);
        LOG.debug("UpdateHostDb: {} hostdatum has been filtered", keyStr);
        return;
      }

      // Get a HostDatum
      hostDatum = (HostDatum)value;
      key.set(buffer);

      // If we're also reading CrawlDb entries, reset db_* statistics because
      // we're aggregating them from CrawlDB anyway
      if (readingCrawlDb) {
        hostDatum.resetStatistics();
      }

      context.write(key, new NutchWritable(hostDatum));

    } else if (value instanceof Text) {
      // a record with host scores

      buffer = filterNormalize(keyStr);

      // Filtered out?
      if (buffer == null) {
        context.getCounter("UpdateHostDb", "filtered_records").increment(1);
        LOG.debug("UpdateHostDb: {} score has been filtered", keyStr);
        return;
      }

      key.set(buffer);

      context.write(key,
        new NutchWritable(new FloatWritable(Float.parseFloat(value.toString()))));
    }
  }
}
