package com.shiwusuo.ReadHdfsToClickHouse.gauss

import java.sql.{Connection, DriverManager}
import java.util.Properties

import org.apache.spark.sql.DataFrame

import scala.collection.mutable.ArrayBuffer;

object GaussDBUtils {

  val urls = application.gaussURL //數(shù)據(jù)庫(kù)URL
  val username = application.gaussName //用戶名
  val password = application.gaussPass //密碼
  val database = application.gaussDatabase
  val gaussCreateDatabase = application.gaussCreateDatabase
  //val driver = "org.postgresql.Driver"
  val driver = application.driver
  var conn: Connection = null;

  def getConnection(): Connection = {
    try {
      Class.forName(driver)
      conn = DriverManager.getConnection(urls, username, password)
      conn
    } catch {
      case e: Throwable => e.printStackTrace()
        println("連接gaussDB失敗===請(qǐng)檢查原因")
        conn
    }
  }

  //創(chuàng)建表
  def createGaussTable(sql: String): Unit = {
    try {
      val connection = getConnection()
      val statement = connection.createStatement()
      statement.executeUpdate(sql)
      statement.close()
    } catch {
      case e: Exception =>
        e.printStackTrace()
    }

  }

  //判斷是否存在該表
  def tableExists(database: String, tablName: String): Boolean = {
    val connection = getConnection()
    val set = connection.getMetaData.getTables(null, database, tablName, null)
    if (set.next()) {
      return true
    } else {
      println("gaussDB中" + tablName + "不存在")
      return false
    }
  }

  //創(chuàng)建數(shù)據(jù)庫(kù)
  def createDatabase(): Unit = {
    val connection = getConnection()
    val statement = connection.createStatement()
    statement.executeUpdate(gaussCreateDatabase)
    statement.close()
  }

  //查詢 、執(zhí)行sql語句
  def ReadFromGaussDB(sql: String): Unit = {
    try {
      val connection = getConnection()
      val statement = connection.createStatement()
      val set = statement.executeQuery(sql)
      while (set.next()) {
        val str = set.getString(2)
        println(str)
      }
    }catch {
      case e: Exception =>
        e.printStackTrace()
    }
  }

  //獲取gaussDB中所有表名
  def getTablesByGaussDB(db: String) :ArrayBuffer[String]={
    val tablesList = new ArrayBuffer[String]()
    val connection = getConnection()
    val set = connection.getMetaData.getTables(null,db,null,null)
    while (set.next()) {
      val str: String = set.getString("TABLE_NAME")
      tablesList += str
      // println(str)
    }
    tablesList
  }

  //根據(jù)resourceId刪除gaussDB數(shù)據(jù)
  def deleteDataByResourceId(sql:String): Unit ={
      val connection = getConnection()
      val statement = connection.createStatement()
      statement.executeUpdate(sql)
  }



  //插入gaussDB庫(kù)中
  def insertgaussDBTable(table: String, df: DataFrame): Unit = {
    val connectionProperties = new Properties()
    connectionProperties.setProperty("user", username)
    connectionProperties.setProperty("password", password)
    println("+++++++開始寫入GaussDB++++++" + table + "共有" + df.count())
    df.write.mode("append").option("batchsize", "50000").option("isolationLevel", "NONE").
      option("numPartitions", "1").jdbc(urls, table, connectionProperties)
    println("=================GaussDB完成寫入========" + table + "====================")
  }


  def main(args: Array[String]): Unit = {
    //val bool = tableExists("gaussdb","test_01087")
    //println(bool)
    //val a = "CREATE TABLE gaussdb.CDPCP\n       (\n         code bigint NULL,\n        ident bigint NULL,\n        len bigint NULL,\n        lineNum1 bigint NULL,\n        lineNum2 bigint NULL,\n        lineNum3 bigint NULL,\n        lineNum4 bigint NULL,\n        lineName varchar(10000000) NULL,\n        begTime bigint NULL,\n        endTime bigint NULL,\n        comDur bigint NULL,\n        meanID varchar(10000000) NULL,\n        siteID varchar(10000000) NULL,\n        unitID varchar(10000000) NULL,\n        taskID varchar(10000000) NULL,\n        guid bigint NULL,\n        storTime bigint NULL,\n        mdSecDeg varchar(10000000) NULL,\n        fileSecDeg varchar(10000000) NULL,\n        secDegPro varchar(10000000) NULL,\n        ipVer bigint NULL,\n        srcAddr varchar(10000000) NULL,\n        dstAddr varchar(10000000) NULL,\n        srcPort bigint NULL,\n        dstPort bigint NULL,\n        protNum bigint NULL,\n        srcAddrV6 varchar(10000000) NULL,\n        dstAddrV6 varchar(10000000) NULL,\n        srcLoc varchar(10000000) NULL,\n        dstLoc varchar(10000000) NULL,\n        srcISP varchar(10000000) NULL,\n        dstISP varchar(10000000) NULL,\n        srcAS varchar(10000000) NULL,\n        dstAS varchar(10000000) NULL,\n        protInfo varchar(10000000) NULL,\n        linkInfo varchar(10000000) NULL,\n        protType varchar(10000000) NULL,\n        protName varchar(10000000) NULL,\n        mulRouFlag bigint NULL,\n        intFlag bigint NULL,\n        strDirec bigint NULL,\n        pktNum bigint NULL,\n        payLen bigint NULL,\n        hashId bigint NULL,\n        sessionID varchar(10000000) NULL,\n        resourceId varchar(10000000) NULL\n       )"
    //createGaussTable(a)
    //createtable
   // ReadFromGaussDB("select * from test_0110.user")
    //val strings = getTablesByGaussDB("test_0301")
    //strings.foreach(println)
    //DELETE FROM test_0110.AOE WHERE resourceId = 518824231216091205
    val b ="DELETE FROM test_0110.\"ANS\" WHERE \"resourceId\" = '518824231216091205'"
    println(b)
    deleteDataByResourceId(b)
  }
}