# -*- coding:UTF-8 -*-
# @Time  : 2020/9/17 16:26
# @File  : output_article_distribution_0917.py
# @email : litao@igengmei.com
# @author : litao


# -*- coding:UTF-8 -*-
# @Time  : 2020/9/11 10:59
# @File  : portary_article_distribution.py
# @email : litao@igengmei.com
# @author : litao


import hashlib
import json

import pymysql
import xlwt, datetime
import redis
from meta_base_code.utils.func_from_redis_get_portrait import user_portrait_scan_info,get_user_portrait_tag3_from_redis
from meta_base_code.utils.func_from_es_get_article import get_es_article_num,get_user_post_from_mysql
# from pyhive import hive
from maintenance.func_send_email_with_file import send_file_email
from typing import Dict, List
from elasticsearch_7 import Elasticsearch
from elasticsearch_7.helpers import scan
import sys
import time
from pyspark import SparkConf
from pyspark.sql import SparkSession, DataFrame


# from pyspark.sql.functions import lit
# import pytispark.pytispark as pti
startTime = time.time()
sparkConf = SparkConf()
sparkConf.set("spark.sql.crossJoin.enabled", True)
sparkConf.set("spark.debug.maxToStringFields", "100")
sparkConf.set("spark.tispark.plan.allow_index_double_read", False)
sparkConf.set("spark.tispark.plan.allow_index_read", True)
sparkConf.set("spark.hive.mapred.supports.subdirectories", True)
sparkConf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive", True)
sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
sparkConf.set("mapreduce.output.fileoutputformat.compress", False)
sparkConf.set("mapreduce.map.output.compress", False)
sparkConf.set("prod.gold.jdbcuri",
              "jdbc:mysql://172.16.30.136/doris_prod?user=doris&password=o5gbA27hXHHm&rewriteBatchedStatements=true")
sparkConf.set("prod.mimas.jdbcuri",
              "jdbc:mysql://172.16.30.138/mimas_prod?user=mimas&password=GJL3UJe1Ck9ggL6aKnZCq4cRvM&rewriteBatchedStatements=true")
sparkConf.set("prod.gaia.jdbcuri",
              "jdbc:mysql://172.16.30.143/zhengxing?user=work&password=BJQaT9VzDcuPBqkd&rewriteBatchedStatements=true")
sparkConf.set("prod.tidb.jdbcuri",
              "jdbc:mysql://172.16.40.158:4000/eagle?user=st_user&password=aqpuBLYzEV7tML5RPsN1pntUzFy&rewriteBatchedStatements=true")
sparkConf.set("prod.jerry.jdbcuri",
              "jdbc:mysql://172.16.40.158:4000/jerry_prod?user=st_user&password=aqpuBLYzEV7tML5RPsN1pntUzFy&rewriteBatchedStatements=true")
sparkConf.set("prod.tispark.pd.addresses", "172.16.40.158:2379")
sparkConf.set("prod.tispark.pd.addresses", "172.16.40.170:4000")
sparkConf.set("prod.tidb.database", "jerry_prod")
sparkConf.setAppName("test")

spark = (SparkSession.builder.config(conf=sparkConf).config("spark.sql.extensions", "org.apache.spark.sql.TiExtensions")
         .config("spark.tispark.pd.addresses", "172.16.40.170:2379").enableHiveSupport().getOrCreate())

spark.sql("ADD JAR hdfs:///user/hive/share/lib/udf/brickhouse-0.7.1-SNAPSHOT.jar")
spark.sql("ADD JAR hdfs:///user/hive/share/lib/udf/hive-udf-1.0-SNAPSHOT.jar")
spark.sql("CREATE TEMPORARY FUNCTION json_map AS 'brickhouse.udf.json.JsonMapUDF'")
spark.sql("CREATE TEMPORARY FUNCTION is_json AS 'com.gmei.hive.common.udf.UDFJsonFormatCheck'")
spark.sql("CREATE TEMPORARY FUNCTION arrayMerge AS 'com.gmei.hive.common.udf.UDFArryMerge'")

task_list = []
task_days = 2


def con_sql(sql):
    # 从数据库的表里获取数据

    db = pymysql.connect(host='172.16.40.158', port=4000, user='st_user', passwd='aqpuBLYzEV7tML5RPsN1pntUzFy',
                         db='jerry_prod')
    cursor = db.cursor()
    cursor.execute(sql)
    result = cursor.fetchall()
    db.close()
    return result


