# -*- coding:UTF-8 -*-
# @Time  : 2020/8/11 14:33
# @File  : crawler_week_report.py
# @email : litao@igengmei.com
# @author : litao

# coding=utf-8
import json

import pymysql
import xlwt, datetime
import redis
# from pyhive import hive
from maintenance.func_send_email_with_file import send_file_email
from typing import Dict, List
from elasticsearch import Elasticsearch
from elasticsearch.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")


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").appName("LR PYSPARK TEST").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'")



es_framework = Elasticsearch(hosts='172.16.32.37', port=9200)
rds = redis.StrictRedis(host='172.16.40.164', port=6379, db=19, password='ReDis!GmTx*0aN12')

conn_mimas = pymysql.connect(host='172.16.30.138', port=3306, user='mimas', passwd='GJL3UJe1Ck9ggL6aKnZCq4cRvM',
                             db='mimas_prod', charset='utf8mb4')
cur_mimas = conn_mimas.cursor()

db_zhengxing_eagle = pymysql.connect(host="172.16.30.136", port=3306, user="doris",
                                     password="o5gbA27hXHHm",
                                     db="doris_prod",
                                     charset='utf8',
                                     cursorclass=pymysql.cursors.DictCursor)
zhengxing_cursor = db_zhengxing_eagle.cursor()

es = Elasticsearch([
    {
        'host': '172.16.31.17',
        'port': 9200,
    }, {
        'host': '172.16.31.11',
        'port': 9200,
    }])


# class Hive_conn():
#     def __init__(self):
#         self._hive_host = "bj-gm-prod-cos-datacenter003"
#         self._hive_port = 10000
#         self._hive_username = "strategy"
#         self._hive_database = "online"
#         self._conn = hive.Connection(host=self._hive_host, port=self._hive_port, username=self._hive_username,
#                                     database=self._hive_database)
#         self.cursor = self._conn.cursor()
#         # self.cursor.execute('show tables')
#
#     def __del__(self):
#         self.cursor.close()
#         self._conn.close()


# def zipDir(dirpath,outFullName):
#     """
#     压缩指定文件夹
#     :param dirpath: 目标文件夹路径
#     :param outFullName: 压缩文件保存路径+xxxx.zip
#     :return: 无
#     """
#     import zipfile
#     zip = zipfile.ZipFile(outFullName,"w",zipfile.ZIP_DEFLATED)
#     for path,dirnames,filenames in os.walk(dirpath):
#         # 去掉目标跟路径，只对目标文件夹下边的文件及文件夹进行压缩
#         fpath = path.replace(dirpath,'')
#
#         for filename in filenames:
#             zip.write(os.path.join(path,filename),os.path.join(fpath,filename))
#     zip.close()

def send_email_tome():
    try:
        date = datetime.datetime.now().date() - datetime.timedelta(days=1)
        fromaddr = 'litao@igengmei.com'
        content = 'hi  all:附件为' + str(date) + '的搜索词数据统计结果以及近一周的数据统计结果，请查收！'
        zipFile = "/srv/apps/crawler/近1周数据统计结果.xls"
        send_file_email("", "", email_group=["<litao@igengmei.com>","‎<duanyingrong@igengmei.com>"], title_str=content
                        , email_msg_body_str=content, file=zipFile)
    except Exception as e:
        print(e)


