package com.aliyun.odps.spark.sparksql

import config.MyConfigSession
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.sql.expressions.{Window, WindowSpec}
import org.apache.spark.sql.functions.lag
import org.apache.spark.storage.StorageLevel
import utils.{DateUtils, StringUtils, UseUtil}

import scala.collection.mutable.ListBuffer

/**
 *
 * 处理埋点流量表,导入到DW层pica_dw.dw_fact_log_session
 *
 * @Author weicheng.mao
 * @Date 2022/02/23 14:23
 * @Version 1.0
 */

object SessionProcess {
  def apply(): SessionProcess = new SessionProcess()

  def main(args: Array[String]): Unit = {

    var scnData: String = DateUtils.getYesterdayDate
    if(args.length>=1){
      scnData = args(0)
    }
    println(s"scnData=${scnData}")
    //设置任务开始时间,格式是2019-09-12 14:03:30
    val startTime: String = DateUtils.getTodayTime
    //存储SQL中的参数
    val insertArr: Array[String] = Array[String](scnData, startTime)

    try {
      val sessionProcess: SessionProcess = SessionProcess()
      //step1:获取源数据,重新分区,产生shuffle,Spark读Hive默认的分区数太少,并对数据去重
      var sourceDF: DataFrame = sessionProcess.sparkSession.sql(MyConfigSession.SOURCE_SQL+s" and created_day='${scnData}'").repartition(200).distinct()
      sourceDF.show()
      var conditionGroup = List( "<='2'","between '3' and '5'","between '6' and '8'","between '9' and 'b'",">='c'" )
      var dataCount = 0
      var index = 0
      sourceDF.persist(StorageLevel.MEMORY_AND_DISK_SER)
      for(condition <- conditionGroup){
        index += 1
        val slideDF = sourceDF.where(s" SUBSTRING(pseudo_session,1,1)  ${condition}").repartition(100)
        //step2:抽取出当天pseudo_session对应的非空的device_token,doctor_id,mobile,补充到对应的pseudo_session下这几项为空的记录中
        val groupRdd = slideDF.rdd.groupBy(r => r.getAs[String]("pseudo_session"))
        val resRdd = groupRdd.flatMap(g => {
          val pseudo_session = g._1
          val resList: ListBuffer[Row] = new ListBuffer[Row]()
          var rowList = g._2
          rowList = rowList.toList.sortWith((x,y)=>x.getAs[String]("created") > y.getAs[String]("created"))//按created由大到小排序
          var thisDeviceToken = ""
          var thisDoctorId = "0"
          var thisMobile = ""

          rowList.foreach(row => {
            var deviceToken = row.getAs[String]("device_token")
            var doctorId = row.getAs[String]("doctor_id")
            var mobile = row.getAs[String]("mobile")
            val created = row.getAs[String]("created")
            if(deviceToken!=null && !deviceToken.equals("") ){
              thisDeviceToken = deviceToken
            }else {
              deviceToken = thisDeviceToken
            }
            if(doctorId!=null && !doctorId.equals("") && !doctorId.equals("0") ){
              thisDoctorId = doctorId
            }else {
              doctorId = thisDoctorId
            }
            if(mobile!=null && !mobile.equals("") ){
              thisMobile = mobile
            }else {
              mobile = thisMobile
            }
            resList += (Row(row.getAs[String]("pseudo_session"),
              doctorId,
              mobile,
              deviceToken,
              row.getAs[String]("user_token_tourist"),
              row.getAs[String]("class_name"),
              row.getAs[String]("view_path"),
              row.getAs[String]("action"),
              row.getAs[String]("component_tag"),
              row.getAs[String]("app_version"),
              row.getAs[String]("device_type"),
              row.getAs[String]("device_brand"),
              row.getAs[String]("device_model"),
              row.getAs[String]("network_type"),
              row.getAs[String]("created")))
          })
          resList.iterator
        })

        import sessionProcess.sparkSession.implicits._
        //step3:根据映射表来进行action_type和class_name数据过滤
        val resDF = sessionProcess.sparkSession.createDataFrame(resRdd,sourceDF.schema)
        resDF.persist(StorageLevel.MEMORY_AND_DISK_SER)
        println("resDF.show=======>")
        //            resDF.show()
        val data: RDD[Row] = resDF.rdd.mapPartitions(sessionProcess.filterRows)
        println("---------------------------------------process columns-------------------------------------------")

        val baseDF: DataFrame = data.mapPartitions(sessionProcess.processColumns)
          .toDF("pseudo_session", "user_id", "mobile", "device_token", "user_token", "view_class", "view_path",
            "action_type", "component_tag", "menu_code", "action_code", "position", "label_value","label_class","module_class1","module_class2", "app_version",
            "device_type", "device_brand", "device_model", "device_system", "net_type", "created_time", "date_time")
        println("----------------------------------compute session id---------------------------------------------")
        val sessionIdDF: DataFrame =  sessionProcess.getSessionId(baseDF,sessionProcess)
        //默认缓存级别是:MEMORY_AND_DISK
        sessionIdDF.persist(StorageLevel.MEMORY_AND_DISK_SER)
        println("sessionIdDF.show=======>")
        println("-------------------------------match user_id 逻辑-------------------------------------------------")
        val dwFactLogSession: DataFrame = sessionProcess.matchUserId(sessionIdDF,sessionProcess.sparkSession,scnData)
        println("dwFactLogSession.show=======>")
        println("-----------------create view fact_log_session and load to dw_fact_log_session--------------------")
        dwFactLogSession.createOrReplaceTempView("fact_log_session")
        var insertMode = "insert overwrite"
        if(index!=1){
          insertMode = "insert into"
        }
        val loadDataSql =
          s"${insertMode} table ${MyConfigSession.HIVE_TABLE1} partition(created_day='${scnData}') select * from fact_log_session distribute by rand()"
        sessionProcess.sparkSession.sql(loadDataSql)
        dataCount = dataCount + dwFactLogSession.count().toInt
      }

      println("----------------------------------update task record table---------------------------------------")

      sessionProcess.sparkSession.stop()
    } catch {
      case e: Exception => {
        println(s"-----------------------------------任务异常---------------------------------------------------")
        e.printStackTrace()

      }
    }
  }



}

