D-Net

package eu.dnetlib.efg.backlinks;

import java.io.StringReader;

import java.io.StringWriter;

import java.util.*;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.ThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

import javax.xml.namespace.QName;

import javax.xml.parsers.DocumentBuilder;

import javax.xml.parsers.DocumentBuilderFactory;

import javax.xml.parsers.ParserConfigurationException;

import javax.xml.stream.*;

import javax.xml.stream.events.EndElement;

import javax.xml.stream.events.StartElement;

import javax.xml.stream.events.XMLEvent;

import javax.xml.transform.*;

import javax.xml.transform.dom.DOMSource;

import javax.xml.transform.stream.StreamResult;

import javax.xml.transform.stream.StreamSource;

import com.mongodb.BasicDBObject;

import com.mongodb.DBObject;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.MongoDatabase;

import com.mongodb.client.model.Filters;

import com.mongodb.client.model.UpdateOptions;

import eu.dnetlib.data.mdstore.modular.action.DoneCallback;

import eu.dnetlib.data.mdstore.modular.mongodb.MDStoreTransactionManagerImpl;

import eu.dnetlib.rmi.data.MDStoreServiceException;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.lang3.StringEscapeUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.conversions.Bson;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Required;

import org.springframework.core.io.Resource;

import org.w3c.dom.Document;

import org.w3c.dom.Element;

import org.w3c.dom.Node;

/**

 * handle backlinks

 * <p/>

 * AvCreation Collection Corporate NonAVCreation Person

 *

 * @author marko

 */

public class MongoBuildBacklinks {

	private static final Log log = LogFactory.getLog(MongoBuildBacklinks.class); // NOPMD by marko on 11/24/08 5:02 PM

	private final int queueSize = 80;

	private final int cpus = Runtime.getRuntime().availableProcessors();

	private final ThreadLocal<XMLInputFactory> factory = new ThreadLocal<XMLInputFactory>() {

		@Override

		protected XMLInputFactory initialValue() {

			return XMLInputFactory.newInstance();

		}

	};

	private final ThreadLocal<XMLOutputFactory> outputFactory = new ThreadLocal<XMLOutputFactory>() {

		@Override

		protected XMLOutputFactory initialValue() {

			return XMLOutputFactory.newInstance();

		}

	};

	private final ThreadLocal<Transformer> serializer = new ThreadLocal<Transformer>() {

		@Override

		protected Transformer initialValue() {

			final TransformerFactory factory = TransformerFactory.newInstance();

			try {

				final Transformer trans = factory.newTransformer();

				trans.setOutputProperty(OutputKeys.OMIT_XML_DECLARATION, "yes");

				return trans;

			} catch (final TransformerConfigurationException e) {

				throw new IllegalStateException(e);

			}

		}

	};

	private final ThreadLocal<DocumentBuilder> docBuilder = new ThreadLocal<DocumentBuilder>() {

		@Override

		protected DocumentBuilder initialValue() {

			final DocumentBuilderFactory dbfac = DocumentBuilderFactory.newInstance();

			try {

				return dbfac.newDocumentBuilder();

			} catch (final ParserConfigurationException e) {

				throw new IllegalStateException(e);

			}

		}

	};

	public UpdateOperation sentinel = new UpdateOperation(null, null);

	@Autowired

	private MDStoreTransactionManagerImpl transactionManager;

	private BacklinkTypeMatcher backlinkTypeMatcher;

	private int MAX_NUMBER_OF_RELS = 1000;

	private List<String> relationTypes = new ArrayList<>();

	private Resource fixLinksXslt;

	private StringTemplate backlinkTemplate;

	private Set<String> entityTypes;

	private List<String> titleTypes;

