// Path: wren/src/main/java/uk/bl/wa/wren/model/CrawlURL.java
// public class CrawlURL implements Serializable {
// 
//     private static final long serialVersionUID = 7452432417457114150L;
// 
//     public String url = "";
// 
//     public String method;
// 
//     public String httpVersion;
// 
//     public List<String> cookies = new ArrayList<String>();
// 
//     public Map<String, String> headers = new HashMap<String, String>();
// 
//     public String parentUrl = "";
// 
//     public String pathFromSeed = "S";
// 
//     public boolean forceFetch = false;
// 
//     public boolean isSeed = false;
// 
//     @JsonProperty
//     public ParentUrlMetadata parentUrlMetadata = new ParentUrlMetadata();
// 
//     public CrawlURL toParent() {
//         CrawlURL nurl = new CrawlURL();
//         nurl.parentUrl = this.url;
//         nurl.parentUrlMetadata.pathFromSeed = this.pathFromSeed;
//         return nurl;
//     }
// 
//     public String toJson() throws JsonProcessingException {
//         ObjectMapper mapper = new ObjectMapper();
//         mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
//         return mapper.writeValueAsString(this);
//     }
// 
//     public static CrawlURL fromJson(String json) throws JsonParseException, JsonMappingException, IOException {
//         ObjectMapper mapper = new ObjectMapper();
//         return mapper.readValue(json, CrawlURL.class);
//     }
// 
//     public static CrawlURL fromCrawlURI(CrawlURI curi) {
//         CrawlURL nurl = new CrawlURL();
//         nurl.url = curi.getURI();
//         nurl.method = "GET";
//         nurl.httpVersion = "1.0";
//         nurl.pathFromSeed = curi.getPathFromSeed();
//         nurl.forceFetch = curi.forceFetch();
//         nurl.isSeed = curi.isSeed();
//         return nurl;
//     }
// 
//     public static CrawlURI toCrawlURI(CrawlURL curi) throws URIException {
//         UURI uuri = UURIFactory.getInstance(curi.url);
//         CrawlURI nuri = new CrawlURI(uuri, curi.pathFromSeed, null,
//                 LinkContext.NAVLINK_MISC);
//         nuri.setSeed(curi.isSeed);
//         nuri.setForceFetch(curi.forceFetch);
//         return nuri;
//     }
// }


import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Map;
import java.util.UUID;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;
import org.jwat.common.Base32;
import org.jwat.common.Base64;
import org.jwat.common.ContentType;
import org.jwat.common.Uri;
import org.jwat.warc.WarcConstants;
import org.jwat.warc.WarcDigest;
import org.jwat.warc.WarcFileNaming;
import org.jwat.warc.WarcFileNamingDefault;
import org.jwat.warc.WarcFileWriter;
import org.jwat.warc.WarcFileWriterConfig;
import org.jwat.warc.WarcHeader;
import org.jwat.warc.WarcRecord;
import org.jwat.warc.WarcWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ibm.icu.util.Calendar;
import uk.bl.wa.wren.model.CrawlURL;
/**
 * 
 */
package uk.bl.wa.wren.bolt;





/**
 * @author Andrew Jackson <Andrew.Jackson@bl.uk>
 *
 */
public class HarToWARCWriterBolt implements IRichBolt {

    private static final long serialVersionUID = 6588071684574175562L;

    private static final Logger LOG = LoggerFactory
            .getLogger(HarToWARCWriterBolt.class);

    private WarcFileWriter writer;

    private OutputCollector _collector;

    private String outputFolder = "warcs";

    private String filePrefix = "BL-TEACUP";

    private String hostname = "localhost";

    private boolean useBase32 = true;

    @Override
    public void prepare(Map stormConf, TopologyContext context,
            OutputCollector collector) {
        // Remember the collector
        _collector = collector;
        // extend prefix using entity ID to guarentee no collisions:
        String componentId = context.getThisComponentId();
        int taskId = context.getThisTaskId();
        this.filePrefix = this.filePrefix + "-" + componentId + "-" + taskId;
        // Set up the writer
        WarcFileNaming warcFileNaming = new WarcFileNamingDefault(filePrefix,
                null, hostname, null);
        WarcFileWriterConfig warcFileConfig = new WarcFileWriterConfig(
                new File(outputFolder), true,
                WarcFileWriterConfig.DEFAULT_MAX_FILE_SIZE, false);
        writer = WarcFileWriter.getWarcWriterInstance(warcFileNaming,
                warcFileConfig);
    }

