<commit_before>/**
 * 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();
        }
        
      };
    <commit_msg>Apply patch in NUTCH-230, which provides additional control over which outlinks are considered for OPIC "cash" value distribution.<commit_after>/**
 * 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.*;
import java.util.ArrayList;

/* 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);
    final boolean countFiltered = job.getBoolean("db.score.count.filtered", false);
    
    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);
          String[] toUrls = new String[links.length];
          int validCount = 0;
          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) validCount++;
            toUrls[i] = toUrl;
          }
          if (countFiltered) {
            score = score / links.length;
          } else {
            score = score / validCount;
          }
          for (int i = 0; i < toUrls.length; i++) {
            if (toUrls[i] == null) continue;
            crawlOut.append(new UTF8(toUrls[i]),
                    new CrawlDatum(CrawlDatum.STATUS_LINKED, interval, score));
          }
        }
        
        public void close(Reporter reporter) throws IOException {
          textOut.close();
          dataOut.close();
          crawlOut.close();
        }
        
      };
    
{'fraction_non_alphanumeric': 0.09845288326300984, 'fraction_numerical': 0.004652169209131234, 'mean_word_length': 2.1303758889265154, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6643100', 'n_tokens_mistral': 2532, 'n_tokens_neox': 2478, 'n_words': 708}