second_demands_zero_dict = {
    # "answer":{},
    "tractate":{},
    # "diary":{},
}
project_zero_dict = {
    # "answer":{},
    "tractate":{},
    # "diary":{},
}
t = 1
day_num = 0 - t
now = (datetime.datetime.now() + datetime.timedelta(days=day_num))
last_30_day_str = (now + datetime.timedelta(days=-31)).strftime("%Y%m%d")
today_str = now.strftime("%Y%m%d")
today_str_format = now.strftime("%Y-%m-%d")
yesterday_str = (now + datetime.timedelta(days=-1)).strftime("%Y%m%d")
yesterday_str_format = (now + datetime.timedelta(days=-1)).strftime("%Y-%m-%d")
one_week_age_str = (now + datetime.timedelta(days=-7)).strftime("%Y%m%d")
sql = """select first_device from online.ml_user_history_detail where partition_date = {today_str} and last_active_date >= {last_30_day_str}
""".format(today_str=today_str,last_30_day_str=last_30_day_str)
print(sql)
new_urser_device_id_df = spark.sql(sql)
new_urser_device_id_df.createOrReplaceTempView("device_id_view")
new_urser_device_id_df.show(1)
sql_res = new_urser_device_id_df.collect()

bulk_dict = {
    0: [0, 0, 0],
    10: [0, 0, 0],
    50: [0, 0, 0],
    100: [0, 0, 0],
    200: [0, 0, 0],
    500: [0, 0, 0],
    1000: [0, 0, 0],
}


task_list = []
# second_demands_count_dict,tags_v3_count_dict=get_user_post_from_mysql()
second_demands_count_dict= {'网红脸': 2, '洗眉': 4, '阴茎美化': 4, '瘦大腿': 120, '注射物取出': 4, '丰臀': 52, '薄唇': 48, '垫鼻基底': 121, '塑臀': 19, 'HPV': 6, '角膜炎': 2, '牙齿矫正': 184, '自体脂肪修复': 1, '瘦身': 404, '美白': 499, '牙齿种植': 2, '私密健康': 44, '祛副乳': 14, '瘦肩': 95, '丰额头': 66, '大眼睛': 534, '瘦下颌缘': 141, '隆鼻': 781, '脱敏': 2, '瘦手臂': 65, '祛法令纹': 39, '阴蒂美化': 39, '下眼睑下至': 66, '嫩唇': 14, '乳头美化': 10, '祛黑眼圈': 28, '产后恢复': 4, '颧骨内推': 259, '开眼角': 193, '鼻部修复': 42, '颌面矫正': 111, '垫眉弓': 62, '丰太阳穴': 109, '植发': 78, '填充卧蚕': 14, '玻尿酸修复': 10, '缩下颌角': 322, '牙病': 64, '牙齿清洁': 30, '祛纹身': 2, '牙齿护理': 2, '祛脂肪粒': 2, '胸部修复': 97, '牙齿美白': 128, '翘睫': 6, '缩毛孔': 180, '眼部修复': 108, '上眼睑提升': 363, '祛黑头': 94, '鼻部缩短': 36, '性快感': 8, '脱毛': 54, '眼线': 8, '健康调理': 19, '面部消脂': 295, '乳腺护理': 20, '修眉': 76, '祛双下巴': 53, '乳牙': 2, '祛肿眼泡': 4, '祛斑': 87, '酒窝': 20, '私密紧致': 97, '鼻尖延长': 4, '填充苹果肌': 70, '祛颈纹': 30, '丰唇': 92, '腿形矫正': 8, '结膜炎': 2, '面部轮廓': 569, '面部修复': 79, '祛痘': 217, '祛颊脂垫': 49, '瘦小腿': 112, '祛红血丝': 21, '人中缩短': 46, '隆胸': 562, '产后塑形': 2, '私密美化': 188, '丰面颊': 38, '紧致提升': 207, '垫下巴': 181, '鼻小柱矫正': 34, '瘦背': 26, '控油': 253, '凸嘴': 2, '缩短下巴': 4, '双眼皮': 493, '生睫毛': 14, '假体取出': 14, '祛疤': 32, '鼻孔矫正': 8, '皮肤病': 3, '鼻形矫正': 90, '处女膜': 49, '祛痣': 35, '祛腋臭': 8, '祛眼纹': 2, '皮肤护理': 85, '祛妊娠纹': 50, '颧骨提升': 8, '祛痘印': 143, '祛眼袋': 90, '嘴角上扬': 77, '智齿': 16, '缩胸': 10, '瘦脸': 646, '鼻中隔延长': 10, '乳晕美化': 70, '补水': 264, '唇腭裂': 6, '耳洞': 2, '祛痘坑': 57, '缩眼距': 302, '胸部提升': 80, '缩鼻头': 32, '缩鼻翼': 78, '生眉毛': 6, '缩鼻背': 6, 'M唇': 77, '填充泪沟': 80, '缩咬肌': 210, '瘦腰腹': 196, '瘦腿': 342, '近视': 6, '人中延长': 2}

