EDIT

/**

* Copyright (C) 2007 EDIT

* European Distributed Institute of Taxonomy

* http://www.e-taxonomy.eu

*

* The contents of this file are subject to the Mozilla Public License Version 1.1

* See LICENSE.TXT at the top of this package for the full license terms.

*/

package eu.etaxonomy.cdm.io.berlinModel.in;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.HashSet;

import java.util.List;

import java.util.Map;

import java.util.Set;

import java.util.SortedSet;

import java.util.TreeSet;

import java.util.UUID;

import org.apache.commons.lang.StringUtils;

import org.apache.log4j.Logger;

import org.springframework.stereotype.Component;

import eu.etaxonomy.cdm.common.CdmUtils;

import eu.etaxonomy.cdm.io.berlinModel.BerlinModelTransformer;

import eu.etaxonomy.cdm.io.berlinModel.in.validation.BerlinModelCommonNamesImportValidator;

import eu.etaxonomy.cdm.io.common.IOValidator;

import eu.etaxonomy.cdm.io.common.ResultSetPartitioner;

import eu.etaxonomy.cdm.io.common.Source;

import eu.etaxonomy.cdm.io.common.TdwgAreaProvider;

import eu.etaxonomy.cdm.model.common.CdmBase;

import eu.etaxonomy.cdm.model.common.Extension;

import eu.etaxonomy.cdm.model.common.ExtensionType;

import eu.etaxonomy.cdm.model.common.Language;

import eu.etaxonomy.cdm.model.common.Marker;

import eu.etaxonomy.cdm.model.common.MarkerType;

import eu.etaxonomy.cdm.model.common.Representation;

import eu.etaxonomy.cdm.model.common.TermType;

import eu.etaxonomy.cdm.model.common.TermVocabulary;

import eu.etaxonomy.cdm.model.description.CommonTaxonName;

import eu.etaxonomy.cdm.model.description.DescriptionElementSource;

import eu.etaxonomy.cdm.model.description.TaxonDescription;

import eu.etaxonomy.cdm.model.location.Country;

import eu.etaxonomy.cdm.model.location.NamedArea;

import eu.etaxonomy.cdm.model.name.TaxonName;

import eu.etaxonomy.cdm.model.reference.Reference;

import eu.etaxonomy.cdm.model.taxon.Taxon;

import eu.etaxonomy.cdm.model.taxon.TaxonBase;

/**

 *

 * @author a.mueller

 * @since 20.03.2008

 */

@Component

public class BerlinModelCommonNamesImport  extends BerlinModelImportBase {

    private static final long serialVersionUID = -8921948187177864321L;

    private static final Logger logger = Logger.getLogger(BerlinModelCommonNamesImport.class);

	public static final UUID REFERENCE_LANGUAGE_ISO639_2_UUID = UUID.fromString("40c4f8dd-3d9c-44a4-b77a-76e137a89a5f");

	public static final UUID REFERENCE_LANGUAGE_STRING_UUID = UUID.fromString("2a1b678f-c27d-48c1-b43e-98fd0d426305");

	public static final UUID COMMONNAME_STATUS_RECOMMENDED_UUID = UUID.fromString("e3f7b80a-1286-458d-812c-5e818f731968");

	public static final UUID COMMONNAME_STATUS_SYNONYM_UUID = UUID.fromString("169b2d97-a706-49de-b28b-c67f0ee6764b");

	public static final String NAMESPACE = "common name";

	private static final String pluralString = "common names";

	private static final String dbTableName = "emCommonName";

	//map that stores the regions (named areas) and makes them accessible via the regionFk

	private Map<String, NamedArea> regionMap = new HashMap<>();

	public BerlinModelCommonNamesImport(){

		super(dbTableName, pluralString);

	}

	@Override

	protected String getIdQuery(BerlinModelImportState state) {

		String result = " SELECT CommonNameId FROM emCommonName WHERE (1=1) ";

		if (isNotBlank(state.getConfig().getCommonNameFilter())){

			result += " AND " + state.getConfig().getCommonNameFilter();

		}

		result += " ORDER BY CommonNameId ";

		return result;

	}

	@Override

