D-Net

package eu.dnetlib.data.mapreduce.hbase.lodImport.utils;

import java.net.URI;

import java.sql.Connection;

import java.sql.Statement;

import java.util.List;

import org.apache.hadoop.conf.Configuration;

import org.apache.log4j.Logger;

import org.json.JSONArray;

import org.json.JSONObject;

import com.jolbox.bonecp.BoneCPDataSource;

public class RDFizer {

	private static Logger log = Logger.getLogger(RDFizer.class);

	public RDFizer() {

	}

//	public static String RDFizeEntityRow(List<String> row,VirtGraph graph, JSONObject mappings, Configuration conf, MapCountries mapCountries,MapLanguages mapLanguages){

	public static String RDFizeEntityRow(List<String> row,BoneCPDataSource ds, JSONObject mappings, Configuration conf, MapCountries mapCountries,MapLanguages mapLanguages){

		//		VirtModel model = new VirtModel(graph);

		String graph = conf.get("lod.defaultGraph");

		String insertQuery ="";

		String baseURI = conf.get("lod.baseURI");

		String id = row.get(1).toString();

		String type = row.get(0).toString();

		String resourceURI = baseURI+type+"/"+id;

//		String askQuery = "SELECT ?o FROM <"+conf.get("lod.defaultGraph")+"> WHERE {<"+resourceURI+"> ?p ?o}";

//		VirtuosoQueryExecution vqe = VirtuosoQueryExecutionFactory.create (askQuery, graph);

		boolean ask = false;

		try{

			ask = resourceExists(ds.getConnection(), resourceURI, graph);

		}catch(Exception e){

			log.error("Could not ASK "+e.toString(),e);

		}

//		ResultSet rs = vqe.execSelect();

//		if(rs.hasNext()) ask=true;

//		vqe.close();

		if(ask){

			try{

				deleteResource(ds.getConnection(), resourceURI, graph);

			}catch(Exception e){

				log.error("Could not DELETE  "+resourceURI+" "+e.toString(),e);

			}

//			String deleteQueryString ="DELETE FROM <"+conf.get("lod.defaultGraph")+"> {<"+resourceURI+"> ?p ?o} WHERE {<"+resourceURI+"> ?p ?o}" ;

//			VirtuosoUpdateRequest vur = VirtuosoUpdateFactory.create(deleteQueryString, graph);

////			System.out.println("DELETED:\t"+deleteQueryString);

//			vur.exec();

		}

	    JSONArray typeMappings = mappings.getJSONArray(type);

		for(int i=0; i<row.size(); i++){

			JSONObject propertyObject = typeMappings.getJSONObject(0);

            String index = new Integer(i).toString();

            String propertyString;

            if(type.equals("project") && i>24) continue;

            if(type.equals("organization") && i>11) continue;

            try{

            	propertyString = propertyObject.getString(index);

            }catch(Exception e){

            	log.error("Could not get the property for type "+type+" :  "+e.toString(),e);

            	continue;

            }

			if(i==0){

				String resourceType = propertyObject.getString(propertyString);

				insertQuery+="<"+resourceURI+"> <"+propertyString+"> <"+resourceType+">";

				continue;

			}

			String value = row.get(i).trim();

			if(value.trim().equals("null") || value==null || value.trim().equals("")) continue;

			if(i==16 && type.equals("result")  && !value.equals("und") || i==11 && type.equals("person")  && !value.equals("und") || i==9 && type.equals("organization") && !value.equals("und")){

				try{

					String countryURI = mapCountries.getCountryURI(value);

					URI uri = new URI(countryURI);

					if(countryURI.equals("")){

						insertQuery+="; <"+propertyString+"> \""+value+"\"";

						continue;

					}

					insertQuery+="; <"+propertyString+"> <"+uri+">";

//					log.info("COUNTRY "+countryURI+"  FROM  "+value);

				}catch(Exception e){

					insertQuery+="; <"+propertyString+"> \""+value+"\"";

					log.error("No country URI for: "+e.toString(),e);

					continue;

				}

				continue;	

			}

			if(i==9 && type.equals("result") && !value.equals("und")){

				try{

					String langURI = mapLanguages.getLangURI(value);

					URI uri = new URI(langURI);

					if(langURI.equals("")){

						insertQuery+="; <"+propertyString+"> \""+value+"\"";

						continue;

					}

					insertQuery+="; <"+propertyString+"> <"+uri+">";

//					log.info("LANG "+langURI+"  FROM  "+value.trim());

				}catch(Exception e){

					insertQuery+="; <"+propertyString+"> \""+value+"\"";

					log.error("No Language URI for: "+e.toString(),e);

					continue;

				}

				continue;	

			}

			if(value.contains(conf.get("lod.seperator"))){

				String[] splittedValue = value.split(conf.get("lod.seperator"));

				for(String v:splittedValue){

					v= v.replace(conf.get("lod.seperator"), "").trim();

					insertQuery+="; <"+propertyString+"> \""+v+"\"";

				}

			}else{

				if(value.startsWith("http://")){

					try{

						URI uri = new URI(value);

						insertQuery+="; <"+propertyString+"> <"+uri+">";

					}catch(Exception e){

						insertQuery+="; <"+propertyString+"> \""+value+"\"";

						log.error("NOT URI "+e.toString(),e);

						continue;

					}

				}

				else insertQuery+="; <"+propertyString+"> \""+value+"\"";

			}	

		}

		insertQuery+=".";

		return insertQuery;

	}

//	public static String RDFizeRelationRow(List<String> row, VirtGraph graph, JSONObject mappings, Configuration conf){