class SessionProcess extends java.io.Serializable{
  def getSparkSession(appName: String): SparkSession = {
//    val sparkSession = SparkSession
//      .builder()
//      .appName("SparkPi")
//      .config("spark.master", "local[1]") // 需设置spark.master为local[N]才能直接运行，N为并发数。
//      .config("spark.hadoop.odps.project.name", "pica")
//      .config("spark.hadoop.odps.access.id", "LTAI5tQEnb9VREcPdDAR32bg")
//      .config("spark.hadoop.odps.access.key", "RYazt1JIYZnRcEdlntarZ1Ov21zoNi")
//      .config("spark.hadoop.odps.end.point", "http://service.cn.maxcompute.aliyun.com/api")
//      .config("spark.sql.catalogImplementation", "odps")
//      .getOrCreate()

        val sparkSession = SparkSession
          .builder()
          .config("spark.hadoop.odps.project.name", "pica")
          .appName("SessionProcess")
          .getOrCreate()

    sparkSession
  }

  val sparkSession: SparkSession = getSparkSession("SessionProcess")

  //获取符合要求的actionType广播变量
  val actionTypeBroad =
    UseUtil.getBroadcast(sparkSession, MyConfigSession.ACTION_TYPE_SQL, "action_type", "is_valid")
  //获取clasName广播变量
  val classNameBroad =
    UseUtil.getBroadcast(sparkSession, MyConfigSession.CLASS_NAME_SQL, "class_name", "is_valid")
  //获取menu_code广播变量
  val menuCodeBroad =
    UseUtil.getBroadcast(sparkSession, MyConfigSession.MENU_CODE_SQL, "view_path", "menu_code")
  //获取actionCategory变量
  val actionCategory =
    UseUtil.getBroadcast(sparkSession,MyConfigSession.ACTION_CATEGORY_SQL,"action_type","action_category")


