package jp.ac.osaka_u.sanken.sparql;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import jp.ac.osaka_u.sanken.sparql.edit.AllegroEditor;

import com.franz.agraph.jena.AGModel;
import com.franz.agraph.jena.AGQuery;
import com.franz.agraph.jena.AGQueryExecutionFactory;
import com.franz.agraph.jena.AGQueryFactory;
import com.hp.hpl.jena.query.Query;
import com.hp.hpl.jena.query.QueryExecution;
import com.hp.hpl.jena.query.QueryExecutionFactory;
import com.hp.hpl.jena.query.QueryFactory;
import com.hp.hpl.jena.query.QuerySolution;
import com.hp.hpl.jena.query.ResultSet;
import com.hp.hpl.jena.query.ResultSetFactory;
import com.hp.hpl.jena.rdf.model.RDFNode;
import com.hp.hpl.jena.rdf.model.impl.ResourceImpl;
/**
 * Sparqlとのデータ
 * @author kato
 *
 */
public class SparqlAccessor {
	
	private EndpointSettings setting;

	/** TSV形式出力 */
	public static final int OUTPUT_TYPE_TSV = 0;
	
	public static final int OUTPUT_TYPE_CSV = 1;

	public static final int OUTPUT_TYPE_XLS = 2;
	
	/** キーワード検索で、項目すべてを検索対象とする（未実装） */
	public static final int FIND_TARGET_ALL = 0;
	/** キーワード検索で、subjectを検索対象とする */
	public static final int FIND_TARGET_SUBJECT = 1;
	/** キーワード検索で、objectを検索対象とする */
	public static final int FIND_TARGET_OBJECT = 2;
	/** キーワード検索で、ラベルobjectを検索対象とする */
	public static final int FIND_TARGET_LABEL_OBJECT = 3;

	
	private SparqlQueryListener queryListener;
	
	private HashMap<Integer, String> separatorMap;

	public SparqlAccessor(EndpointSettings endpoint, SparqlQueryListener queryListener){
		this.queryListener = queryListener;

		this.setting = endpoint;
		
		separatorMap = new HashMap<Integer, String>();
		separatorMap.put(OUTPUT_TYPE_TSV, "\t");
		separatorMap.put(OUTPUT_TYPE_CSV, ",");
}

	public SparqlAccessor(EndpointSettings endpoint){
		this(endpoint, null);
	}
	
	
	public EndpointSettings getSetting(){
		return this.setting;
	}
	/**
	 * queryを叩いて結果を返す
	 * @param query
	 * @return
	 */
	private QueryExecution makeQuery(String queryString){
		QueryExecution qe = null;
		if (this.setting.isEditable()){
			// TODO 本来ならendpointの実装に応じてfactoryから固有のAccessorを取得する形にすべきだが、
			// 現状は「editable=true」の場合はAllegroGraphに決め打ちしている。
			AllegroEditor ae;
			try {
				ae = new AllegroEditor(this, this.setting.getRepositoryURL(), setting.getRepository(), setting.getUser(), setting.getPass());
				AGQuery sparql = AGQueryFactory.create(queryString);
				qe = AGQueryExecutionFactory.create(sparql, (AGModel)ae.getModel());
			} catch (Exception e) {
				e.printStackTrace();
			}
		} else {
			Query query = QueryFactory.create(queryString);
			qe = QueryExecutionFactory.sparqlService(setting.getEndpoint(), query);
		}

		return qe;
	}

	/**
	 * Threadを起動してquery実行を行う。結果はlistenerに返る。
	 * @param queryString	query
	 * @param resultListener		結果を受け取るlistener
	 */
	public void executeQuery(String queryString, SparqlResultListener resultListener){
		
		Thread thread = new QueryThread(queryString, resultListener){
			public void run(){
				try {
					getSparqlResultListener().resultReceived(new SparqlResultSet(executeQuery(getQueryString())));
				} catch (Exception e) {
					throw new RuntimeException(e);
				}
			}
		};
		thread.setUncaughtExceptionHandler(resultListener);
		thread.start();
	}
	