	public void process(final String mdId, final DoneCallback doneCallback) throws MDStoreServiceException {

		try {

			log.debug("generating backlink for mdstore ID:" + mdId);

			ThreadPoolExecutor executor = createExecutor();

			final String internalId = transactionManager.readMdStore(mdId);

			final MongoDatabase db = transactionManager.getDb();

			final MongoCollection<DBObject> backlinks = db.getCollection(mdId + "backlinks", DBObject.class);

			backlinks.deleteMany(new BasicDBObject());

			final BlockingQueue<UpdateOperation> emitQueue = new ArrayBlockingQueue<>(500);

			final MongoCollection<DBObject> data = db.getCollection(internalId, DBObject.class);

			final long size = data.count();

			double currentPosition = 0;

			backlinks.drop();

			backlinks.createIndex(new BasicDBObject("id", 1));

			data.createIndex(new BasicDBObject("id", 1));

			data.createIndex(new BasicDBObject("originalId", 1));

			log.info("start background updating thread");

			final Thread backgroundUpdating = startBackgroundUpdating(emitQueue, backlinks);

			log.info("start parsing links");

			currentPosition = parseLinks(currentPosition, executor, data, emitQueue);

			executor.shutdown();

			executor.awaitTermination(10, TimeUnit.MINUTES);

			log.info("finish parallel part: grouping " + executor.isShutdown());

			emitQueue.put(sentinel);

			backgroundUpdating.join();

			log.info("finish background backlinks storing");

			executor = createExecutor();

			log.info("new executor " + executor.isShutdown());

			fixLinks(data, size, backlinks, executor, currentPosition);

			log.info("waiting for other executor");

			executor.shutdown();

			executor.awaitTermination(10, TimeUnit.MINUTES);

			log.info("finish parallel part: fixing");

			backlinks.createIndex(new BasicDBObject("id", 1));

			data.createIndex(new BasicDBObject("id", 1));

			data.createIndex(new BasicDBObject("originalId", 1));

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

			doneCallback.call(paramsOut);

		} catch (final InterruptedException e) {

			throw new IllegalStateException(e);

		}

	}

	private void fixLinks(final MongoCollection<DBObject> data,

			final long size,

			final MongoCollection<DBObject> backlinks,

			final ThreadPoolExecutor executor,

			double currentPosition

	) {

		final double step = (1.0 * size) / backlinks.count();

		log.info("FIXLINKS");

		for (final DBObject link : backlinks.find()) {

			log.debug("FIXING LINK " + link);

			try {

				executor.submit((Runnable) () -> {

					try {

						@SuppressWarnings("unchecked")

						List<DBObject> links = (List<DBObject>) link.get("links");

						fixLinks(data, (String) link.get("id"), links);

					} catch (final Throwable e) {

						log.warn("problems fixing", e);

					}

				});

				currentPosition += step;

			} catch (final Throwable e) {

				log.warn("problems fixing", e);

			}

		}

		log.info("FINISH FIXLINKS");

	}

	private double parseLinks(double currentPosition,

			final ThreadPoolExecutor executor,

			final MongoCollection<DBObject> data,

			final BlockingQueue<UpdateOperation> emitQueue) {

		// log.info("PARSING LINKS " + collection);

		for (final DBObject current : data.find()) {

			try {

				// log.info("PARSING LINK " + current.get("originalId"));

				executor.submit((Runnable) () -> parseLinks((String) current.get("body"), emitQueue));

				currentPosition++;

			} catch (final Throwable e) {

				log.warn("problems parsing", e);

			}

		}

		return currentPosition;

	}

	private Thread startBackgroundUpdating(final BlockingQueue<UpdateOperation> queue, final MongoCollection<DBObject> coll) {

		final Thread background = new Thread(() -> {

			while (true) {

				try {

					final UpdateOperation record = queue.take();

					if (record == sentinel) {

						break;

					}

					final UpdateOptions op = new UpdateOptions();

					op.upsert(true);

					coll.updateOne((Bson) record.find, (Bson) record.update, op);

				} catch (final InterruptedException e) {

					log.fatal("got exception in background thread", e);

					throw new IllegalStateException(e);

				}

			}

		});

		background.start();

		return background;

	}

