source: gs3-extensions/maori-lang-detection/src/org/greenstone/atea/WETProcessor.java@ 33471

package org.greenstone.atea;


import java.io.*;
import java.util.Properties;
import java.util.zip.GZIPInputStream;

import org.apache.log4j.Logger;

/**
 * The main() method of this class takes a folder of warc.wet(.gz) files and goes through
 * the WET records in each, putting each WET record into a file. Each file is put into a
 * keep or discard folder, based on content-length and number of lines.
 * A single instance of the WETProcessor class processes a single unzipped warc.wet file.
 *
 * To compile, including the jars in lib/ for compiling.
 *      maori-lang-detection/src$ javac -cp ".:../lib/*" org/greenstone/atea/WETProcessor.java 
 *
 * To run, passing the log4j and other properties files in conf/ folder:
 *      maori-lang-detection/src$ java -cp ".:../conf:../lib/*" org.greenstone.atea.WETProcessor <folder containing warc.wet(.gz) files> <outputFolder>
 *
 * e.g.
 *    - java -cp ".:../conf:../lib/*" org.greenstone.atea.WETProcessor ../tmp/processWET /Scratch/ak19/gs3-extensions/maori-lang-detection/tmp/processedWET
 *    - java -cp ".:../conf:../lib/*" org.greenstone.atea.WETProcessor ../tmp/processWET /Scratch/ak19/gs3-extensions/maori-lang-detection/tmp/processedWET 2>&1 | less
 *
*/
public class WETProcessor {
    private static Logger logger = Logger.getLogger(org.greenstone.atea.WETProcessor.class.getName());    
    private static Properties configProperties = new Properties();

    // In Java, can initialize static final variables inside a static block
    // But the unavoidable try/catch in this static block prevents initialization of
    // the static final int variables further below that therefore need to be declared
    // and initialized thereafter.
    static {    
    
    // load up the properties from the config file
    try (InputStream infile = org.greenstone.atea.WETProcessor.class.getClassLoader().getResourceAsStream("config.properties")) {
        configProperties = new Properties();
        configProperties.load(infile);      
        //infile.close();
        
    } catch(Exception e) {
        System.err.println("Exception attempting to read properties from config.properties.");
        logger.error("Exception attempting to read properties from config.properties.");
        e.printStackTrace();
    }
    }
    
    // Providing fall-back cuttoff values if config.properties doesn't load
    // or doesn't have the named props. But what happens when Integer.parseInt throws an exception?
    private static final int MIN_CONTENT_LENGTH = Integer.parseInt(configProperties.getProperty("WETprocessor.min.content.length", "100"));
    private static final int MIN_LINE_COUNT= Integer.parseInt(configProperties.getProperty("WETprocessor.min.line.count", "2"));
    private static final int MIN_CONTENT_LENGTH_WRAPPED_LINE = Integer.parseInt(configProperties.getProperty("WETprocessor.min.content.length.wrapped.line", "500"));
    private static final int MIN_SPACES_IN_A_WRAPPED_LINE = Integer.parseInt(configProperties.getProperty("WETprocessor.min.spaces.per.wrapped.line", "10"));

    // File paths shared across WETProcessor instances
    private static File discardFolder;
    private static File keepFolder;
    private static File keepURLsFile;
    private static File discardURLsFile;

    // WARC WET header lines and header line prefixes of interest
    static final String WARC_RECORD_START = "WARC/1.0";
    static final String WARC_INFO_HEADER = "WARC-Type: warcinfo";
    static final String WARC_TARGET_URI_HEADER_PREFIX = "WARC-Target-URI:";
    static final String WARC_CONTENT_LENGTH_HEADER_PREFIX = "Content-Length:";
    
    // Keep a count of all the records that all WETProcessors instantiated
    // by our main method combined have processed
    //private static int recordCount = 0;

    private final File outputFolder;
    private final String WETFileID;