    @Override
    public void execute(Tuple input) {
// Path: wren/src/main/java/uk/bl/wa/wren/model/CrawlURL.java
// public class CrawlURL implements Serializable {
// 
//     private static final long serialVersionUID = 7452432417457114150L;
// 
//     public String url = "";
// 
//     public String method;
// 
//     public String httpVersion;
// 
//     public List<String> cookies = new ArrayList<String>();
// 
//     public Map<String, String> headers = new HashMap<String, String>();
// 
//     public String parentUrl = "";
// 
//     public String pathFromSeed = "S";
// 
//     public boolean forceFetch = false;
// 
//     public boolean isSeed = false;
// 
//     @JsonProperty
//     public ParentUrlMetadata parentUrlMetadata = new ParentUrlMetadata();
// 
//     public CrawlURL toParent() {
//         CrawlURL nurl = new CrawlURL();
//         nurl.parentUrl = this.url;
//         nurl.parentUrlMetadata.pathFromSeed = this.pathFromSeed;
//         return nurl;
//     }
// 
//     public String toJson() throws JsonProcessingException {
//         ObjectMapper mapper = new ObjectMapper();
//         mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
//         return mapper.writeValueAsString(this);
//     }
// 
//     public static CrawlURL fromJson(String json) throws JsonParseException, JsonMappingException, IOException {
//         ObjectMapper mapper = new ObjectMapper();
//         return mapper.readValue(json, CrawlURL.class);
//     }
// 
//     public static CrawlURL fromCrawlURI(CrawlURI curi) {
//         CrawlURL nurl = new CrawlURL();
//         nurl.url = curi.getURI();
//         nurl.method = "GET";
//         nurl.httpVersion = "1.0";
//         nurl.pathFromSeed = curi.getPathFromSeed();
//         nurl.forceFetch = curi.forceFetch();
//         nurl.isSeed = curi.isSeed();
//         return nurl;
//     }
// 
//     public static CrawlURI toCrawlURI(CrawlURL curi) throws URIException {
//         UURI uuri = UURIFactory.getInstance(curi.url);
//         CrawlURI nuri = new CrawlURI(uuri, curi.pathFromSeed, null,
//                 LinkContext.NAVLINK_MISC);
//         nuri.setSeed(curi.isSeed);
//         nuri.setForceFetch(curi.forceFetch);
//         return nuri;
//     }
// }


// Path: wren/src/main/java/uk/bl/wa/wren/bolt/HarToWARCWriterBolt.java
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Map;
import java.util.UUID;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;
import org.jwat.common.Base32;
import org.jwat.common.Base64;
import org.jwat.common.ContentType;
import org.jwat.common.Uri;
import org.jwat.warc.WarcConstants;
import org.jwat.warc.WarcDigest;
import org.jwat.warc.WarcFileNaming;
import org.jwat.warc.WarcFileNamingDefault;
import org.jwat.warc.WarcFileWriter;
import org.jwat.warc.WarcFileWriterConfig;
import org.jwat.warc.WarcHeader;
import org.jwat.warc.WarcRecord;
import org.jwat.warc.WarcWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ibm.icu.util.Calendar;
import uk.bl.wa.wren.model.CrawlURL;

/**
 * 
 */
package uk.bl.wa.wren.bolt;





/**
 * @author Andrew Jackson <Andrew.Jackson@bl.uk>
 *
 */
public class HarToWARCWriterBolt implements IRichBolt {

    private static final long serialVersionUID = 6588071684574175562L;

    private static final Logger LOG = LoggerFactory
            .getLogger(HarToWARCWriterBolt.class);

    private WarcFileWriter writer;

    private OutputCollector _collector;

    private String outputFolder = "warcs";

    private String filePrefix = "BL-TEACUP";

    private String hostname = "localhost";

    private boolean useBase32 = true;

    @Override
    public void prepare(Map stormConf, TopologyContext context,
            OutputCollector collector) {
        // Remember the collector
        _collector = collector;
        // extend prefix using entity ID to guarentee no collisions:
        String componentId = context.getThisComponentId();
        int taskId = context.getThisTaskId();
        this.filePrefix = this.filePrefix + "-" + componentId + "-" + taskId;
        // Set up the writer
        WarcFileNaming warcFileNaming = new WarcFileNamingDefault(filePrefix,
                null, hostname, null);
        WarcFileWriterConfig warcFileConfig = new WarcFileWriterConfig(
                new File(outputFolder), true,
                WarcFileWriterConfig.DEFAULT_MAX_FILE_SIZE, false);
        writer = WarcFileWriter.getWarcWriterInstance(warcFileNaming,
                warcFileConfig);
    }

    @Override
    public void execute(Tuple input) {
        CrawlURL url = (CrawlURL) input.getValueByField("url");