import pymysql
from pyspark.conf import SparkConf
import pytispark.pytispark as pti
from pyspark.sql import SparkSession
import datetime
import pandas as pd
import time
from pyspark import StorageLevel


def all_click(x):
    total = []
    sum = 0
    date = (datetime.date.today() - datetime.timedelta(days=x)).strftime("%Y%m%d")
    total.append(date)

    print("美购首页美购列表卡片点击")
    tmp = spark.sql("select count(*) from online.bl_hdfs_maidian_updates "
                    "where partition_date='{}' "
                    "and action = 'goto_welfare_detail' and params['from'] = 'welfare_home_list_item'"
                    .format(date)).rdd.map(lambda x: x[0]).collect()[0]
    total.append(tmp)
    sum = sum + tmp

    print("美券相关的美购列表页美购卡片点击")
    tmp = spark.sql("select count(*) from online.bl_hdfs_maidian_updates "
                    "where partition_date='{}' "
                    "and action = 'goto_welfare_detail' and params['from'] = 'coupon_welfare_list'"
                    .format(date)).rdd.map(lambda x: x[0]).collect()[0]
    total.append(tmp)
    sum = sum + tmp

    print("新美购首页-固定ICON美购卡片点击")
    tmp = spark.sql("select count(*) from online.bl_hdfs_maidian_updates "
                    "where partition_date='{}' "
                    "and action = 'goto_welfare_detail' and params['from'] = 'welfare_list'"
                    .format(date)).rdd.map(lambda x: x[0]).collect()[0]
    total.append(tmp)
    sum = sum + tmp

    print("首页-品类模块点击跳转到品类聚合美购卡片点击")
    tmp = spark.sql("select count(*) from online.bl_hdfs_maidian_updates "
                    "where partition_date='{}' "
                    "and action = 'goto_welfare_detail' and params['from'] = 'category'"
                    .format(date)).rdd.map(lambda x: x[0]).collect()[0]
    total.append(tmp)
    sum = sum + tmp

    print("serach")
    tmp = spark.sql("select count(*) from online.bl_hdfs_maidian_updates "
                    "where partition_date='{}' "
                    "and action = 'search_result_welfare_click_item'"
                    .format(date)).rdd.map(lambda x: x[0]).collect()[0]
    total.append(tmp)
    sum = sum + tmp
    total.append(sum)

    return sum


def cpc_click(x):
    cpc = []
    date = (datetime.date.today() - datetime.timedelta(days=x)).strftime("%Y%m%d")
    cpc.append(date)
    sum = 0

    print("美购首页美购列表卡片点击")
    tmp = spark.sql("select count(*) from online.bl_hdfs_maidian_updates "
                    "where partition_date='{}' "
                    "and action = 'goto_welfare_detail' and params['from'] = 'welfare_home_list_item' "
                    "and params['is_cpc'] = '1'"
                    .format(date)).rdd.map(lambda x: x[0]).collect()[0]
    cpc.append(tmp)
    sum = sum + tmp

    print("美券相关的美购列表页美购卡片点击")
    tmp = spark.sql("select count(*) from online.bl_hdfs_maidian_updates "
                    "where partition_date='{}' "
                    "and action = 'goto_welfare_detail' and params['from'] = 'coupon_welfare_list' "
                    "and params['is_cpc'] = '1'"
                    .format(date)).rdd.map(lambda x: x[0]).collect()[0]
    cpc.append(tmp)
    sum = sum + tmp

    print("新美购首页-固定ICON美购卡片点击")
    tmp = spark.sql("select count(*) from online.bl_hdfs_maidian_updates "
                    "where partition_date='{}' "
                    "and action = 'goto_welfare_detail' and params['from'] = 'welfare_list' "
                    "and params['is_cpc'] = '1'"
                    .format(date)).rdd.map(lambda x: x[0]).collect()[0]
    cpc.append(tmp)
    sum = sum + tmp

    print("首页-品类模块点击跳转到品类聚合美购卡片点击")
    tmp = spark.sql("select count(*) from online.bl_hdfs_maidian_updates "
                    "where partition_date='{}' "
                    "and action = 'goto_welfare_detail' and params['from'] = 'category' "
                    "and params['is_cpc'] = '1'"
                    .format(date)).rdd.map(lambda x: x[0]).collect()[0]
    cpc.append(tmp)
    sum = sum + tmp

    print("serach")
    tmp = spark.sql("select count(*) from online.bl_hdfs_maidian_updates "
                    "where partition_date='{}' "
                    "and action = 'search_result_welfare_click_item' "
                    "and params['is_cpc'] = '1'"
                    .format(date)).rdd.map(lambda x: x[0]).collect()[0]
    cpc.append(tmp)
    sum = sum + tmp

    cpc.append(sum)

    return sum



if __name__ == '__main__':
    sparkConf = SparkConf().set("spark.hive.mapred.supports.subdirectories", "true") \
        .set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive", "true") \
        .set("spark.tispark.plan.allow_index_double_read", "false") \
        .set("spark.tispark.plan.allow_index_read", "true") \
        .set("spark.sql.extensions", "org.apache.spark.sql.TiExtensions") \
        .set("spark.tispark.pd.addresses", "172.16.40.170:2379").set("spark.io.compression.codec", "lzf") \
        .set("spark.driver.maxResultSize", "8g").set("spark.sql.avro.compression.codec", "snappy")

    spark = SparkSession.builder.config(conf=sparkConf).enableHiveSupport().getOrCreate()
    clicks = []
    cpcs = []
    dates = []
    for i in range(1,3):
        clicks.append(all_click(i))
        cpcs.append(cpc_click(i))
        date_str = (datetime.date.today() - datetime.timedelta(days=i)).strftime("%Y%m%d")
        dates.append(date_str)
    print("clicks")
    print(clicks)

    print("cpcs")
    print(cpcs)
    df = pd.DataFrame({'date': dates,
                   'clicks': clicks,"cpc":cpcs})
    df.to_csv('/home/gmuser/cpc.csv', index=False)
    # rdd = spark.sparkContext.parallelize(cpcs)
    # df = spark.createDataFrame(rdd).toDF.toPandas()
    # df.to_csv('/home/gmuser/cpc.csv',index=False)
    #
    # rdd = spark.sparkContext.parallelize(clicks)
    # df = spark.createDataFrame(rdd).toDF.toPandas()
    # df.to_csv('/home/gmuser/clicks.csv', index=False)

    spark.stop()