	protected String getRecordQuery(BerlinModelImportConfigurator config) {

		String recordQuery = "";

		recordQuery =

				" SELECT rel.RelPTaxonId, rel.RelQualifierFk, acc.RIdentifier accTaxonId, factTaxon.RIdentifier factTaxonId, accName.NameId, f.FactId, " +

				           "       cn.CommonNameId, cn.CommonName, tax.RIdentifier AS taxonId, cn.PTNameFk, cn.RefFk AS refId, cn.Status, cn.RegionFks, cn.MisNameRefFk, " +

					       "       cn.NameInSourceFk, cn.Created_When, cn.Updated_When, cn.Created_Who, cn.Updated_Who, cn.Note AS Notes, languageCommonName.Language, " +

					       "       languageCommonName.LanguageOriginal, languageCommonName.ISO639_1, languageCommonName.ISO639_2,   " +

					       "       emLanguageReference.RefFk AS languageRefRefFk, emLanguageReference.ReferenceShort, emLanguageReference.ReferenceLong,  " +

					       "       emLanguageReference.LanguageFk, languageReferenceLanguage.Language AS refLanguage, languageReferenceLanguage.ISO639_2 AS refLanguageIso639_2,  "+

					       "       misappliedTaxon.RIdentifier AS misappliedTaxonId " +

				  " FROM  PTaxon AS misappliedTaxon RIGHT OUTER JOIN " +

					       "      emLanguage AS languageReferenceLanguage RIGHT OUTER JOIN " +

			               "      emLanguageReference ON languageReferenceLanguage.LanguageId = emLanguageReference.LanguageFk RIGHT OUTER JOIN " +

			               "      emCommonName AS cn INNER JOIN " +

			               "      PTaxon AS tax ON cn.PTNameFk = tax.PTNameFk AND cn.PTRefFk = tax.PTRefFk ON  " +

			               "      emLanguageReference.ReferenceId = cn.LanguageRefFk LEFT OUTER JOIN " +

			               "      emLanguage AS languageCommonName ON cn.LanguageFk = languageCommonName.LanguageId ON misappliedTaxon.PTNameFk = cn.NameInSourceFk AND  " +

			               "      misappliedTaxon.PTRefFk = cn.MisNameRefFk " +

	                     "     LEFT OUTER JOIN Fact f ON cn.CommonNameId = f.ExtensionFk " +

	                     "     LEFT OUTER JOIN PTaxon factTaxon ON factTaxon.PTNameFk = f.PTNameFk AND factTaxon.PTRefFk = f.PTRefFk " +

	                     "     LEFT OUTER JOIN RelPTaxon rel ON rel.PTNameFk1 = tax.PTNameFk AND rel.PTRefFk1 = tax.PTRefFk AND rel.RelQualifierFk IN (2,6,7) " +

                         "     LEFT OUTER JOIN PTaxon acc ON rel.PTNameFk2 = acc.PTNameFk AND rel.PTRefFk2 = acc.PTRefFk " +

                         "     LEFT OUTER JOIN Name accName ON accName.NameId = acc.PTNameFk " +

			        " WHERE cn.CommonNameId IN (" + ID_LIST_TOKEN + ") " +

			        " ORDER BY cn.CommonNameId ";

		return recordQuery;

	}

	@Override

	protected void doInvoke(BerlinModelImportState state) {

		try {

			makeRegions(state);

		} catch (Exception e) {

			logger.error("Error when creating common name regions:" + e.getMessage());

			e.printStackTrace();

			state.setUnsuccessfull();

		}

		super.doInvoke(state);

		return;

	}

	/**

	 * @param state

	 *

	 */

	private void makeRegions(BerlinModelImportState state) {

		try {

			SortedSet<Integer> regionFks = new TreeSet<>();

			Source source = state.getConfig().getSource();

			//fill set with all regionFk from emCommonName.regionFks

			getRegionFks(state, regionFks, source);

			//concat filter string

			String sqlWhere = getSqlWhere(regionFks);

			//get E+M - TDWG Mapping

			Map<String, String> emTdwgMap = getEmTdwgMap(source);

			//fill regionMap

			fillRegionMap(state, sqlWhere, emTdwgMap);

			return;

		} catch (NumberFormatException e) {

			e.printStackTrace();

			state.setUnsuccessfull();

			return;

		} catch (SQLException e) {

			e.printStackTrace();

			state.setUnsuccessfull();

			return;

		}

	}

	@Override