  //定义函数式变量,过滤映射表数据
  val filterRows: Iterator[Row] => Iterator[Row] = (rows: Iterator[Row]) => {
    val rowList: ListBuffer[Row] = new ListBuffer[Row]()
    val actionTypeMap: Map[String, String] = actionTypeBroad.value
    val classNameMap: Map[String, String] = classNameBroad.value
    //关联到action_category的映射表广播变量
    val actionCategoryMap: Map[String, String] = actionCategory.value
    rows.toList.foreach(row => {
      //筛选action的条件
      val action: String = StringUtils.getNotNullString(row.getAs[String]("action"))
      //说明该action类型即为所要的
      if (actionTypeMap.getOrElse(action, "-1").equals("1")) {
        //将action转化为映射表中对应的标准actionCategory
        val action_type: String = actionCategoryMap.getOrElse(action,"ACTION")
        //                val action_type = UseUtil.getActionType(action)
        //action为其中的任何一个
        if (action_type.equals("ACTION_CLICK") || action_type.equals("ACTION_EXPOSE")) {
          //判断 component_tag 必须要包含 "#"
          if (row.getAs[String]("component_tag") != null
            && row.getAs[String]("component_tag").contains("#")) {
            rowList += row
          }
        } else if (action_type.equals("ACTION_VIEW")) {
          rowList += row
          //非上述三种action_type,那么需要过滤掉映射表中class_name为"0"对应的那些数据
        } else if (row.getAs[String]("class_name") != null
          && !classNameMap.getOrElse(row.getAs[String]("class_name"), "-1").equals("0")) {
          rowList += row
        }
      }
    })
    rowList.iterator
  }

