func_from_es_to_csv.py

# -*- coding:utf-8 -*-
# @Time : 2019/10/28 15:48 
# @Author : litao
import elasticsearch
import datetime
import pandas as pd
from elasticsearch.helpers import scan
from func_cal_doc_id import cal_doc_id
from write_data_into_es.func_get_releaser_id import *
import copy, os, time, json
import zipfile
import redis
from maintenance.send_email_with_file_auto_task import write_email_task_to_redis

hosts = '192.168.17.11'
port = 80
user = 'zhouyujiang'
passwd = '8tM9JDN2LVxM'
http_auth = (user, passwd)
es = elasticsearch.Elasticsearch(hosts=hosts, port=port, http_auth=http_auth)
from redis.sentinel import Sentinel
sentinel = Sentinel([('192.168.17.65', 26379),
                     ('192.168.17.66', 26379),
                     ('192.168.17.67', 26379)
             ],socket_timeout=0.5)
# 查看master节点
master = sentinel.discover_master('ida_redis_master')
# 查看slave 节点
slave = sentinel.discover_slaves('ida_redis_master')
# 连接数据库
rds = sentinel.master_for('ida_redis_master', socket_timeout=0.5, db=2, decode_responses=True)
# es = elasticsearch.Elasticsearch(hosts=hosts, port=port, http_auth=http_auth)
# rds = redis.StrictRedis(host='192.168.17.60', port=6379, db=2, decode_responses=True)


def get_project_releaser(project_name):
    res_list = []
    search_body = {
            "query": {
                    "match": {
                            "project_tags.keyword": project_name
                    }
            }
    }
    res_scan = scan(client=es, query=search_body, index="target_releasers")
    for res in res_scan:
        res_list.append(res["_source"])
    return res_list


def zipDir(dirpath, outFullName):
    """
    压缩指定文件夹
    :param dirpath: 目标文件夹路径
    :param outFullName: 压缩文件保存路径+xxxx.zip
    :return: 无
    """
    remove_list = []
    if ".zip" not in outFullName:
        outFullName += ".zip"
    file_path = os.path.split(outFullName)
    is_path = os.path.exists(file_path[0])
    if not is_path:
        os.mkdir(file_path[0])
        time.sleep(0.5)
    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))
            remove_list.append(os.path.join(path, filename))
    zip.close()

    for p in remove_list:
        os.remove(p)
    time.sleep(0.5)
    os.removedirs(dirpath)
    return outFullName


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] - 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] + 1))
    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 = 'short-video-weekly'
    this_week_doc = 'daily-url-' + str(year) + '_w' + format(this_week, '>02d') + '_s1'
    last_week_index = 'releaser-weekly-short-video'
    last_week_doc = 'doc'
    if compare_type == "new_released":
        this_week_index = last_week_index
        this_week_doc = last_week_doc
    return this_week_index, this_week_doc, last_week_index, last_week_doc, fisrt_day_ts, last_day_ts, this_week, last_week, last_year


def month_num(year=None, cycle=None, cycle_num=None, compare_type=None):
    now = datetime.datetime.now()
    if not year:
        if cycle_num <= now.month:
            year = now.year
        else:
            year = now.year - 1
    if not cycle_num:
        this_mon = now.month - 1 if now.month != 1 else 12
        last_mon = this_mon - 1 if this_mon > 1 else this_mon - 1 + 12
        if last_mon == 12:
            last_year = year - 1
        else:
            last_year = year
    else:
        this_mon = cycle_num
        last_mon = cycle_num - 1 if this_mon > 1 else this_mon - 1 + 12
        if last_mon == 12:
            last_year = year - 1
        else:
            last_year = year
    first_day_ts = int(datetime.datetime(year=year, month=this_mon, day=1).timestamp() * 1e3)
    if this_mon == 12:
        next_year = last_year + 1
        next_month = 1
    else:
        next_year = year
        next_month = this_mon + 1
    last_day_ts = int(datetime.datetime(year=next_year, month=next_month, day=1).timestamp() * 1e3)
    this_mon_index = "short-video-production-%s" % year
    this_mon_doc = "daily-url-%s" % (
            datetime.datetime(year=next_year, month=next_month, day=1) + datetime.timedelta(days=-1)).strftime(
            "%Y-%m-%d")
    last_mon_index = "releaser"
    last_mon_doc = "releasers"
    if compare_type == "new_released":
        this_mon_index = last_mon_index
        this_mon_doc = last_mon_doc
    return this_mon_index, this_mon_doc, last_mon_index, last_mon_doc, first_day_ts, last_day_ts, this_mon, last_mon, last_year


def quarter_num(year=None, cycle=None, cycle_num=None, compare_type=None):
    now = datetime.datetime.now()
    if not cycle_num:
        this_quarter = int(now.month / 3) + 1
    else:
        this_quarter = cycle_num
    last_quarter = this_quarter - 1 if cycle_num > 1 else 4
    if last_quarter == 4:
        last_year = year - 1
    else:
        last_year = year
    first_day_ts = int(datetime.datetime(year=year, month=(this_quarter - 1) * 3 + 1, day=1).timestamp() * 1e3)
    last_day_ts = int(datetime.datetime(year=year, month=this_quarter * 3 + 1, day=1).timestamp() * 1e3)
    this_quarter_index = "short-video-quarter-%s" % year
    this_quarter_doc = "daily-url-2019-Q%s" % this_quarter
    last_quarter_index = "releaser"
    last_quarter_doc = "releasers-%s-Q%s" % (last_year, last_quarter)
    if compare_type == "new_released":
        this_quarter_index = last_quarter_index
        this_quarter_doc = last_quarter_doc
    return this_quarter_index, this_quarter_doc, last_quarter_index, last_quarter_doc, first_day_ts, last_day_ts, this_quarter, last_quarter, last_year


