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

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.OptionBuilder;
import org.apache.commons.cli.Options;
import org.apache.commons.lang3.time.StopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.ObjectWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.MapFileOutputFormat;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.nutch.util.FSUtils;
import org.apache.nutch.util.NutchConfiguration;
import org.apache.nutch.util.NutchJob;
import org.apache.nutch.util.URLUtil;

public class LinkRank extends Configured implements Tool {

  private static final Logger LOG = LoggerFactory
      .getLogger(MethodHandles.lookup().lookupClass());
  private static final String NUM_NODES = "_num_nodes_";

  /**
   * Runs the counter job. The counter job determines the number of links in the
   * webgraph. This is used during analysis.
   * 
   * @param fs
   *          The job file system.
   * @param webGraphDb
   *          The web graph database to use.
   * 
   * @return The number of nodes in the web graph.
   * @throws IOException
   *           If an error occurs while running the counter job.
   */
  private int runCounter(FileSystem fs, Path webGraphDb) throws IOException,
      ClassNotFoundException, InterruptedException {

    // configure the counter job
    Path numLinksPath = new Path(webGraphDb, NUM_NODES);
    Path nodeDb = new Path(webGraphDb, WebGraph.NODE_DIR);
    Job counter = Job.getInstance(getConf(), "Nutch LinkRank: counter " + webGraphDb);
    Configuration conf = counter.getConfiguration();
    FileInputFormat.addInputPath(counter, nodeDb);
    FileOutputFormat.setOutputPath(counter, numLinksPath);
    counter.setInputFormatClass(SequenceFileInputFormat.class);
    counter.setJarByClass(Counter.class);
    counter.setMapperClass(Counter.CountMapper.class);
    counter.setCombinerClass(Counter.CountReducer.class);
    counter.setReducerClass(Counter.CountReducer.class);
    counter.setMapOutputKeyClass(Text.class);
    counter.setMapOutputValueClass(LongWritable.class);
    counter.setOutputKeyClass(Text.class);
    counter.setOutputValueClass(LongWritable.class);
    counter.setNumReduceTasks(1);
    counter.setOutputFormatClass(TextOutputFormat.class);
    conf.setBoolean("mapreduce.fileoutputcommitter.marksuccessfuljobs",
        false);

    // run the counter job, outputs to a single reduce task and file
    LOG.info("Starting link counter job");
    try {
      boolean success = counter.waitForCompletion(true);
      if (!success) {
        String message = NutchJob.getJobFailureLogMessage("Link counter",
            counter);
        LOG.error(message);
        throw new RuntimeException(message);
      }
    } catch (IOException | InterruptedException | ClassNotFoundException e) {
      LOG.error("Link counter job failed:", e);
      throw e;
    }
    
    LOG.info("Finished link counter job");

    // read the first (and only) line from the file which should be the
    // number of links in the web graph
    FileStatus[] numLinksFiles = fs.listStatus(numLinksPath);
    if (numLinksFiles.length == 0) {
      throw new IOException("Failed to read numlinks temp file: "
          + " no file found in " + numLinksPath);
    } else if (numLinksFiles.length > 1) {
      throw new IOException("Failed to read numlinks temp file: "
          + " expected only one file but found " + numLinksFiles.length
          + " files in folder " + numLinksPath);
    }
    Path numLinksFile = numLinksFiles[0].getPath();
    LOG.info("Reading numlinks temp file {}", numLinksFile);
    FSDataInputStream readLinks = fs.open(numLinksFile);
    CompressionCodecFactory cf = new CompressionCodecFactory(conf);
    CompressionCodec codec = cf.getCodec(numLinksFiles[0].getPath());
    InputStream streamLinks;
    if (codec == null) {
      LOG.debug("No compression codec found for {}, trying uncompressed",
          numLinksFile);
      streamLinks = readLinks;
    } else {
      LOG.info("Compression codec of numlinks temp file: {}",
          codec.getDefaultExtension());
      readLinks.seek(0);
      streamLinks = codec.createInputStream(readLinks);
    }
    BufferedReader buffer = new BufferedReader(
        new InputStreamReader(streamLinks));

    String numLinksLine = buffer.readLine();
    readLinks.close();

    // check if there are links to process, if none, webgraph might be empty
    if (numLinksLine == null || numLinksLine.length() == 0) {
      LOG.error(
          "Failed to determine number of links because of empty line in input {}",
          numLinksFile);
      fs.delete(numLinksPath, true);
      throw new IOException("No links to process, is the webgraph empty?");
    }

    // delete temp file and convert and return the number of links as an int
    LOG.info("Deleting numlinks temp file");
    fs.delete(numLinksPath, true);
    String numLinks = numLinksLine.split("\\s+")[1];
    return Integer.parseInt(numLinks);
  }