	protected ThreadPoolExecutor createExecutor() {

		return new ThreadPoolExecutor(cpus, cpus, 5, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueSize, true),

				new ThreadPoolExecutor.CallerRunsPolicy());

	}

	private void fixLinks(final MongoCollection<DBObject> collection, final String id, final List<DBObject> links) {

		final Bson query = Filters.eq("originalId", id);

		final BasicDBObject source = (BasicDBObject) collection.find(query).first();

		if (source == null) {

			log.warn("object with originalId " + id + " doesn't exist or doesn't belong to this collection");

			return;

		}

		final String origBody = (String) source.get("body");

		log.info("UPDATING LINKS " + id);

		final String updated = updateLinks(id, origBody, links);

		// for (DBObject link : links) {

		// log.info(" link from " + id + " to: " + link.get("target"));

		// }

		if (!origBody.equalsIgnoreCase(updated)) {

			collection.updateOne(query, new BasicDBObject("$set", new BasicDBObject("body", updated)));

		}

	}

	public boolean isEfgEntity(final StartElement element) {

		return isEfgEntity(element.getName());

	}

	public boolean isEfgEntity(final EndElement element) {

		return isEfgEntity(element.getName());

	}

	public boolean isEfgEntity(final QName name) {

		return entityTypes.contains(name.getLocalPart());

	}

	public boolean isRelation(final XMLEvent event) {

		if (event.isStartElement()) return isRelation(event.asStartElement());

		return false;

	}

	public boolean isEndRelation(final XMLEvent event) {

		if (event.isEndElement()) return isRelation(event.asEndElement());

		return false;

	}

	public boolean isRelation(final StartElement element) {

		return isRelation(element.getName());

	}

	public boolean isRelation(final EndElement element) {

		return isRelation(element.getName());

	}

	public boolean isRelation(final QName name) {

		return name.getLocalPart().startsWith("rel");

	}

	public XMLEventReader fragment(final String source) throws XMLStreamException {

		return fragment(factory.get().createXMLEventReader(new StringReader(source)));

	}

	public XMLEventReader fragment(final XMLEventReader reader) throws XMLStreamException {

		return factory.get().createFilteredReader(reader, new EventFilter() {

			@Override

			public boolean accept(final XMLEvent evt) {

				return !(evt.isStartDocument() || evt.isEndDocument());

			}

		});

	}

	public String linkToFragment(final DBObject link) {

		StringTemplate template = new StringTemplate(backlinkTemplate.getTemplate());

		template.setAttribute("relType", prepareValueForTemplate(link.get("type")));

		template.setAttribute("targetId", prepareValueForTemplate(link.get("target")));

		template.setAttribute("title", prepareValueForTemplate(link.get("title")));

		template.setAttribute("name", prepareValueForTemplate(link.get("name")));

		template.setAttribute("type", prepareValueForTemplate(backlinkTypeMatcher.getInverseType(link.get("elementType").toString())));

		template.setAttribute("itemTypes", prepareValueForTemplate(link.get("itemTypes")));

		String res = template.toString();

		if (log.isDebugEnabled()) {

			log.debug("UPDATING adding link: " + res.replaceAll("\n", ""));

		}

		return res;

	}

	private Object prepareValueForTemplate(final Object obj) {

		if ((obj != null) && (obj instanceof String)) return StringEscapeUtils.escapeXml11(obj.toString());

		return obj;

	}

	private String maybeIdentifier(final XMLEventReader reader, final XMLEventWriter writer, final XMLEvent event) throws XMLStreamException {

		if (event.isStartElement() && event.asStartElement().getName().getLocalPart().equalsIgnoreCase("identifier")) {

			XMLEvent next = reader.peek();

			if (next.isCharacters()) return next.asCharacters().getData().trim();

		}

		return null;

	}

	public DBObject findLink(final String currentRelId, final List<DBObject> links) {

		for (DBObject link : links)

			if (link.get("target").equals(currentRelId)) return link;

		return null;

	}

	@SuppressWarnings("unchecked")

	public String updateLinks(final String id, final String body, final List<DBObject> links) {

		log.info("UPDATING links " + id);

		try {

			StringWriter res = new StringWriter();

			XMLEventReader reader = factory.get().createXMLEventReader(new StringReader(body));

			XMLEventWriter writer = outputFactory.get().createXMLEventWriter(res);

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

			boolean inRelation = false;

			boolean hasItemType = false;

			String currentRelId = null;

			XMLEvent event = null;

			while (!(event = reader.nextEvent()).isEndDocument()) {

				if (isRelation(event)) {

					inRelation = true;

					currentRelId = null;

				} else if (isEndRelation(event)) {

					inRelation = false;

					if (!hasItemType) {

						DBObject obj = findLink(currentRelId, links);

						if (obj != null) {

							final Object itemTypes = obj.get("itemTypes");

							if (itemTypes != null) {

								for (String itemType : (List<String>) itemTypes) {

									writer.add(fragment("<efg:itemType xmlns:efg=\"http://www.europeanfilmgateway.eu/efg\" generated=\"true\">" + itemType

											+ "</efg:itemType>"));

								}

							}

						}

					}

				}

				if (inRelation) {

					// if current element is the identifier, mark the relation as seen.

					String relId = maybeIdentifier(reader, writer, event);

					if (relId != null) {

						currentRelId = relId;

						present.add(relId);

					}

					if (event.isStartElement() && event.asStartElement().getName().getLocalPart().equalsIgnoreCase("itemType")) {

						hasItemType = true;

					}

				}

				// flush new backlinks at the end of the record

				if (event.isEndElement()) {

					EndElement end = event.asEndElement();

					if (isEfgEntity(end)) {

						log.info("UPDATING end element is efgEntity");

						for (int i = 0; i < Math.min(links.size(), MAX_NUMBER_OF_RELS); i++) {

							DBObject link = links.get(i);

							// generate only those (back)links which are not already present.

							if (!present.contains(link.get("target"))) {

								writer.add(fragment(linkToFragment(link)));

							}

						}

					}

				}

				writer.add(event);

			}

			writer.close();

			return res.toString();

		} catch (Exception e) {

			throw new IllegalStateException(e);

		}

	}

	protected Document createLinksParams(final List<DBObject> links) throws ParserConfigurationException {

		final Document doc = docBuilder.get().newDocument();

		createLinksParams(links, doc);

		return doc;

	}

	protected Document createLinksParams(final List<DBObject> links, final Document doc) throws ParserConfigurationException {

		return createLinksParams(links, doc, doc);

	}

	protected Document createLinksParams(final List<DBObject> links, final Document doc, final Node container) throws ParserConfigurationException {

		final Element root = doc.createElement("links");

		container.appendChild(root);

		for (final DBObject link : links) {

			final Element node = doc.createElement("link");

			node.setAttribute("target", (String) link.get("target"));

			node.setAttribute("type", (String) link.get("type"));

			Arrays.asList("name", "title").stream().filter(attr -> link.get(attr) != null).forEach(attr -> {

				node.setAttribute(attr, (String) link.get(attr));

			});

			root.appendChild(node);

		}

		return doc;

	}

	protected String serialize(final Node doc) {

		try {

			// create string from xml tree

			final StringWriter sw = new StringWriter();

			final StreamResult result = new StreamResult(sw);

			final DOMSource source = new DOMSource(doc);

			serializer.get().transform(source, result);

			return sw.toString();

		} catch (final TransformerException e) {

			log.warn("cannot serialize", e);

		}

		return "";

	}

	public boolean moreSpecific(final String next, final String old, final List<String> categories) {

		if (old == null) return true;

		if (next == null) return false;

		return categories.indexOf(next.toLowerCase()) > categories.indexOf(old.toLowerCase());

	}

	private void parseLinks(final String record, final BlockingQueue<UpdateOperation> emitQueue) {

		// log.info("record: " + record);

		String recordId = null;

		String recordType = null;

		String relType = null;

		final List<ToEmit> toEmit = new ArrayList<>();

		final Map<String, String> params = new HashMap<>();

		// for persons:

		String firstName = null;

		String lastName = "";

		// for creations

		String title = null;

		String titleType = null;

		Stack<String> elementStack = new Stack<String>();

		elementStack.push("/");

		try {

			final XMLStreamReader parser = factory.get().createXMLStreamReader(new StreamSource(new StringReader(record)));

			while (parser.hasNext()) {

				int event = parser.next();

				if (event == XMLStreamConstants.END_ELEMENT) {

					pop(elementStack);

				} else if (event == XMLStreamConstants.START_ELEMENT) {

					final String localName = parser.getLocalName();

					push(elementStack, localName);

					if ("efgEntity".equalsIgnoreCase(localName)) {

						while (parser.hasNext()) {

							event = parser.next();

							if (event == XMLStreamConstants.START_ELEMENT) {

								recordType = parser.getLocalName();

								relType = normalizeRecordType(recordType);

								push(elementStack, recordType);

								break;

							}

						}

					}

					if ("identifier".equalsIgnoreCase(localName) && "efgEntity".equalsIgnoreCase(grandParent(elementStack))) {

						parser.next();

						recordId = parser.getText();

					} else if ("identifier".equalsIgnoreCase(localName)) {

						parser.next();

						// log.info("CCCCC: is an id but not the right depth: " + parser.getText());

						// log.info("stack " + elementStack);

					} else if ("person".equalsIgnoreCase(recordType)) {

						if ("name".equalsIgnoreCase(localName)) {

							final String part = parser.getAttributeValue(null, "part");

							parser.next();

							String value = parser.getText();

							if (value != null) {

								value = value.trim();

							}

							if ("forename".equalsIgnoreCase(part) || "Forename(s)".equalsIgnoreCase(part)) {

								firstName = value;

							} else if ("surname".equalsIgnoreCase(part) || "Family Name".equalsIgnoreCase(part)) {

								lastName = value;

							} else if ("stem".equalsIgnoreCase(part)) {

								lastName = value;

							}

						}

					} else if (isCreation(recordType)) {

						if ("title".equalsIgnoreCase(localName)) {

							String tmpTitle = null;

							String tmpType = null;

							while (parser.hasNext()) {

								event = parser.next();

								if (event == XMLStreamConstants.START_ELEMENT) {

									push(elementStack, parser.getLocalName());

									if ("text".equalsIgnoreCase(parser.getLocalName())) {

										event = parser.next();

										if (event == XMLStreamConstants.CHARACTERS) {

											tmpTitle = parser.getText();

										}

									} else if ("relation".equalsIgnoreCase(parser.getLocalName())) {

										event = parser.next();

										if (event == XMLStreamConstants.CHARACTERS) {

											tmpType = parser.getText();

										}

									}

								} else if (event == XMLStreamConstants.END_ELEMENT) {

									pop(elementStack);

									if ("title".equalsIgnoreCase(parser.getLocalName())) {

										tmpType = (tmpType == null) ? "n/a" : tmpType.trim();

										tmpTitle = (tmpTitle == null) ? "n/a" : tmpTitle.trim();

										if (moreSpecific(tmpType, titleType, titleTypes)) {

											title = tmpTitle;

											titleType = tmpType;

										}

										break;

									}

								}

							}

						}

					}

					if (relationTypes.contains(localName)) {

						ToEmit nextEmission = null;

						while (parser.hasNext()) {

							event = parser.next();

							if (event == XMLStreamConstants.START_ELEMENT) {

								push(elementStack, parser.getLocalName());

								if ("identifier".equalsIgnoreCase(parser.getLocalName())) {

									parser.next();

									// toEmit.add(new ToEmit(parser.getText().trim(), "n/a"));

									nextEmission = new ToEmit(parser.getText().trim(), "n/a");

									// break;

								} else if ("type".equalsIgnoreCase(parser.getLocalName())) {

									parser.next();

									nextEmission.setElementType(parser.getText().trim());

									toEmit.add(nextEmission);

									break;

								}

							} else if (event == XMLStreamConstants.END_ELEMENT) {

								pop(elementStack);

							}

						}

					}

				}

			}

			if ("person".equalsIgnoreCase(recordType)) {

				String personName = lastName;

				if (firstName != null) {

					personName = firstName + " " + lastName;

				}

				params.put("name", personName);

			} else if (isCreation(recordType) && (title != null)) {

				params.put("title", title);

			}

			if (toEmit.isEmpty() && log.isDebugEnabled()) {

				log.debug("emit queue is empty for record " + recordId.trim());

			}

			for (final ToEmit emitting : toEmit) {

				emitBacklink(emitQueue, emitting.targetId, recordId.trim(), relType, emitting.elementType, params);

			}

		} catch (final XMLStreamException e) {

			log.info("some parsing exception, moving along");

		}

	}

	private String grandParent(final Stack<String> elementStack) {

		if (elementStack.size() <= 3) return "";

		String res = elementStack.get(elementStack.size() - 3);

		return res;

	}

	private void pop(final Stack<String> elementStack) {

		elementStack.pop();

	}

	private void push(final Stack<String> elementStack, final String localName) {

		elementStack.push(localName);

	}

	private boolean isCreation(final String recordType) {

		return "avcreation".equalsIgnoreCase(recordType) || "nonavcreation".equalsIgnoreCase(recordType);

	}

	private String normalizeRecordType(final String localName) {

		final String relName = "rel" + localName;

		for (final String type : relationTypes) {

			if (relName.equalsIgnoreCase(type)) return type;

		}

		return relName;

	}

	private void emitBacklink(final BlockingQueue<UpdateOperation> emitQueue,

			final String source,

			final String target,

			final String targetType,

			final String elementType,

			final Map<String, String> params) {

		final DBObject pair = new BasicDBObject();

		pair.put("target", target);

		pair.put("type", targetType);

		pair.put("elementType", elementType);

		pair.putAll(params);

		final DBObject find = new BasicDBObject("id", source);

		final DBObject update = new BasicDBObject("$push", new BasicDBObject("links", pair));

		try {

			emitQueue.put(new UpdateOperation(find, update));

		} catch (final InterruptedException e) {

			throw new IllegalStateException(e);

		}

	}

	public List<String> getRelationTypes() {

		return relationTypes;

	}

	public void setRelationTypes(final List<String> relationTypes) {

		this.relationTypes = relationTypes;

	}

	public Resource getFixLinksXslt() {

		return fixLinksXslt;

	}

	@Required

	public void setFixLinksXslt(final Resource fixLinksXslt) {

		this.fixLinksXslt = fixLinksXslt;

	}

	public StringTemplate getBacklinkTemplate() {

		return backlinkTemplate;

	}

	@Required

	public void setBacklinkTemplate(final StringTemplate backlinkTemplate) {

		this.backlinkTemplate = backlinkTemplate;

	}

	public List<String> getTitleTypes() {

		return titleTypes;

	}

	public void setTitleTypes(final List<String> titleTypes) {

		this.titleTypes = titleTypes;

	}

	public Set<String> getEntityTypes() {

		return entityTypes;

	}

	public void setEntityTypes(final Set<String> entityTypes) {

		this.entityTypes = entityTypes;

	}

	public BacklinkTypeMatcher getBacklinkTypeMatcher() {

		return backlinkTypeMatcher;

	}

	@Required

	public void setBacklinkTypeMatcher(final BacklinkTypeMatcher backlinkTypeMatcher) {

		this.backlinkTypeMatcher = backlinkTypeMatcher;

	}

	public int getMAX_NUMBER_OF_RELS() {

		return MAX_NUMBER_OF_RELS;

	}

	@Required

	public void setMAX_NUMBER_OF_RELS(final int mAXNUMBEROFRELS) {

		MAX_NUMBER_OF_RELS = mAXNUMBEROFRELS;

	}

	public static class UpdateOperation {

		private DBObject find;

		private DBObject update;

		public UpdateOperation(final DBObject find, final DBObject update) {

			super();

			this.find = find;

			this.update = update;

		}

		public DBObject getFind() {

			return find;

		}

		public void setFind(final DBObject find) {

			this.find = find;

		}

		public DBObject getUpdate() {

			return update;

		}

		public void setUpdate(final DBObject update) {

			this.update = update;

		}

	}

	static class ToEmit {