def create_index(year=None, cycle="week", cycle_num=None, compare_type=None, project_type="short_video",
                 data_type="detail", **kwargs):
    """

    :param year: 年份 可为空
    :param cycle: 周期 week/month/quarter/year/("all-time" 用于微信导出自定义数据)
    :param cycle_num: 周期 1/2/3/4/5/6
    :param compare_type:默认为空 可不填
    :param project_type:数据平台 short_video/weibo/weixin/all-time
    :param data_type:数据类型 detail/summary
    :param kwargs: 自定义索引和切片 参数如下
    :return:
    """
    first_day_ts = ""
    last_day_ts = ""
    this_cycle = ""
    try:
        if kwargs:
            this_cycle_index = kwargs.get("this_cycle_index")
            this_cycle_doc = kwargs.get("this_cycle_doc")
            last_cycle_index = kwargs.get("last_cycle_index")
            last_cycle_doc = kwargs.get("last_cycle_doc")
            first_day_ts = kwargs.get("first_day_ts")
            last_day_ts = kwargs.get("last_day_ts")
        else:
            this_cycle_index = None
    except:
        this_cycle_index = None
    one_line = {}
    if not this_cycle_index:
        if cycle == "week":
            this_cycle_index, this_cycle_doc, last_cycle_index, last_cycle_doc, first_day_ts, last_day_ts, this_cycle, last_cycle, last_year = week_num(
                    year, cycle, cycle_num, compare_type)
        elif cycle == "month":
            this_cycle_index, this_cycle_doc, last_cycle_index, last_cycle_doc, first_day_ts, last_day_ts, this_cycle, last_cycle, last_year = month_num(
                    year, cycle, cycle_num, compare_type)
        elif cycle == "quarter":
            this_cycle_index, this_cycle_doc, last_cycle_index, last_cycle_doc, first_day_ts, last_day_ts, this_cycle, last_cycle, last_year = quarter_num(
                    year, cycle, cycle_num, compare_type)
        elif cycle == "year":
            pass

    # print(this_cycle_index, this_cycle_doc, last_cycle_index, last_cycle_doc, fisrt_day_ts, last_day_ts, this_cycle,
    #      last_cycle, last_year)
    if data_type == "detail":
        if project_type == "short_video":
            return this_cycle_index, this_cycle_doc, first_day_ts, last_day_ts, this_cycle
        elif project_type == "weibo":
            return "ronghe_weibo_monthly", "doc", first_day_ts, last_day_ts, this_cycle
        elif project_type == "weixin":
            return "ronghe_weixin", "doc", first_day_ts, last_day_ts, "ronghe_weixin"
        elif project_type == "all-time":
            return "short-video-all-time-url", "all-time-url", None, None, None
        elif project_type == "target_releasers":
            return "target_releasers", "doc", None, None, None
        elif project_type == "purchased_releasers":
            return "purchased_releasers", "doc", None, None, None
    else:
        if project_type == "short_video":
            return last_cycle_index, last_cycle_doc, first_day_ts, last_day_ts, this_cycle
        elif project_type == "weibo":
            return "ronghe_weibo_releaser", "doc", first_day_ts, last_day_ts, this_cycle
        elif project_type == "weixin":
            if cycle == "all-time":
                return "ronghe_weixin", "doc", first_day_ts, last_day_ts, "ronghe_weixin"
            return "ronghe_weixin_releaser", "doc", first_day_ts, last_day_ts, this_cycle
        elif project_type == "all-time":
            return "short-video-all-time-url", "all-time-url", None, None, None
        elif project_type == "purchased_releasers":
            return "purchased_releasers", "doc", None, None, None

def create_search_body(year=None, cycle="week", cycle_num=None, project_type="short_video", data_type="detail",
                       start_ts=None, end_ts=None, releaser=None, platform=None, releaser_id_str=None, **kwargs):
    if data_type == "detail":
        search_body = {
                "query": {
                        "bool": {
                                "filter": [
                                        {"term": {"releaser_id_str": releaser_id_str}},
                                        {"range": {"release_time": {"gte": start_ts, "lt": end_ts}}},
                                        {"range": {"duration": {"lte": 600}}}
                                ]
                        }
                }
        }
        if project_type == "weibo":
            search_body["query"]["bool"]["filter"].pop(0)
            search_body["query"]["bool"]["filter"].pop(1)
            search_body["query"]["bool"]["filter"].append({"term": {"UID.keyword": releaser_id_str}})
            if cycle == "month":
                search_body["query"]["bool"]["filter"].append({"term": {"data_month": cycle_num}})
                search_body["query"]["bool"]["filter"].append({"term": {"data_year": year}})
        elif project_type == "target_releasers":
            search_body = kwargs.get("extra_dic").get("search_body")

        elif project_type == "weixin":
            search_body["query"]["bool"]["filter"].pop(2)
            search_body["query"]["bool"]["filter"].pop(0)
            search_body["query"]["bool"]["filter"].append({"term": {"releaser_id_str.keyword": releaser_id_str}})
        # elif cycle == "month" and project_type == "short_video":
        #     search_body["query"]["bool"]["filter"].pop(0)
        #     search_body["query"]["bool"]["filter"].append({"term": {"releaser_id_str.keyword": releaser_id_str}})
    else:
        search_body = {
                "query": {
                        "bool": {
                                "filter": [
                                        {"term": {"releaser_id_str.keyword": releaser_id_str}}
                                ]
                        }
                }
        }
        if project_type == "weibo":
            search_body["query"]["bool"]["filter"].pop(0)
            search_body["query"]["bool"]["filter"].append({"term": {"UID.keyword": releaser_id_str}})
            if cycle == "month":
                search_body["query"]["bool"]["filter"].append({"term": {"data_month": cycle_num}})
                search_body["query"]["bool"]["filter"].append({"term": {"data_year": year}})
        elif project_type == "short_video":
            if cycle == "month":
                search_body["query"]["bool"]["filter"].append({"term": {"data_month": cycle_num}})
                search_body["query"]["bool"]["filter"].append({"term": {"data_year": year}})
            elif cycle == "week":
                search_body["query"]["bool"]["filter"].append({"term": {"data_week_num": cycle_num}})
                search_body["query"]["bool"]["filter"].append({"term": {"data_week_year": year}})
        elif project_type == "weixin":
            if cycle == "month":
                search_body["query"]["bool"]["filter"].append({"term": {"data_month": cycle_num}})
                search_body["query"]["bool"]["filter"].append({"term": {"data_year": year}})
            elif cycle == "all-time":
                search_body["query"]["bool"]["filter"].append(
                        {"range": {"release_time": {"gte": start_ts, "lt": end_ts}}})
                search_body["aggs"] = {
                        "sum_play": {
                                "sum": {
                                        "field": "play_count"
                                }
                        },
                        "sum_favorite": {
                                "sum": {
                                        "field": "favorite_count"
                                }
                        },
                        "sum_comment": {
                                "sum": {
                                        "field": "comment_count"
                                }
                        },
                        "sum_repost": {
                                "sum": {
                                        "field": "repost_count"
                                }
                        }

                }
        elif project_type == "all-time":
            search_body["query"]["bool"]["filter"].pop(0)
            search_body["query"]["bool"]["filter"].append({"term": {"releaser_id_str": releaser_id_str}})
            search_body["query"]["bool"]["filter"].append({"range": {"duration": {"lte": 600}}})
            search_body["query"]["bool"]["filter"].append({"range": {"release_time": {"gte": start_ts, "lt": end_ts}}})
            search_body["aggs"] = {
                    "sum_play": {
                            "sum": {
                                    "field": "play_count"
                            }
                    },
                    "sum_favorite": {
                            "sum": {
                                    "field": "favorite_count"
                            }
                    },
                    "sum_comment": {
                            "sum": {
                                    "field": "comment_count"
                            }
                    },
                    "sum_repost": {
                            "sum": {
                                    "field": "repost_count"
                            }
                    }

            }
        elif project_type == "purchased_releasers":
            search_body["query"]["bool"]["filter"].pop(0)
            search_body["query"]["bool"]["filter"].append({"term": {"releaser_id_str": releaser_id_str}})
    print(search_body)
    return search_body