  //处理字段,得到需要的字段值
  val processColumns = (rows: Iterator[Row]) => {
    //        val baseList = new ListBuffer[(String, String, String, String, String, String, String, String, String,
    //                String, String, String, String, String, String, String, String, String, String, String, String,String,String,String)]()
    val baseList = new ListBuffer[SessionLog]()
    //关联到menu_code的映射表广播变量
    val path_menu: Map[String, String] = menuCodeBroad.value
    //关联到action_category的映射表广播变量
    val actionCategoryMap: Map[String, String] = actionCategory.value
    rows.toList.foreach(row => {
      //1.获取网络类型
      //2G,3G,4G,2G/3G/4G,WIFI,WLAN,或者为空字符串
      val net_type = UseUtil.netTypeMatch(StringUtils.getNotNullString(row.getAs[String]("network_type")))

      //2.修改action类型
      var action_type: String = ""
      if (row.getAs[String]("action") !=null) {
        action_type = actionCategoryMap.getOrElse(row.getAs[String]("action"),"ACTION")
      }
      //3.拆分 component_tag字段
      val component_tag: String = StringUtils.getNotNullString(row.getAs[String]("component_tag"))
      var menu_code: String = ""
      var action_code: String = ""
      var position: String = ""
      var label_value: String = ""
      var label_class:String = ""
      var module_class1:String=""
      var module_class2:String=""
      //将符合要求的component_tag进行切割,获取 aciton_code,label_value
      if (component_tag.contains("#")) {
        //按照#号切割
        val strs: Array[String] = component_tag.split("#")
        strs.length match {
          case 1 => {
            menu_code = strs(0)
          }
          case 2 => {
            menu_code = strs(0)
            action_code = strs(1)
          }
          case 3 => {
            menu_code = strs(0)
            action_code = strs(1)
            position = strs(2)
          }
          case 4 => {
            menu_code = strs(0)
            action_code = strs(1)
            position = strs(2)
            label_value = strs(3).substring(0,math.min(250,strs(3).length))
          }
          case 5 => {
            menu_code = strs(0)
            action_code = strs(1)
            position = strs(2)
            label_value = strs(3).substring(0,math.min(250,strs(3).length))
            label_class = strs(4)
          }
          case 6 => {
            menu_code = strs(0)
            action_code = strs(1)
            position = strs(2)
            label_value = strs(3).substring(0,math.min(250,strs(3).length))
            label_class = strs(4)
            module_class1=strs(5)
          }
          case _ => {
            menu_code = strs(0)
            action_code = strs(1)
            position = strs(2)
            label_value = strs(3).substring(0,math.min(250,strs(3).length))
            label_class = strs(4)
            module_class1=strs(5)
            module_class2=strs(6)
          }
        }
      }
      //匹配menu_code:如果上述截取出来的menu_code为(''||null||0||length(menu_code)>3 ) and action is ACTION_VIEW
      if ((menu_code.equals("")|| menu_code.equals("null") || menu_code.equals("0") || (menu_code.length> 3 && !component_tag.contains("#") ))
        && action_type.equals("ACTION_VIEW")) {
        menu_code = "0" //关联不上的显示为0
        import scala.util.control.Breaks._
        breakable {
          //利用menu_code映射表匹配
          for (tuple <- path_menu) {
            //源数据view_path的字符串包含映射表view_path的字符串
            if (StringUtils.getNotNullString(row.getAs[String]("view_path")).contains(tuple._1)) {
              //满足条件后,修改源数据的menu_code
              menu_code = tuple._2
              println("--------------------menu_code match successfully-----------------------")
              //结束遍历
              break()
            }
          }
          //经过上述匹配,如果menu_code仍然为空串,那么置为component_tag字段一样
          if (menu_code.equals("")) {
            menu_code = component_tag
          }
        }
      }
      //一行数据添加到List中
      baseList += SessionLog(StringUtils.getNotNullString(row.getAs[String]("pseudo_session")),
        StringUtils.getNotNullString(row.getAs[String]("doctor_id")),
        StringUtils.getNotNullString(row.getAs[String]("mobile")),
        StringUtils.getNotNullString(row.getAs[String]("device_token")),
        StringUtils.getNotNullString(row.getAs[String]("user_token_tourist")),
        StringUtils.getNotNullString(row.getAs[String]("class_name")),
        StringUtils.getNotNullString(row.getAs[String]("view_path")),
        action_type, component_tag, menu_code, action_code, position, label_value,label_class,module_class1,module_class2,
        StringUtils.getNotNullString(row.getAs[String]("app_version")),
        StringUtils.getNotNullString(row.getAs[String]("device_type")),
        StringUtils.getNotNullString(row.getAs[String]("device_brand")),
        StringUtils.getNotNullString(row.getAs[String]("device_model")),
        "", net_type,
        StringUtils.getNotNullString(row.getAs[String]("created")),
        DateUtils.milliSecondsFormatTime(StringUtils.getNotNullString(row.getAs[String]("created"))))
    })
    baseList.iterator
  }
  //按照time_gap 切割session,计算session_id
  val computeSessionId = (tuple: (String, Iterable[Row])) => {
    //先按照 pseudo_session 的值命名 sessionID
    val sessionID: String = tuple._1
    val rowList: Iterable[Row] = tuple._2
    //定义一个累加量
    var count: Int = 0
    //存储一行的数据
    //        val list = new ListBuffer[(String, String, String, String, String, String, String, String, String,
    //                String, String, String, String, String, String, String, String, String, String, String, String, String,String, String)]()
    val list = new ListBuffer[SessionLog]()
    rowList.toList.foreach(row => {
      val created: String = StringUtils.getNotNullString(row.getAs[String]("created_time"))
      val refer_created: String = StringUtils.getNotNullString(row.getAs[String]("refer_created"))
      val time_diff: Long = created.toLong - refer_created.toLong
      //相邻的时间差小于等于30分钟,就是同一个 sessionID
      if (time_diff > MyConfigSession.SESSION_GAP) {
        count = count + 1
      }
      //添加到List
      list +=  SessionLog(sessionID + count,
        StringUtils.getNotNullString(row.getAs[String]("user_id")),
        StringUtils.getNotNullString(row.getAs[String]("mobile")),
        StringUtils.getNotNullString(row.getAs[String]("device_token")),
        StringUtils.getNotNullString(row.getAs[String]("user_token")),
        StringUtils.getNotNullString(row.getAs[String]("view_class")),
        StringUtils.getNotNullString(row.getAs[String]("view_path")),
        StringUtils.getNotNullString(row.getAs[String]("action_type")),
        StringUtils.getNotNullString(row.getAs[String]("component_tag")),
        StringUtils.getNotNullString(row.getAs[String]("menu_code")),
        StringUtils.getNotNullString(row.getAs[String]("action_code")),
        StringUtils.getNotNullString(row.getAs[String]("position")),
        StringUtils.getNotNullString(row.getAs[String]("label_value")),
        StringUtils.getNotNullString(row.getAs[String]("label_class")),
        StringUtils.getNotNullString(row.getAs[String]("module_class1")),
        StringUtils.getNotNullString(row.getAs[String]("module_class2")),
        StringUtils.getNotNullString(row.getAs[String]("app_version")),
        StringUtils.getNotNullString(row.getAs[String]("device_type")),
        StringUtils.getNotNullString(row.getAs[String]("device_brand")),
        StringUtils.getNotNullString(row.getAs[String]("device_model")),
        StringUtils.getNotNullString(row.getAs[String]("device_system")),
        StringUtils.getNotNullString(row.getAs[String]("net_type")),
        created, StringUtils.getNotNullString(row.getAs[String]("date_time")))
    })
    list
  }
  /**
   * @Description     按照TimeGap切割session,重命名session_id
   * @param dataFrame 要处理的DataFrame
   * @param sessionProcess   SessionProcess对象,包含SparkSession 环境
   * @return org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>
   **/
  def getSessionId(dataFrame: DataFrame,sessionProcess: SessionProcess):DataFrame = {
    import sessionProcess.sparkSession.implicits._

    //先按照 pseudo_session 分组,然后按照 created 排序,组件一个窗口
    val pSessionWinSpec: WindowSpec = Window.partitionBy("pseudo_session").orderBy("created_time")
    //增加一个字段 refer_created ,这个字段的值是上一条记录 created 字段的值,方便后面通过两者差值计算出 session_id
    val rcreDF: DataFrame =
      dataFrame.withColumn("refer_created", lag(dataFrame("created_time"), 1).over(pSessionWinSpec))

    //执行COALESCE,目的是为了去掉 refer_created 为Null的值
    val coalesceDF: DataFrame = rcreDF.selectExpr(
      "pseudo_session", "user_id", "mobile", "device_token",
      "user_token", "view_class", "view_path", "action_type",
      "component_tag", "menu_code", "action_code", "position", "label_value","label_class","module_class1","module_class2",
      "app_version", "device_type", "device_brand", "device_model",
      "device_system", "net_type", "created_time", "date_time",
      "COALESCE(refer_created,created_time) as refer_created")

    //按照 pseudo_session聚合,计算 session_id
    val groupRDD: RDD[(String, Iterable[Row])] =
      coalesceDF.rdd.groupBy(row => row.getAs[String]("pseudo_session"))

    //计算两者之差,这时候就得到了 session_id
    val sessionIdDF: DataFrame = groupRDD.map(sessionProcess.computeSessionId).flatMap(it => it)
      .toDF("session_id", "user_id", "mobile", "device_token", "user_token",
        "view_class", "view_path", "action_type", "component_tag",
        "menu_code", "action_code", "position", "label_value","label_class","module_class1","module_class2",
        "app_version", "device_type", "device_brand", "device_model",
        "device_system", "net_type", "created_time", "date_time")
    sessionIdDF
  }