	/**
	 * query実行を行う。結果は戻り値として返る。
	 * @param queryString
	 * @return
	 * @throws Exception 
	 */
	public List<Map<String, RDFNode>> executeQuery(String queryString) throws Exception{
		
		List<Map<String, RDFNode>> ret = new ArrayList<Map<String, RDFNode>>();
		QueryExecution qe = makeQuery(queryString);

		if (qe == null){
			throw new Exception("Can't connect to endpoint");
		}
		
		try {
//			System.out.println("query:"+queryString);
			if (this.queryListener != null){
				queryListener.sparqlExecuted(queryString);
			}
			ResultSet results = null;
			try {
				if (!setting.isUseCustomParam()){
					results = qe.execSelect();
				} else {
					results = customQuery(queryString);
				}
			} catch(Exception e){
				e.printStackTrace();
//				results = customQuery(queryString);
			}
			List<String> keys = results.getResultVars();
			while(results.hasNext()){
				QuerySolution result = results.next();
				HashMap<String, RDFNode> map = new HashMap<String, RDFNode>(); 
				for (String key : keys){
					RDFNode node = result.get(key);
					map.put(key, node);
				}
				ret.add(map);
			}
		} catch(Exception e){
			e.printStackTrace();
			throw e;
		} finally {
			qe.close();
		}
		
		return ret;
	}
	
	private ResultSet customQuery(String query) throws Exception {
	
		URL url = new URL(this.setting.getEndpoint() + "?" +setting.getQueryKey() + "=" + URLEncoder.encode(query, setting.getEncoding()) + "&" + setting.getOption());//POSTするデータ
		HttpURLConnection conn = (HttpURLConnection)url.openConnection();
		conn.setRequestProperty("Accept-Language", "ja");// ヘッダを設定
		conn.setRequestProperty("Referer", setting.getEndpoint());// ヘッダを設定

		int resultType = setting.getResultType();
		ResultSet ret = null;
		
		if (resultType == EndpointSettings.RESULT_TYPE_JSON){
			ret = ResultSetFactory.fromJSON(conn.getInputStream());
		} else if (resultType == EndpointSettings.RESULT_TYPE_XML){
			ret = ResultSetFactory.fromXML(conn.getInputStream());
		} else if (resultType == EndpointSettings.RESULT_TYPE_SSE){
			ret = ResultSetFactory.fromSSE(conn.getInputStream());
		} else if (resultType == EndpointSettings.RESULT_TYPE_TSV){
			ret = ResultSetFactory.fromTSV(conn.getInputStream());
		}
		conn.disconnect();

		return ret;
	}