def detail_to_dic(search_res, data_type="short_video"):
    if data_type == "target_releasers":
        for one_scan in search_res:
            project_tags = one_scan["_source"].get("project_tags")
            department_tags = one_scan["_source"].get("department_tags")
            project_type = ""
            department_type = ""
            if project_tags:
                for i in project_tags:
                    if not project_type:
                        project_type += i
                    else:
                        project_type += "/" + i
            if department_tags:
                for department in department_tags:
                    if not department_type:
                        department_type += department
                    else:
                        department_type += "/" + department
            try:
                re_time = datetime.datetime.fromtimestamp(one_scan['_source'].get("post_time") / 1000)
                time_str = re_time.strftime('%Y-%m-%d %H:%M:%S')
            except:
                time_str = ""

            line_dict = ({"id": one_scan['_id'],
                          'releaser': one_scan['_source'].get("releaser"),
                          'platform': one_scan['_source'].get("platform"),
                          "post_time": time_str,
                          "releaserUrl": one_scan['_source'].get("releaserUrl"),
                          "releaser_id_str": one_scan['_source'].get("releaser_id_str"),
                          "project_type": project_type,
                          "department_tags": department_type
                          })
            if not line_dict:
                print(line_dict)
            yield line_dict

    else:
        if data_type != "weibo":
            for one_scan in search_res:
                try:
                    re_time = datetime.datetime.fromtimestamp(one_scan['_source'].get("release_time") / 1000)
                    time_str = re_time.strftime('%Y-%m-%d %H:%M:%S')
                    fc_time = datetime.datetime.fromtimestamp(one_scan['_source'].get("fetch_time") / 1000)
                    fc_str = fc_time.strftime('%Y-%m-%d %H:%M:%S')
                except:
                    time_str = ""
                    fc_str = ""

                try:
                    title = one_scan['_source'].get("title")
                    title = title.replace("\n", "")
                    title = title.replace("\r", "")
                except:
                    title = ""
                line_dict = ({"id": one_scan['_id'],
                              'releaser': one_scan['_source'].get("releaser"),
                              'title': title,
                              'platform': one_scan['_source'].get("platform"),
                              'url': one_scan['_source'].get("url"),
                              "release_time": time_str,
                              "releaserUrl": one_scan['_source'].get("releaserUrl"),
                              "releaser_id_str": one_scan['_source'].get("releaser_id_str"),
                              "fetch_time": fc_str,
                              "play_count": one_scan['_source'].get("play_count"),
                              "favorite_count": one_scan['_source'].get("favorite_count"),
                              "comment_count": one_scan['_source'].get("comment_count"),
                              "repost_count": one_scan['_source'].get("repost_count"),
                              "duration": one_scan['_source'].get("duration"),
                              "top": one_scan['_source'].get("top")
                              })
                yield line_dict
        else:
            for one_scan in search_res:
                try:
                    re_time = datetime.datetime.fromtimestamp(one_scan['_source'].get("release_time") / 1000)
                    time_str = re_time.strftime('%Y-%m-%d %H:%M:%S')
                    fc_time = datetime.datetime.fromtimestamp(one_scan['_source'].get("fetch_time") / 1000)
                    fc_str = fc_time.strftime('%Y-%m-%d %H:%M:%S')
                except:
                    time_str = ""
                    fc_str = ""
                try:
                    title = one_scan['_source'].get("wb_text").replace("\r", "").replace("\n", "")
                except:
                    title = ""
                line_dict = ({
                        'ID': one_scan['_source'].get("wb_bowen_id"),
                        "UID": str(one_scan['_source']['UID']),
                        '微博昵称': one_scan['_source'].get("wb_name"),
                        '微博内容': title,
                        'platform': one_scan['_source'].get("platform"),
                        '博文地址': one_scan['_source'].get("wb_bowen_address"),
                        "发布日期": time_str,
                        "抓取时间": fc_str,
                        # "play_count": one_scan['_source'].get("play_count"),
                        "点赞量": one_scan['_source'].get("favorite_count"),
                        "评论量": one_scan['_source'].get("comment_count"),
                        "转发量": one_scan['_source'].get("repost_count"),
                        # "duration": one_scan['_source'].get("duration"),
                        "配图连接": one_scan['_source'].get("video_address"),
                        "博文配图": one_scan['_source'].get("wb_pic"),
                        "博文长度": one_scan['_source'].get("wb_bowen_length"),
                        "短链标题": one_scan['_source'].get("wb_type_text"),
                        "短链播放量-视频": one_scan['_source'].get("wb_video_number"),
                        "短链地址-视频": one_scan['_source'].get("video_address"),
                        "短链地址": one_scan['_source'].get("wb_type_address"),
                        "是否原创": one_scan['_source'].get("wb_father_UID"),
                        "原创账户": one_scan['_source'].get("wb_father_bowen_id"),
                        "博文类型": one_scan['_source'].get("wb_bowen_type"),

                })
                yield line_dict