  /**
   * Runs the initializer job. The initializer job sets up the nodes with a
   * default starting score for link analysis.
   * 
   * @param nodeDb
   *          The node database to use.
   * @param output
   *          The job output directory.
   * 
   * @throws IOException
   *           If an error occurs while running the initializer job.
   */
  private void runInitializer(Path nodeDb, Path output) throws IOException,
     InterruptedException, ClassNotFoundException {

    // configure the initializer
    Job initializer = Job.getInstance(getConf(), "Nutch LinkRank: initializer " + nodeDb);
    Configuration conf = initializer.getConfiguration();
    FileInputFormat.addInputPath(initializer, nodeDb);
    FileOutputFormat.setOutputPath(initializer, output);
    initializer.setJarByClass(Initializer.class);
    initializer.setInputFormatClass(SequenceFileInputFormat.class);
    initializer.setMapperClass(Initializer.class);
    initializer.setMapOutputKeyClass(Text.class);
    initializer.setMapOutputValueClass(Node.class);
    initializer.setOutputKeyClass(Text.class);
    initializer.setOutputValueClass(Node.class);
    initializer.setOutputFormatClass(MapFileOutputFormat.class);
    conf.setBoolean("mapreduce.fileoutputcommitter.marksuccessfuljobs",
        false);

    // run the initializer
    LOG.info("Starting initialization job");
    try {
      boolean success = initializer.waitForCompletion(true);
      if (!success) {
        String message = NutchJob.getJobFailureLogMessage("Initialization",
            initializer);
        LOG.error(message);
        throw new RuntimeException(message);
      }
    } catch (IOException | InterruptedException | ClassNotFoundException e) {
      LOG.error("Initialization job failed:", e);
      throw e;
    }
    LOG.info("Finished initialization job.");
  }

  /**
   * Runs the inverter job. The inverter job flips outlinks to inlinks to be
   * passed into the analysis job.
   * 
   * @param nodeDb
   *          The node database to use.
   * @param outlinkDb
   *          The outlink database to use.
   * @param output
   *          The output directory.
   * 
   * @throws IOException
   *           If an error occurs while running the inverter job.
   */
  private void runInverter(Path nodeDb, Path outlinkDb, Path output)
      throws IOException, InterruptedException, ClassNotFoundException {

    // configure the inverter
    Job inverter = Job.getInstance(getConf(),
        "Nutch Linkrank: inverter nodedb: " + nodeDb + " outlinkdb: " + outlinkDb);
    Configuration conf = inverter.getConfiguration();
    FileInputFormat.addInputPath(inverter, nodeDb);
    FileInputFormat.addInputPath(inverter, outlinkDb);
    FileOutputFormat.setOutputPath(inverter, output);
    inverter.setInputFormatClass(SequenceFileInputFormat.class);
    inverter.setJarByClass(Inverter.class);
    inverter.setMapperClass(Inverter.InvertMapper.class);
    inverter.setReducerClass(Inverter.InvertReducer.class);
    inverter.setMapOutputKeyClass(Text.class);
    inverter.setMapOutputValueClass(ObjectWritable.class);
    inverter.setOutputKeyClass(Text.class);
    inverter.setOutputValueClass(LinkDatum.class);
    inverter.setOutputFormatClass(SequenceFileOutputFormat.class);
    conf.setBoolean("mapreduce.fileoutputcommitter.marksuccessfuljobs",
        false);

    // run the inverter job
    LOG.info("Starting inverter job");
    try {
      boolean success = inverter.waitForCompletion(true);
      if (!success) {
        String message = NutchJob.getJobFailureLogMessage("Inverter", inverter);
        LOG.error(message);
        throw new RuntimeException(message);
      }
    } catch (IOException | InterruptedException | ClassNotFoundException e) {
      LOG.error("Inverter job failed:", e);
      throw e;
    }
    LOG.info("Finished inverter job.");
  }