	/**
	 * word文字列を含むsubjectを取得して返す
	 * @param word
	 * @param fullMatch	完全一致検索か
	 * @param limit		検索最大数
	 * @param offset	検索オフセット
	 * @param type	検索対象種別
	 * @return
	 * @throws Exception 
	 */
	public SparqlResultSet findSubject(String word, boolean fullMatch, Integer limit, Integer offset, int type) throws Exception{
		String query;
		if (!fullMatch){
			if (type == SparqlAccessor.FIND_TARGET_SUBJECT){
				query = 
					"select distinct ?s where {\n" +
					"?s <http://www.w3.org/2000/01/rdf-schema#label> ?o \n" +
					"FILTER(regex(str(?s), \""+word+"\", \"m\"))\n" +
					"}";
			} else if (type == SparqlAccessor.FIND_TARGET_OBJECT){
				query = 
					"select distinct ?s where {\n" +
					"?s ?p ?o \n" +
					"FILTER(regex(str(?o), \""+word+"\", \"m\"))\n" +
					"}";
			} else if (type == SparqlAccessor.FIND_TARGET_LABEL_OBJECT){
				query = 
					"select distinct ?s where {\n" +
					"?s <http://www.w3.org/2000/01/rdf-schema#label> ?o \n" +
					"FILTER(regex(str(?o), \""+word+"\", \"m\"))\n" +
					"}";
			} else {
				// TODO
				query = 
					"select distinct ?s where {" +
					"?s <http://www.w3.org/2000/01/rdf-schema#label> ?o " +
					"FILTER(regex(str(?o), \""+word+"\", \"m\"))" +
					"}";
			}
		} else {
			if (type == SparqlAccessor.FIND_TARGET_OBJECT){
				query = 
					"select distinct ?s where \n{" +
					"{?s ?p \""+word+"\"} UNION \n" +
					"{?s ?p \""+word+"\"@en} UNION \n" +
					"{?s ?p \""+word+"\"@ja} ";
				String[] namespaces = setting.getNamespaceList();
				if (namespaces != null && namespaces.length > 0){
					query += "UNION \n";
				}
				for (int i=0; i<namespaces.length; i++){
					String ns = namespaces[i];
					query += "{?s ?p <"+ns+"/"+word+">} ";
					if (i != namespaces.length-1){
						query += "UNION \n";
					} else {
						query += "\n";
					}
				}
				
				query += "}";
			} else if (type == SparqlAccessor.FIND_TARGET_LABEL_OBJECT){
				query = 
					"select distinct ?s where \n{" +
					"{?s <http://www.w3.org/2000/01/rdf-schema#label> \""+word+"\"} UNION \n" +
					"{?s <http://www.w3.org/2000/01/rdf-schema#label> \""+word+"\"@en} UNION \n" +
					"{?s <http://www.w3.org/2000/01/rdf-schema#label> \""+word+"\"@ja} ";
				
				query += "}";
			} else if (type == SparqlAccessor.FIND_TARGET_SUBJECT){
				String[] namespaces = setting.getNamespaceList();
				
				List<Map<String, RDFNode>> ret = new ArrayList<Map<String,RDFNode>>();
				for (int i=0; i<namespaces.length; i++){
					String ns = namespaces[i];
					query = "select ?o where {\n"+
					"<"+ns+"/"+word+"> ?p ?o \n";
					query += "}";
					List<Map<String, RDFNode>> temp = executeQuery(query);
					if (temp != null && temp.size() > 0){
						HashMap<String, RDFNode> node = new HashMap<String, RDFNode>();
						node.put("s", new ResourceImpl(ns+"/"+word));
						ret.add(node);
					}
					
				}
				// TODO 結果の最大数はnamespaces.sizeなので厳密にはLIMIT, OFFSETも定義できるようにしておいた方が良い
				return new SparqlResultSet(ret, false);
			} else {
				// TODO
				query = 
					"select distinct ?s where {\n" +
					"{?s ?p \""+word+"\"} UNION \n" +
					"{?s ?p \""+word+"\"@en} UNION \n" +
					"{?s ?p \""+word+"\"@ja}\n" +
					"}";
			}
		}
		if (limit != null && limit > 0){
			query +="\n LIMIT " + String.valueOf(limit+1);
		}
		if (offset != null && offset > 0){
			query += "\n OFFSET " + String.valueOf(offset);
		}

		List<Map<String, RDFNode>> result = executeQuery(query);
		SparqlResultSet ret = new SparqlResultSet(result);
		if (result != null && result.size() > limit){
			result = result.subList(0, limit);
			ret.setResult(result);
			ret.setHasNext(true);
		}
		
		return ret;
	}

	/**
	 * Threadを起動してword文字列を含むsubjectを取得して返す。結果はlistenerに返る。
	 * @param word			検索文字列
	 * @param resultListener		結果を受け取るlistener
	 */
	public void findSubject(String word, boolean fullMatch, Integer limit, Integer offset, int type, SparqlResultListener resultListener){
		
		Thread thread = new QueryThread(word, new Object[]{new Boolean(fullMatch), limit, offset, new Integer(type)}, resultListener){
			public void run(){
				try {
					Boolean fullMatch = (Boolean)((Object[])getOption())[0];
					Integer limit = (Integer)((Object[])getOption())[1];
					Integer offset = (Integer)((Object[])getOption())[2];
					Integer type = (Integer)((Object[])getOption())[3];
					getSparqlResultListener().resultReceived(findSubject(getQueryString(), fullMatch, limit, offset, type));
				} catch(Exception e){
					throw new RuntimeException(e);
				}
			}
		};
		thread.setUncaughtExceptionHandler(resultListener);
		thread.start();
	}
	
