catalogue-util-library/src/main/java/org/gcube/datacatalogue/ckanutillibrary/server/DataCatalogueImpl.java

package org.gcube.datacatalogue.ckanutillibrary.server;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;

import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.gcube.common.authorization.library.provider.SecurityTokenProvider;
import org.gcube.common.scope.impl.ScopeBean;
import org.gcube.common.scope.impl.ScopeBean.Type;
import org.gcube.datacatalogue.ckanutillibrary.ckan.ExtendCkanClient;
import org.gcube.datacatalogue.ckanutillibrary.ckan.MarshUnmarshCkanObject;
import org.gcube.datacatalogue.ckanutillibrary.ckan.MarshUnmarshCkanObject.METHOD;
import org.gcube.datacatalogue.ckanutillibrary.db.DBCaller;
import org.gcube.datacatalogue.ckanutillibrary.gcat.GCatCaller;
import org.gcube.datacatalogue.ckanutillibrary.server.DataCatalogueRunningCluster.ACCESS_LEVEL_TO_CATALOGUE_PORTLET;
import org.gcube.datacatalogue.ckanutillibrary.server.utils.CatalogueUtilMethods;
import org.gcube.datacatalogue.ckanutillibrary.server.utils.url.EntityContext;
import org.gcube.datacatalogue.ckanutillibrary.shared.LandingPages;
import org.gcube.datacatalogue.ckanutillibrary.shared.ResourceBean;
import org.gcube.datacatalogue.ckanutillibrary.shared.RolesCkanGroupOrOrg;
import org.gcube.datacatalogue.ckanutillibrary.shared.State;
import org.json.simple.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import eu.trentorise.opendata.jackan.exceptions.JackanException;
import eu.trentorise.opendata.jackan.internal.org.apache.http.HttpResponse;
import eu.trentorise.opendata.jackan.internal.org.apache.http.client.methods.HttpPost;
import eu.trentorise.opendata.jackan.internal.org.apache.http.entity.ContentType;
import eu.trentorise.opendata.jackan.internal.org.apache.http.entity.StringEntity;
import eu.trentorise.opendata.jackan.internal.org.apache.http.impl.client.CloseableHttpClient;
import eu.trentorise.opendata.jackan.internal.org.apache.http.impl.client.HttpClientBuilder;
import eu.trentorise.opendata.jackan.internal.org.apache.http.util.EntityUtils;
import eu.trentorise.opendata.jackan.model.CkanDataset;
import eu.trentorise.opendata.jackan.model.CkanGroup;
import eu.trentorise.opendata.jackan.model.CkanLicense;
import eu.trentorise.opendata.jackan.model.CkanOrganization;
import eu.trentorise.opendata.jackan.model.CkanPair;
import eu.trentorise.opendata.jackan.model.CkanResource;
import eu.trentorise.opendata.jackan.model.CkanTag;
import eu.trentorise.opendata.jackan.model.CkanUser;
import net.htmlparser.jericho.Renderer;
import net.htmlparser.jericho.Segment;
import net.htmlparser.jericho.Source;

/**
 * This is the Ckan Utils implementation class.
 * 
 * revisited by
 * @author Francesco Mangiacrapa at ISTI-CNR
 * 
 * @author Costantino Perciante at ISTI-CNR (costantino.perciante@isti.cnr.it)
 */
public class DataCatalogueImpl implements DataCatalogue {

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

	private String CKAN_CATALOGUE_URL;
	private String CKAN_DB_NAME;
	private String CKAN_DB_USER;
	private String CKAN_DB_PASSWORD;
	private String CKAN_DB_URL;
	private Integer CKAN_DB_PORT;
	private String PORTLET_URL_FOR_SCOPE;
	private String SOLR_URL;
	private String CKAN_TOKEN_SYS;
	private String CKAN_EMAIL;
	private String URI_RESOLVER_URL;
	private boolean MANAGE_PRODUCT_BUTTON;
	private boolean SOCIAL_POST;
	private boolean ALERT_USERS_ON_POST_CREATION;
	private String CONTEXT;
	private Map<String, String> extendRoleInOrganization;

	public Map<ACCESS_LEVEL_TO_CATALOGUE_PORTLET, String> mapAccessURLToCatalogue;
	

	private static final String CATALOGUE_TAB_ENDING_URL = "/catalogue";
	
	// gCat client
	private GCatCaller gCatCaller;
	
	// db client
	private DBCaller dbCaller;
	
	// ckan client
	private ExtendCkanClient ckanCaller;
	
	// hashmap for ckan api keys
	private ConcurrentHashMap<String, CKANTokenBean> apiKeysMap;

	//http ckan caller
	//private DirectCkanCaller directCkanCaller; //is not needed anymore?
	
	// apikey bean expires after X minutes in the above map
	private static final int EXPIRE_KEY_TIME = 60 * 60 * 1000;