  /**
   * Runs the link analysis job. The link analysis job applies the link rank
   * formula to create a score per url and stores that score in the NodeDb.
   * 
   * Typically the link analysis job is run a number of times to allow the link
   * rank scores to converge.
   * 
   * @param nodeDb
   *          The node database from which we are getting previous link rank
   *          scores.
   * @param inverted
   *          The inverted inlinks
   * @param output
   *          The link analysis output.
   * @param iteration
   *          The current iteration number.
   * @param numIterations
   *          The total number of link analysis iterations
   * 
   * @throws IOException
   *           If an error occurs during link analysis.
   */
  private void runAnalysis(Path nodeDb, Path inverted, Path output,
      int iteration, int numIterations, float rankOne) 
      throws IOException, InterruptedException, ClassNotFoundException {

    Job analyzer = Job.getInstance(getConf(),
        "Nutch LinkRank: analysis iteration" + (iteration + 1) + " of " + numIterations);
    Configuration conf = analyzer.getConfiguration();
    conf.set("link.analyze.iteration", String.valueOf(iteration + 1));
    FileInputFormat.addInputPath(analyzer, nodeDb);
    FileInputFormat.addInputPath(analyzer, inverted);
    FileOutputFormat.setOutputPath(analyzer, output);
    conf.set("link.analyze.rank.one", String.valueOf(rankOne));
    analyzer.setMapOutputKeyClass(Text.class);
    analyzer.setMapOutputValueClass(ObjectWritable.class);
    analyzer.setInputFormatClass(SequenceFileInputFormat.class);
    analyzer.setJarByClass(Analyzer.class);
    analyzer.setMapperClass(Analyzer.AnalyzerMapper.class);
    analyzer.setReducerClass(Analyzer.AnalyzerReducer.class);
    analyzer.setOutputKeyClass(Text.class);
    analyzer.setOutputValueClass(Node.class);
    analyzer.setOutputFormatClass(MapFileOutputFormat.class);
    conf.setBoolean("mapreduce.fileoutputcommitter.marksuccessfuljobs",
        false);

    LOG.info("Starting analysis job");
    try {
      boolean success = analyzer.waitForCompletion(true);
      if (!success) {
        String message = NutchJob.getJobFailureLogMessage("Analysis", analyzer);
        LOG.error(message);
        throw new RuntimeException(message);
      }
    } catch (IOException | InterruptedException | ClassNotFoundException e) {
      LOG.error("Analysis job failed:", e);
      throw e;
    }
    LOG.info("Finished analysis job.");
  }

  /**
   * The Counter job that determines the total number of nodes in the WebGraph.
   * This is used to determine a rank one score for pages with zero inlinks but
   * that contain outlinks.
   */
  private static class Counter {

    private static Text numNodes = new Text(NUM_NODES);
    private static LongWritable one = new LongWritable(1L);

    /**
     * Outputs one for every node.
     */
    public static class CountMapper extends
        Mapper<Text, Node, Text, LongWritable> {
      @Override
      public void map(Text key, Node value,
          Context context)
          throws IOException, InterruptedException {
        context.write(numNodes, one);
      }
    }

    /**
     * Totals the node number and outputs a single total value.
     */
    public static class CountReducer extends
        Reducer<Text, LongWritable, Text, LongWritable> {
      @Override
      public void reduce(Text key, Iterable<LongWritable> values,
          Context context)
          throws IOException, InterruptedException {

        long total = 0;
        for (LongWritable val : values) {
          total += val.get();
        }
        context.write(numNodes, new LongWritable(total));
      }
    }

  }

