crawler_week_report.py

# -*- 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_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")


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'")




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():
    es_framework = Elasticsearch(hosts='172.16.32.37', port=9200)
    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):
    es_framework = Elasticsearch(hosts='172.16.32.37', port=9200)
    search_body = {
        "query": {
            "bool": {
                "filter": [
                    {"range": {"fetch_time": {"gte": int(start_ts)}}}
                ], "must_not": [
                    {"exists": {"field": "search_word"}}
                ]
            }
        }, "size": 0, "aggs": {
            "NAME": {
                "terms": {
                    "field": "platform",
                    "size": 10
                }
            }
        }
    }
    print(search_body)
    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)
        start_date_str = start_date.strftime("%Y-%m-%d %H:%M:%S")
        match_all_sql = """select count(id) as nums from api_tractate where create_time >= '{time}' and platform = {platfrom}""".format(
            time=start_date_str, 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 pgc_type = 1""".format(
            time=start_date_str)
        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 = str(round(sorteds/exposure_count * 100, 2)) + "%"
            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()
    start_ts = 1597593600000
    craw_one_week = get_how_word_crawler_count(data_index, start_ts, end_ts, week_num, last_week_num, year)
    word_count = get_keyword_ctr(start_ts, end_ts)
    # 一周爬虫抓取热点数

    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()