import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.{ProcessingTime, Trigger}
import org.json._
import utils._


/**
  * Created by gaowei
  */
object ZooMessage {
  case class Log(timestamp : String, hostname : String, name : String, version : String, index : String,
                  input_type : String, offset : String, source : String, types : String, uid : String,
                  zid : String, text : String, sessionKey:String,ds:String)

  def logParing(log : String) = {
      val str = Config.decodeUFT8(log)
      try {
        var json = new JSONObject()
        if (str.startsWith("\"")){
          json = new JSONObject(str.replace("\\", "").dropRight(1).drop(1))
        }
        json =  new JSONObject(str)
        val timestamp = json.optString("@timestamp", "")
        val ds = timestamp.split("T")()
        val beat = json.getJSONObject("beat")
        val hostname = beat.optString("hostname", "")
        val name = beat.optString("name", "")
        val version = beat.optString("version", "")
        val index = json.optString("index", "")
        val input_type = json.optString("input_type", "")
        val message = json.optString("message", "")
        var uid = ""
        var zid = ""
        var sessionKey = ""
        val userPatterns = "user:[0-9]+".r
        var text = ""
        if (message.contains("建立连接")) {
          val zooPatterns = "聊天室:[0-9]+".r
          val sessionPatterns = "sessionKey:[^,\"}]+".r
          uid = if (userPatterns.findFirstMatchIn(message).isDefined) userPatterns.findFirstMatchIn(message).get.toString().substring(5) else "None"
          zid = if (zooPatterns.findFirstMatchIn(message).isDefined) zooPatterns.findFirstMatchIn(message).get.toString().substring(4) else "None"
          sessionKey = if (sessionPatterns.findFirstMatchIn(message).isDefined) sessionPatterns.findFirstMatchIn(message).get.toString().substring(12) else "None"
          text = "Connection"
        }
        else {
          val zooPatterns = "zoo:[0-9]+".r
          val sessionPatterns = "sessionKey:[^,\"}\\s+]+".r
          uid = if (userPatterns.findFirstMatchIn(message).isDefined) userPatterns.findFirstMatchIn(message).get.toString().substring(5) else "None"
          zid = if (zooPatterns.findFirstMatchIn(message).isDefined) zooPatterns.findFirstMatchIn(message).get.toString().substring(4) else "None"
          sessionKey = if (sessionPatterns.findFirstMatchIn(message).isDefined) sessionPatterns.findFirstMatchIn(message).get.toString().substring(11) else "None"
          text = "Leave"
        }
        val offset = json.optString("offset", "")
        val source = json.optString("source", "")
        val types = json.optString("type", "")
        Log( timestamp, hostname, name, version, index, input_type,  offset, source, types, uid, zid, text, sessionKey, ds)
      }
      catch {
        case ex : JSONException => Log("1","1","1","1","1","1","1","1","1","1","1","1","1","1")
      }
  }

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("ZooMessage").getOrCreate()
    val brokers = "xxx.xxx.xxx.xx.xx:9092,xxx.xxx.xxx.xx.xx::9092"
    val topics = "test"
    val df = spark.readStream.format("kafka").
      option("kafka.bootstrap.servers", brokers).
      option("subscribe", topics).
     load()

    import spark.implicits._
    val kafkaDf = df.selectExpr("CAST(value AS STRING)").as[String].
      filter(f => f.contains("cn.idongjia.zoo.handler.AbstractZooPackWebSocketHandler.afterConnectionEstablished") ||
        f.contains("cn.idongjia.zoo.handler.AbstractZooPackWebSocketHandler.afterConnectionClosed")).
      filter(_.contains("/data/servers/zoo/logs/service.log"))
    import spark.implicits._

    val dataFrame = kafkaDf.map {
      f =>
        logParing(f)
    }
    
    val query =  dataFrame.coalesce(1).writeStream.outputMode("append").partitionBy("ds").trigger(Trigger.ProcessingTime("600 seconds")).format("parquet").
      option("checkpointLocation", "/user/gaowei/auctionCheck").
      option("path", "/user/gaowei/auctionMini").start()
    query.awaitTermination()
  }
}

class MysqlSink(url: String, user: String, pwd: String) extends ForeachWriter[Row] {
    var conn: Connection = _

    override def open(partitionId: Long, epochId: Long): Boolean = {
      Class.forName("com.mysql.jdbc.Driver")
      conn = DriverManager.getConnection(url, user, pwd)
      true
    }

    override def process(value: Row): Unit = {
      val p = conn.prepareStatement("replace into test(pid,pv) values(?,?)")
      p.setString(1, value(0).toString)
      p.setLong(2, value(1).toString.toLong)
      p.execute()
    }

    override def close(errorOrNull: Throwable): Unit = {
      conn.close()
    }
  }

create table kafka_source (
messageKey varbinary,
message varbinary,
topic varchar,
`partition` int,
`offset` bigint
) with (
type = 'kafka010',
topic = 'topic1',
bootstrap.servers = 'test',
`group.id` = 'test',
startupMode = 'TIMESTAMP',
starttime = '2020-03-31 00:00:00'
);

create table pv_user_id (  
  `user_id` varchar,
  `view_from` varchar,
  `create_date` varchar,
  `pv` bigint,
  PRIMARY KEY (`user_id`,`view_from`,`create_date`)
 ) with (
   type='mysql',
   dburl='jdbc:mysql://test:3306/gw?useUnicode=true&characterEncoding=UTF-8&useSSL=true',
   drivername='com.mysql.jdbc.Driver',
   tablename='pv_user_id',
   username='test',
   password='test'
 );

insert into pv_user_id
select user_id,
view_from,
create_date,
count(1) as pv
from (
select
JSON_VALUE(cast(message as varchar), '$.data.otherUId') as user_id,
case when JSON_VALUE(cast(message as varchar), '$.data.from') is null then '' 
else JSON_VALUE(cast(message as varchar), '$.data.from') end as view_from,
from_unixtime(cast(cast((JSON_VALUE(cast(message as varchar), '$.date')) as bigint)/1000 as bigint)+28800, 'yyyy-MM-dd') as `create_date`
from kafka_source
where JSON_VALUE(cast(message as varchar), '$.platform') = 'IOS'
and JSON_VALUE(cast(message as varchar), '$.typeid') = 'view'
and JSON_VALUE(cast(message as varchar), '$.data.otherUId') is not null
and JSON_VALUE(cast(message as varchar), '$.date') is not null
)
group by user_id,view_from,create_date