def get_es_word(word, start_ts):
    # tractate oneline
    results = es.search(
        index='gm-dbmw-tractate-read',
        doc_type='tractate',
        timeout='10s',
        size=0,
        body={
            "query": {
                "bool": {
                    "minimum_should_match": 1,
                    "should": [{"match_phrase": {"content": {"query": word, "analyzer": "gm_default_index"}}}, {
                        "match_phrase": {"tractate_tag_name": {"query": word, "analyzer": "gm_default_index"}}}, {
                                   "match_phrase": {"tractate_tag_name_content": {"query": word,
                                                                                  "analyzer": "gm_default_index"}}}],
                    "must": [{"term": {"is_online": True}}, {
                        "terms":
                            {"content_level": [6, 5, 4, 3]}
                    }]
                }
            }, "size": 0, "aggs": {
                "content_level": {
                    "terms": {
                        "field": "content_level",
                        "size": 10
                    }
                }
            }
        }
    )
    tractate_content_num = results["hits"]["total"]
    content_level_count = results["aggregations"]["content_level"]["buckets"]

    # tractate craw
    results = es.search(
        index='gm-dbmw-tractate-read',
        doc_type='tractate',
        timeout='10s',
        size=0,
        body={
            "query": {
                "bool": {
                    "minimum_should_match": 1,
                    "should": [{"match_phrase": {"content": {"query": word, "analyzer": "gm_default_index"}}}, {
                        "match_phrase": {"tractate_tag_name": {"query": word, "analyzer": "gm_default_index"}}}, {
                                   "match_phrase": {"tractate_tag_name_content": {"query": word,
                                                                                  "analyzer": "gm_default_index"}}}],
                    "must": [
                        {"terms": {"platform": [13, 14, 15, 16, 17]}},
                        {"range": {"create_time_epoch": {"gte": start_ts / 1e3}}}
                    ]
                }
            }, "size": 0, "aggs": {
                "content_level": {
                    "terms": {
                        "field": "content_level",
                        "size": 10
                    },
                    "aggs": {
                        "is_online": {
                            "terms": {
                                "field": "is_online",
                                "size": 10
                            }
                        }
                    }
                }
            }
        }
    )
    craw_tractate_num = results["hits"]["total"]
    craw_level_count = results["aggregations"]["content_level"]["buckets"]
    res_list = []
    # 可见3456星帖子
    try:
        for content_level in ("3", "4", "5", "6"):
            doc_count = 0
            for content_level_dic in content_level_count:
                if content_level_dic["key"] == content_level:
                    doc_count = content_level_dic["doc_count"]
                    break
            res_list.append(doc_count)
    except:
        print("craw_level_count error", craw_level_count)
    res_list.append(craw_tractate_num)
    for content_level in ("3", "4", "5", "6"):
        doc_count = 0
        for content_level_dic in craw_level_count:
            if content_level_dic["key"] == content_level:
                for buck in content_level_dic["is_online"]["buckets"]:
                    if buck["key"] == 1:
                        doc_count = buck["doc_count"]
                        break
        res_list.append(doc_count)

    return res_list


class WritrExcel():
    def __init__(self):
        self.workbook = xlwt.Workbook(encoding='utf-8')

    def set_style(self, name, height, bold=False):
        style = xlwt.XFStyle()  # 初始化样式
        font = xlwt.Font()  # 为样式创建字体
        font.name = name
        font.bold = bold
        font.color_index = 4
        font.height = height
        style.font = font
        return style

    # 写入Excel
    def write_excel(self, sheet_name, rows):
        # 创建工作簿
        # 创建sheet
        data_sheet = self.workbook.add_sheet(sheet_name)
        # 将样式定义在循环之外
        default = self.set_style('Times New Roman', 220, True)
        j = k = 0
        # 循环读取每一行数据并写入Excel
        for row in rows[:65530]:
            for i in range(len(row)):
                try:
                    # 写入
                    data_sheet.write((j + k), i, row[i], default)
                except:
                    print(i)
                    raise
                # data_sheet.write(1, i, row1[i], self.set_style('Times New Roman', 220, True))
            k = k + 1
        print("写入sheet成功，共" + str(k) + "行数据")

    def save_excel(self, path):
        self.workbook.save(path)


def get_search_keywrod_dict():
    search_body = {
        "query": {
            "bool": {
                "filter": [

                ]
            }
        }
    }
    return_dict = {}
    scan_res = scan(es_framework, query=search_body, index="search_word")
    for res in scan_res:
        return_dict[res["_source"]["title"]] = {}
    return return_dict


def get_hot_crawler_data_count(index, start_ts):
    search_body = {
        "query": {
            "bool": {
                "filter": [
                    {"range": {"fetch_time": {"gte": start_ts}}}
                ], "must_not": [
                    {"exists": {"field": "search_word"}}
                ]
            }
        }, "size": 0, "aggs": {
            "NAME": {
                "terms": {
                    "field": "platform",
                    "size": 10
                }
            }
        }
    }
    search_res = es_framework.search(body=search_body, index=index)
    platform_count = search_res["aggregations"]["NAME"]["buckets"]
    return platform_count