	public boolean doPartition(ResultSetPartitioner partitioner, BerlinModelImportState state)  {

		boolean success = true ;

		Set<TaxonBase> taxaToSave = new HashSet<>();

		@SuppressWarnings("unchecked")

        Map<String, Taxon> taxonMap = partitioner.getObjectMap(BerlinModelTaxonImport.NAMESPACE);

		@SuppressWarnings("unchecked")

        Map<String, TaxonName> taxonNameMap = partitioner.getObjectMap(BerlinModelTaxonNameImport.NAMESPACE);

		@SuppressWarnings("unchecked")

        Map<String, Reference> refMap = partitioner.getObjectMap(BerlinModelReferenceImport.REFERENCE_NAMESPACE);

		Map<String, Language> iso6392Map = new HashMap<>();

	//	logger.warn("MisappliedNameRefFk  not yet implemented for Common Names");

		ResultSet rs = partitioner.getResultSet();

		Integer lastCommonNameId = null;

		try{

			while (rs.next()){

				//create TaxonName element

				Integer commonNameId = rs.getInt("CommonNameId");

				int taxonId = rs.getInt("taxonId");

				Integer factTaxonId = nullSafeInt(rs, "factTaxonId");

				Integer accTaxonId = nullSafeInt(rs, "accTaxonId");  //if common name is related to synonym this is the accepted taxon id

				Integer refId = nullSafeInt(rs, "refId");

//				Integer ptNameFk = nullSafeInt(rs,"PTNameFk");

				String commonNameString = rs.getString("CommonName");

				String iso639_2 = rs.getString("ISO639_2");

				String iso639_1 = rs.getString("ISO639_1");

				String languageString = rs.getString("Language");

				String originalLanguageString = rs.getString("LanguageOriginal");

				Integer misNameRefFk = nullSafeInt(rs, "MisNameRefFk");

				Integer languageRefRefFk = nullSafeInt(rs, "languageRefRefFk");

				String refLanguage = rs.getString("refLanguage");

				String refLanguageIso639_2 = rs.getString("refLanguageIso639_2");

				String status = rs.getString("Status");

				Integer nameInSourceFk = nullSafeInt( rs, "NameInSourceFk");

				Integer misappliedTaxonId = nullSafeInt( rs, "misappliedTaxonId");

				if (commonNameId == lastCommonNameId){

				    logger.warn("CommonNameId >1 times in query. This may happen due to LEFT JOINS to fact and/or accepted taxon and e.g. multiple taxon relationships. 2018-04-01 no such double relation existed in E+M. ");

				}else{

				    lastCommonNameId = commonNameId;

				}

				final String NO_REGION = "";

				//regions

				String regionFks  = rs.getString("RegionFks");

				String[] regionFkSplit = (regionFks==null)? new String[]{NO_REGION} : regionFks.trim().split(",");

				if (regionFkSplit.length == 0){

				    String message = "regionFkSplit should never be empty but was for common name id " + commonNameId;

                    logger.warn(message);

				}

				//commonNameString

				if (isBlank(commonNameString)){

					String message = "CommonName is empty or null. Do not import record for taxon " + taxonId;

					logger.warn(message);

					continue;

				}

				//taxon

				Taxon taxon = null;

				TaxonBase<?> taxonBase  = taxonMap.get(String.valueOf(taxonId));

				if (taxonBase == null){

					logger.warn("Taxon (" + taxonId + ") could not be found. Common name " + commonNameString + "(" + commonNameId + ") not imported");

					continue;

				}else if (taxonBase.isInstanceOf(Taxon.class)){

				    taxon = CdmBase.deproxy(taxonBase, Taxon.class);

				    if (factTaxonId != null && !factTaxonId.equals(taxonId)){

				        logger.warn("Fact taxon ("+factTaxonId+") for common name "+commonNameId+" differs from common name taxon " + taxonId);

				    }

				}else{

				    Taxon factTaxon = null;

				    if (factTaxonId != null && factTaxonId.equals(accTaxonId)){

				        factTaxon = taxonMap.get(String.valueOf(factTaxonId));

				    }

				    if (factTaxon != null){

				        taxon = factTaxon;

				    }else{

				        logger.warn("taxon (" + taxonId + ") is not accepted. Can't import common name " +  commonNameId + ". FactTaxonId= " +  factTaxonId + "; accTaxonId = " + accTaxonId);

				        continue;

				    }

				}

				//Language

				Language language = getAndHandleLanguage(iso6392Map, iso639_2, iso639_1, languageString, originalLanguageString, state);

				//CommonTaxonName

				List<CommonTaxonName> commonTaxonNames = new ArrayList<>();

				for (String regionFk : regionFkSplit){ //

					CommonTaxonName commonTaxonName;

					if (commonTaxonNames.size() == 0){

						commonTaxonName = CommonTaxonName.NewInstance(commonNameString, language);

					}else{

						commonTaxonName = (CommonTaxonName)commonTaxonNames.get(0).clone();

					}

					commonTaxonNames.add(commonTaxonName);

					regionFk = regionFk.trim();

					NamedArea area = regionMap.get(regionFk);

					if (area == null){

						if (isNotBlank(regionFk) && regionFk != NO_REGION){

							logger.warn("Area for " + regionFk + " not defined in regionMap.");

						}else{

							//no region is defined

						}

					}else{

						commonTaxonName.setArea(area);

					}

					TaxonDescription description = getDescription(taxon);

					description.addElement(commonTaxonName);

				}

				//Reference/Source

				if (! CdmUtils.nullSafeEqual(refId, languageRefRefFk)){

					//use strRefId if languageRefFk is null

					if (languageRefRefFk == null){

						languageRefRefFk = refId;

					}else{

						logger.warn("CommonName.RefFk (" + CdmUtils.Nz(refId) + ") and LanguageReference.RefFk " + languageRefRefFk  + " are not equal. I will import only languageReference.RefFk");

					}

				}

				Reference reference = refMap.get(String.valueOf(languageRefRefFk));

				if (reference == null && languageRefRefFk != null){

					logger.warn("CommonName reference was null but reference exists. languageRefRefFk = " + languageRefRefFk + "; commonNameId = " + commonNameId);

				}

				String microCitation = null;

				String originalNameString = null;

				TaxonName nameUsedInSource = taxonNameMap.get(String.valueOf(nameInSourceFk));

				if (nameInSourceFk != null && nameUsedInSource == null){

					if (nameInSourceFk != -1 || !state.getConfig().isEuroMed()){

					    logger.warn("Name used in source (" + nameInSourceFk + ") was not found for common name " + commonNameId);

					}

				}

				for (CommonTaxonName commonTaxonName : commonTaxonNames){

				    DescriptionElementSource source = DescriptionElementSource.NewPrimarySourceInstance(reference, microCitation, nameUsedInSource, originalNameString);

	                commonTaxonName.addSource(source);

				}

				//MisNameRef

				if (misNameRefFk != null){

					//Taxon misappliedName = getMisappliedName(biblioRefMap, nomRefMap, misNameRefFk, taxon);

					Taxon misappliedNameTaxon = null;

					if (misappliedTaxonId != null){

						TaxonBase<?> misTaxonBase =  taxonMap.get(String.valueOf(misappliedTaxonId));

						if (misTaxonBase == null){

							logger.warn("MisappliedName not found for misappliedTaxonId " + misappliedTaxonId + "; commonNameId: " + commonNameId);

						}else if (misTaxonBase.isInstanceOf(Taxon.class)){

							misappliedNameTaxon = CdmBase.deproxy(misTaxonBase, Taxon.class);

						}else{

							logger.warn("Misapplied name taxon is not of type Taxon but " + misTaxonBase.getClass().getSimpleName());

						}

					}else{

						Reference sec = refMap.get(String.valueOf(misNameRefFk));

						if (nameUsedInSource == null || sec == null){

							logger.warn("Taxon name or misapplied name reference is null for common name " + commonNameId);

						}else{

							misappliedNameTaxon = Taxon.NewInstance(nameUsedInSource, sec);

							MarkerType misCommonNameMarker = getMarkerType(state, BerlinModelTransformer.uuidMisappliedCommonName,"Misapplied Common Name in Berlin Model", "Misapplied taxon was automatically created by Berlin Model import for a common name with a misapplied name reference", "MCN");

							Marker marker = Marker.NewInstance(misCommonNameMarker, true);

							misappliedNameTaxon.addMarker(marker);

							taxaToSave.add(misappliedNameTaxon);

							logger.warn("Misapplied name taxon could not be found in database but misapplied name reference exists for common name. " +

									"New misapplied name for misapplied reference common name was added. CommonNameId: " + commonNameId);

						}

					}

					if (misappliedNameTaxon != null){

						if (! taxon.getMisappliedNames(false).contains(misappliedNameTaxon)){

							taxon.addMisappliedName(misappliedNameTaxon, state.getTransactionalSourceReference(), null);

							logger.warn("Misapplied name for common name was not found related to the accepted taxon. Created new relationship. CommonNameId: " + commonNameId);

						}

						//add common name also to missaplied taxon

						//TODO is this really wanted

						TaxonDescription misappliedNameDescription = getDescription(misappliedNameTaxon);

						for (CommonTaxonName commonTaxonName : commonTaxonNames){

							CommonTaxonName commonNameClone = (CommonTaxonName)commonTaxonName.clone();

							misappliedNameDescription.addElement(commonNameClone);

							doIdCreatedUpdatedNotes(state, commonNameClone, rs, String.valueOf(commonNameId), NAMESPACE);

						}

					}else{

						//wird schon oben gelogged

					    //logger.warn("Misapplied name is null for common name " + commonNameId);

					}

				}

				//reference extensions

				if (reference != null){

					if (isNotBlank(refLanguage) && !reference.hasExtension(REFERENCE_LANGUAGE_STRING_UUID, refLanguage)){

						ExtensionType refLanguageExtensionType = getExtensionType( state, REFERENCE_LANGUAGE_STRING_UUID, "reference language","The language of the reference","ref. lang.");

						Extension.NewInstance(reference, refLanguage, refLanguageExtensionType);

					}

					if (isNotBlank(refLanguageIso639_2) && !reference.hasExtension(REFERENCE_LANGUAGE_ISO639_2_UUID, refLanguage)){

						ExtensionType refLanguageIsoExtensionType = getExtensionType( state, REFERENCE_LANGUAGE_ISO639_2_UUID, "reference language iso 639-2","The iso 639-2 code of the references language","ref. lang. 639-2");

						Extension.NewInstance(reference, refLanguageIso639_2, refLanguageIsoExtensionType);

					}

				}else if (isNotBlank(refLanguage) || isNotBlank(refLanguageIso639_2)){

					logger.warn("Reference is null (" + languageRefRefFk + ") but refLanguage (" + CdmUtils.Nz(refLanguage) + ") or iso639_2 (" + CdmUtils.Nz(refLanguageIso639_2) + ") was not null for common name ("+ commonNameId +")");

				}

				//status

				if (isNotBlank(status)){

				    TermVocabulary<MarkerType> markerTypeVoc = getVocabulary(TermType.MarkerType, BerlinModelTransformer.uuidVocEMMarkerType,

				            "Euro+Med marker type vocabulary", "E+M marker types", null, null, false, MarkerType.COMPLETE());

				    MarkerType recommendedMarkerType = getMarkerType( state, COMMONNAME_STATUS_RECOMMENDED_UUID, "recommended","If the common name has the status recommended (see also status 'synonym', if none of them is true the default status is 'unassessed')",

				            "recommended", markerTypeVoc);

					MarkerType synonymMarkerType = getMarkerType( state, COMMONNAME_STATUS_SYNONYM_UUID, "synonym","If the common name has the status synonym (see also status 'recommended', if none of them is true the default status is 'unassessed')",

					        "synonym", markerTypeVoc);

                    for (CommonTaxonName commonTaxonName : commonTaxonNames){

                        Marker marker = null;

                        if (status.equals("recommended")){

						    marker = Marker.NewInstance(recommendedMarkerType, true);

                        }else if (status.equals("synonym")){

                            marker = Marker.NewInstance(synonymMarkerType, true);

                        }else if (status.equals("unassessed")){

                            //do nothing

                        }else{

						    logger.warn("Unknown common name status: " + status);

						}

                        if (marker != null){

                            commonTaxonName.addMarker(marker);

                        }

					}

				}

				//Notes

				for (CommonTaxonName commonTaxonName : commonTaxonNames){

					doIdCreatedUpdatedNotes(state, commonTaxonName, rs, String.valueOf(commonNameId), NAMESPACE);

				}

				partitioner.startDoSave();

				taxaToSave.add(taxon);

			}

		} catch (SQLException e) {

			logger.error("SQLException:" +  e);

			return false;

		} catch (ClassCastException e) {

			e.printStackTrace();

		} catch (Exception e) {

            throw e;

        }

		//	logger.info( i + " names handled");

		getTaxonService().save(taxaToSave);

		return success;

	}