	/**
	 * The ckan catalogue url and database will be discovered in this scope
	 * @param scope
	 * @throws Exception if unable to find datacatalogue info
	 */
	public DataCatalogueImpl(String scope) throws Exception{

		DataCatalogueRunningCluster runningInstance = new DataCatalogueRunningCluster(scope);

		// save information
		CKAN_DB_URL = runningInstance.getDatabaseHosts().get(0).trim();
		CKAN_DB_PORT = runningInstance.getDatabasePorts().get(0);
		CKAN_DB_NAME = runningInstance.getDataBaseName().trim();
		CKAN_DB_USER = runningInstance.getDataBaseUser().trim();
		CKAN_DB_PASSWORD = runningInstance.getDataBasePassword().trim();

		CKAN_TOKEN_SYS = runningInstance.getSysAdminToken().trim();
		CKAN_EMAIL = runningInstance.getEmailCatalogue().trim();

		CKAN_CATALOGUE_URL = runningInstance.getDataCatalogueUrl().get(0).trim();
		PORTLET_URL_FOR_SCOPE = runningInstance.getPortletUrl().trim();
		mapAccessURLToCatalogue = runningInstance.getMapAccessURLToCatalogue();
		MANAGE_PRODUCT_BUTTON = runningInstance.isManageProductEnabled();
		URI_RESOLVER_URL = runningInstance.getUrlResolver();
		SOCIAL_POST = runningInstance.isSocialPostEnabled();
		ALERT_USERS_ON_POST_CREATION = runningInstance.isAlertEnabled();
		SOLR_URL = runningInstance.getUrlSolr();

		LOG.info("In the scope: "+scope+", I read the catalogue URL: " + CKAN_CATALOGUE_URL);
		
		// build the clients
		gCatCaller = new GCatCaller(CKAN_CATALOGUE_URL);
		dbCaller = new DBCaller(CKAN_DB_URL, CKAN_DB_PORT, CKAN_DB_NAME, CKAN_DB_USER, CKAN_DB_PASSWORD);
		ckanCaller = new ExtendCkanClient(CKAN_CATALOGUE_URL);
		//directCkanCaller = new DirectCkanCaller(CKAN_CATALOGUE_URL); //is not needed anymore?
		// init map
		apiKeysMap = new ConcurrentHashMap<String, CKANTokenBean>();
		// save the context
		CONTEXT = scope;
		// extended roles
		extendRoleInOrganization = runningInstance.getExtendRoleInOrganization();
	}
	