def get_count_hot_crawler_data_from_mysql(start_ts: int) -> Dict:
    data_platform_id_list = [11, 12]
    res_dict = {}
    for platform_id in data_platform_id_list:
        start_date = datetime.datetime.fromtimestamp(start_ts / 1e3)
        match_all_sql = """select count(id) as nums from api_tractate where create_time >= '{time}' and platform = {platfrom}""".format(
            time=str(start_date), platfrom=platform_id)
        cur_mimas.execute(match_all_sql)
        data = cur_mimas.fetchall()
        match_all_num = data[0][0]
        # print(match_all_num)
        oneline_sql = """select count(id) as nums from api_tractate where create_time >= '{time}' and is_online = 1 and platform = {platfrom}""".format(
            time=str(start_date), platfrom=platform_id)
        cur_mimas.execute(oneline_sql)
        data = cur_mimas.fetchall()
        match_oneline_num = data[0][0]
        # print(match_oneline_num)
        if platform_id == 11:
            platform = 'weibo'
        elif platform_id == 12:
            platform = "douban"
        res_dict[platform] = []
        res_dict[platform].append(match_all_num)
        res_dict[platform].append(match_oneline_num)
    return res_dict


def week_num(year=None, cycle=None, cycle_num=None, compare_type=None):
    now = datetime.datetime.now()
    now_canlendar = now.isocalendar()
    if not cycle_num:
        week_canlendar = now_canlendar
    else:
        week_canlendar = (now.year, cycle_num + 1, 0)
    year = week_canlendar[0]
    this_week = week_canlendar[1]
    if this_week == 0:
        last_year = year - 1
        this_week = 1
    else:
        last_year = year
    if this_week == 1:
        last_week = "52"
    else:
        last_week = this_week - 1
    today = datetime.datetime(datetime.datetime.now().year, datetime.datetime.now().month, datetime.datetime.now().day)
    # today = datetime.datetime(year=2018, month=12, day=25)
    first_day_in_week = today - datetime.timedelta(
        days=now_canlendar[2] + 7 * (now_canlendar[1] - week_canlendar[1]) + 2)
    fisrt_day_ts = int(first_day_in_week.timestamp() * 1e3)
    last_day_in_week = first_day_in_week + datetime.timedelta(days=7)
    last_day_ts = int(last_day_in_week.timestamp() * 1e3)

    this_week_index = 'crawler-data-raw'
    return this_week_index, fisrt_day_ts, last_day_ts, this_week, last_week, last_year


def get_how_word_crawler_count(data_index, start_ts, end_ts, week_num, last_week_num, year):
    res_data = [("周", "抓取平台", "入库内容数", "推送候选内容入库数", "上线内容数", "推送内容数")]
    craw_data_count = get_hot_crawler_data_count(data_index, start_ts)
    res_dict = get_count_hot_crawler_data_from_mysql(start_ts)
    for platform in ("douban", "weibo"):
        craw_one_week_count = ""
        for buck in craw_data_count:
            if buck["key"] == platform:
                craw_one_week_count = buck["doc_count"]
        push_num = res_dict[platform][0]
        oneline_num = res_dict[platform][1]
        new_line = (week_num, platform, craw_one_week_count, push_num, oneline_num, "")
        res_data.append(new_line)
    return res_data