def summary_to_dic(search_res, fans_count=None, fans_date=None, releaser_id_str=None, platform=None,start_ts=None, end_ts=None, **kwargs):
    extra_dic = kwargs.get("extra_dic")
    if search_res:
        line_dict = {}
        video_num = search_res['hits']['total']
        play_count_sum = search_res['aggregations']['sum_play']['value']
        favorite_count_sum = search_res['aggregations']['sum_favorite']['value']
        comment_count_sum = search_res['aggregations']['sum_comment']['value']
        repost_count_sum = search_res['aggregations']['sum_repost']['value']
        month_str = ""
        if start_ts:
            re_time = datetime.datetime.fromtimestamp(start_ts/ 1000)
            month_str += re_time.strftime('%Y-%m-%d') + "至"
        if end_ts:
            re_time = datetime.datetime.fromtimestamp(end_ts / 1000)
            month_str += re_time.strftime('%Y-%m-%d')
        line_dict.update({'releaser': "",
                          'platform': platform,
                          'releaserUrl': "",
                          'video_num': video_num,
                          'play_count_sum': play_count_sum,
                          'favorite_count_sum': favorite_count_sum,
                          'comment_count_sum': comment_count_sum,
                          'repost_count_sum': repost_count_sum,
                          "data_month": month_str,
                          "this_fans": fans_count,
                          "production_org_category": "",
                          "tv_station": "",
                          "releaser_id_str": releaser_id_str,
                          "fans_date": fans_date
                          })
        line_dict.update(extra_dic)
        return line_dict


def purchased_releaser_to_dic(search_res, releaser_is_str=None, line_dict=None,releaser=None,platform=None,releaserUrl=None,**kwargs):
    count_has = 0
    line_dict_new = copy.deepcopy(line_dict)
    for res in search_res:
        count_has = 1
        departments = res["_source"].get("departments")
        department_str = ""
        if departments:
            for department in departments:
                if not department_str:
                    department_str += department
                else:
                    department_str += "/" + department

        try:
            re_time = datetime.datetime.fromtimestamp(res['_source'].get("end_purchase_time") / 1000)
            time_str_end = re_time.strftime('%Y-%m-%d %H:%M:%S')
        except:
            time_str_end = ""
        try:
            re_time = datetime.datetime.fromtimestamp(res['_source'].get("start_purchase_time") / 1000)
            time_str_start = re_time.strftime('%Y-%m-%d %H:%M:%S')
        except:
            time_str_start = ""
        line_dict_new = {
                      'releaser': releaser,
                      'platform': platform,
                      "start_purchase_time": time_str_start,
                      "end_purchase_time": time_str_end,
                      "releaserUrl": releaserUrl,
                      "releaser_id_str": releaser_is_str,
                      "is_purchased" : 1,
                      "departments": department_str,
                      "is_purchased_str": res["_source"].get("is_purchased_str")
                      }
        return line_dict_new
    if not count_has:
        line_dict_new = {
                      'releaser': releaser,
                      'platform': platform,
                      "start_purchase_time": "",
                      "end_purchase_time": "",
                      "releaserUrl": releaserUrl,
                      "releaser_id_str": releaser_is_str,
                      "is_purchased": "",
                      "departments": "",
                      "is_purchased_str": ""
                      }
        return line_dict_new


def releaser_to_dic(search_res, releaser_is_str=None, line_dict=None, fans=None, fc_str=None, **kwargs):
    count_has = 0
    line_dict_list = []
    for res in search_res:
        line_dict_new = copy.deepcopy(line_dict)
        count_has = 1
        print(res)
        video_num = res['_source']['video_num']
        play_count_sum = res['_source']['play_count_sum']
        favorite_count_sum = res['_source']['favorite_count_sum']
        comment_count_sum = res['_source']['comment_count_sum']
        repost_count_sum = res['_source'].get('repost_count_sum')
        ranking_by_play_count_sum = res['_source'].get('ranking_by_play_count_sum')
        stats_type = res['_source']['stats_type']
        _id = res['_id']
        line_dict_new.update({
                'video_num': video_num,
                'play_count_sum': play_count_sum,
                'favorite_count_sum': favorite_count_sum,
                'comment_count_sum': comment_count_sum,
                "stats_type": stats_type,
                "repost_count_sum": repost_count_sum,
                "ranking_by_play_count_sum": ranking_by_play_count_sum,
                "_id": _id,
                "releaser_followers_count": fans,
                "this_fans": fc_str,
                "releaser_id_str": releaser_is_str
        })
        yield line_dict_new
    if not count_has:
        new_dic = {
                'video_num': 0,
                'play_count_sum': 0,
                'favorite_count_sum': 0,
                'comment_count_sum': 0,
                "stats_type": "new_released",
                "repost_count_sum": 0,
                "releaser_id_str": releaser_is_str,
        }
        new_dic.update(line_dict)
        ob_dic = {
                'video_num': 0,
                'play_count_sum': 0,
                'favorite_count_sum': 0,
                'comment_count_sum': 0,
                "stats_type": "observed",
                "repost_count_sum": 0,
                "releaser_id_str": releaser_is_str,
                "releaser_followers_count": fans,
                "this_fans": fc_str,
        }
        ob_dic.update(line_dict)
        line_dict_list.append(new_dic)
        line_dict_list.append(ob_dic)
        for dic in line_dict_list:
            yield dic