    /**

	 * @param iso6392Map

	 * @param iso639_2

	 * @param languageString

	 * @param originalLanguageString

	 * @param state

	 * @return

	 */

	private Language getAndHandleLanguage(Map<String, Language> iso639Map,	String iso639_2, String iso639_1, String languageString, String originalLanguageString, BerlinModelImportState state) {

		Language language;

		if (isNotBlank(iso639_2)|| isNotBlank(iso639_1)  ){

			//TODO test performance, implement in state

			language = getLanguageFromIsoMap(iso639Map, iso639_2, iso639_1);

			if (language == null){

				language = getTermService().getLanguageByIso(iso639_2);

				iso639Map.put(iso639_2, language);

				if (language == null){

					try {

						language = getTermService().getLanguageByIso(iso639_1);

					} catch (Exception e) {

						// TODO Auto-generated catch block

						// TODO remove if problem with duplicate DescElement_Annot id is solved

						e.printStackTrace();

					}

					iso639Map.put(iso639_1, language);

				}

				if (language == null){

					logger.warn("Language for code ISO693-2 '" + iso639_2 + "' and ISO693-1 '" + iso639_1 + "' was not found");

				}

			}

		} else if ("unknown".equals(languageString)){

			language = Language.UNKNOWN_LANGUAGE();

		} else if ("Majorcan".equalsIgnoreCase(languageString)){

			language = getLanguage(state, BerlinModelTransformer.uuidLangMajorcan, "Majorcan", "Majorcan (original 'mallorqu\u00EDn')", null);

		}else{

			logger.warn("language ISO 639_1 and ISO 639_2 were empty for " + languageString);

			language = null;

		}

		addOriginalLanguage(language, originalLanguageString);

		return language;

	}

