from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import pymysql
import smtplib
from email.mime.text import MIMEText
from email.utils import formataddr
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
import redis
import datetime
from pyspark import SparkConf
import time
from pyspark.sql import SparkSession
import json
import numpy as np
import pandas as pd
from pyspark.sql.functions import lit
from pyspark.sql.functions import concat_ws


def compute_henqiang(x):
    score = 15-x*((15-0.5)/180)
    if score>0.5:
        return score
    else:
        return 0.5
def compute_jiaoqiang(x):
    score = 12-x*(12/180)
    if score>0.5:
        return score
    else:
        return 0.5
def compute_ruoyixiang(x):
    score = 5-x*((5-0.5)/180)
    if score>0.5:
        return score
    else:
        return 0.5
def compute_validate(x):
    score = 10-x*((10-0.5)/180)
    if score>0.5:
        return score
    else:
        return 0.5
def tag_list2dict(lst, size):
    result = []
    if lst:
        for i in lst:
            tmp = dict()
            tmp["content"] = i["tag_id"]
            if isinstance(i['tag_id'], int):
                tmp["type"] = "tag"
            else:
                tmp["type"] = "search_word"
            tmp["score"] = i["tag_score"]
            result.append(tmp)
    return result[:size]


def get_device_user_id(cl_id):
    try:
        db_zhengxing = pymysql.connect(host="172.16.30.141", port=3306, user="work",
                                       password="BJQaT9VzDcuPBqkd",
                                       db="zhengxing", cursorclass=pymysql.cursors.DictCursor)
        cur_zhengxing = db_zhengxing.cursor()

        sql = """select user_id from statistic_device_user where device_id = (select id from statistic_device where device_id = "{}")""".format(cl_id)
        cur_zhengxing.execute(sql)
        cl_id = cur_zhengxing.fetchall()
        if cl_id:
            return cl_id[0]["user_id"]
        else:
            return ""
    except Exception as e:
        print(e)


def get_user_tag_score(cl_id, all_log_df, stat_date, size=10):
    try:
        db_jerry_test = pymysql.connect(host='172.16.40.158', port=4000, user='root', passwd='3SYz54LS9#^9sBvC',
                                        db='jerry_test', charset='utf8')
        cur_jerry_test = db_jerry_test.cursor()

        user_log_df = all_log_df.loc[all_log_df['cl_id'] == cl_id]
        if not user_log_df.empty:
            user_log_df["tag_id"] = np.where(user_log_df["action"] == "do_search",user_log_df["tag_referrer"],user_log_df["tag_id"])
            user_log_df["days_diff_now"] = round((int(time.time())-user_log_df["time"]) / (24*60*60))
            user_log_df["tag_score"] = user_log_df.apply(
                lambda x: compute_henqiang(x.days_diff_now) if x.score_type == "henqiang" else (
                    compute_jiaoqiang(x.days_diff_now) if x.score_type == "jiaoqiang" else (
                        compute_ruoyixiang(x.days_diff_now) if x.score_type == "ruoyixiang" else compute_validate(x.days_diff_now))), axis=1)
            finally_score = user_log_df.sort_values(by=["tag_score","time"],ascending=False)
            finally_score.drop_duplicates(subset="tag_id", inplace=True)
            finally_score_lst = finally_score[["tag_id","tag_score"]].to_dict('record')
            tag_id_list = tag_list2dict(finally_score_lst, size)

            user_id = get_device_user_id(cl_id)
            replace_sql = """replace into user_portrait_tags (stat_date, cl_id, user_id, tag_list) values("{stat_date}","{cl_id}",{user_id},"{tag_list}")"""\
                .format(stat_date=stat_date, cl_id=cl_id, user_id=user_id, tag_list=tag_id_list)
            cur_jerry_test.execute(replace_sql)
            db_jerry_test.commit()
            db_jerry_test.close()
            return "sucess"
    except Exception as e:
        return 'pass'


if __name__ == '__main__':
    try:
        db_jerry_test = pymysql.connect(host='172.16.40.158', port=4000, user='root', passwd='3SYz54LS9#^9sBvC',
                                        db='jerry_test', charset='utf8')
        cur_jerry_test = db_jerry_test.cursor()

        # 获取所有用户的设备id
        sql_device_ids = "select distinct cl_id from user_new_tag_log"
        cur_jerry_test.execute(sql_device_ids)
        device_ids_lst = [i[0] for i in cur_jerry_test.fetchall()]

        # 获取所有用户的行为日志
        sql_all_log = "select time,cl_id,score_type,tag_id,tag_referrer,action from user_new_tag_log"
        cur_jerry_test.execute(sql_all_log)
        all_log = cur_jerry_test.fetchall()
        db_jerry_test.close()
        all_log_df = pd.DataFrame(list(all_log))
        all_log_df.columns = ["time", "cl_id", "score_type","tag_id","tag_referrer","action"]
        stat_date = datetime.datetime.today().strftime('%Y-%m-%d')
        # rdd
        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.158: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()
        spark.sparkContext.setLogLevel("WARN")

        device_ids_lst_rdd = spark.sparkContext.parallelize(device_ids_lst)
        gm_kv_cli = redis.Redis(host="172.16.40.135", port=5379, db=6, socket_timeout=2000)
        result = device_ids_lst_rdd.repartition(100).map(lambda x: get_user_tag_score(x, all_log_df, stat_date))
        result.collect()

        # result_last = result_rename.withColumn("stat_date", lit(stat_date))
        # result_last.show()
        # df = result_last.select("stat_date", "cl_id", concat_ws(',', 'tag_list').alias("tag_list"))
        # df.show()
        # df.write.jdbc(
        #     mode="overwrite",
        #     url="jdbc:mysql://172.16.40.158:4000/jerry_test?user=root&password=3SYz54LS9#^9sBvC&useSSL=true",
        #     table="user_portrait_tags",
        #     properties={"driver": 'com.mysql.jdbc.Driver'})
    except Exception as e:
        print(e)