def get_keyword_ctr(start_ts,end_ts):
    start_date = datetime.datetime.fromtimestamp(start_ts/1e3)
    end_date = datetime.datetime.fromtimestamp(end_ts/1e3)
    start_date_str = start_date.strftime("%Y%m%d")
    end_date_str = end_date.strftime("%Y%m%d")
    data_dic = {}
    # --query词曝光
    baoguang_sql = """
SELECT card_id as query,count(*) as query_count FROM online.ml_community_precise_exposure_detail
WHERE partition_date>='{start_date}' AND partition_date<'{end_date}' and page_name in ('search_home','search_home_more','search_home_welfare','search_home_diary','search_home_wiki','search_home_post','search_home_hospital','search_home_doctor') group by query
""".format(start_date=str(start_date_str),end_date=str(end_date_str))
    device_df = spark.sql(baoguang_sql)
    device_df.show(1, False)
    sql_res = device_df.collect()
    print("-----------------------------------------------------------------------------")
    for res in sql_res:
        data_dic[res.query] = res.query_count
    # --query词曝光
    query_sql = """
SELECT params['query_words'] as query_words
FROM online.bl_hdfs_maidian_updates
WHERE partition_date>='{start_date}' AND partition_date<'{end_date}'
AND action = 'report_status'
AND page_name in ('search_home','search_home_more','search_home_welfare','search_home_diary','search_home_wiki','search_home_post','search_home_hospital','search_home_doctor')
    """.format(start_date=start_date_str,end_date=end_date_str)

    device_df = spark.sql(query_sql)
    device_df.show(1, False)
    sql_res = device_df.collect()
    for res in sql_res:
        try:
            res_json = json.loads(res.query_words)
            # print(res_json, type(res_json))
        except:
            continue
        for single_keyword in res_json:
            data_count = data_dic.get(single_keyword)
            if data_count:
                data_dic[single_keyword] = data_dic[single_keyword] + 1
            else:
                data_dic[single_keyword] = 1
    print(data_dic)
    return data_dic


def craw_query_one_week(data_index, start_ts, end_ts, week_num, last_week_num, year,word_count):

    tag_names_list_week = []
    one_month_ago = datetime.datetime.now().date() - datetime.timedelta(days=30)
    search_keyword_dict = get_search_keywrod_dict()
    sql = 'select keywords,sum(sorted) as nums,sum(uv) as uvs  from api_search_words where is_delete = 0 and create_time >= "' + str(
        one_month_ago) + '" group by keywords  order by  nums  desc'

    print(sql)
    zhengxing_cursor.execute("set names 'UTF8'")
    zhengxing_cursor.execute(sql)
    data = zhengxing_cursor.fetchall()

    tup_title = ("关键词", "近一周搜索内容ctr", "过去30天搜索pv", "过去30天搜索uv", "帖子-3星内容数量", "帖子-4星内容数量",
                 "帖子-5星内容数量", "帖子-6星内容数量", "近一周抓取数量", "近一周审核上线3星数量",
                 "近一周审核上线4星数量", "近一周审核上线5星数量", "近一周审核上线6星数量")
    tag_names_list_week.append(tup_title)
    for name in list(data):
        word = name.get("keywords", None)
        if word in search_keyword_dict:
            sorteds = name.get("nums", 0)
            uv = name.get("uvs", 0)
            tractate_content_num = get_es_word(word, start_ts)
            exposure_count = word_count.get(word)
            if not exposure_count:
                ctr = ""
            else:
                ctr = int(sorteds/exposure_count)
            new_line = [word,ctr, sorteds, uv]
            tag_names_list_week.append(tuple(new_line + tractate_content_num))
    for word in search_keyword_dict:
        tractate_content_num = get_es_word(word, start_ts)
        if tractate_content_num[0] != 0:
            new_line = [word, 0, 0, 0]
            tag_names_list_week.append(tuple(new_line + tractate_content_num))
    return tag_names_list_week


if __name__ == "__main__":
    data_index, start_ts, end_ts, week_num, last_week_num, year = week_num()
    word_count = get_keyword_ctr(start_ts, end_ts)
    # 一周爬虫抓取热点数
    craw_one_week = get_how_word_crawler_count(data_index, start_ts, end_ts, week_num, last_week_num, year)
    print(craw_one_week)
    # query 一周抓取详情
    all_data_week = craw_query_one_week(data_index, start_ts, end_ts, week_num, last_week_num, year,word_count)
    path = "近1周数据统计结果.xls"
    exl = WritrExcel()
    exl.write_excel("热点内容抓取周报", tuple(craw_one_week))
    exl.write_excel("query抓取周报", tuple(all_data_week))
    exl.save_excel(path)
    print(u'创建demo.xls文件成功')
    send_email_tome()