tags_v3_count_dict = {'颧骨修复': 38, '焕肤美白': 2, '玻尿酸垫下巴': 58, '补牙': 14, '手术缩鼻翼': 78, '阴唇漂红': 60, '吸脂瘦腰腹': 169, '激光祛眼袋': 34, '埋线祛颈纹': 6, '口腔综合治疗': 2, '胸部假体取出': 12, 'G点注射': 8, '皮肤综合护理': 85, '下巴前推': 4, '黄金微雕': 6, '真皮填充卧蚕': 2, '瓷贴面': 47, '颧骨提高': 8, '婴儿针': 2, '手术缩鼻头': 2, '打耳洞': 2, '眶隔脂肪释放': 26, '美白针': 170, '龋齿': 14, '激光祛脂肪粒': 2, '胶原蛋白隆鼻': 2, '美白': 167, '开眼角修复': 10, '胶原蛋白填充泪沟': 6, '激光祛黑眼圈': 28, '激光脱敏': 2, '吸脂塑臀': 19, '面部线雕': 159, '玻尿酸丰额头': 16, '硅胶垫眉弓': 22, '激光溶脂瘦腰腹': 27, '童颜针': 18, '药物脱毛': 2, '射频溶脂瘦腰腹': 31, '乳晕缩小': 3, '自体脂肪丰太阳穴': 70, '玻尿酸填充泪沟': 56, '身体检查': 19, '牙齿隐形矫正': 177, '玻尿酸填充卧蚕': 12, '点阵激光': 8, '产后塑形': 2, '切眉': 38, '硅胶隆鼻': 108, '鼻形矫正': 58, '手术祛副乳': 10, '阴茎增大': 4, '自体脂肪祛法令纹': 15, '下颚前突': 2, '植发': 78, '鼻孔矫正': 8, '开外眼角': 64, '胶原蛋白隆胸': 10, '种植牙': 16, '腿形矫正': 8, '埋线瘦腰腹': 6, '胶原蛋白垫下巴': 4, '生长因子': 2, '假体丰太阳穴': 28, '手术紧缩阴道': 91, '玻尿酸精华导入': 6, '玻尿酸隆鼻': 54, '下颌角整形': 312, '注射祛腋臭': 6, '面膜': 4, '眼综合': 205, '埋线隆胸': 12, '毛囊检测': 2, '手术祛腋臭': 4, '吸脂瘦全身': 224, '胸部修复': 55, '半飞秒': 4, '翘睫': 6, '缩毛孔': 22, '自体脂肪垫眉弓': 10, '祛颈纹': 6, 'HPV检查': 6, '小V脸': 115, '玻尿酸丰太阳穴': 18, '激光洗眉': 2, '镜片近视矫正': 2, '耳软骨垫鼻头': 6, '激光溶脂瘦脸': 32, '自体脂肪祛黑眼圈': 16, '阴茎增粗': 4, 'PRP美肤': 8, '冷光美白': 36, '胸部注射物取出': 4, '鼻中隔软骨隆鼻': 10, '自体脂肪丰唇': 2, '吸脂瘦肩': 6, '微晶瓷隆鼻': 2, '丰唇': 42, '根管治疗': 2, '玻尿酸垫鼻基底': 26, '切开双眼皮': 220, '洗牙': 26, '祛痘': 129, '削磨祛斑': 2, '祛颊脂垫': 49, '颧骨颧弓降低': 237, '半永久纹眉': 76, '植鬓角': 34, '产后修复': 4, '光纤溶脂祛双下巴': 2, '玻尿酸祛颈纹': 16, '胶原蛋白注射': 6, '下巴修复': 6, '外眼角修复': 2, '埋线祛皱': 72, '鼻小柱矫正': 34, '吸脂瘦腿': 215, '面部溶脂针': 37, '内吸祛眼袋': 76, '牙龈炎': 2, '自体脂肪祛颈纹': 2, '吸脂瘦手臂': 46, '耳软骨隆鼻': 110, '嘟嘟唇': 14, '自体脂肪隆胸': 393, '刷酸祛斑': 2, '激光脱腋毛': 12, '硅胶丰额头': 6, '硅胶丰太阳穴': 8, '正颌': 6, '玻尿酸垫眉弓': 12, '胶原蛋白祛法令纹': 4, '玻尿酸丰面颊': 8, '激光祛斑': 87, '全身塑形': 247, '牙齿抛光': 2, '肤质检测': 2, '玻尿酸填充苹果肌': 20, '祛痘印': 79, '黄金微针': 72, '微针美塑': 4, '自体脂肪填充修复': 1, '瘦脸针': 210, '人工骨隆鼻': 165, '激光脱发际线': 16, '乳头再造': 4, '注射缩鼻头': 2, '处女膜修复': 49, '光纤溶脂瘦脸': 2, '综合面部整形': 93, '埋线双眼皮': 70, '射频溶脂祛双下巴': 2, '吸脂祛副乳': 10, '热拉提': 43, '激光紧缩阴道': 35, '祛痘坑': 53, '唇珠唇弓': 10, '假体垫眉弓': 40, '缩鼻头': 32, '激光美白': 4, '膨体丰额头': 2, '除皱针': 48, '缩鼻背': 6, '埋线隆鼻': 18, '超声波洗牙': 14, '智齿': 16, '植发际线': 54, '植睫毛': 10, '假体隆胸': 400, '人中延长': 2, '角膜炎': 2, '肉毒素缩鼻头': 2, '综合祛疤': 30, '激光脱私处毛发': 20, 'OK镜近视矫正': 2, '缩下巴': 4, '玻尿酸溶解酶': 10, '假体垫鼻基底': 91, '超声溶脂': 19, '唇腭裂': 6, '玻尿酸祛皱': 8, '下颚提升': 2, '菲洛嘉': 26, '激光祛妊娠纹': 22, '假体丰臀': 12, '颌面修复': 14, '激光脱全身毛发': 48, '拔牙': 28, '发质护理': 8, '激光祛痣': 27, '膨体丰太阳穴': 14, '外切祛眼袋': 34, '激光祛皱': 10, '全飞秒': 4, '像素激光': 66, '玻尿酸填充面部': 96, '定点双眼皮': 280, '微针水光': 56, '开内眼角': 175, '树脂贴面': 4, '白瓷娃娃': 24, '无针水光': 15, '自体脂肪丰面颊': 30, '假体垫下巴': 104, '超声溶脂瘦腿': 8, '溶脂针瘦小腿': 2, '漂唇': 14, '激光祛红血丝': 21, '腹壁成形术': 10, '超声提升': 56, '双眼皮修复': 92, '膨体垫下巴': 24, '肋软骨隆鼻': 148, '瘦肩针': 89, '下眼睑下至': 66, '隆胸修复': 3, '小腿神经阻断术': 6, '面部紧致提升': 44, '乳头内陷矫正': 8, '颧骨内推': 255, '鼻部修复': 36, '吸脂祛双下巴': 51, '自体真皮隆鼻': 4, '人工骨垫下巴': 2, '乳房缩小': 10, '玻尿酸丰唇': 38, '皮秒激光': 84, '硅胶隆胸': 2, '植眉': 6, '激光祛颈纹': 12, '自体脂肪丰额头': 50, '综合祛皱': 158, '硅胶垫下巴': 62, '射频溶脂': 28, '卧蚕修复': 2, '祛肿眼泡': 4, '全飞秒激光术': 4, '冷冻溶脂': 54, '烤瓷牙': 22, '热玛吉': 101, '少女针': 22, '鼻部假体取出': 12, '膨体隆鼻': 275, '上眼睑提肌': 22, '皮肤病': 3, '射频提升': 36, '祛黑头': 20, '牙齿综合': 2, '乳晕淡化': 10, '激光美肤': 10, '胶原蛋白祛黑眼圈': 10, '自体脂肪填充苹果肌': 50, '提眉修复': 12, '结膜炎': 2, '半永久纹发际线': 6, '自体脂肪填充泪沟': 40, '乳房再造': 34, '埋线瘦身': 24, '瘦腿针': 108, '手术祛疤': 16, '自体软骨垫鼻尖': 4, '假体填充苹果肌': 2, '面部不对称改善': 25, '激光溶脂瘦腿': 31, '射频溶脂瘦腿': 14, '私密清洁': 2, '乳头缩小': 14, '光纤溶脂': 61, '埋线缩鼻翼': 2, '果酸焕肤': 48, '激光脱臂毛': 12, '精华导入': 6, '牙髓炎': 2, 'M唇': 45, '软骨隆鼻': 379, '私密护理': 1, '人中缩短': 46, '手术近视矫正': 2, '优立塑': 12, '半永久纹眼线': 8, '牙周病': 2, '凸嘴矫正': 79, '激光脱腿毛': 14, '下颌内缩': 309, '黑脸娃娃': 13, '牙齿托槽矫正': 99, '溶脂针祛双下巴': 2, '阴唇整形': 152, '微笑唇': 32, '光子嫩肤': 79, '睫毛增长': 4, '注射祛疤': 12, '厚唇改薄': 48, '自体脂肪丰臀': 48, '水光针': 135, '软骨垫鼻基底': 38, '玻尿酸祛黑眼圈': 6, '吸脂瘦脸': 149, '自体脂肪垫鼻基底': 20, '自体脂肪填充面部': 387, '膨体垫眉弓': 38, '黄金微针祛妊娠纹': 36, '自体脂肪垫下巴': 26, '胶原蛋白丰唇': 2, '乳晕漂红': 61, '自体脂肪填充卧蚕': 6, '鼻综合': 681, '激光近视矫正': 6, '吸脂瘦背': 26, '乳腺疏通': 20, '口腔溃疡': 8, '女性私密检查': 42, '射频溶脂瘦脸': 2, '激光祛疤': 26, '激光溶脂瘦手臂': 2, '假体祛法令纹': 2, '假体丰额头': 6, '激光祛纹身': 2, '补水': 89, '颌面正畸': 18, '酒窝成形': 20, '全瓷牙': 22, '射频溶脂瘦手臂': 21, '皓齿美白': 103, '激光脱唇毛': 12, '缩鼻翼': 66, '乳房提升': 4, '下巴假体取出': 2, '半飞秒激光术': 4, '射频祛眼袋': 10, '阴蒂整形': 39, '玻尿酸祛法令纹': 29, '洗眉': 4, '玻尿酸隆胸': 12, '假体隆鼻': 323, '小气泡': 32, '自体脂肪隆鼻': 195, '水氧活肤': 4, '微针美肤': 2, '自体脂肪祛皱': 97}
print(second_demands_count_dict,tags_v3_count_dict)
time.sleep(10)

