/**
 * Copyright 2005 The Apache Software Foundation
 *
 * Licensed 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.parse;

import org.apache.hadoop.io.*;
import org.apache.nutch.crawl.CrawlDatum;
import org.apache.nutch.fetcher.Fetcher;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.mapred.*;
import org.apache.nutch.util.StringUtil;
import org.apache.nutch.net.*;

import java.io.*;

/* Parse content in a segment. */
public class ParseOutputFormat implements OutputFormat {

  private UrlNormalizer urlNormalizer;
  private JobConf jobConf;
  private URLFilters filters;


  public RecordWriter getRecordWriter(FileSystem fs, JobConf job,
                                      String name) throws IOException {

    this.jobConf = job;
    this.urlNormalizer = new UrlNormalizerFactory(job).getNormalizer();
    this.filters = new URLFilters(job);
    final float interval = job.getFloat("db.default.fetch.interval", 30f);
    final float extscore = job.getFloat("db.score.link.external", 1.0f);
    
    File text =
      new File(new File(job.getOutputDir(), ParseText.DIR_NAME), name);
    File data =
      new File(new File(job.getOutputDir(), ParseData.DIR_NAME), name);
    File crawl =
      new File(new File(job.getOutputDir(), CrawlDatum.PARSE_DIR_NAME), name);
    
    final MapFile.Writer textOut =
      new MapFile.Writer(fs, text.toString(), UTF8.class, ParseText.class);
    
    final MapFile.Writer dataOut =
      new MapFile.Writer(fs, data.toString(), UTF8.class,ParseData.class,true);
    
    final SequenceFile.Writer crawlOut =
      new SequenceFile.Writer(fs, crawl.toString(),
                              UTF8.class, CrawlDatum.class);
    
    return new RecordWriter() {


        public void write(WritableComparable key, Writable value)
          throws IOException {
          
          Parse parse = (Parse)value;
          
          textOut.append(key, new ParseText(parse.getText()));
          dataOut.append(key, parse.getData());
          
          // recover the signature prepared by Fetcher or ParseSegment
          String sig = parse.getData()
                            .getContentMeta().get(Fetcher.SIGNATURE_KEY);
          if (sig != null) {
            byte[] signature = StringUtil.fromHexString(sig);
            if (signature != null) {
              // append a CrawlDatum with a signature
              CrawlDatum d = new CrawlDatum(CrawlDatum.STATUS_SIGNATURE, 0.0f);
              d.setSignature(signature);
              crawlOut.append(key, d);
            }
          }

          // collect outlinks for subsequent db update
          Outlink[] links = parse.getData().getOutlinks();

          // compute OPIC score contribution
          String scoreString = parse.getData()
                                    .getContentMeta().get(Fetcher.SCORE_KEY);
          float score = extscore;
          // this may happen if there was a fetch error.
         if (scoreString != null) score = Float.parseFloat(scoreString);
          score /= links.length;
                          
          for (int i = 0; i < links.length; i++) {
            String toUrl = links[i].getToUrl();
            try {
              toUrl = urlNormalizer.normalize(toUrl); // normalize the url
              toUrl = filters.filter(toUrl);   // filter the url
            } catch (Exception e) {
              toUrl = null;
            }
            if (toUrl != null)
              crawlOut.append(new UTF8(toUrl),
                              new CrawlDatum(CrawlDatum.STATUS_LINKED,
                                             interval, score));
          }
        }
        
        public void close(Reporter reporter) throws IOException {
          textOut.close();
          dataOut.close();
          crawlOut.close();
        }
        
      };
    
  }

}

@@ -912,16 +912,44 @@
 va.io.*;
+%0Aimport java.util.ArrayList;
 %0A%0A/* Par
@@ -1551,24 +1551,108 @@
 al%22, 1.0f);%0A
+    final boolean countFiltered = job.getBoolean(%22db.score.count.filtered%22, false);%0A
     %0A    Fil
@@ -3563,24 +3563,25 @@
 r.%0A         
+ 
 if (scoreStr
@@ -3646,17 +3646,37 @@
     
-score /= 
+String%5B%5D toUrls = new String%5B
 link
@@ -3683,16 +3683,17 @@
 s.length
+%5D
 ;%0A      
@@ -3696,32 +3696,35 @@
         
-                
+int validCount = 0;
 %0A       
@@ -4085,171 +4085,416 @@
 ull)
-%0A              crawlOut.append(new UTF8(toUrl),%0A                              new CrawlDatum(CrawlDatum.STATUS_LINKED,%0A                                            
+ validCount++;%0A            toUrls%5Bi%5D = toUrl;%0A          %7D%0A          if (countFiltered) %7B%0A            score = score / links.length;%0A          %7D else %7B%0A            score = score / validCount;%0A          %7D%0A          for (int i = 0; i %3C toUrls.length; i++) %7B%0A            if (toUrls%5Bi%5D == null) continue;%0A            crawlOut.append(new UTF8(toUrls%5Bi%5D),%0A                    new CrawlDatum(CrawlDatum.STATUS_LINKED,
  int

Apply patch in NUTCH-230, which provides additional control over which outlinks are considered for OPIC "cash" value distribution.