  private static class Initializer extends Mapper<Text, Node, Text, Node> {

    private Configuration conf;
    private float initialScore = 1.0f;

    @Override
    public void setup(Mapper<Text, Node, Text, Node>.Context context) {
      conf = context.getConfiguration();
      initialScore = conf.getFloat("link.analyze.initial.score", 1.0f);
    }

    @Override
    public void map(Text key, Node node, Context context) 
        throws IOException, InterruptedException {

      String url = key.toString();
      Node outNode = WritableUtils.clone(node, conf);
      outNode.setInlinkScore(initialScore);

      context.write(new Text(url), outNode);
    }

  }

  /**
   * Inverts outlinks and attaches current score from the NodeDb of the
   * WebGraph. The link analysis process consists of inverting, analyzing and
   * scoring, in a loop for a given number of iterations.
   */
  private static class Inverter {

    /**
     * Convert values to ObjectWritable
     */
    public static class InvertMapper extends 
        Mapper<Text, Writable, Text, ObjectWritable> {

      @Override
      public void setup(Mapper<Text, Writable, Text, ObjectWritable>.Context context) {
      }

      @Override
      public void map(Text key, Writable value,
          Context context)
          throws IOException, InterruptedException {

        ObjectWritable objWrite = new ObjectWritable();
        objWrite.set(value);
        context.write(key, objWrite);
      }
    }

    /**
     * Inverts outlinks to inlinks, attaches current score for the outlink from
     * the NodeDb of the WebGraph.
     */
    public static class InvertReducer extends
        Reducer<Text, ObjectWritable, Text, LinkDatum> {

      private Configuration conf;      

      @Override
      public void setup(Reducer<Text, ObjectWritable, Text, LinkDatum>.Context context) {
        conf = context.getConfiguration();
      }

      @Override
      public void reduce(Text key, Iterable<ObjectWritable> values,
          Context context)
          throws IOException, InterruptedException {

        String fromUrl = key.toString();
        List<LinkDatum> outlinks = new ArrayList<>();
        Node node = null;

        // aggregate outlinks, assign other values
        for (ObjectWritable write : values) {
          Object obj = write.get();
          if (obj instanceof Node) {
            node = (Node) obj;
          } else if (obj instanceof LinkDatum) {
            outlinks.add(WritableUtils.clone((LinkDatum) obj, conf));
          }
        }

        // get the number of outlinks and the current inlink and outlink scores
        // from the node of the url
        int numOutlinks = node.getNumOutlinks();
        float inlinkScore = node.getInlinkScore();
        float outlinkScore = node.getOutlinkScore();
        LOG.debug("{}: num outlinks {}", fromUrl, numOutlinks);

        // can't invert if no outlinks
        if (numOutlinks > 0) {
          for (int i = 0; i < outlinks.size(); i++) {
            LinkDatum outlink = outlinks.get(i);
            String toUrl = outlink.getUrl();

            outlink.setUrl(fromUrl);
            outlink.setScore(outlinkScore);

            // collect the inverted outlink
            context.write(new Text(toUrl), outlink);
            LOG.debug(
                "{}: inverting inlink from {} origscore: {} numOutlinks: {} inlinkscore: {}",
                toUrl, fromUrl, inlinkScore, numOutlinks, outlinkScore);
          }
        }
      }
    }
  }

  /**
   * Runs a single link analysis iteration.
   */
  private static class Analyzer {

    /**
     * Convert values to ObjectWritable
     */
    public static class AnalyzerMapper extends 
        Mapper<Text, Writable, Text, ObjectWritable> {

      private Configuration conf;

      /**
       * Configures the job mapper, sets the damping factor, rank one score, and other
       * needed values for analysis.
       */
      @Override
      public void setup(Mapper<Text, Writable, Text, ObjectWritable>.Context context) {
        conf = context.getConfiguration();
      }

      @Override
      public void map(Text key, Writable value,
          Context context)
          throws IOException, InterruptedException {

        ObjectWritable objWrite = new ObjectWritable();
        objWrite.set(WritableUtils.clone(value, conf));
        context.write(key, objWrite);
      }
    }