	/**

	 * @param iso639Map

	 * @param iso639_2

	 * @param iso639_1

	 * @return

	 */

	private Language getLanguageFromIsoMap(Map<String, Language> iso639Map,	String iso639_2, String iso639_1) {

		Language language;

		language = iso639Map.get(iso639_2);

		if (language == null){

			language = iso639Map.get(iso639_1);

		}

		return language;

	}

	/**

	 * @param language

	 * @param originalLanguageString

	 */

	private void addOriginalLanguage(Language language,	String originalLanguageString) {

		if (isBlank(originalLanguageString)){

			return;

		}else if (language == null){

			logger.warn("Language could not be defined, but originalLanguageString exists: " + originalLanguageString);

		}else {

			Representation representation = language.getRepresentation(language);

			if (representation == null){

				language.addRepresentation(Representation.NewInstance(originalLanguageString, originalLanguageString, originalLanguageString, language));

				getTermService().saveOrUpdate(language);

			}

		}

	}

	/**

	 * Fills the regionFks with all regionFks from emCommonName. Comma separated regionFks will be split.

	 * @param state

	 * @param regionFks

	 * @param source

	 * @return

	 * @throws SQLException

	 *

	 */

	private void getRegionFks(BerlinModelImportState state, SortedSet<Integer> regionFks, Source source) throws SQLException {

		String sql = " SELECT DISTINCT RegionFks FROM emCommonName";

		if (state.getConfig().getCommonNameFilter() != null){

			sql += " WHERE " + state.getConfig().getCommonNameFilter();

		}

		ResultSet rs = source.getResultSet(sql);

		while (rs.next()){

			String strRegionFks = rs.getString("RegionFks");

			if (isBlank(strRegionFks)){

				continue;

			}

			String[] regionFkArray = strRegionFks.split(",");

			for (String regionFk: regionFkArray){

				regionFk = regionFk.trim();

				if (! StringUtils.isNumeric(regionFk) || "".equals(regionFk)  ){

					state.setUnsuccessfull();

					logger.warn("RegionFk is not numeric: " + regionFk +  " ( part of " + strRegionFks + ")");

				}else{

					regionFks.add(Integer.valueOf(regionFk));

				}

			}

		}

		return;

	}