    /**
     * WET processor processes a single warc.wet file containing multiple WET records
     * containing text identified as primary langcode=mri. Each individual WET record is written
     * out to a uniquely named file in either the keep or discard folder depending on the WET
     * record's content length and number of lines of actual content (excluding WARC headers).
     */
    public WETProcessor(File inFile, File outFolder) {
    this.outputFolder = outFolder;
    
    StringBuilder record = null;
    String line = null; 
    boolean readingRecord = false;

    String WARCtargetURI = "";
    
    int recordCount = 0;
    
    int contentLength = -1; // of record
    int lineCount = -1; // actual number of non-empty lines in record body (i.e. excludes WET/WARC headers)

    // We just want a unique recordID prefix, which we get from the wet file name suffix:
    // inFile name looks something like MAORI-CC-2019-30-20190902100139-000000.warc.wet
    // the prefix will be everything after the last hyphen and without file extension,
    // so "000000" in our example. Then suffix the recordCount (keeping track of the current
    // WET record) to get a unique filename to store each WET record into.

    String fileID = inFile.getName();
    fileID = fileID.substring(fileID.lastIndexOf("-")+1);
    fileID = fileID.substring(0, fileID.indexOf("."));
    this.WETFileID = fileID;

        
    // read from WETfile
    try (
         BufferedReader reader = new BufferedReader(new FileReader(inFile));
         BufferedWriter keepURLsWriter = new BufferedWriter(new FileWriter(keepURLsFile, true));
         BufferedWriter discardURLsWriter = new BufferedWriter(new FileWriter(discardURLsFile, true)); // true to append
         ) {
        
        while((line = reader.readLine()) != null) { // readLine removes newline separator

        if(line.startsWith(WARC_INFO_HEADER)) { // "WARC-Type: warcinfo"
            readingRecord = false;
            record = null; // drop this record, which is just an info record not actual web page's text
            recordCount--;
            continue;
        }
        
        if(line.startsWith(WARC_RECORD_START)) { // "WARC/1.0" means finished previous WET record
            // process any previous record
            if(record != null) {
            processWETrecord(keepURLsWriter, discardURLsWriter,
                     recordCount, contentLength, lineCount,
                     WARCtargetURI, record.toString());
            record = null;
            contentLength = -1;
            lineCount = -1;
            }
            
            recordCount++;          
            // get ready to start a new record
            readingRecord = true;
            record = new StringBuilder();
        }

        if(readingRecord) { // append current line to current record

            if(line.startsWith(WARC_TARGET_URI_HEADER_PREFIX)) { // "WARC-Target-URI:"
            // get and store the value
            WARCtargetURI = line.substring(WARC_TARGET_URI_HEADER_PREFIX.length()).trim();
            }
            
            record.append(line + "\n"); // add back (unix style) line ending

            // if the line is non-empty
            // AND if we've started counting lines, which happens only when the current
            // line is past WARC/WET headers and we're into the actual body portion
            // of the WET record,
            // start incrementing the line counter.
            if(lineCount >= 0 && !line.trim().equals("")) {
            lineCount++;
            }
            else if(line.startsWith(WARC_CONTENT_LENGTH_HEADER_PREFIX)) { // "Content-Length:"
            String contentLengthStr = line.substring(WARC_CONTENT_LENGTH_HEADER_PREFIX.length()).trim();
            contentLength = Integer.parseInt(contentLengthStr);
            lineCount = 0;
            }
            
        }
        
        }

        // flush the last record. If it was a warcinfo record, record would be null here
        if(record != null) {
        processWETrecord(keepURLsWriter, discardURLsWriter,
                 recordCount, contentLength, lineCount,
                 WARCtargetURI, record.toString());
        record = null;
        }
        
    } catch(IOException ioe) {
        ioe.printStackTrace();
    }
    }

    /**
     * Determines if a WET record belongs in the keep or discard pile depending on if it
     * contains enough text, based on contentLength and line count of the record body.
     * Then writes out the WET record to a uniquely named file in the keep or discard folder,
     * and writes out the WET record's URL to the keepURLs.txt file or discardURLs.txt file.
     */
    private void processWETrecord(BufferedWriter keepURLsWriter, BufferedWriter discardURLsWriter,
                  int recordID, int contentLength, int lineCount,
                  String recordURI, String record)
    {
    System.err.println("WET #" + this.WETFileID + " record #" + recordID
               + " - contentLength: " + contentLength
               + " - lineCount: " + lineCount);
    System.err.println("URI: " + recordURI);
    //System.err.println(record);
    //System.err.println("--------------------------");

    String paddedFileName = String.format("%04d.txt", recordID);
    
    File parentFolder = null;

    // want to match "product(s)" but not "production"
    
    //if(recordURI.matches(".*/?product[^a-rt-z].*")) {//if(recordURI.matches(".*/?products?/?.*")) {
    if(recordURI.contains("product") && !recordURI.contains("production")) {

        // don't want a "translated" product site/online store
        // These curiously often tend to have "product(s)" in the URL
        parentFolder = WETProcessor.discardFolder;
    }   
    else if(lineCount >= MIN_LINE_COUNT && contentLength >= MIN_CONTENT_LENGTH) {
        parentFolder = WETProcessor.keepFolder;
        System.err.println("@@@KEEPING");
    } else if(contentLength >= MIN_CONTENT_LENGTH_WRAPPED_LINE) {
        int countSpaces = 0;
        for(int i = 0; i < record.length(); i++) {
        if(record.charAt(i) == ' ') countSpaces++;
        }
        if(countSpaces >= MIN_SPACES_IN_A_WRAPPED_LINE) {
        // So we have at least 500 chars (possibly on a single wrapped line)
        // containing at least 10 spaces. Such a record is also worth keeping.
        parentFolder = WETProcessor.keepFolder;
        }
    }

    // if parentFolder still not set, set to discard pile folder
    if(parentFolder == null) {
        parentFolder = WETProcessor.discardFolder;
        System.err.println("@@@DISCARDING");
    }

    try {
        if (parentFolder == WETProcessor.keepFolder) {
        keepURLsWriter.write(recordURI + "\n");
        } else {
        discardURLsWriter.write(recordURI + "\n");
        }
    } catch(Exception e) {
        System.err.println("Unable to write URL");
        e.printStackTrace();
    }
   
    System.err.println("--------------------------");
    
    File outFile = new File(parentFolder, this.WETFileID + "-" + paddedFileName);
    
    try (BufferedWriter writer = new BufferedWriter(new FileWriter(outFile))) {
        writer.write(record);
        // Try-with-resources examples don't call close() explicitly:
        // https://docs.oracle.com/javase/tutorial/essential/exceptions/tryResourceClose.html
        //writer.close();
    } catch(IOException ioe) {
        ioe.printStackTrace();
        System.err.println("\n@@@@@@@@@ Error writing to file " + outFile);
    }
    }
    