	/**
	 * The ckan catalogue url and database will be discovered in this scope.
	 *
	 * @param scope the scope
	 * @param sysAuthentication if true perform the sys authentication
	 * @throws Exception if unable to find datacatalogue info
	 */
	public DataCatalogueImpl(String scope, boolean sysAuthentication) throws Exception{	
		this(scope);
		
		if(sysAuthentication)
			ckanCaller = new ExtendCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);
	}
	
	@Override
	public String getCatalogueUrl() {
		return CKAN_CATALOGUE_URL;
	}


	@Override
	public String getPortletUrl() {

		//PATCHED By Francesco
		ScopeBean context = new ScopeBean(CONTEXT);
		
		if(context.is(Type.INFRASTRUCTURE)) {
			LOG.info("Working with the {} scope returning the path read from GR 'Ckan-Porltet': {}", Type.INFRASTRUCTURE.toString(), PORTLET_URL_FOR_SCOPE);
			return PORTLET_URL_FOR_SCOPE;
		}
		
		String vreNameLower = context.name().toLowerCase();
		//CHECKING IF THE PORTLET URL CONTAINS THE VRE NAME INTO URL
		if(PORTLET_URL_FOR_SCOPE.toLowerCase().contains(vreNameLower)){
			//THE PORLTET URL READ FROM GENERIC RESOUCE 'CkanPortlet' SHOULD BE ALREADY VALID, POITING TO CKAN PORTLET
			return PORTLET_URL_FOR_SCOPE;
		}else{
			//ADDING VRE getApiKeyFromUsernameNAME AND THE SUFFIX 'CATALOGUE_TAB_ENDING_URL' TO URL
			String buildedUrl = PORTLET_URL_FOR_SCOPE.endsWith("/") ? PORTLET_URL_FOR_SCOPE : PORTLET_URL_FOR_SCOPE + "/";
			String defaultSuffix = vreNameLower + CATALOGUE_TAB_ENDING_URL;
			buildedUrl+= defaultSuffix;
			LOG.warn("The Portlet URL read from Generic Resource 'Ckan-Porltet' does not contain the portlet suffix, so I added the default: "+defaultSuffix);
			return buildedUrl;
		}
	}


	@Override
	public String findLicenseIdByLicenseTitle(String chosenLicense) {
		LOG.debug("Requested license id");

		// checks
		checkNotNull(chosenLicense);

		//retrieve the list of available licenses
		List<CkanLicense> licenses = ckanCaller.getLicenseList();

		for (CkanLicense ckanLicense : licenses) {
			if(ckanLicense.getTitle().equals(chosenLicense))
				return ckanLicense.getId();
		}

		return null;
	}

	@Override
	public List<String> getLicenseTitles() {

		LOG.debug("Request for CKAN licenses");
		// get the url and the api key of the user
		List<String> result = new ArrayList<String>();
		//retrieve the list of available licenses
		List<CkanLicense> licenses = ckanCaller.getLicenseList();

		for (CkanLicense ckanLicense : licenses) {

			result.add(ckanLicense.getTitle());
			LOG.debug("License is " + ckanLicense.getTitle() + " and id " + ckanLicense.getId());

		}

		return result;
	}

	@Override
	public List<CkanLicense> getLicenses() {
		LOG.debug("Request for CKAN licenses (original jackan objects are going to be retrieved)");
		//retrieve the list of available licenses
		return ckanCaller.getLicenseList();
	}
	
	@Override
	public CkanDataset getDataset(String datasetId, String apiKey) {
		LOG.info("Request ckan dataset with id " + datasetId);

		// checks
		checkNotNull(datasetId);
		checkArgument(!datasetId.isEmpty());

		try{
			
			if(apiKey!=null && !apiKey.isEmpty()) {
				LOG.info("API-KEY found. Calling the "+ExtendCkanClient.class.getSimpleName());
				ExtendCkanClient client = new ExtendCkanClient(CKAN_CATALOGUE_URL, apiKey);
				return client.getDataset(datasetId);
			}
			
			String authzToken = SecurityTokenProvider.instance.get();
			if(authzToken!=null && !authzToken.isEmpty()) {
				LOG.info("gcube-token found. Calling the gCat client");
				String jsonDataset = gCatCaller.getDatasetForName(datasetId);
				return MarshUnmarshCkanObject.toCkanDataset(jsonDataset, METHOD.TO_READ);
			}
			
			LOG.info("No api-key or gcube-token found. Calling Ckan Client without API-KEY");
			return ckanCaller.getDataset(datasetId);
			
		}catch(Exception e){
			LOG.error("Unable to retrieve such dataset, returning null ...", e);
		}

		return null;
	}
	

	@Override
	public String getUnencryptedUrlFromDatasetIdOrName(String datasetIdOrName) {
		LOG.debug("Request coming for getting dataset url (not encrypted) of dataset with name/id " + datasetIdOrName);

		// checks
		checkNotNull(datasetIdOrName);
		checkArgument(!datasetIdOrName.isEmpty());
		String url = null;
		
		try{
			// get the dataset from name
			ExtendCkanClient client = new ExtendCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);
			CkanDataset dataset = client.getDataset(datasetIdOrName);
			String name = dataset.getName();

			if(dataset != null){

				if(getUriResolverUrl() != null)
					url = getUrlForProduct(CONTEXT, EntityContext.DATASET, name);

				if(url == null || url.isEmpty())
					url = getPortletUrl() + "?" + URLEncoder.encode("path=/dataset/"  + name, "UTF-8");

			}
		}catch(Exception e){
			LOG.error("Error while retrieving dataset with id/name=" + datasetIdOrName, e);
		}			//requestEntity.put("clear_url", Boolean.toString(unencrypted));
		return url;
	}

	
	@Override
	public Map<String, Map<CkanGroup, RolesCkanGroupOrOrg>> getUserRoleByGroup(String username) {
		LOG.info("Get user role by group called. The username is: "+username);
		
		checkNotNull(username);

		Map<String, Map<CkanGroup, RolesCkanGroupOrOrg>> toReturn = new HashMap<String, Map<CkanGroup,RolesCkanGroupOrOrg>>();

		try{

			String ckanUsername = CatalogueUtilMethods.fromUsernameToCKanUsername(username);
			Map<String, RolesCkanGroupOrOrg> partialResult = dbCaller.getGroupsByUserFromDB(ckanUsername);
			
			for (String groupID : partialResult.keySet()) {

				CkanGroup group = ckanCaller.getGroup(groupID);
				HashMap<CkanGroup, RolesCkanGroupOrOrg> subMap = new HashMap<CkanGroup, RolesCkanGroupOrOrg>();
				subMap.put(group, partialResult.get(groupID));
				toReturn.put(groupID, subMap);
			}
			
			LOG.debug("Returning map " + toReturn);
			LOG.info("Found " + toReturn.size() +" group/s with the user "+username);
		}catch(Exception e){
			LOG.error("Failed to retrieve roles of user in his/her own groups",e);
		}

		return toReturn;
		
	}
	
	@Override
	public Map<String, Map<CkanOrganization, RolesCkanGroupOrOrg>> getUserRoleByOrganization(
			String username) {
		LOG.info("Get user role by organization called. The username is: "+username);
		
		checkNotNull(username);

		Map<String, Map<CkanOrganization, RolesCkanGroupOrOrg>> toReturn = new HashMap<String, Map<CkanOrganization,RolesCkanGroupOrOrg>>();

		try{

			String ckanUsername = CatalogueUtilMethods.fromUsernameToCKanUsername(username);
			Map<String, RolesCkanGroupOrOrg> partialResult = dbCaller.getOrganizationsByUserFromDB(ckanUsername);

			for (String orgID : partialResult.keySet()) {

				CkanOrganization org = ckanCaller.getOrganization(orgID);
				HashMap<CkanOrganization, RolesCkanGroupOrOrg> subMap = new HashMap<CkanOrganization, RolesCkanGroupOrOrg>();
				subMap.put(org, partialResult.get(orgID));
				toReturn.put(orgID, subMap);
				LOG.debug("For organisation: " + org.getName() + ", the user "+username+" has role: "+subMap);
			}

			LOG.debug("Returning map " + toReturn);
			LOG.info("Found " + toReturn.size() +" organization/s with the user "+username);
		}catch(Exception e){
			LOG.error("Failed to retrieve roles of user in his/her own groups",e);
		}

		return toReturn;
	}
	
	
	/**
	 * Retrieve an url for the tuple scope, entity, entity name
	 * @param context
	 * @param entityContext
	 * @param entityName
	 */
	private String getUrlForProduct(String context, EntityContext entityContext, String entityName){

		String toReturn = null;

		try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {

			HttpPost httpPostRequest = new HttpPost(getUriResolverUrl());

			JSONObject requestEntity = new JSONObject();
			requestEntity.put("gcube_scope", context);
			requestEntity.put("entity_context", entityContext.toString());
			requestEntity.put("entity_name", entityName);

			StringEntity params = new StringEntity(requestEntity.toJSONString(), ContentType.APPLICATION_JSON);
			httpPostRequest.setEntity(params);

			HttpResponse response = httpClient.execute(httpPostRequest);

			if(response.getStatusLine().getStatusCode() != 200)
				throw new Exception("There was an error while creating an url " + response.getStatusLine());

			toReturn = EntityUtils.toString(response.getEntity());
			LOG.debug("Result is " + toReturn);

		}catch(Exception e){
			LOG.error("Failed to get an url for this product", e);
		}

		return  toReturn;
	}
	
	@Override
	public LandingPages getLandingPages() throws Exception {

		LandingPages landingPages = new LandingPages();
		landingPages.setUrlGroups(PORTLET_URL_FOR_SCOPE + "?path=/group/");
		landingPages.setUrlItems(PORTLET_URL_FOR_SCOPE + "?path=/dataset/");
		landingPages.setUrlOrganizations(PORTLET_URL_FOR_SCOPE + "?path=/organization/");
		landingPages.setUrlTypes(PORTLET_URL_FOR_SCOPE + "?path=/type/");
		return landingPages;
	}
	
	

	@Override
	public String getUriResolverUrl() {
		return URI_RESOLVER_URL;
	}

	/**
	 * Check if the manage product is enabled
	 * @return
	 */
	@Override
	public boolean isManageProductEnabled() {
		return MANAGE_PRODUCT_BUTTON;
	}
	
	@Override
	public List<CkanOrganization> getOrganizationsByUser(String username) {

		LOG.debug("Requested organizations for user " + username);

		// checks
		checkNotNull(username);
		// in order to avoid errors, the username is always converted
		String ckanUsername = CatalogueUtilMethods.fromUsernameToCKanUsername(username);
		// list to return
		List<CkanOrganization> toReturn = new ArrayList<CkanOrganization>();

		try{

			// get the list of all organizations
			List<CkanOrganization> organizations = ckanCaller.getOrganizationList();

			// iterate over them
			for (CkanOrganization ckanOrganization : organizations) {
				// get the list of users in it (if you try ckanOrganization.getUsers() it returns null.. maybe a bug TODO)
				List<CkanUser> users = ckanCaller.getOrganization(ckanOrganization.getName()).getUsers();
				// check if the current user is among them
				for (CkanUser ckanUser : users) {
					
					if(ckanUser.getName().equals(ckanUsername)){

						LOG.debug("User " + ckanUsername + " is into " + ckanOrganization.getName());
						toReturn.add(ckanOrganization);
						break;

					}
				}
			}
		}catch(Exception e){
			LOG.error("Unable to get user's organizations", e);
		}
		return toReturn;
	}
	
	@Override
	public List<CkanGroup> getGroupsByUser(String username) {
		LOG.debug("Requested groups for user " + username);

		// checks
		checkNotNull(username);
		// in order to avoid errors, the username is always converted
		String ckanUsername = CatalogueUtilMethods.fromUsernameToCKanUsername(username);
		// list to return
		List<CkanGroup> toReturn = new ArrayList<CkanGroup>();

		try{

			// get the list of all organizations
			List<CkanGroup> groups = ckanCaller.getGroupList();

			// iterate over them
			for (CkanGroup ckanGroup : groups) {

				List<CkanUser> users = ckanCaller.getGroup(ckanGroup.getName()).getUsers();

				// check if the current user is among them
				for (CkanUser ckanUser : users) {
					
					if(ckanUser.getName().equals(ckanUsername)){

						LOG.debug("User " + ckanUsername + " is into " + ckanGroup.getName());
						toReturn.add(ckanGroup);
						break;

					}
				}
			}
		}catch(Exception e){
			LOG.error("Unable to get user's groups", e);
		}
		return toReturn;
	}
	
	@Override
	public List<String> getOrganizationsIds(){

		List<String> toReturn = new ArrayList<String>();
		List<CkanOrganization> orgs = ckanCaller.getOrganizationList();

		for (CkanOrganization ckanOrganization : orgs) {
			LOG.debug("Retrieved org " + ckanOrganization.getName());
			toReturn.add(ckanOrganization.getId());
		}

		return toReturn;
	}

	@Override
	public List<String> getOrganizationsNames(){

		List<String> toReturn = new ArrayList<String>();
		List<CkanOrganization> orgs = ckanCaller.getOrganizationList();

		for (CkanOrganization ckanOrganization : orgs) {
			LOG.debug("Retrieved org " + ckanOrganization.getName());
			toReturn.add(ckanOrganization.getName());
		}

		return toReturn;
	}

	@Override
	public List<String> getOrganizationsNamesByUser(String username) {

		LOG.debug("Requested organizations for user " + username);

		// checks
		checkNotNull(username);

		// in order to avoid errors, the username is always converted
		String ckanUsername = CatalogueUtilMethods.fromUsernameToCKanUsername(username);
		List<CkanOrganization> orgs = getOrganizationsByUser(ckanUsername);
		List<String> orgsName = new ArrayList<String>();

		for (CkanOrganization ckanOrganization : orgs) {
			orgsName.add(ckanOrganization.getName());
			LOG.debug("Organization name is " + ckanOrganization.getName());
		}

		return orgsName;

	}
	
	@Override
	public String getRoleOfUserInOrganization(String username, String orgName) {

		String toReturn = null;

		String apiKey = getApiKeyFromUsername(username);
		String usernameCkan = CatalogueUtilMethods.fromUsernameToCKanUsername(username);

		try{

			ExtendCkanClient client = new ExtendCkanClient(CKAN_CATALOGUE_URL, apiKey);
			List<CkanUser> users = client.getOrganization(orgName).getUsers();
			for (CkanUser ckanUser : users) {
				if(ckanUser.getName().equals(usernameCkan)){
					toReturn = ckanUser.getCapacity();
					break;
				}
			}

		}catch(Exception e){
			LOG.error("Unable to retrieve the role the user has into organization: "+orgName, e);
		}

		return toReturn;
	}
	
	
	/**
	 * Check if the user is valid by checking if its API_KEY is present into DB.
	 *
	 * @param username the username
	 * @return true, if successful
	 */
	@Override
	public boolean checkValidUser(String username) {
		
		String apiKey = getApiKeyFromUsername(username);
		return apiKey != null;
		
	}

	private String getApiKeyFromUsername(String username) {

		LOG.debug("Request api key for user = " + username);

		// checks
		checkNotNull(username);
		checkArgument(!username.isEmpty());

		// in order to avoid errors, the username is always converted
		String ckanUsername = CatalogueUtilMethods.fromUsernameToCKanUsername(username);

		// check in the hashmap first
		if(apiKeysMap.containsKey(ckanUsername)){
			CKANTokenBean bean = apiKeysMap.get(ckanUsername);
			if(bean.timestamp + EXPIRE_KEY_TIME > System.currentTimeMillis()){ // it's still ok
				return bean.apiKey;
			}
		}

		LOG.debug("Api key was not in cache or it expired");

		try{
			
			String apiToReturn = dbCaller.getApiKeyFromUsername(username, State.ACTIVE.name().toLowerCase());

			// save into the hashmap
			if(apiToReturn != null)
				apiKeysMap.put(ckanUsername, new CKANTokenBean(apiToReturn, System.currentTimeMillis()));

			return apiToReturn;

		}catch(Exception e){
			LOG.error("Unable to retrieve key for user " + ckanUsername, e);
		}
		
		return null;
	}
	
	@Override
	public boolean existProductWithNameOrId(String nameOrId) {
		
		checkNotNull(nameOrId);
		checkArgument(!nameOrId.isEmpty());

		try{
			ExtendCkanClient client = new ExtendCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);
			CkanDataset product = client.getDataset(nameOrId);
			return product != null;
		}catch(Exception e){
			LOG.debug("A dataset with name " + nameOrId + " doesn't exist");
			return false;
		}
	}
	
	@Override
	public CkanOrganization getOrganizationByIdOrName(String idOrName) {

		checkNotNull(idOrName);

		String ckanName = idOrName.toLowerCase();
		try{

			return ckanCaller.getOrganization(ckanName);

		}catch(Exception e){
			LOG.warn("Failed to retrieve the organization with name " +idOrName+ " on the ckan: "+ckanCaller.getCatalogUrl(), e);
		}
		return null;
	}

	
	/*
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 * WRITE OPERATIONS
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 * 
	 */
	
	@Override
	public String createCkanDatasetMultipleCustomFields(String username, String title, String name, String organizationNameOrId,
			String author, String authorMail, String maintainer, String maintainerMail, long version,
			String description, String licenseId, List<String> tags, Map<String, List<String>> customFieldsMultiple,
			List<ResourceBean> resources, boolean setPublic, boolean setSearchable, boolean socialPost) throws Exception {
		LOG.info("Called createCKanDatasetMultipleCustomFields");
		
		// checks (minimum)
		checkNotNull(username);
		checkNotNull(organizationNameOrId);
		checkArgument(!organizationNameOrId.isEmpty());
		checkArgument(!(title == null && name == null || title.isEmpty() && name.isEmpty()), "Name and Title cannot be empty/null at the same time!");

		// String ckanUsername = getUserFromApiKey(apiKey).getName();
		CkanDataset dataset = convertToCkanDataset(username, title, name, organizationNameOrId, author, authorMail,
				maintainer, maintainerMail, version, description, licenseId, tags, null, customFieldsMultiple,
				resources, setPublic, setSearchable);
		
		// trying to create by gCat
		String jsonValueDataset = MarshUnmarshCkanObject.toJsonValueDataset(dataset,METHOD.TO_CREATE);
		LOG.info("Serialized dataset is: " + jsonValueDataset);
		jsonValueDataset = gCatCaller.createDataset(jsonValueDataset,socialPost);
		LOG.debug("Created dataset is: " + jsonValueDataset);

		if(jsonValueDataset != null){
			CkanDataset toCkanDataset = MarshUnmarshCkanObject.toCkanDataset(jsonValueDataset,METHOD.TO_READ);
			LOG.info("Dataset with name " + toCkanDataset.getName() + " has been created correctly");

			//TODO FOLLOWING OPERATIONS NOT SHOULD LONGER NEEDED ON DATASET CREATION, MUST BE CHECKED with @LUCA
			
//			// set visibility
//			boolean visibilitySet = setDatasetPrivate(
//					!setPublic, // swap to private
//					res.getOrganization().getId(),
//					res.getId(),
//					CKAN_TOKEN_SYS); // use sysadmin api key to be sure it will be set
//
//			logger.info("Was visibility set to " + (setPublic ? "public" : "private") + "? " + visibilitySet);
//
//			// set searchable to true if dataset visibility is private
//			if(!setPublic){ // (swap to private)
//				boolean searchableSet = setSearchableField(res.getId(), true);
//				logger.info("Was searchable set to True? " + searchableSet);
//			}
//			return res.getId();

	
			return toCkanDataset.getId();
		}

		return null;
	}
	
	/**
	 * Convert to ckan dataset. This method creates the body of the dataset
	 * 
	 * @param username the username
	 * @param title the title
	 * @param name the name
	 * @param organizationNameOrId the organization name or id
	 * @param author the author
	 * @param authorMail the author mail
	 * @param maintainer the maintainer
	 * @param maintainerMail the maintainer mail
	 * @param version the version
	 * @param description the description
	 * @param licenseId the license id
	 * @param tags the tags
	 * @param customFields the custom fields
	 * @param customFieldsMultipleValues the custom fields multiple values
	 * @param resources the resources
	 * @param setPublic the set public
	 * @param setSearchable the set searchable
	 * @return the ckan dataset
	 */
	private CkanDataset convertToCkanDataset(String username, String title, String name, String organizationNameOrId, String author,
			String authorMail, String maintainer, String maintainerMail, long version, String description,
			String licenseId, List<String> tags, Map<String, String> customFields,
			Map<String, List<String>> customFieldsMultipleValues, List<ResourceBean> resources, boolean setPublic, boolean setSearchable) {
		
		LOG.info("Called convert data to CkanDataset");

		//ExtendCkanClient client = new ExtendCkanClient(CKAN_CATALOGUE_URL, apiKey);

		//String ckanUsername = getUserFromApiKey(apiKey).getName();
		CkanDataset dataset = new CkanDataset();
		
		// in order to avoid errors, the username is always converted
		String ckanUsername = CatalogueUtilMethods.fromUsernameToCKanUsername(username);

		String nameToUse = name;
		if(nameToUse == null)
			nameToUse = CatalogueUtilMethods.fromProductTitleToName(title);

		LOG.debug("Name of the dataset is going to be " + nameToUse + ". Title is going to be " + title);

		dataset.setName(nameToUse);
		dataset.setTitle(title);

		CkanOrganization orgOwner = ckanCaller.getOrganization(organizationNameOrId);
		dataset.setOwnerOrg(orgOwner.getId());
		dataset.setAuthor(author);
		dataset.setAuthorEmail(authorMail);
		dataset.setMaintainer(maintainer);
		dataset.setMaintainerEmail(maintainerMail);
		dataset.setVersion(String.valueOf(version));
		
		//set is private
		dataset.setPriv(!setPublic);

		// description must be escaped
		if(description != null && !description.isEmpty()){
			Source descriptionEscaped = new Source(description);
			Segment htmlSeg = new Segment(descriptionEscaped, 0, descriptionEscaped.length());
			Renderer htmlRend = new Renderer(htmlSeg);
			dataset.setNotes(htmlRend.toString());
			LOG.debug("Description escaped " + htmlRend.toString());
		}

		dataset.setLicenseId(licenseId);

		// set the tags, if any
		if(tags != null && !tags.isEmpty()){
			List<CkanTag> ckanTags = new ArrayList<CkanTag>(tags.size());
			for (String stringTag : tags) {
				ckanTags.add(new CkanTag(stringTag));
			}
			dataset.setTags(ckanTags);
			dataset.setNumTags(ckanTags.size());
		}

		// set the custom fields, if any
		List<CkanPair> extras = new ArrayList<CkanPair>();

		if(customFields != null && !customFields.isEmpty()){

			Iterator<Entry<String, String>> iterator = customFields.entrySet().iterator();

			while (iterator.hasNext()) {
				Map.Entry<String, String> entry = iterator.next();
				extras.add(new CkanPair(entry.getKey(), entry.getValue()));
			}

		}else if(customFieldsMultipleValues != null && !customFieldsMultipleValues.isEmpty()){

			Iterator<Entry<String, List<String>>> iterator = customFieldsMultipleValues.entrySet().iterator();
			while (iterator.hasNext()) {
				Map.Entry<String, List<String>> entry = iterator.next();
				List<String> valuesForEntry = entry.getValue();
				for (String value : valuesForEntry) {
					extras.add(new CkanPair(entry.getKey(), value));
				}
			}
		}

		dataset.setExtras(extras);

		// check if we need to add the resources
		if(resources != null && !resources.isEmpty()){

			LOG.debug("We need to add resources to the dataset");

			try{
				List<CkanResource> resourcesCkan = new ArrayList<CkanResource>();
				for(ResourceBean resource: resources){

					LOG.debug("Going to add resource described by " + resource);
					CkanResource newResource = new CkanResource();
					newResource.setDescription(resource.getDescription());
					newResource.setId(resource.getId());
					newResource.setUrl(resource.getUrl());
					newResource.setName(resource.getName());
					newResource.setMimetype(resource.getMimeType());
					newResource.setFormat(resource.getMimeType());
					newResource.setOwner(ckanUsername);
					resourcesCkan.add(newResource);
				}

				// add to the dataset
				dataset.setResources(resourcesCkan);
				dataset.setNumResources(resourcesCkan.size());

			}catch(Exception e){
				LOG.error("Unable to add those resources to the dataset", e);
			}

		}
		
		return dataset;

	}
	
	@Override 
	public boolean patchFieldsForDataset(String datasetId, Map<String, String> mapFields) throws Exception {
		LOG.info("Called patch the fields "+mapFields+" for dataset id: "+datasetId);
		
		checkNotNull(datasetId);
		checkNotNull(mapFields);
		
		try {
			JSONObject jsonObj = new JSONObject();
			for (String key : mapFields.keySet()) {
				jsonObj.put(key, mapFields.get(key));
			}
			
			LOG.debug("Json Dataset is: " + jsonObj);
			gCatCaller.patchDataset(datasetId, jsonObj);
			LOG.info("Patch operation for dataset "+datasetId+" terminates without errors");
			return true;
		}catch (Exception e) {
			LOG.error("Error occurred trying to patch the dataset with id: "+datasetId, e.getMessage());
			throw new Exception("Unable to patch the dataset. Error: "+e.getMessage());
		}
		
	}
	
	@Override
	public boolean setSearchableFieldForDataset(String datasetId, boolean searchable) throws Exception {
		LOG.info("Set searchalbe field as "+searchable+" for dataset id: "+datasetId);
		
		checkNotNull(datasetId);
		
		try {
			String searchableAsString = searchable ? "True" : "False";
			JSONObject jsonObj = new JSONObject();
			jsonObj.put("id", datasetId); //just to be sure
			jsonObj.put("searchable", searchableAsString);
			LOG.debug("Json Dataset is: " + jsonObj);
			gCatCaller.patchDataset(datasetId, jsonObj);
			LOG.info("Set 'searchable' field for dataset "+datasetId+" terminates without errors");
			return true;
		}catch (Exception e) {
			LOG.error("Error occured trying to set the 'searchable' field for the dataset with id: "+datasetId, e.getMessage());
			throw new Exception("Unable to set the 'searchable' field for this dataset. Error: "+e.getMessage());
		}
		
	}


	@Override
	public String addResourceToDataset(ResourceBean resourceBean) throws Exception {
		LOG.info("Request to add a resource described by this bean " + resourceBean);

		// checks
		checkNotNull(resourceBean);

		if(CatalogueUtilMethods.resourceExists(resourceBean.getUrl())){

			// in order to avoid errors, the username is always converted
			String ckanUsername = CatalogueUtilMethods.fromUsernameToCKanUsername(resourceBean.getOwner());

			CkanResource resource = new CkanResource(CKAN_CATALOGUE_URL, resourceBean.getDatasetId());
			resource.setName(resourceBean.getName());

			// escape description
			Source description = new Source(resourceBean.getDescription());
			Segment htmlSeg = new Segment(description, 0, description.length());
			Renderer htmlRend = new Renderer(htmlSeg);

			resource.setDescription(htmlRend.toString());
			resource.setUrl(resourceBean.getUrl());
			resource.setOwner(ckanUsername);
			
			String jsonValueResource = MarshUnmarshCkanObject.toJsonValueResource(resource);
			LOG.trace("Serialized resource is: " + jsonValueResource);
			jsonValueResource = gCatCaller.addResourceToDataset(resourceBean.getDatasetId(), jsonValueResource);
			LOG.debug("Added resource to dataset is: " + jsonValueResource);

			CkanResource createdRes = MarshUnmarshCkanObject.toCkanResource(jsonValueResource);

			if(createdRes != null){
				LOG.debug("Resource  " + createdRes.getName() + " is now available");
				return createdRes.getId();

			}
		}else
			throw new Exception("It seems there is no is no resource at this url " + resourceBean.getUrl());

		return null;
	}

	@Override
	public boolean deleteResourceFromDataset(String resourceId) throws Exception {
		LOG.info("Request to delete a resource with id " + resourceId);

		// checks
		checkNotNull(resourceId);
		checkArgument(!resourceId.isEmpty());

		try{
			CkanResource theResource = ckanCaller.getResource(resourceId);
			gCatCaller.deleteResource(theResource.getPackageId(), resourceId);
			return true;

		}catch(Exception e){
			LOG.error("Unable to delete resource whose id is " + resourceId, e);
		}

		return false;
	}

	
	@Override
	public CkanGroup createGroup(String name, String title, String description) throws Exception {

		// checks
		checkNotNull(name);
		checkArgument(!name.trim().isEmpty());

		// check if it exists
		CkanGroup toCreate = null;
		LOG.debug("Request for creating group with name " + name + " title " + title + " and description " + description);
		String normalizedName = CatalogueUtilMethods.fromGroupTitleToName(name);
		if((toCreate = groupExists(name))!= null)
			return toCreate;
		else{
			try{

				CkanGroup group = new CkanGroup(normalizedName);
				group.setTitle(title);
				group.setDisplayName(title);
				group.setDescription(description);
				String jsonValueGroup = MarshUnmarshCkanObject.toJsonValueGroup(group);
				LOG.trace("Serialized group is: " + jsonValueGroup);
				
				String theGroup = gCatCaller.createGroup(jsonValueGroup);
				toCreate = MarshUnmarshCkanObject.toCkanGroup(theGroup);
				LOG.info("Created the group with id: " + toCreate.getId() + ", name: "+toCreate.getName());

			}catch(JackanException je){
				LOG.error("Unable to create such group", je);
			}
		}

		return toCreate;
	}
	
	/**
	 * Just check if the group exists
	 * @param nameOrId
	 * @param client
	 * @return
	 */
	private CkanGroup groupExists(String nameOrId){

		CkanGroup toReturn = null;

		try{
			ExtendCkanClient client = new ExtendCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);
			toReturn = client.getGroup(nameOrId);

		}catch(JackanException je){
			LOG.error("The group "+nameOrId+" doesn't exist");
		}

		return toReturn;
	}