	/**

	 * Fills the {@link #regionMap} by all emLanguageRegion regions defined in the sql filter.

	 * {@link #regionMap} maps emLanguageRegion.RegionId to named areas.

	 * @param state

	 * @param sqlWhere

	 * @param emTdwgMap

	 * @throws SQLException

	 */

	private void fillRegionMap(BerlinModelImportState state, String sqlWhere,

			Map<String, String> emTdwgMap) throws SQLException {

		Source source = state.getConfig().getSource();

		String sql;

		ResultSet rs;

		sql = " SELECT RegionId, Region FROM emLanguageRegion WHERE RegionId IN ("+ sqlWhere+ ") ";

		rs = source.getResultSet(sql);

		while (rs.next()){

			Object regionId = rs.getObject("RegionId");

			String region = rs.getString("Region");

			String[] splitRegion = region.split("-");

			if (splitRegion.length <= 1){

				NamedArea newArea = getNamedArea(state, null, region, "Language region '" + region + "'", null, null, null);

//				getTermService().save(newArea);

				regionMap.put(String.valueOf(regionId), newArea);

				logger.info("Found new area: " +  region);

			}else if (splitRegion.length == 2){

				String emCode = splitRegion[1].trim();

				String tdwgCode = emTdwgMap.get(emCode);

				if (isNotBlank(tdwgCode) ){

					NamedArea tdwgArea = getNamedArea(state, tdwgCode);

					regionMap.put(String.valueOf(regionId), tdwgArea);

				}else {

					NamedArea area = getOtherAreas(state, emCode, tdwgCode);

					if (area != null){

						regionMap.put(String.valueOf(regionId), area);

					}else{

						logger.warn("emCode did not map to valid tdwgCode: " +  CdmUtils.Nz(emCode) + "->" + CdmUtils.Nz(tdwgCode));

					}

				}

			}

		}

	}

