package com.telefonica.sittel.udf

import org.apache.spark.sql.api.java.UDF2

import scala.util.Try

import org.json4s.jackson.JsonMethods
import org.json4s.jackson.JsonMethods._

class FilterJSON extends UDF2[String, String, Boolean] {


    def isEquals(field: String, condition: String, row: JValue, objFilter: JValue, arrResult: List[Boolean]): Boolean = {
        var arrCondition = condition.split(".")
        var value = objFilter \ arrCondition(0)
        if(arrCondition.size == 2) {
            value = objFilter \ arrCondition(0) \ arrCondition(1)
        }
        if(value != JNothing) {
            val rowValue =  row \ field
            if (rowValue != JNothing) {
                arrResult.append(rowValue.toString().trim().equals(value.toString().trim()))
            }
        }

    }


    override def call(row: String, filter: String): Boolean = {
        val rowJSON    = JsonMethods.parse(row)
        val filterJSON = JsonMethods.parse(filter)

        val arrResult: List[Boolean] = List()
        val filtered = for {
            JObject(obj) <- filterJSON
            isEquals("num_terminal_linha_assinante", "numero_terminal", rowJSON, filterJSON, arrResult)
            //isEquals("rpon"             , "numero_terminal", rowJSON, filterJSON, arrResult)
            //isEquals("rpon_migrado_rnc" , "numero_terminal", rowJSON, filterJSON, arrResult)
            //isEquals("designador"       , "numero_terminal", rowJSON, filterJSON, arrResult)
            //isEquals("cpf_cnpj"         , "CNPJ" , rowJSON, filterJSON, arrResult)
            //isEquals("cpf_cnpj"         , "CPF"  , rowJSON, filterJSON, arrResult)
            //isEquals("iccid"            , "iccid", rowJSON, filterJSON, arrResult)
            val arrDocument: List[Boolean] = List()
            //isEquals("documento_assinante" , "documento.numero", rowJSON, filterJSON, arrDocument)
            //isEquals("tipo_documento"      , "documento.tipo"  , rowJSON, filterJSON, arrDocument)
            //if(arrDocument.size > 0) {
            //    arrResult.append( arrDocument.filter(_ == true) == arrDocument.size )
            //}
            val arrEndereco: List[Boolean] = List()
            //isEquals("bairro"        , "endereco.bairro",   rowJSON, filterJSON, arrEndereco)
            //isEquals("tipo_endereco" , "endereco.tipo_endereco", rowJSON, filterJSON, arrEndereco)
            //isEquals("endereco"      , "endereco.endereco", rowJSON, filterJSON, arrEndereco)
            //isEquals("uf"            , "endereco.uf",       rowJSON, filterJSON, arrEndereco)
            //isEquals("cep"           , "endereco.cep",      rowJSON, filterJSON, arrEndereco)
            //if(arrEndereco.size > 0) {
            //    arrResult.append( arrDocument.filter(_ == true) == arrDocument.size )
            // }

            JField(res) <- arrResult.filter(_ == true).size == arrResult.size
        } yield res

        return res
  }
}