def get_follwer_num(releaser_id_str=None, ts=None):
    search_short_body_fans = {
            "query": {
                    "bool": {
                            "filter": [
                                    {"term": {"releaser_id_str.keyword": releaser_id_str}},
                                    {"range": {"fetch_time": {"gte": ts}}}
                            ]
                    }
            }, "sort": [
                    {
                            "timestamp": {
                                    "order": "asc"
                            }
                    }
            ]
    }

    try:
        fans_count = es.search(index="releaser_fans", doc_type="doc", body=search_short_body_fans)
        fans = fans_count["hits"]["hits"][0]["_source"]["releaser_followers_count"]
        fans_date = datetime.datetime.fromtimestamp(
                fans_count["hits"]["hits"][0]["_source"]["fetch_time"] / 1000)
        fc_str = fans_date.strftime('%Y-%m-%d %H:%M:%S')
    except:
        fans = None
        fc_str = None
    return fans, fc_str


def func_es_to_csv(file_path, read_index, read_doc, year=None, cycle="week", cycle_num=None, project_type="short_video",
                   data_type="detail",
                   start_ts=None, end_ts=None, save_file_path="", file_task_name=None, this_cycle=None, **kwargs):
    all_dic = {
            "toutiao": [],
            "抖音": [],
            "腾讯视频": [],
            "腾讯新闻": [],
            "haokan": [],
            "miaopai": [],
            "网易新闻": [],
            "new_tudou": [],
            "pearvideo": [],
            "kwai": [],
            "weibo": [],
            "weixin": [],
            "tencent_combine": [],
            "purchased_releasers": [],
    }
    all_plant = []
    if not cycle_num:
        cycle_num = this_cycle

    if file_path:
        try:
            f = open(file_path, 'r', encoding="gb18030")
            head = f.readline()
            head_list = head.strip().split(',')
        except:
            f = file_path
        for line, i in enumerate(f):
            if type(file_path) != list:
                try:
                    line_list = i.strip().split(',')
                    line_dict = dict(zip(head_list, line_list))
                except:
                    line_dict = f
            else:
                line_dict = i
            releaser_id_str = line_dict.get("releaser_id_str")
            platform = line_dict.get("platform")
            releaserUrl = line_dict.get("releaserUrl")
            releaser = line_dict.get("releaser")
            if not releaser_id_str:
                releaser_id_str = line_dict.get("releaser_id")
            if not releaser_id_str:
                releaser_id = get_releaser_id(platform=platform, releaserUrl=releaserUrl)
                if releaser_id:
                    if platform != "weixin" and platform != "weibo":
                        releaser_id_str = platform + '_' + releaser_id
                    else:
                        releaser_id_str = releaser_id
                else:
                    releaser_id_str = ""
            search_body = create_search_body(year=year, cycle=cycle, cycle_num=cycle_num, project_type=project_type,
                                             data_type=data_type, start_ts=start_ts, end_ts=end_ts, platform=platform,
                                             releaser=releaser, releaser_id_str=releaser_id_str,**kwargs)
            print(search_body)
            search_res = scan(es, query=search_body, index=read_index, doc_type=read_doc,preserve_order=True)

            # file_name = "%s_%s_%s_%s_%s" % (project_type, year, cycle, cycle_num, data_type)
            if data_type == "detail":
                for data in detail_to_dic(search_res, data_type=project_type):
                    all_plant.append(data)
                    all_dic[line_dict["platform"]].append(data)
            else:
                if project_type == "all-time" or cycle == "all-time":
                    fans, fc_str = get_follwer_num(releaser_id_str=releaser_id_str, ts=end_ts)
                    search_res = es.search(body=search_body, index=read_index, doc_type=read_doc)
                    res_dic = summary_to_dic(search_res, fans_count=fans, fans_date=fc_str, releaser=releaser,
                                             releaser_id_str=releaser_id_str, releaserUrl=releaserUrl,
                                             platform=platform, extra_dic=line_dict,start_ts=start_ts, end_ts=end_ts)
                    all_plant.append(res_dic)
                elif project_type == "purchased_releasers":
                    res_dic = purchased_releaser_to_dic(search_res,releaser=releaser,platform=platform,releaserUrl=releaserUrl,releaser_is_str=releaser_id_str,line_dict=line_dict)
                    all_plant.append(res_dic)
                else:
                    fans, fc_str = get_follwer_num(releaser_id_str=releaser_id_str, ts=end_ts)
                    for dic in releaser_to_dic(search_res, line_dict=line_dict, fans=fans, fc_str=fc_str,
                                               releaser_is_str=releaser_id_str):
                        all_plant.append(dic)
        if data_type == "detail":
            if project_type != "weibo":
                columns = ["id", 'releaser', 'platform', 'title', "releaserUrl", 'url', 'releaser_id_str',
                           "release_time",
                           "fetch_time",
                           "play_count",
                           "favorite_count",
                           "comment_count", "repost_count", "duration", "top"
                           ]
            else:
                columns = ['ID', "UID",
                           '微博昵称',
                           '微博内容',
                           'platform',
                           '博文地址',
                           "发布日期",
                           "抓取时间",
                           "点赞量",
                           "评论量",
                           "转发量",
                           "配图连接",
                           "博文配图",
                           "博文长度",
                           "短链标题",
                           "短链播放量-视频",
                           "短链地址-视频",
                           "短链地址",
                           "是否原创",
                           "原创账户",
                           "博文类型",
                           ]
        else:
            if project_type == "purchased_releasers":
                columns = ['releaser','platform',"start_purchase_time","end_purchase_time","releaserUrl","releaser_id_str","is_purchased","departments","is_purchased_str"]
            else:
                columns = ['releaser',
                       "ranking_by_play_count_sum",
                       "total_releaser_num_on_platform",
                       'platform',
                       "video_num",
                       "play_count_sum",
                       "play_count_sum_total_on_platform",
                       "video_num_sum_total_on_platform",
                       "favorite_count_sum",
                       "comment_count_sum",
                       "repost_count_sum",
                       "releaser_followers_count",
                       "this_fans",
                       "last_fans",
                       "fans_change",
                       "data_year",
                       "data_month",
                       "stats_type",
                       "production_org_category",
                       "releaserUrl",
                       "releaser_id_str",
                        "fans_date",
                       "top"
                       ]
        for d in line_dict:
            if d not in columns:
                columns.append(d)
        if not all_plant:
            return save_file_path + file_task_name
        data = pd.DataFrame(all_plant)
        s = datetime.datetime.now()
        ss = str(s)[0:19].replace(' ', '-').replace(':', '-')
        try:
            os.mkdir(save_file_path + file_task_name)
        except Exception as e:
            print("save error 716", e)
            pass
        save_path = '%s%s/%s_%s_%s_%s.csv' % (
        save_file_path, file_task_name, file_task_name, project_type, data_type, ss)
        print(save_path)
        data.to_csv(save_path, encoding='gb18030', columns=columns)
        if data_type == "detail" and project_type == "short_video":
            for d in all_dic:
                if all_dic[d]:
                    data = pd.DataFrame(all_dic[d])
                    s = datetime.datetime.now()
                    ss = str(s)[0:19].replace(' ', '-').replace(':', '-')
                    data.to_csv('%s/%s_%s_%s.csv' % (save_file_path + file_task_name, file_task_name, d, ss),
                                encoding='gb18030', columns=columns)
        time.sleep(0.5)
        return save_file_path + file_task_name
    else:
        search_body = kwargs.get("search_body")
        search_res = scan(es, query=search_body, index=read_index, doc_type=read_doc)
        file_name = "%s_%s" % (project_type, data_type)
        if data_type == "detail":
            for data in detail_to_dic(search_res, data_type=project_type):
                if data:
                    all_plant.append(data)

        data = pd.DataFrame(all_plant)
        s = datetime.datetime.now()
        ss = str(s)[0:19].replace(' ', '-').replace(':', '-')
        try:
            os.mkdir(save_file_path + file_task_name)
        except Exception as e:
            print(e)
        save_path = '%s%s/%s_%s.csv' % (save_file_path, file_task_name, file_name, ss)
        print(save_path)
        data.to_csv(save_path, encoding='gb18030')
        return save_file_path + file_task_name