	/**

	 * Returns the are for a given TDWG code. See {@link #getEmTdwgMap(Source)} for exceptions from

	 * the TDWG code

	 * @param state

	 * @param tdwgCode

	 */

	private NamedArea getNamedArea(BerlinModelImportState state, String tdwgCode) {

		NamedArea area;

		if (tdwgCode.equalsIgnoreCase("Ab")){

			area = getNamedArea(state, BerlinModelTransformer.uuidAb, "Azerbaijan & Nakhichevan", "Azerbaijan (including Nakhichevan)",  "Ab", null, null);

			getTermService().saveOrUpdate(area);

		}else if (tdwgCode.equalsIgnoreCase("Uk")){

			area = getNamedArea(state, BerlinModelTransformer.uuidUk , "Ukraine & Crimea", "Ukraine (including Crimea)", "Uk", null, null);

			getTermService().saveOrUpdate(area);

		}else if (tdwgCode.equalsIgnoreCase("Rf")){

//			area = getNamedArea(state, BerlinModelTransformer.uuidRf , "Ukraine & Crimea", "Ukraine (including Crimea)", "Uk", null, null);

//			getTermService().saveOrUpdate(area);

			area = Country.RUSSIANFEDERATION();

		}else if (tdwgCode.equalsIgnoreCase("Gg")){

			area = Country.GEORGIA();

		}else if (tdwgCode.equalsIgnoreCase("SM")){

            area = getNamedArea(state, BerlinModelTransformer.uuidSM , "Serbia & Montenegro", "Serbia & Montenegro", "SM", null, null);

            getTermService().saveOrUpdate(area);

        }else if (tdwgCode.equalsIgnoreCase("Tu")){

            area = Country.TURKEYREPUBLICOF();

        }else{

			area = TdwgAreaProvider.getAreaByTdwgAbbreviation(tdwgCode);

		}

		if (area == null){

			logger.warn("Area is null for " + tdwgCode);

		}

		return area;

	}

	/**

	 * @param regionFks

	 * @return

	 */

	private String getSqlWhere(SortedSet<Integer> regionFks) {

		String sqlWhere = "";

		for (Integer regionFk : regionFks){

			sqlWhere += regionFk + ",";

		}

		sqlWhere = sqlWhere.substring(0, sqlWhere.length()-1);

		return sqlWhere;

	}

	/**

	 * Returns a map which is filled by the emCode->TdwgCode mapping defined in emArea.

	 * Some exceptions are defined for emCode 'Ab','Rf','Uk' and some additional mapping is added

	 * for 'Ab / Ab(A)', 'Ga / Ga(F)', 'It / It(I)', 'Ar / Ar(A)','Hs / Hs(S)'

	 * @param source

	 * @throws SQLException

	 */