    //public static int getRecordCount() { return recordCount; }
    
    public static void printUsage() {
    System.err.println("Run this program as:");
    System.err.println("\tWetProcessor <folder containing wet(.gz) files> <output folder path>");   
    }

    /** Filename filter to only list warc.wet files or else warc.wet.gz files
     * for which unzipped warc.wet equivalents don't yet exist.
     */
    private static class WETFilenameFilter implements FilenameFilter {
    
    public boolean accept(File dir, String name) {
        if(name.endsWith(".warc.wet")) {
        logger.debug("Will include " + name + " for processing.");
        return true;
        }

        if(name.endsWith(".warc.wet.gz")) {
        String nameWithoutGZext = name.substring(0, name.lastIndexOf(".gz"));
        File unzippedVersion = new File(dir, nameWithoutGZext);
        if(unzippedVersion.exists()) {
            logger.debug("--- Unzipped version " + unzippedVersion + " exists.");
            logger.debug("Skipping " + name);
            return false; // don't count gzipped version if unzipped version exists.
        }
        else {
            logger.debug("Only zipped version " + name + " exists.");
            return true; // No unzipped version, so have to work with gzipped version
        }
        }

        // we're not even interested in any other file extensions
        logger.debug("Not a WET file. Skipping " + name);
        return false;
    }
    }
    
    public static void main(String[] args) {
    if(args.length != 2) {
        printUsage();
        return;
    }

    
    File WETFileDir = new File(args[0]);
    if(!WETFileDir.exists() || !WETFileDir.isDirectory()) {
        System.out.println("Error: " + args[0] + " does not exist or is not a directory");
        return;
    }
    
    File outFolder = new File(args[1]);
    if(!outFolder.exists() || !outFolder.isDirectory()) {
        System.out.println("Error: " + args[1] + " does not exist or is not a directory.");
        return;
    }   

    // static folders and files to be shared across all WETProcessor instances
    WETProcessor.discardFolder = new File(outFolder, "discard");
    if(!WETProcessor.discardFolder.exists()) {
        WETProcessor.discardFolder.mkdir();
    }   
    WETProcessor.keepFolder = new File(outFolder, "keep");
    if(!WETProcessor.keepFolder.exists()) {
        WETProcessor.keepFolder.mkdir();
    }

    WETProcessor.keepURLsFile = new File(outFolder, "keepURLs.txt");    
    if(WETProcessor.keepURLsFile.exists() && !WETProcessor.keepURLsFile.delete()) {
        System.err.println("Unable to delete " + WETProcessor.keepURLsFile + ". Unable to proceed.");
        return;
    }
    WETProcessor.discardURLsFile = new File(outFolder, "discardURLs.txt");
    if(WETProcessor.discardURLsFile.exists() && !WETProcessor.discardURLsFile.delete()) {
        System.err.println("Unable to delete " + WETProcessor.discardURLsFile + ". Unable to proceed.");
        return;
    }
    
    // Will list all the warc.wet files in the input directory or else their gzipped versions
    File[] WETFiles = WETFileDir.listFiles(new WETFilenameFilter());

    for(int i = 0; i < WETFiles.length; i++) {
        File WETFile = WETFiles[i];     
        logger.debug("Processing WETfile: " + WETFile);

        // Any .gz files listed means they haven't been unzipped yet. So unzip.
        String WETFilename = WETFile.toString();
        if(WETFilename.endsWith(".gz")) {
        File GZippedWETFile = WETFile;
        String WETGZippedFilename = WETFilename;
        WETFilename = WETFilename.substring(0, WETFilename.lastIndexOf(".gz"));     

        WETFile = new File(WETFilename);
        Utility.unzipFile(GZippedWETFile, WETFile);
        }
        // hereafter all WETFiles should refer to the unzipped version
        // Check the unzipped WETFile exists        

        if(!WETFile.exists() || !WETFile.isFile()) {
        System.err.println("Error: " + WETFile + " does not exist (failure to unzip?)");
        logger.error("Error: " + WETFile + " does not exist (failure to unzip?)");
        return;
        }

        // Finally, we can process this WETFile's records into the keep and discard pile
        logger.debug("Off to process " + WETFile);
        WETProcessor processor = new WETProcessor(WETFile, outFolder);      
        
    }
    
    return; 
    }
}