    /**
     * Performs a single iteration of link analysis. The resulting scores are
     * stored in a temporary NodeDb which replaces the NodeDb of the WebGraph.
     */
    public static class AnalyzerReducer extends
        Reducer<Text, ObjectWritable, Text, Node> {

      private Configuration conf;
      private float dampingFactor = 0.85f;
      private float rankOne = 0.0f;
      private int itNum = 0;
      private boolean limitPages = true;
      private boolean limitDomains = true;

      /**
       * Configures the job reducer, sets the damping factor, rank one score, and other
       * needed values for analysis.
       */
      @Override
      public void setup(
          Reducer<Text, ObjectWritable, Text, Node>.Context context) {
        conf = context.getConfiguration();
        dampingFactor = conf.getFloat("link.analyze.damping.factor", 0.85f);
        rankOne = conf.getFloat("link.analyze.rank.one", 0.0f);
        itNum = conf.getInt("link.analyze.iteration", 0);
        limitPages = conf.getBoolean("link.ignore.limit.page", true);
        limitDomains = conf.getBoolean("link.ignore.limit.domain", true);
      }

      @Override
      public void reduce(Text key, Iterable<ObjectWritable> values,
          Context context)
          throws IOException, InterruptedException {

        String url = key.toString();
        Set<String> domains = new HashSet<>();
        Set<String> pages = new HashSet<>();
        Node node = null;

        // a page with zero inlinks has a score of rankOne
        int numInlinks = 0;
        float totalInlinkScore = rankOne;

        for (ObjectWritable next : values) {

          Object value = next.get();
          if (value instanceof Node) {
            node = (Node) value;
          } else if (value instanceof LinkDatum) {

            LinkDatum linkDatum = (LinkDatum) value;
            float scoreFromInlink = linkDatum.getScore();
            String inlinkUrl = linkDatum.getUrl();
            String inLinkDomain = URLUtil.getDomainName(inlinkUrl);
            String inLinkPage = URLUtil.getPage(inlinkUrl);

            // limit counting duplicate inlinks by pages or domains
            if ((limitPages && pages.contains(inLinkPage))
                || (limitDomains && domains.contains(inLinkDomain))) {
              LOG.debug("{}: ignoring {} from {}, duplicate page or domain",
                  url, scoreFromInlink, inlinkUrl);
              continue;
            }

            // aggregate total inlink score
            numInlinks++;
            totalInlinkScore += scoreFromInlink;
            domains.add(inLinkDomain);
            pages.add(inLinkPage);
            LOG.debug("{}: adding {} from {}, total: {}", url, scoreFromInlink,
                inlinkUrl, totalInlinkScore);
          }
        }

        // calculate linkRank score formula
        float linkRankScore = (1 - dampingFactor)
            + (dampingFactor * totalInlinkScore);

        LOG.debug("{}: score: {} num inlinks: {} iteration: {}", url,
            linkRankScore, numInlinks, itNum);

        // store the score in a temporary NodeDb
        Node outNode = WritableUtils.clone(node, conf);
        outNode.setInlinkScore(linkRankScore);
        context.write(key, outNode);
      }
    }
  }

  /**
   * Default constructor.
   */
  public LinkRank() {
    super();
  }

  /**
   * Configurable constructor.
   * @param conf a populated {@link org.apache.hadoop.conf.Configuration}
   */
  public LinkRank(Configuration conf) {
    super(conf);
  }