	public static String RDFizeRelationRow(List<String> row, JSONObject mappings, Configuration conf){

			String insertQuery = "";

			String baseURI = conf.get("lod.baseURI");

			JSONObject typeMappings = mappings.getJSONArray(row.get(0)).getJSONObject(0);

			String sourceType = typeMappings.getString("sourceType");

			String sourceId = typeMappings.getString("sourceId");

			String targetType = typeMappings.getString("targetType");

			String targetId = typeMappings.getString("targetId");

			String property = typeMappings.getString("property");

			String sourceURI = baseURI+row.get(Integer.parseInt(sourceType))+"/"+row.get(Integer.parseInt(sourceId));

			String targetURI = baseURI+row.get(Integer.parseInt(targetType))+"/"+row.get(Integer.parseInt(targetId));

			insertQuery= "<"+sourceURI+"> <"+property+"> <"+targetURI+">. ";

			return insertQuery;

		}

	public static boolean resourceExists(Connection conn, String resourceURI, String graph){

		boolean exists = false;

		String askQuery = "SPARQL SELECT ?o FROM <"+graph+"> WHERE {<"+resourceURI+"> ?p ?o}";

		Statement stmt;

		try {

			stmt = conn.createStatement();

//			ResultSet rs=	stmt.executeQuery(askQuery);

			exists = stmt.execute(askQuery);

//			if(rs.next()) exists=true;

			stmt.close();

			conn.commit();

			conn.close();

		} catch (Exception e) {

			log.error("Virtuoso ask Query failed. Checkpoint was "+ askQuery +"\n" + e.toString(), e);

		}

		return exists;

	}

	public static void deleteResource(Connection conn, String resourceURI, String graph){

		String deleteQueryString ="SPARQL DELETE FROM <"+graph+"> {<"+resourceURI+"> ?p ?o} WHERE {<"+resourceURI+"> ?p ?o}" ;

		Statement stmt;

		try {

			stmt = conn.createStatement();

			stmt.execute(deleteQueryString);

			stmt.close();

			conn.commit();

			conn.close();

		} catch (Exception e) {

			log.error("Virtuoso ask Query failed. Checkpoint was "+ deleteQueryString +"\n" + e.toString(), e);

		}

	}

	public static void clearGraph(String graph, Connection conn){

		String clearQuery="SPARQL DEFINE sql:log-enable 3 CLEAR GRAPH  <"+graph+">";

		Statement stmt;

		try {

			stmt = conn.createStatement();

			stmt.execute(clearQuery);

			stmt.close();

			conn.commit();

			conn.close();

		} catch (Exception e) {

			log.error("Virtuoso FAILED TO CLEAR graph "+ clearQuery +"\n" + e.toString(), e);

		}

	}

	public static void setCheckpoint(Connection conn, String checkpointValue){

		Statement stmt;

		try {

			stmt = conn.createStatement();

			stmt.execute("checkpoint_interval("+checkpointValue+");");

			stmt.close();

			conn.commit();

			conn.close();

		} catch (Exception e) {

			log.error("Virtuoso set checkpoint failed. Checkpoint was "+ checkpointValue +"\n" + e.toString(), e);

		}

	}

}