package mobvista.dmp.datasource.app_info_tag
import java.util.Properties

import com.alibaba.fastjson.{JSONArray, JSONObject}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{SparkSession, _}
import org.apache.spark.sql.types._
/**
 *
 * mysql -h adn-data-foronlinetest.c5yzcdreb1xr.us-east-1.rds.amazonaws.com -P 3306 -u adnro  -pYcM123glh
 *
 */

object Offer_adr {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    val spark = SparkSession.builder().appName("JdbcOperation").config(conf)
      .getOrCreate()
    val loadTime = spark.conf.get("spark.app.loadTime")
    var year = loadTime.substring(0, 4)
    var month = loadTime.substring(4, 6)
    val day = loadTime.substring(6, 8)
    val output_path = spark.conf.get("spark.app.output_path")
    try {
      /**
       * *
       * U>
       * 1:  第一步获得字典， 广播字典
       *
       *
       */
      val dim_category_rdd = spark.sparkContext
        .textFile("s3://mob-emr-test/dataplatform/DataWareHouse/data/dwh/dim_category/dim_category.txt")
      val dict_rdd = dim_category_rdd.map { line =>
        {
          val p = line.split("\t")
          val id = p(0)
          val store = p(1)
          val tag_code = p(5)
          var mv_first_tag_1 = ""
          try {
            mv_first_tag_1 = p(6)
          } catch {
            case e: Exception =>
              e.printStackTrace()
          } finally {
          }
          var mv_second_tag_1 = ""
          try {
            mv_second_tag_1 = p(7)
          } catch {
            case e: Exception =>
              e.printStackTrace()
          } finally {
          }
          var mv_first_tag_2 = ""
          try {
            mv_first_tag_2 = p(8)
          } catch {
            case e: Exception =>
              e.printStackTrace()
          } finally {
          }
          var mv_second_tag_2 = ""
          try {
            mv_second_tag_2 = p(9)
          } catch {
            case e: Exception =>
              e.printStackTrace()
          } finally {
          }

          val tag_array = new JSONArray()
          if (mv_second_tag_1.nonEmpty || mv_first_tag_1.nonEmpty) {
            var tag = ""
            if (mv_second_tag_1.nonEmpty) {
              tag = mv_second_tag_1
            } else if (mv_first_tag_1.nonEmpty) {
              tag = mv_first_tag_1
            }
            val tag_json = new JSONObject()
            tag_json.put("code", tag_code)
            tag_json.put("name", tag)
            tag_array.add(tag_json)
          }

          if (mv_second_tag_2.nonEmpty || mv_first_tag_2.nonEmpty) {
            var tag = ""
            if (mv_second_tag_2.nonEmpty) {
              tag = mv_second_tag_2
            } else if (mv_first_tag_2.nonEmpty) {
              tag = mv_first_tag_1
            }
            val tag_json = new JSONObject()
            tag_json.put("code", tag_code)
            tag_json.put("name", tag)
            tag_array.add(tag_json)
          }
          Row(
            tag_code,
            tag_array.toString())
        }
      }
      var dict_json = new JSONObject()
      dict_rdd.collect().foreach { x =>
        {
          val id = x.getString(0)
          val tag = x(1)
          // val tag = x.getJ
          /**
           * val tag_json = new JSONObject()
           * tag_json.put("code", tag_code)
           * tag_json.put("name", tag)
           * **
           */
          dict_json.put(id, tag)
        }
      }
      val dict_json_brocast = spark.sparkContext.broadcast(dict_json)

      /**
       *
       * 2: 第二步， 处理publisher数据
       *
       */

      val sqlContext = spark.sqlContext
      val properties = new Properties()
      properties.put("user", "adnro")
      properties.put("password", "YcM123glh")
      val url = "jdbc:mysql://adn-data-foronlinetest.c5yzcdreb1xr.us-east-1.rds.amazonaws.com:3306/mob_adn"
      //val campaign_list_DF = sqlContext.read.jdbc(url, "campaign_list", properties).select("id","app_name", "platform", "preview_url").filter("platform='1'")
      val campaign_list_DF = sqlContext.read.jdbc(url, "campaign_list", "id", 1, 200000000, 500, properties).select("id", "app_name", "platform", "preview_url").filter("platform='1' and status=1")
        .toDF("id", "app_name", "platform", "track_view_url").coalesce(500)
      // val campaign_list_cpi_DF = sqlContext.read.jdbc(url, "campaign_list_cpi", properties).select("campaign_id","city_code","device_type","device_id","sub_category_v2").filter("sub_category_v2!=''")
      val campaign_list_cpi_DF = sqlContext.read.jdbc(url, "campaign_list_cpi", "id", 1, 50000000, 500, properties).select("campaign_id", "city_code", "device_type", "device_id", "sub_category_v2").filter("sub_category_v2!=''")
        .toDF("campaign_id", "city_code", "device_type", "device_id", "sub_category_v2").coalesce(500)

      val table_DF = campaign_list_DF.join(campaign_list_cpi_DF, campaign_list_DF("id") === campaign_list_cpi_DF("campaign_id"), "inner")
        .select(campaign_list_DF.col("app_name"), campaign_list_DF.col("platform"), campaign_list_DF.col("track_view_url"), campaign_list_cpi_DF.col("sub_category_v2"))
        .toDF("app_name", "platform", "track_view_url", "sub_category_v2").repartition(500)
      val result_rdd = table_DF.rdd.map { p =>
        {
          val dict_json = dict_json_brocast.value
          val app_name = p.getString(0)
          var platform = ""
          try {
            platform = p.getInt(1).toString()
          } catch {
            case e: Exception =>
              try {
                platform = p.getString(1)
              } finally {
              }
              e.printStackTrace()
          } finally {
          }
          val confirmed_url = p.getString(2)
          val category_array = p.getString(3).split(",")
          var package_name = ""
          if (platform.equals("1")) {
            try {
              platform = "android"
              package_name = confirmed_url.substring(confirmed_url.indexOf("id=") + 3, confirmed_url.length())
              package_name = package_name.split("&")(0).split(";")(0)
            } catch {
              case e: Exception =>
                package_name = confirmed_url
                e.printStackTrace()
            } finally {
            }
          } else if (platform.equals("2")) {
            try {
              platform = "ios"
              package_name = confirmed_url.substring(0, confirmed_url.lastIndexOf("?"))
              package_name = package_name.substring(package_name.lastIndexOf("/") + 1, package_name.length())
            } catch {
              case e: Exception =>
                package_name = confirmed_url
                e.printStackTrace()
            } finally {
            }
          }
          val category_list = new JSONArray()
          for (a <- 0 to category_array.length - 1) {
            val catrgory_id = category_array(a)
            if (dict_json.get(catrgory_id) != null) {
              val json_array_data = dict_json.getJSONArray(catrgory_id)
              for (i <- 0 until json_array_data.size()) {
                category_list.add(json_array_data.getJSONObject(i))
              }
            }
          }

          Row(
            package_name,
            app_name,
            category_list.toString(),
            "",
            "",
            confirmed_url,
            "")
        }
      }
      val schema = StructType(Array(
        StructField("package_name", StringType, true),
        StructField("app_name", StringType, true),
        StructField("category_list", StringType, true),
        StructField("user_rating_count", StringType, true),
        StructField("download", StringType, true),
        StructField("track_view_url", StringType, true),
        StructField("update_time", StringType, true)))

      var df = spark.createDataFrame(result_rdd, schema)
      df.coalesce(20).write.format("orc").mode(SaveMode.Overwrite).save(output_path)
    } catch {
      case e: Exception =>
        e.printStackTrace()
    } finally {
    }
    
    spark.sparkContext.stop()
  }

}