//	@Override
//	public boolean patchProductCustomFields(String productId, Map<String, List<String>> customFieldsToChange,
//			boolean removeOld) {
//		return false;
//	}
//
	
//	@Override
	public boolean assignDatasetToGroup(String groupNameOrId, String datasetNameOrId) {

		return assignDatasetToGroupBody(groupNameOrId, datasetNameOrId, "", false);

	}

//	@Override
	public boolean assignDatasetToGroup(String groupNameOrId, String datasetNameOrId, boolean addOnParents) {

		return assignDatasetToGroupBody(groupNameOrId, datasetNameOrId, "", addOnParents);

	}
	
	/**
	 * The real body of the assignDatasetToGroup
	 * @param groupNameOrId
	 * @param datasetNameOrId
	 * @param apiKey
	 * @param addOnParents
	 * @return
	 */
	private boolean assignDatasetToGroupBody(String groupNameOrId, String datasetNameOrId, String apiKey, boolean addOnParents) {

		return false;
//		// checks
//		checkNotNull(groupNameOrId);
//		checkArgument(!groupNameOrId.isEmpty());
//		checkNotNull(datasetNameOrId);
//		checkArgument(!datasetNameOrId.isEmpty());
//		checkNotNull(apiKey);
//		checkArgument(!apiKey.isEmpty());
//
//		String groupNameToCheck = CatalogueUtilMethods.fromGroupTitleToName(groupNameOrId);
//
//		try(CloseableHttpClient httpClient = HttpClientBuilder.create().build();){
//
//			ExtendCkanClient client = new ExtendCkanClient(CKAN_CATALOGUE_URL, CKAN_TOKEN_SYS);
//
//			// check the group exists
//			CkanGroup group = client.getGroup(groupNameToCheck);
//
//			// move to a list
//			List<String> groupNames = new ArrayList<String>();
//			groupNames.add(group.getName());
//			if(group != null && addOnParents){
//				findHierarchyGroups(groupNames, CKAN_TOKEN_SYS);
//			}
//
//			// we need to use the apis to make it
//			String pathPackageShow = CKAN_CATALOGUE_URL + "/api/3/action/package_show?id=" + datasetNameOrId;
//			HttpGet getRequest = new HttpGet(pathPackageShow);
//			getRequest.addHeader("Authorization", CKAN_TOKEN_SYS);
//			HttpResponse response = httpClient.execute(getRequest);
//
//			logger.debug("Response is " + response.getStatusLine().getStatusCode() + " and message is " + response.getStatusLine().getReasonPhrase());
//
//			// read the json dataset and fetch the groups and fetch the groups' names, if any
//			if(response.getStatusLine().getStatusCode() == HttpStatus.SC_OK){
//
//				// parse the json and convert to java beans
//				String jsonAsString = EntityUtils.toString(response.getEntity());
//				JSONParser parser = new JSONParser();
//				JSONObject json = (JSONObject) parser.parse(jsonAsString);
//				JSONObject resultJson =  (JSONObject) json.get("result");
//				JSONArray groupsJson = (JSONArray)resultJson.get("groups");
//				Iterator<JSONObject> it =  groupsJson.iterator();
//
//				while (it.hasNext()) {
//					JSONObject object = it.next();
//					try{
//						if(object.containsKey("name"))
//							groupNames.add((String)object.get("name"));
//					}catch(Exception e){
//						logger.error("Error", e);
//					}
//				}
//
//				// remove duplicates
//				Set<String> groupNamesSet = new HashSet<String>(groupNames);
//
//				logger.debug("Groups to be added are " + groupNamesSet);
//
//				// now we patch the dataset with the new group
//				String pathUpdatePatch = CKAN_CATALOGUE_URL + "/api/3/action/package_patch";
//
//				JSONObject req = new JSONObject();
//				req.put("id", datasetNameOrId);
//
//				JSONArray groups = new JSONArray();
//				Iterator<String> iteratorNameSet = groupNamesSet.iterator();
//				while (iteratorNameSet.hasNext()) {
//					String groupName = iteratorNameSet.next();
//					JSONObject groupJSON = new JSONObject();
//					groupJSON.put("name", groupName);
//					groups.add(groupJSON);
//				}
//				req.put("groups", groups);
//
//				logger.debug("Request for patch is going to be " + req.toJSONString());
//
//				HttpPost request = new HttpPost(pathUpdatePatch);
//				request.addHeader("Authorization", CKAN_TOKEN_SYS);
//				StringEntity params = new StringEntity(req.toJSONString());
//				request.setEntity(params);
//				HttpResponse responsePatch = httpClient.execute(request);
//				logger.debug("Response code is " + responsePatch.getStatusLine().getStatusCode() + " and response message is " + responsePatch.getStatusLine().getReasonPhrase());
//
//				if(responsePatch.getStatusLine().getStatusCode() == HttpStatus.SC_OK){
//					logger.info("Dataset Added to the group!!");
//					return true;
//				}
//
//			}
//
//		}catch(Exception e){
//			logger.error("Unable to make this association", e);
//		}
//
//		return false;
	}



}