def timestamp_range(start_timestamp, end_timestamp):
    time_stamp_list = []
    start_date = datetime.datetime.fromtimestamp(start_timestamp / 1e3)
    start_year = start_date.year
    start_month = start_date.month
    end_date = datetime.datetime.fromtimestamp(end_timestamp / 1e3)
    if start_date.day == 1 and end_date.day == 1:
        while start_timestamp <= end_timestamp:
            time_stamp_list.append(start_timestamp)
            if start_month == 12:
                start_month = 1
                start_year += 1
            else:
                start_month += 1
            start_date = datetime.datetime(year=start_year, month=start_month, day=1)
            start_timestamp = int(start_date.timestamp() * 1e3)
        return time_stamp_list
    else:
        return [start_timestamp, end_timestamp]


def es_to_csv(year=None, cycle="week", cycle_num=None, compare_type=None, project_type="short_video",
              data_type_list=["detail"], file_path=None, file_task_name=None, save_file_path=None, start_timestamp=None,
              end_timestamp=None, split_month=False, project_name=None, **kwargs):
    """
    :param year: int 年份 可为空
    :param cycle: str 周期 week/month/quarter/year/(all-time 微信使用自定义时间)/
    :param cycle_num: int  周期 1/2/3/4/5/6
    :param compare_type:默认为空 可不填
    :param project_type:str 数据平台 short_video/weibo/weixin/all-time/target_relasers/purchased_releasers(data_type_list需为summary)
    :param data_type_list: list 数据类型 ["detail","summary"]
    :param file_path: str 传入文件的路径
    :param file_task_name: str 传入的任务名
    :param split_month: bool 是否进行分月 只能在all-time参数下使用
    :param project_name: Str 如果传入的是项目名,则不使用文件
    :param kwargs:dict search_body = {}  自定义请求体
    :return:
    """
    project_dic = {
            "weixin": [],
            "weibo": [],
            "short_video": [],
            "all-time":[],
            "target_releasers":[],
            "purchased_releasers":[]
    }
    #
    if project_name:
        file_path = get_project_releaser(project_name)
    # file_path = r"D:\wxfile\WeChat Files\litaolemo\FileStorage\File\2020-01\微博模板 (1)84781418.csv"
    if not save_file_path:
        save_file_path = "/opt/code/dataManagementSys/dataManagementSys/dataManagementSys/static/zipfile/"
        # save_file_path = r"G:\releaserSystem\dataManagementSys\dataManagementSys\static\zipfile\\"
    if not year:
        year = datetime.datetime.now().year

    try:
        f = open(file_path, 'r', encoding="gb18030")
        head = f.readline()
        head_list = head.strip().split(',')
    except:
        f = file_path
    if file_path:
        for line, i in enumerate(f):
            if type(file_path) != list:
                try:
                    line_list = i.strip().split(',')
                    line_dict = dict(zip(head_list, line_list))
                except:
                    line_dict = f
            else:
                line_dict = i
            if line_dict["platform"] == "weixin":
                project_dic["weixin"].append(line_dict)
            elif line_dict["platform"] == "weibo":
                project_dic["weibo"].append(line_dict)
            if project_type == "purchased_releasers":
                project_dic["purchased_releasers"].append(line_dict)
            else:
                if project_type == "short_video":
                    project_dic["short_video"].append(line_dict)
                else:
                    project_dic["all-time"].append(line_dict)
    else:
        if project_type == "target_releasers":
            project_dic["target_releasers"].append(None)

    for platform_type in project_dic:
        if not project_dic[platform_type]:
            continue
        if platform_type == "weixin" or platform_type == "weibo":
            project_type_temp = platform_type
        else:
            project_type_temp = project_type
        if project_type != platform_type:
            continue
        for data_type in data_type_list:
            index, doc, start_ts, end_ts, this_cycle = create_index(year=year, cycle=cycle, cycle_num=cycle_num,
                                                                    compare_type=compare_type,
                                                                    project_type=project_type_temp,
                                                                    data_type=data_type, first_day_ts=start_timestamp,
                                                                    last_day_ts=end_timestamp, **kwargs)
            print(index, doc, start_ts, end_ts, this_cycle)
            try:
                if not split_month:
                    if start_timestamp and end_timestamp:
                        start_ts = start_timestamp
                        end_ts = end_timestamp
                    try:
                        if platform_type == "target_releasers":
                            project_dic[platform_type] = []
                        res_path = func_es_to_csv(project_dic[platform_type], index, doc, year=year, cycle=cycle,
                                                  cycle_num=cycle_num,
                                                  project_type=project_type_temp,
                                                  data_type=data_type, start_ts=start_ts, end_ts=end_ts,
                                                  save_file_path=save_file_path, file_task_name=file_task_name,
                                                  this_cycle=this_cycle, **kwargs)
                    except Exception as e:
                        print("zip file eroor ",e)
                else:
                    month_timenstamp_list = timestamp_range(start_timestamp, end_timestamp)
                    for s, start_ts in enumerate(month_timenstamp_list):
                        if s < len(month_timenstamp_list) - 1:
                            try:
                                res_path = func_es_to_csv(project_dic[platform_type], index, doc, year=year, cycle=cycle,
                                                          cycle_num=cycle_num,
                                                          project_type=project_type_temp,
                                                          data_type=data_type, start_ts=start_ts,
                                                          end_ts=month_timenstamp_list[s + 1],
                                                          save_file_path=save_file_path, file_task_name=file_task_name,
                                                          this_cycle=this_cycle, **kwargs)
                                time.sleep(2)
                            except Exception as e:
                                print("zip file eroor ", e)
            except Exception as e:
                print("line 769 eroor", e)
                # zipDir(res_path, save_file_path + "down_task/" + file_task_name)
                # raise FileNotFoundError("file didn't find")
        time.sleep(3)
    return zipDir(res_path, save_file_path + "down_task/" + file_task_name)