  /**
   * @Description         匹配user_id,补全数据中的user_id字段
   * @param dataFrame     筛选后的数据
   * @param sparkSQLSession  SparkSession 环境
   * @param created_day   当前数据的日期,格式 "2020-03-01"
   * @return org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>
   **/
  def matchUserId(dataFrame: DataFrame,sparkSQLSession: SparkSession,created_day:String):DataFrame={
    //追加:将dataFrame与pica_ds.pica_doctor根据user_id进行匹配,匹配不上的user_id置为'0'
    println("matchUserId开始执行-----------------------------------")
    dataFrame.createOrReplaceTempView(MyConfigSession.VIEW_SESSION_ODS)
    val DF = sparkSQLSession.sql(MyConfigSession.INIT_USER_ID_SQL)
    //以下的所有逻辑是为了补全user_id字段
    //第一步：首先筛选出不符合的use_id数据,将这些user_id置为字符串'0'
    val noMatchUserIdDF: Dataset[Row] = DF.where("user_id ='' OR user_id = '0' OR LENGTH(user_id) = 24")
      .selectExpr("session_id","'0' as user_id", "mobile", "device_token", "user_token",
        "view_class", "view_path", "action_type", "component_tag",
        "menu_code", "action_code", "position", "label_value","label_class","module_class1","module_class2",
        "app_version", "device_type", "device_brand", "device_model",
        "device_system", "net_type", "created_time", "date_time")
    noMatchUserIdDF.createOrReplaceTempView(MyConfigSession.VIEW_SESSION_NO_MATCH)

    //1.筛选出上一步没有匹配到的user_id,先按照mobile_phone进行匹配
    val mobilePhoneDF: DataFrame = sparkSQLSession.sql(MyConfigSession.MOBILE_PHONE_SQL)
    mobilePhoneDF.createOrReplaceTempView(MyConfigSession.VIEW_MOBILE_PHONE)
    //2.使用临时表equiment,筛选出为1的那条最新数据
    var equipmentInfoSql =  MyConfigSession.EQUIPMENT_INFO_SQL
    if(!created_day.equals(DateUtils.getYesterdayDate)){//如果不是跑昨天的数据,使用equipment拉链表
      equipmentInfoSql =  MyConfigSession.EQUIPMENT_INFO_SQL_ARGS+ s"'${created_day}'"
    }
    println(s"equipmentInfoSql=${equipmentInfoSql}")
    val equipmentDF: DataFrame = sparkSQLSession.sql(equipmentInfoSql).where("row_d =1")
    equipmentDF.createOrReplaceTempView(MyConfigSession.VIEW_EQUIPMENT_INFO)

    //3.将第2步筛选出来的数据按照device_token进行匹配,获得user_id
    val deviceTokenDF: DataFrame = sparkSQLSession.sql(MyConfigSession.DEVICE_TOKEN_SQL)

    //4.将上述三者union,最终导入表中的数据
    val rightUserId: Dataset[Row] = DF.where("user_id !='' and user_id != '0' and LENGTH(user_id) !=24")
    val mobilePhoneResDF: Dataset[Row] = mobilePhoneDF.where("user_id !='0'")
    val dwFactLogSession: Dataset[Row] = rightUserId.union(mobilePhoneResDF).union(deviceTokenDF)
    dwFactLogSession.createOrReplaceTempView(MyConfigSession.VIEW_DEVICE_TOKEN)
    //根据pica_doctor补充user_id_int字段（字段类型转换成int型）, 限制 delete_flag = 1 and creat_time截止昨日创建,未关联上显示为0
    val USER_ID_INT_SQL:String=
      s"""
         |SELECT   concat(regexp_replace( '${created_day}',"-","") ,cast(row_number() over(partition by 1 order by created_time) as string)) as id,
         |ss.session_id, ss.user_id,COALESCE(b.id,0) user_id_int, ss.mobile, ss.device_token, ss.user_token,
         |ss.view_class,ss.view_path,ss.action_type,ss.component_tag, ss.menu_code,
         |ss.action_code, ss.position,ss.label_value,ss.label_class,ss.module_class1,ss.module_class2,ss.app_version, ss.device_type,
         |ss.device_brand, ss.device_model,ss.device_system,ss.net_type,ss.created_time,
         |ss.date_time from ${MyConfigSession.VIEW_DEVICE_TOKEN}  AS ss
         |left join (select  id,cast(id as string) id_str  from pica.ods_pica_p_doctor a where a.delete_flag = 1 and to_date(a.creat_time) <= '${created_day}') AS b on ss.user_id = b.id_str
         |""".stripMargin
    val userIdDF: DataFrame = sparkSQLSession.sql(USER_ID_INT_SQL)
    userIdDF
  }
}
case class SessionLog(pseudo_session: String,
                      user_id: String,
                      mobile: String,
                      device_token: String,
                      user_token: String,
                      view_class: String,
                      view_path: String,
                      action_type: String,
                      component_tag: String,
                      menu_code: String,
                      action_code: String,
                      position: String,
                      label_value: String,
                      label_class: String,
                      module_class1: String,
                      module_class2: String,
                      app_version: String,
                      device_type: String,
                      device_brand: String,
                      device_model: String,
                      device_system: String,
                      net_type: String,
                      created_time: String,
                      date_time: String)


