package org.hathitrust.extractedfeatures;

import java.util.ArrayList;
import java.util.Iterator;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.util.DoubleAccumulator;
import org.json.JSONArray;
import org.json.JSONObject;

/*
class PagedJSON implements Function<String, Boolean> {

	private static final long serialVersionUID = 1L;

	public Boolean call(String s) { return s.contains("a"); }
}
 */


class __PerPageJSONForeach implements VoidFunction<JSONObject> 
//public class PagedJSON implements VoidFunction<String> 
{
	private static final long serialVersionUID = 1L;
	
	protected String _input_dir;
	protected String _solr_url;
	protected String _output_dir;
	protected int    _verbosity;
	
	protected DoubleAccumulator _progress_accum;
	protected double            _progress_step;
	
	
	public __PerPageJSONForeach(String input_dir, String solr_url, String output_dir, int verbosity, 
					        DoubleAccumulator progress_accum, double progress_step)
	{
		_input_dir  = input_dir;
		_solr_url   = solr_url;
		_output_dir = output_dir;
		_verbosity  = verbosity;
		
		_progress_accum = progress_accum;
		_progress_step  = progress_step;
	}
	
	public void call(JSONObject solr_add_doc_json) 
	{ 
		String output_json_bz2 = solr_add_doc_json.getString("filename_json_bz2");
		solr_add_doc_json.remove("filename_json_bz2");
		
		boolean random_test = (Math.random()>0.999); // every 1000
		
		if ((_verbosity >=2) && (random_test)) {
			System.out.println("==================");
			System.out.println("Sample output Solr add JSON [random test 1/1000]: " + solr_add_doc_json.toString());
			System.out.println("==================");
		}
		
					
		if (_solr_url != null) {
			if ((_verbosity >=2) && (random_test)) {
				System.out.println("==================");
				System.out.println("Posting to: " + _solr_url);
				System.out.println("==================");
			}
			SolrDocJSON.postSolrDoc(_solr_url, solr_add_doc_json);
		}

		if (_output_dir != null) {
			if ((_verbosity >=2) && (random_test)) {
				System.out.println("==================");
				System.out.println("Saving to: " + _output_dir);
				System.out.println("==================");
			}
			SolrDocJSON.saveSolrDoc(solr_add_doc_json, _output_dir + "/" + output_json_bz2);
		}
	}
	
}