def write_csv_task_to_redis(year=None, cycle="week", cycle_num=None, compare_type=None, project_type="short_video",
                            data_type_list=["detail"], file_path=None, file_task_name=None, save_file_path=None,
                            start_timestamp=None, end_timestamp=None, split_month=False, email_group=[],
                            project_name=None, **kwargs):
    """
        :param year: int 年份 可为空
        :param cycle: str 周期 week/month/quarter/year
        :param cycle_num: int  周期 1/2/3/4/5/6
        :param compare_type:默认为空 可不填
        :param project_type:str 数据平台 short_video/weibo/weixin/all-time/target_relasers
        :param data_type_list: list 数据类型 ["detail","summary"]
        :param file_path: str 传入文件的路径
        :param file_task_name: str 传入的任务名
        :param split_month: bool 是否进行分月 只能在all-time参数下使用
        :param project_name: Str 传入项目标签
        :param kwargs:dict search_body = {}  自定义请求体
        :return:
        """
    now = int(datetime.datetime.now().timestamp() * 1e3)
    mapping_dic = {
            "year": year,
            "cycle": cycle,
            "cycle_num": cycle_num,
            "compare_type": compare_type,
            "project_type": project_type,
            "data_type_list": data_type_list,
            "file_path": file_path,
            "file_task_name": file_task_name,
            "save_file_path": save_file_path,
            "start_timestamp": start_timestamp,
            "end_timestamp": end_timestamp,
            "split_month": split_month,
            "email_group": email_group,
            "project_name": project_name,
            "kwargs": kwargs
    }
    for k in mapping_dic:
        mapping_dic[k] = json.dumps(mapping_dic[k])
    file_name_msg = rds.hmset(file_task_name + "%s_csv" % now, mapping_dic)
    csv_task_msg = rds.rpush("csv_task", file_task_name + "%s_csv" % str(now))
    res = rds.hgetall(file_task_name + "%s_csv" % now)
    if res:
        print(res)
        print("file_name_msg", file_name_msg)
        print("csv_task_msg", csv_task_msg)
        return True
    else:
        return False


def get_task_name_from_redis():
    res = rds.llen("csv_task")
    if res != 0:
        data_dic = {}
        one_project_name = rds.lpop("csv_task")
        time.sleep(3)
        project_data = rds.hgetall(one_project_name)
        rds.delete(one_project_name)
        for k in project_data:
            data_dic[k] = json.loads(project_data[k])
        task_name = data_dic.get("file_task_name")
        print(data_dic)
        try:
            res_path = es_to_csv(year=data_dic.get("year"),
                                 cycle=data_dic.get("cycle"),
                                 cycle_num=data_dic.get("cycle_num"),
                                 compare_type=data_dic.get("compare_type"),
                                 project_type=data_dic.get("project_type"),
                                 data_type_list=data_dic.get("data_type_list"),
                                 file_path=data_dic.get("file_path"),
                                 file_task_name=data_dic.get("file_task_name"),
                                 save_file_path=data_dic.get("save_file_path"),
                                 start_timestamp=data_dic.get("start_timestamp"),
                                 end_timestamp=data_dic.get("end_timestamp"),
                                 split_month=data_dic.get("split_month"),
                                 email_group=data_dic.get("email_group"),
                                 project_name=data_dic.get("project_name"),
                                 **data_dic.get("kwargs")
                                 )
        except Exception as e:
            print("save error ", e)
            mapping_dic = {
                    "task_name": task_name,
                    "file_path": None,
                    "data_str": "导出任务 %s 失败" % task_name,
                    "email_group": data_dic.get("email_group"),
                    "email_msg_body_str": "问好:\n        导出任务 %s 失败\n        库中没有对应数据" % task_name,
                    "title_str": "导出任务 %s 失败" % task_name,
                    "cc_group": ["litao@csm.com.cn", "zhouyujiang@csm.com.cn", "gengdi@csm.com.cn"],
                    "sender": data_dic.get("email_group")[0]
            }

            delete_task_form_redis(task_name, mapping_dic, "", hasdata=False)
            return None
        mapping_dic = {
                "task_name": task_name,
                "file_path": None,
                "data_str": "导出任务 %s 已完成" % task_name,
                "email_group": data_dic.get("email_group"),
                "email_msg_body_str": "问好:\n        导出任务 %s 已完成" % task_name,
                "title_str": "导出任务 %s 已完成" % task_name,
                "cc_group": ["litao@csm.com.cn", "liushuangdan@csm.com.cn", "gengdi@csm.com.cn"],
                "sender": data_dic.get("email_group")[0]
        }
        print(task_name, mapping_dic, res_path)
        delete_task_form_redis(task_name, mapping_dic, res_path, hasdata=True)
        return res_path
    else:
        return None