  /**
   * Runs the complete link analysis job. The complete job determins rank one
   * score. Then runs through a given number of invert and analyze iterations,
   * by default 10. And finally replaces the NodeDb in the WebGraph with the
   * link rank output.
   * 
   * @param webGraphDb
   *          The WebGraph to run link analysis on.
   * 
   * @throws IOException
   *           If a fatal I/O runtime error occurs during link analysis.
   * @throws InterruptedException if the Job is interrupted during execution
   * @throws ClassNotFoundException if classes required to run 
   * the Job cannot be located
   */
  public void analyze(Path webGraphDb) throws IOException, 
      ClassNotFoundException, InterruptedException {

    StopWatch stopWatch = new StopWatch();
    stopWatch.start();
    LOG.info("LinkRank Analysis: starting");

    // store the link rank under the webgraphdb temporarily, final scores get
    // upddated into the nodedb
    Path linkRank = new Path(webGraphDb, "linkrank");
    Configuration conf = getConf();
    FileSystem fs = linkRank.getFileSystem(conf);

    // create the linkrank directory if needed
    if (!fs.exists(linkRank)) {
      fs.mkdirs(linkRank);
    }

    // the webgraph outlink and node database paths
    Path wgOutlinkDb = new Path(webGraphDb, WebGraph.OUTLINK_DIR);
    Path wgNodeDb = new Path(webGraphDb, WebGraph.NODE_DIR);
    Path nodeDb = new Path(linkRank, WebGraph.NODE_DIR);

    // get the number of total nodes in the webgraph, used for rank one, then
    // initialze all urls with a default score
    int numLinks = runCounter(fs, webGraphDb);
    runInitializer(wgNodeDb, nodeDb);
    float rankOneScore = (1f / (float) numLinks);

    LOG.info("Analysis: Number of links: {}", numLinks);
    LOG.info("Analysis: Rank One: {}", rankOneScore);

    // run invert and analysis for a given number of iterations to allow the
    // link rank scores to converge
    int numIterations = conf.getInt("link.analyze.num.iterations", 10);
    for (int i = 0; i < numIterations; i++) {

      // the input to inverting is always the previous output from analysis
      LOG.info("Analysis: Starting iteration {} of {}", (i + 1), numIterations);
      Path tempRank = new Path(linkRank + "-"
          + Integer.toString(new Random().nextInt(Integer.MAX_VALUE)));
      fs.mkdirs(tempRank);
      Path tempInverted = new Path(tempRank, "inverted");
      Path tempNodeDb = new Path(tempRank, WebGraph.NODE_DIR);

      // run invert and analysis
      runInverter(nodeDb, wgOutlinkDb, tempInverted);
      runAnalysis(nodeDb, tempInverted, tempNodeDb, i, numIterations,
          rankOneScore);

      // replace the temporary NodeDb with the output from analysis
      LOG.info("Analysis: Installing new link scores");
      FSUtils.replace(fs, linkRank, tempRank, true);
      LOG.info("Analysis: finished iteration {} of {}", (i + 1), numIterations);
    }

    // replace the NodeDb in the WebGraph with the final output of analysis
    LOG.info("Analysis: Installing web graph nodes");
    FSUtils.replace(fs, wgNodeDb, nodeDb, true);

    // remove the temporary link rank folder
    fs.delete(linkRank, true);
    stopWatch.stop();
    LOG.info("LinkRank Analysis: finished, elapsed: {} ms", stopWatch.getTime(
        TimeUnit.MILLISECONDS));
  }

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

  /**
   * Runs the LinkRank tool.
   */
  @Override
  public int run(String[] args) throws Exception {

    Options options = new Options();
    OptionBuilder.withArgName("help");
    OptionBuilder.withDescription("show this help message");
    Option helpOpts = OptionBuilder.create("help");
    options.addOption(helpOpts);

    OptionBuilder.withArgName("webgraphdb");
    OptionBuilder.hasArg();
    OptionBuilder.withDescription("the web graph db to use");
    Option webgraphOpts = OptionBuilder.create("webgraphdb");
    options.addOption(webgraphOpts);

    CommandLineParser parser = new GnuParser();
    try {

      CommandLine line = parser.parse(options, args);
      if (line.hasOption("help") || !line.hasOption("webgraphdb")) {
        HelpFormatter formatter = new HelpFormatter();
        formatter.printHelp("LinkRank", options);
        return -1;
      }

      String webGraphDb = line.getOptionValue("webgraphdb");

      analyze(new Path(webGraphDb));
      return 0;
    } catch (Exception e) {
      LOG.error("LinkAnalysis:", e);
      return -2;
    }
  }
}