for redis_count,spark_res in enumerate(sql_res):
    # if redis_count >= 50:break
    second_demands = []
    projects = []
    total_answer_content_num = 0
    total_tractate_content_num = 0
    total_diary_content_num = 0
    # print(sql_res)
    try:
        res = get_user_portrait_tag3_from_redis(spark_res.first_device)
    except:
        continue
    if res.get("second_demands"):
        second_demands = res.get("second_demands")
        # print(count_res)
        for tag in second_demands:
            if tag in tags_v3_count_dict:
                total_tractate_content_num += second_demands_count_dict[tag]
    if res.get("projects"):
        projects = res.get("projects")
        # print(count_res)
        for tag in projects:
            if tag in tags_v3_count_dict:
                total_tractate_content_num += tags_v3_count_dict[tag]
    # print(total_answer_content_num, total_tractate_content_num, total_diary_content_num)
    tmp_count_num = 0


    if 0 <= total_tractate_content_num < 10:
        bulk_dict[0][1] += 1
    elif 10 <= total_tractate_content_num < 50:
        bulk_dict[10][1] += 1
    elif 50 <= total_tractate_content_num < 100:
        bulk_dict[50][1] += 1
    elif 100 <= total_tractate_content_num < 200:
        bulk_dict[100][1] += 1
    elif 200 <= total_tractate_content_num < 500:
        bulk_dict[200][1] += 1
    elif 500 <= total_tractate_content_num < 1000:
        bulk_dict[500][1] += 1
    else:
        bulk_dict[1000][1] += 1

    if redis_count % 5000 == 0:
        print(bulk_dict)


print(bulk_dict)