def delete_task_form_redis(one_project_name, email_dic, csv_path, hasdata=True):
    print(one_project_name)
    rds.delete(one_project_name)
    dump_str = json.dumps(
            {"timestamp": int(datetime.datetime.now().timestamp() * 1e3), "csv_path": csv_path, "has_data": hasdata})
    rds.hset("csv_task_task_dowm", one_project_name, dump_str)
    if email_dic:
        write_email_task_to_redis(**email_dic)


def get_csv_task_down():
    res = rds.hgetall("csv_task_task_dowm")
    if res:
        for key in res:
            rds.hdel("csv_task_task_dowm", key)
            rds.delete(key)
        return res
    else:
        return None


def do_task():
    now = datetime.datetime.now()
    while True:
        try:
            if rds.llen("csv_task") != 0:
                get_task_name_from_redis()
            else:
                now = datetime.datetime.now()
                print(now.strftime("%Y-%m-%d %H:%M:%S"))
                time.sleep(5)
        except Exception as e:
            print(e)
            continue


# while True:
#     do_task()

if __name__ == "__main__":
    # es_to_csv(year=2019, cycle="month", project_type="purchased_releasers", data_type_list=["summary"],
    #           file_path=r"D:\work_file\发布者账号\一次性需求附件\CCR2月以来采购账号.csv", save_file_path=r"./", file_task_name="156",
    #          )
    # a =  {'kwargs': {}, 'split_month': False, 'project_name': None, 'email_group': ['gengdi@csm.com.cn'], 'start_timestamp': None, 'project_type': 'purchased_releasers', 'compare_type': None, 'save_file_path': '/opt/code/dataManagementSys/dataManagementSys/dataManagementSys/static/zipfile/down_task/', 'cycle_num': None, 'file_path': '/opt/code/dataManagementSys/dataManagementSys/dataManagementSys/static/purchaseCsv/', 'year': None, 'data_type_list': ['summary'], 'file_task_name': 'is_purchase_gengdi2020412', 'cycle': 'week', 'end_timestamp': None}
    #
    # write_csv_task_to_redis(**a)
    # write_csv_task_to_redis(year=2019, cycle="month",cycle_num=9,project_type="target_releasers",data_type_list=["detail"],save_file_path=r"E:\test",file_task_name="1789",start_timestamp=1567267200000,end_timestamp=1572537600000,email_group=["litao@csm.com.cn",],search_body={
    #       "query": {
    #         "bool": {
    #           "filter": [
    #              {"term": {"releaser.keyword": "BTV财经"}}
    #             ]
    #         }
    #       },"sort": [
    #         {
    #           "timestamp": {
    #             "order": "desc"
    #           }
    #         }
    #       ]
    #     })
    # get_task_name_from_redis()
    # time.sleep(2)
    # print(get_csv_task_down())
    ###  写入redis任务 调用 write_csv_task_to_redis()
    ### 查询完成状态  调用 get_csv_task_down()
    # print(month_num(year=2020,cycle_num=1))
    # print(month_num(year=2019,cycle_num=12))
    # do_task()
    # get_csv_task_down()r
    # dic = {'start_timestamp': 1577808000000, 'cycle': 'week', 'compare_type': None, 'project_type': 'all-time', 'year': None,
    #  'file_path': r'D:\wxfile\WeChat Files\litaolemo\FileStorage\File\2020-04\brief2-影视综账号抖音快手.csv',
    #  'end_timestamp': 1585670400000,
    #  'save_file_path': './',
    #  'data_type_list': ['summary', 'detail'], 'file_task_name': '306_天津brief二影视综短视频一至三月', 'email_group': [],
    #  'cycle_num': None, 'project_name': None, 'split_month': True, 'kwargs': {}}
    # es_to_csv(**dic)
    do_task()
    # res= es_to_csv(email_group=[],project_type='weixin',
    #           compare_type=None, save_file_path="./", cycle_num=2,
    #          year=2020,project_name="中山",
    # data_type_list=['summary','detail'], file_task_name= '中山', cycle='all-time',start_timestamp=1577808000000,end_timestamp=1580486400000)
    # dic = {'kwargs': {'search_body': {'query': {'bool': {'filter': [], 'should': [{'match': {'project_tags.keyword': '城市媒体融合'}}], 'minimum_should_match': 1}}}}, 'split_month': False, 'project_name': None, 'email_group': ['litao@csm.com.cn'], 'start_timestamp': None, 'project_type': 'target_releasers', 'compare_type': None, 'save_file_path': './', 'cycle_num': None, 'file_path': None, 'year': None, 'data_type_list': ['detail'], 'file_task_name': 'litao202048', 'cycle': 'week', 'end_timestamp': None}
    # es_to_csv(**dic)
    # res = timestamp_range(1577808000000,1585670400000)
    # print(res)