	private Map<String, String> getEmTdwgMap(Source source) throws SQLException {

		String sql;

		ResultSet rs;

		Map<String, String> emTdwgMap = new HashMap<>();

		sql = " SELECT EmCode, TDWGCode FROM emArea ";

		rs = source.getResultSet(sql);

		while (rs.next()){

			String emCode = rs.getString("EMCode");

			String TDWGCode = rs.getString("TDWGCode");

			if (isNotBlank(emCode) ){

				emCode = emCode.trim();

				if (emCode.equalsIgnoreCase("Ab") || emCode.equalsIgnoreCase("Rf")||

						emCode.equalsIgnoreCase("Uk") || emCode.equalsIgnoreCase("Gg")

						|| emCode.equalsIgnoreCase("SM") || emCode.equalsIgnoreCase("Tu")){

					emTdwgMap.put(emCode, emCode);

				}else if (isNotBlank(TDWGCode)){

					emTdwgMap.put(emCode, TDWGCode.trim());

				}

			}

		}

		emTdwgMap.put("Ab / Ab(A)", "Ab");

		emTdwgMap.put("Ga / Ga(F)", "FRA-FR");

		emTdwgMap.put("It / It(I)", "ITA");

		emTdwgMap.put("Uk / Uk(U)", "Uk");

		emTdwgMap.put("Ar / Ar(A)", "TCS-AR");

		emTdwgMap.put("Hs / Hs(S)", "SPA-SP");

		emTdwgMap.put("Hb / Hb(E)", "IRE-IR");

		return emTdwgMap;

	}

	/**

	 * Returns the first non-image gallery description. Creates a new one if no description exists.

	 * @param taxon

	 * @return

	 */

	private TaxonDescription getDescription(Taxon taxon) {

		TaxonDescription result = null;

		for (TaxonDescription taxonDescription : taxon.getDescriptions()){

			if (! taxonDescription.isImageGallery()){

				result = taxonDescription;

			}

		}

		if (result == null){

			result = TaxonDescription.NewInstance(taxon);

		}

		return result;

	}

	@Override

	public Map<Object, Map<String, ? extends CdmBase>> getRelatedObjectsForPartition(ResultSet rs, BerlinModelImportState state) {

		String nameSpace;

		Class<?> cdmClass;

		Set<String> idSet;

		Map<Object, Map<String, ? extends CdmBase>> result = new HashMap<>();

		String pos = "0";

		try{

			Set<String> taxonIdSet = new HashSet<>();

			Set<String> nameIdSet = new HashSet<>();

			Set<String> referenceIdSet = new HashSet<>();

			while (rs.next()){

				handleForeignKey(rs, taxonIdSet, "taxonId");

				handleForeignKey(rs, taxonIdSet, "factTaxonId");

                handleForeignKey(rs, taxonIdSet, "misappliedTaxonId");

				handleForeignKey(rs, referenceIdSet, "refId");

				handleForeignKey(rs, referenceIdSet, "languageRefRefFk");

				handleForeignKey(rs, nameIdSet, "NameInSourceFk");

				handleForeignKey(rs, nameIdSet, "PTNameFk");

				handleForeignKey(rs, referenceIdSet, "MisNameRefFk");

			}

			//name map

			nameSpace = BerlinModelTaxonNameImport.NAMESPACE;

			cdmClass = TaxonName.class;

			idSet = nameIdSet;

			@SuppressWarnings("unchecked")

            Map<String, TaxonName> nameMap = (Map<String, TaxonName>)getCommonService().getSourcedObjectsByIdInSource(cdmClass, idSet, nameSpace);

			result.put(nameSpace, nameMap);

			//taxon map

			nameSpace = BerlinModelTaxonImport.NAMESPACE;

			cdmClass = TaxonBase.class;

			idSet = taxonIdSet;

			@SuppressWarnings("unchecked")

            Map<String, TaxonBase<?>> taxonMap = (Map<String, TaxonBase<?>>)getCommonService().getSourcedObjectsByIdInSource(cdmClass, idSet, nameSpace);

			result.put(nameSpace, taxonMap);

			//reference map

			nameSpace = BerlinModelReferenceImport.REFERENCE_NAMESPACE;

			cdmClass = Reference.class;

			idSet = referenceIdSet;

			@SuppressWarnings("unchecked")

            Map<String, Reference> referenceMap = (Map<String, Reference>)getCommonService().getSourcedObjectsByIdInSource(cdmClass, idSet, nameSpace);

			result.put(nameSpace, referenceMap);

			// TODO remove if problem with duplicate DescElement_Annot id is solved

		} catch (SQLException e) {

			throw new RuntimeException("pos: " +pos, e);

		} catch (NullPointerException nep){

			logger.error("NullPointerException in getRelatedObjectsForPartition()");

		}

		return result;

	}

	@Override

	protected boolean doCheck(BerlinModelImportState state){

		IOValidator<BerlinModelImportState> validator = new BerlinModelCommonNamesImportValidator();

		return validator.validate(state);

	}

	@Override

	protected boolean isIgnore(BerlinModelImportState state){

		return ! state.getConfig().isDoCommonNames();

	}

}