/**
 * Licensed to DigitalPebble Ltd under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * DigitalPebble licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.digitalpebble.storm.crawler.parse.filter;

import java.io.IOException;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.Set;

import javax.xml.namespace.QName;
import javax.xml.xpath.XPath;
import javax.xml.xpath.XPathConstants;
import javax.xml.xpath.XPathExpression;
import javax.xml.xpath.XPathExpressionException;
import javax.xml.xpath.XPathFactory;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.xml.serialize.Method;
import org.apache.xml.serialize.OutputFormat;
import org.apache.xml.serialize.XMLSerializer;
import org.codehaus.jackson.JsonNode;
import org.w3c.dom.Document;
import org.w3c.dom.DocumentFragment;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;

import com.digitalpebble.storm.crawler.parse.ParseFilter;
import com.digitalpebble.storm.crawler.util.KeyValues;

/**
 * Simple ParseFilter to illustrate and test the interface. Reads a XPATH
 * pattern from the config file and stores the value as metadata
 **/

public class XPathFilter implements ParseFilter {

    private enum EvalFunction {

        NONE, STRING, SERIALIZE;

        public QName getReturnType() {
            switch (this) {
            case STRING:
                return XPathConstants.STRING;
            default:
                return XPathConstants.NODESET;
            }
        }
    }

    public static final Log LOG = LogFactory.getLog(XPathFilter.class);

    private static XPathFactory factory = XPathFactory.newInstance();
    private static XPath xpath = factory.newXPath();

    private List<LabelledExpression> expressions = new ArrayList<LabelledExpression>();

    private class LabelledExpression {

        private String key;
        private EvalFunction evalFunction;
        private XPathExpression expression;

        private LabelledExpression(String key, String expression) throws XPathExpressionException {
            this.key = key;
            if (expression.startsWith("string(")) {
                evalFunction = EvalFunction.STRING;
            }
            else if (expression.startsWith("serialize(")) {
                expression = expression.substring(10, expression.length() -1);
                evalFunction = EvalFunction.SERIALIZE;
            }
            else {
                evalFunction = EvalFunction.NONE;
            }
            this.expression = xpath.compile(expression);
        }

        private Set<String> evaluate(DocumentFragment doc) throws XPathExpressionException, IOException
        {
            Object evalResult = expression.evaluate(doc, evalFunction.getReturnType());
            Set<String> values = new HashSet<String>();
            switch (evalFunction) {
            case STRING:
                if (evalResult != null) {
                    values.add((String) evalResult);
                }
                break;
            case SERIALIZE:
                NodeList nodesToSerialize = (NodeList) evalResult;
                StringWriter out = new StringWriter();
                OutputFormat format = new OutputFormat( Method.XHTML, null, false);
                format.setOmitXMLDeclaration(true);
                XMLSerializer serializer = new XMLSerializer(out, format);
                for (int i = 0; i < nodesToSerialize.getLength(); i++) {
                    Node node = nodesToSerialize.item(i);
                    switch (node.getNodeType()) {
                    case Node.ELEMENT_NODE:
                        serializer.serialize((Element) node);
                        break;
                    case Node.DOCUMENT_NODE:
                        serializer.serialize((Document) node);
                        break;
                    case Node.DOCUMENT_FRAGMENT_NODE:
                        serializer.serialize((DocumentFragment) node);
                        break;
                    }
                    String serializedValue = out.toString();
                    if (serializedValue.length() > 0) {
                        values.add(serializedValue);
                    }
                    out.getBuffer().setLength(0);
                }
                break;
            default:
                NodeList nodes = (NodeList) evalResult;
                for (int i = 0; i < nodes.getLength(); i++) {
                    Node node = nodes.item(i);
                    values.add(node.getTextContent());
                }
            }
            return values;
        }
    }

    @Override
    public void filter(String URL, byte[] content, DocumentFragment doc,
            HashMap<String, String[]> metadata) {

        // applies the XPATH expression in the order in which they are produced

        java.util.Iterator<LabelledExpression> iter = expressions.iterator();
        while (iter.hasNext()) {
            LabelledExpression le = iter.next();
            try {
                Set<String> values = le.evaluate(doc);
                KeyValues.addValues(le.key, metadata, values);
            } catch (XPathExpressionException e) {
                LOG.error(e);
            } catch (IOException e) {
                LOG.error(e);
            }
        }

    }

    @Override
    public void configure(JsonNode paramNode) {
        java.util.Iterator<Entry<String, JsonNode>> iter = paramNode
                .getFields();
        while (iter.hasNext()) {
            Entry<String, JsonNode> entry = iter.next();
            String key = entry.getKey();
            String xpathvalue = entry.getValue().asText();
            try {
                LabelledExpression lexpression = new LabelledExpression(key, xpathvalue);
                expressions.add(lexpression);
            } catch (XPathExpressionException e) {
                throw new RuntimeException("Can't compile expression : "
                        + xpathvalue, e);
            }
        }

    }

    @Override
    public boolean needsDOM() {
        return true;
    }

}

@@ -1284,93 +1284,8 @@
 y;%0A%0A
-import org.apache.commons.logging.Log;%0Aimport org.apache.commons.logging.LogFactory;%0A
 impo
@@ -1447,24 +1447,81 @@
 n.JsonNode;%0A
+import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A
 import org.w
@@ -2302,16 +2302,19 @@
 inal Log
+ger
  LOG = L
@@ -2315,16 +2315,19 @@
 OG = Log
+ger
 Factory.
@@ -2332,16 +2332,19 @@
 y.getLog
+ger
 (XPathFi
@@ -5966,32 +5966,67 @@
       LOG.error(
+%22Error evaluating %7B%7D: %7B%7D%22, le.key, 
 e);%0A            
@@ -6077,16 +6077,51 @@
 G.error(
+%22Error evaluating %7B%7D: %7B%7D%22, le.key, 
 e);%0A    

Switch XPAthFilter logging to sl4j to be consistent with the rest of the code