	/**
	 * 指定されたsubjectを持つtriple（subjectは確定しているのでpropertyとobjectのみ）を返す
	 * @param triple
	 * @return
	 * @throws Exception 
	 */
	public List<Map<String, RDFNode>> findTripleFromSubject(String subject) throws Exception{
		String query = 
			"select ?p ?o where {\n" +
			"<" + subject + "> ?p ?o\n"+
			"}";
		
		return executeQuery(query);
	}

	public void findTripleFromSubject(String subject, SparqlResultListener listener){
		
		Thread thread = new QueryThread(subject, listener){
			public void run(){
				try {
					getSparqlResultListener().resultReceived(new SparqlResultSet(findTripleFromSubject(getQueryString())));
				} catch(Exception e){
					throw new RuntimeException(e);
				}
			}
		};
		thread.setUncaughtExceptionHandler(listener);
		thread.start();
	}

	/**
	 * 指定したストリームに結果を文字列として出力する
	 * @param results
	 * @param type
	 * @param stream
	 * @return
	 */
	public boolean saveResult(List<Map<String, RDFNode>> results, int type, OutputStream stream){
		/**
		 * typeによってresult formatを変えて再POSTすることも考えられるが、普遍性がないので保留とし、
		 * 変換は自前で実装する。
		 */
		if (type == OUTPUT_TYPE_TSV || type == OUTPUT_TYPE_CSV){
			String separator = separatorMap.get(type);
			if (separator == null){
				return false;
			}
			SeparatedValuesExporter expo = new SeparatedValuesExporter(separator, results);
			try {
				expo.export(stream);
				return true;
			} catch (IOException e) {
				e.printStackTrace();
			}
			return true;
		}
		
		return false;
	}

	public boolean saveResult(List<Map<String, RDFNode>> results,  int type, File file) throws FileNotFoundException{
		return saveResult(results, type, new FileOutputStream(file));
	}

	
	private class QueryThread extends Thread {
		
		private String queryString;
		private Object option;
		private SparqlResultListener listener;

		public QueryThread(String queryString, Object option, SparqlResultListener listener){
			this.queryString = queryString;
			this.option = option;
			this.listener = listener;
		}

		public QueryThread(String queryString, SparqlResultListener listener){
			this(queryString, null, listener);
		}
		
		protected String getQueryString(){
			return this.queryString;
		}

		protected Object getOption(){
			return this.option;
		}
		
		protected SparqlResultListener getSparqlResultListener(){
			return this.listener;
		}

	}
	


	public static void main(String[] args){

		try {
			URL url = new URL("http://www.wikipediaontology.org/query/?q=" + URLEncoder.encode("select * {?s ?p ?o}", "UTF-8") + "&type=xml&LIMIT=100");//POSTするデータ
			HttpURLConnection conn = (HttpURLConnection)url.openConnection();
			conn.setRequestProperty("User-Agent", "test");// ヘッダを設定
			conn.setRequestProperty("Accept-Language", "ja");// ヘッダを設定
			conn.setRequestProperty("Referer", "http://www.wikipediaontology.org/query/");// ヘッダを設定
			InputStreamReader isr = new java.io.InputStreamReader(conn.getInputStream(), "UTF-8");
			BufferedReader br = new java.io.BufferedReader(isr);

			// 受信したストリームを表示
			String line = null;
			while (null != (line = br.readLine())) {
			    System.out.println(line);
			}

			// ストリームならびに接続をクローズ
			br.close();
			conn.disconnect();
		} catch(Exception e){
			e.printStackTrace();
		}
		
	}
	
	
}
