# coding=utf-8
from __future__ import absolute_import, unicode_literals
from __future__ import print_function

import codecs
import collections
import datetime
import hashlib
import json
import os
import random
import requests
import time
import urllib
import urllib2
from datetime import timedelta
from xml.etree import ElementTree

from celery import shared_task
from django.conf import settings
from django.core.mail import send_mail
from django.db.models import Avg
from django.db.models import Count
from django.db.models import Q
from django.db.models import Min, Max
from django.http import HttpResponse
from django.utils import timezone
from gm_protocol import GmProtocol
from gm_types.doctor import DOCTOR_TITLE_TYPE
from gm_types.gaia import (
    RESERVATION_STATUS,
    COUPON_GIFT_TYPES,
    DOCTOR_TYPE,
    RESERVATION_TYPE,
    PLATFORM_CHOICES,
    TOPIC_TYPE,
    MEMBERSHIP_LEVEL,
    DIARY_CONTENT_LEVEL,
    SERVICE_SELL_TYPE,
    ORDER_CATEGORY,
    MOMO_STAT_LOG_TYPE,
    PLATFORM_CHANNEL,
    BENEFIT_TYPE
)
from gm_types.push import PUSH_INFO_TYPE
from gm_types.push import AUTOMATED_PUSH

from api.models import COUPON_STATUS
from api.models import CouponInfo, CouponGift, GiftToCoupon, Coupon, DoctorCouponInfoRecord
from api.models import (
    PAYMENT_TYPE,
    ORDER_STATUS_USED,
    Tag,
    TAG_TYPE,
    Reservation,
    Doctor,
    STAGE_STATUS,
    REFUND_STATUS,
    Order,
    RefundOrder,
    Service,
    ORDER_STATUS,
    Hospital,
    Special,
    SpecialItem,
    ServiceItem,
    SeckillNotify,
    SmartRank,
    FamousDoctorClient,
    WikiStruct,
    WIKI_LEVEL,
    WikiAdver,
    SERVICE_ORDER_TYPE,
    QQCouponRecord,
    UserExtra,
    ServiceComment,
    OrderCategory,
    College,
)
from api.models import Person
from api.tasks.push_task import allocate_push_task_one, automate_push_task
from api.tool import notification_tool
from api.tool.city_tool import native_and_oversea_area, NATIVE_OVERSEA_KEY, \
    NATIVE_OVERSEA_KEY_v2
from api.tool.coupon_tool import can_qq_coupon_verificate, get_coupon_gift_id_for_doctor_detail
from api.tool.datetime_tool import add_day
from api.tool.filter_tool import (
    COMMON_AREA_CACHE_KEY,
    COMMON_TAG_CACHE_KEY,
    SERVICE_AREA_CACHE_KEY,
    SERVICE_TAG_CACHE_KEY,
    DOCTOR_AREA_CACHE_KEY,
    SERVICE_AREA_CACHE_KEY_V1,
    SERVICE_AREA_CACHE_KEY_V2,
    DOCTOR_AREA_CACHE_KEY_V1,
    DOCTOR_AREA_CACHE_KEY_V2,
    COMMON_AREA_CACHE_KEY_V1
)
from api.tool.filter_tool import build_special_area_filter
from api.tool.filter_tool import build_special_tag_filter
from api.tool.filter_tool import generate_area_filter_by_doctors, generate_doctor_area_filter, \
    generate_doctor_area_filter_v2
from api.tool.filter_tool import generate_area_filter_by_service, generate_service_area_filter, \
    generate_service_area_filter_v2
from api.tool.filter_tool import generate_tag_filter_by_service
from api.tool.filter_tool import get_all_areas, get_all_areas_v1
from api.tool.filter_tool import get_all_tags
from api.tool.jike_tool import Jike
from api.tool.log_tool import (
    logging_exception,
    info_logger,
    momo_stat_logger,
    cache_skus_logger,
)
from api.tool.order_tool import (
    send_remind_to_evaluate_order,
    send_repurchase_reminder_msg,
)
from api.tool.push_tool import NotifyManager
from api.tool.user_tool import AddFansConfig
from gaia.routers import thread_local
from pay.tasks.alter_task import remind_coupon_expire_alert
from rpc.all import get_rpc_remote_invoker
from rpc.cache import (
    doctor_service_tags_distribution_cache as doctor_cache,
    hospital_service_tags_distribution_cache as hospital_cache
)
from rpc.cache import (
    filter_cache,
    hospital_doctor_service_cache,
    tag_child_tags_cache,
    itemwiki_cache,
    tag_top_sale_cache,
    service_home_city_cache,
    service_score_tag_avg,
    sleep_user_cache,
    citycache,
    recommend_college_cache,
)
from rpc.tool.log_tool import period_task_logger
from search.utils.service import filter_service
from services.custom_phone_service import (
    Phone400Service,
)
from sms.utils.smsfactory import send_sms
from social.models import SocialInfo
# TODO zhangyunyu
from talos import TagService
from talos.models.topic import TopicReply
from talos.tools.push_tool import special_push_limit
from utils.qq_connect_verify import QQConnect


@shared_task
def sms_service(debug=False):
    """
        短信服务剩余条数
    """
    data = {}
    data['sn'] = u'SDK-BBX-010-19243'
    data['pwd'] = u'f4e-8016'
    url = "http://sdk.entinfo.cn:8060/webservice.asmx/GetBalance"
    para_data = urllib.urlencode(data)
    f = urllib2.urlopen(url, para_data)
    content = f.read()
    period_task_logger.info('sms service test')
    r = ElementTree.fromstring(content)
    num_left = int(r.text)
    message = u'短信服务剩余条数:{}'.format(num_left)
    if debug:
        period_task_logger.info(u'短信服务剩余条数:{}'.format(num_left))
    if num_left < 500:
        send_mail(u'充值提醒', message, 'doctor@wanmeizhensuo.com',
                  ['linlin@wanmeizhensuo.com', 'xuepengfei@wanmeizhensuo.com'])
    return HttpResponse(num_left)


@shared_task
def remind_doctor_reservation():
    """ 给医生未处理预约的提醒
    """
    unhandles = Reservation.objects.filter(status=RESERVATION_STATUS.RESERVING)
    if not unhandles:
        return

    doctors = {}
    for unhandle in unhandles:
        doctor_id = unhandle.schedule.doctor.id
        if doctor_id not in doctors:
            doctors[doctor_id] = 1
        else:
            doctors[doctor_id] += 1

    for doctor_id, num in doctors.iteritems():
        doctor = Doctor.objects.get(pk=doctor_id)
        try:
            send_sms(doctor.user.person.phone, 12,
                     [{'doctor_name': doctor.name}, {'number': num}])
        except:
            logging_exception()


@shared_task
def remind_order_expired_before_15_days():
    # 这个方法已经被废弃
    return

    days = 15
    today = datetime.date.today()
    fifteen_days_later = today + timedelta(days=days)

    orders = Order.objects.filter(status=ORDER_STATUS.PAID).filter(
        expired_date__range=(today, fifteen_days_later)
    )

    if not orders.count():
        return

    for order in orders:
        service_snapshot = json.loads(order.service_snapshot)
        payment_type = service_snapshot.get('payment_type')
        if payment_type not in (PAYMENT_TYPE.FULL_PAYMENT, PAYMENT_TYPE.PREPAYMENT):
            continue

        # update expired_date
        days = order.service.valid_duration or 30
        order.expired_date += datetime.timedelta(days=days)
        order.save()


@shared_task
def query_and_set_order_stage_status():
    """
    查询 GX0304 接口，并设置api-order数据库 'stage_status' 和 'stage_app_id' 字段值
    """
    step = 10
    jike = Jike(
        url=settings.JIKE_URL,
        user=settings.JIKE_SERVICE_ID,
        jike_hash=settings.JIKE_SERVICE_HASH,
        userid=settings.JIKE_USERID
    )

    orders = Order.objects.filter(is_stage=True).exclude(
        stage_status__in=[STAGE_STATUS.DECL, STAGE_STATUS.POFF])
    order_groups = [orders[i:i + step] for i in xrange(0, len(orders), step)]

    for group in order_groups:
        order_ids = [order.id for order in group]
        ids = ','.join(map(str, order_ids))
        jike_data = jike.get_jike_info_by_order_ids(order_ids=ids)
        result = jike_data['result']

        for data in result:
            try:
                order = Order.objects.get(pk=int(data['ORDER_ID']))
                order.stage_status = STAGE_STATUS[data['APP_STATUS']]
                if not order.stage_app_id:
                    order.stage_app_id = data['APP_ID']
                order.save()
            except Order.DoesNotExist:
                logging_exception()


@shared_task
def order_evaluate_remind():
    now = timezone.now()
    days = settings.ORDER_EVALUATE_REMIND_TIME
    start_time = now - timedelta(days=days + 1)
    end_time = now - timedelta(days=days)
    orders = Order.objects.filter(
        validate_time__range=(start_time, end_time),
        status__in=ORDER_STATUS_USED
    )
    for order in orders:
        try:
            diary = order.diary
        except:
            continue

        if diary.has_commented:
            continue

        # send push/notification
        send_remind_to_evaluate_order(order)


@shared_task
def remind_doctor_refund():
    unhandle_refunds = RefundOrder.objects.filter(
        status=REFUND_STATUS.PROCESSING)
    refund_list = {}

    for unhandle_refund in unhandle_refunds:
        try:
            doctor = unhandle_refund.order.service.doctor.id
        except:
            logging_exception()
            continue
        if doctor not in refund_list:
            refund_list[doctor] = 0
        refund_list[doctor] += 1

    for doctor_id, times in refund_list.iteritems():
        if not doctor_id or not times:
            continue
        doctor = Doctor.objects.get(pk=doctor_id)
        try:
            phone = doctor.user.person.phone
            send_sms(phone, 10, [{'number': times}])
        except:
            logging_exception()


# @shared_task
# def calc_service_rating():
#    online_services_ids = Service.objects.filter(is_online=True).values_list('id', flat=True)
#    online_services_diaries = Diary.objects.filter(service_id__in=online_services_ids)
#    with_rating_diaries = online_services_diaries.exclude(rating=0)
#    total_avg = with_rating_diaries.aggregate(Avg('rating'))['rating__avg']
#
#    services = Service.objects.filter(is_online=True)
#    i = 1
#    for service in services:
#        rating = 0
#        operation_effect = 0
#        doctor_attitude = 0
#        hospital_env = 0
#
#        operation_diaries = Diary.objects.filter(
#            service_id=service.id,
#            operation_effect_rating=0
#        )
#
#        if operation_diaries:
#            calc_operation = operation_diaries.aggregate(
#                Avg('operation_effect_rating')
#            )
#            operation_effect = calc_operation[
#                'operation_effect_rating__avg'
#            ]
#
#        attitude_diaries = Diary.objects.filter(
#            service_id=service.id,
#        ).exclude(
#            doctor_attitude_rating=0
#        )
#
#        if attitude_diaries:
#            calc_attitude = attitude_diaries.aggregate(
#                Avg('doctor_attitude_rating')
#            )
#            doctor_attitude = calc_attitude[
#                'doctor_attitude_rating__avg'
#            ]
#
#        env_diaries = Diary.objects.filter(
#            service_id=service.id,
#        ).exclude(
#            hospital_env_rating=0
#        )
#
#        if env_diaries:
#            calc_env = env_diaries.aggregate(
#                Avg('hospital_env_rating')
#            )
#            hospital_env = calc_env[
#                'hospital_env_rating__avg'
#            ]
#
#        rating_diaries = Diary.objects.filter(
#            service_id=service.id,
#        ).exclude(rating=0)
#        if rating_diaries:
#            calc_rating = rating_diaries.aggregate(Avg('rating'))
#            rating = calc_rating['rating__avg']
#        i += 1
#
#        if rating and operation_effect and doctor_attitude and hospital_env:
#            # WR = (v/(v+m)) * R + (m/(v+m)) * C
#            # WR 美购最终分数
#            # R 美购平均评分
#            # C 所有有评价的美购日记本的平均得分
#            # v 该美购有评价的日记本数
#            # m min[5, v]
#            v = rating_diaries.count() * 1.0
#            R = rating * 1.0
#            C = total_avg * 1.0
#            m = min(5, v) * 1.0
#            service_rating = (v / (v + m)) * R + (m / (v + m)) * C
#
#            service.rating = service_rating
#            service.hospital_env_rating = hospital_env
#            service.doctor_attitude_rating = doctor_attitude
#            service.operation_effect_rating = operation_effect
#            service.save()
#
#
# @shared_task
# def calc_doctor_rating():
#    doctors = Doctor.objects.all()
#    service_w = SERVICE_DIARY_WEIGHT
#    n_service_w = NOT_SERVICE_DIARY_WEIGHT
#
#    # http://wiki.wanmeizhensuo.com/pages/viewpage.action?pageId=1052411
#    for doctor in doctors:
#        service_diarys = Diary.objects.filter(doctor_id=doctor.id, is_online=True, service__isnull=False)
#        service_diary_score = get_score(service_diarys)
#
#        not_service_diarys = Diary.objects.filter(doctor_id=doctor.id, is_online=True, service__isnull=True)
#        not_service_diary_score = get_score(not_service_diarys)
#
#        (rate, env_rating, attitude_rating, effect_rating) = get_last_score(service_diary_score,
#                                                                            not_service_diary_score, service_w,
#                                                                            n_service_w)
#
#        doctor.rate = rate
#        doctor.env_rating = env_rating
#        doctor.attitude_rating = attitude_rating
#        doctor.effect_rating = effect_rating
#        doctor.save()
#
#
# @shared_task
# def calc_hospital_rating():
#    hospitals = Hospital.objects.all()
#    service_w = SERVICE_DIARY_WEIGHT
#    n_service_w = NOT_SERVICE_DIARY_WEIGHT
#
#    for hospital in hospitals:
#        service_diarys = Diary.objects.filter(hospital_id=hospital.id, service__isnull=False)
#        service_diary_score = get_score(service_diarys)
#
#        not_service_diarys = Diary.objects.filter(hospital_id=hospital.id, service__isnull=True)
#        not_service_diary_score = get_score(not_service_diarys)
#
#        (rate, env_rating, attitude_rating, effect_rating) = get_last_score(service_diary_score,
#                                                                            not_service_diary_score, service_w,
#                                                                            n_service_w)
#
#        hospital.rate = rate
#        hospital.env_rating = env_rating
#        hospital.attitude_rating = attitude_rating
#        hospital.effect_rating = effect_rating
#        hospital.save()


def get_score(service_diarys):
    service_diary_score = {
        'rate': {
            'val': float(0),
            'count': 0,
        },
        'env_rating': {
            'val': float(0),
            'count': 0,
        },
        'attitude_rating': {
            'val': float(0),
            'count': 0,
        },
        'effect_rating': {
            'val': float(0),
            'count': 0,
        },
    }

    for service_diary in service_diarys:
        if service_diary.rating > 0:
            service_diary_score['rate']['val'] += service_diary.rating
            service_diary_score['rate']['count'] += 1
        if service_diary.operation_effect_rating > 0:
            service_diary_score['effect_rating']['val'] += service_diary.operation_effect_rating
            service_diary_score['effect_rating']['count'] += 1
        if service_diary.hospital_env_rating > 0:
            service_diary_score['env_rating']['val'] += service_diary.hospital_env_rating
            service_diary_score['env_rating']['count'] += 1
        if service_diary.doctor_attitude_rating > 0:
            service_diary_score['attitude_rating']['val'] += service_diary.doctor_attitude_rating
            service_diary_score['attitude_rating']['count'] += 1

    return service_diary_score


def get_calc_score(service_rating, service_diary_count, not_service_rating, not_service_diary_count, service_w,
                   n_service_w):
    if service_diary_count == 0 and not_service_diary_count == 0:
        return float(0)

        # [m*(x1+x2+.........+xp)+n*(y1+y2+......yq)]/(m*p+n*q)
    rating = (service_w * service_rating + n_service_w * not_service_rating) / \
             (service_w * service_diary_count + n_service_w * not_service_diary_count)

    return rating


def get_last_score(service_diary_score, not_service_diary_score, service_w, n_service_w):
    rate = get_calc_score(
        service_diary_score['rate']['val'], service_diary_score['rate']['count'],
        not_service_diary_score['rate']['val'], not_service_diary_score['rate']['count'],
        service_w, n_service_w
    )
    env_rating = get_calc_score(
        service_diary_score['env_rating']['val'], service_diary_score['env_rating']['count'],
        not_service_diary_score['env_rating']['val'], not_service_diary_score['env_rating']['count'],
        service_w, n_service_w
    )
    attitude_rating = get_calc_score(
        service_diary_score['attitude_rating']['val'], service_diary_score['attitude_rating']['count'],
        not_service_diary_score['attitude_rating']['val'], not_service_diary_score['attitude_rating']['count'],
        service_w, n_service_w
    )
    effect_rating = get_calc_score(
        service_diary_score['effect_rating']['val'], service_diary_score['effect_rating']['count'],
        not_service_diary_score['effect_rating']['val'], not_service_diary_score['effect_rating']['count'],
        service_w, n_service_w
    )
    return (rate, env_rating, attitude_rating, effect_rating)


@shared_task
def repurchase_reminder():
    # get all tags with repurchase days set, if it a subbodypart tag,
    # add it's children tags
    final_tags = {}
    level_2_tags = set()

    tags = Tag.objects.filter(re_purchase_days__gt=0)
    for tag in tags:
        final_tags[tag.id] = tag.re_purchase_days

        if tag.tag_type == TAG_TYPE.BODY_PART_SUB_ITEM:
            level_2_tags.add(tag)

    for tag in level_2_tags:
        subtags = tag.online_child_tags()
        for subtag in subtags:
            if subtag.re_purchase_days > tag.re_purchase_days:
                final_tags[subtag.id] = subtag.re_purchase_days
            else:
                final_tags[subtag.id] = tag.re_purchase_days

    # get all services have tag in final_tags
    services_re_purchase_days = {}
    ss = Service.objects.filter(tags__in=final_tags.keys())
    for s in ss:
        days = max(final_tags.get(x.id, 0) for x in s.tags.all())
        if not days:
            continue

        services_re_purchase_days[s.id] = days

    # we get all the services, now, we can get all orders
    now = timezone.now()
    beginning_of_day = datetime.datetime.combine(now, datetime.time.min)
    end_of_day = datetime.datetime.combine(now, datetime.time.max)

    order_ids_sent_notification = {}

    for s in ss:
        if s.id not in services_re_purchase_days:
            continue

        days = services_re_purchase_days[s.id]
        order_range = (
            beginning_of_day - timedelta(days=days),
            end_of_day - timedelta(days=days)
        )
        orders = s.order_set.filter(
            status__in=ORDER_STATUS_USED,
            validate_time__range=order_range
        )

        for order in orders:
            if order.id in order_ids_sent_notification:
                continue

            info_logger.info('order %s, repurchase: %s' % (order.id, days))
            send_repurchase_reminder_msg(order, days)
            order_ids_sent_notification[order.id] = True


@shared_task
@thread_local(DB_FOR_READ_OVERRIDE=settings.SLAVE_DB_NAME)
def remind_coupon_expire():
    _remind_coupon_expire_according_to_value()

# todo 用了_remind_coupon_expire_according_to_value， 这个不需要的话后面可以删掉
def _remind_coupon_expire_by_days(days):
    today = datetime.datetime.now().date()
    one_day = datetime.timedelta(days=1)
    tomorrow = today + one_day
    remind_days = one_day * days
    will_expire_coupons = CouponInfo.objects.filter(
        status=COUPON_STATUS.CLAIMED).filter(
        end_time__gte=(today + remind_days)).filter(
        end_time__lt=(tomorrow + remind_days)
    )
    extra = {
        "type": PUSH_INFO_TYPE.GM_PROTOCOL,
        "pushUrl": "gengmei://coupon_list",
    }
    result = will_expire_coupons.values('user_id').annotate(dcount=Count('user_id'))
    for dic in result:
        if not special_push_limit(user_id=dic.get('user_id'), push_type='overdue_coupon'):
            continue
        will_expire_coupons_count = dic.get('dcount')
        if not will_expire_coupons:
            continue
        alert = u"您有{}张美券将于{}天后到期，不用会损失上千元哦，心疼啊！把握最后机会，不要后悔～".format(will_expire_coupons_count, days)
        user_id = dic.get('user_id')

        allocate_push_task_one(user_id=user_id, extra=extra, alert=alert,
                               labels={'event_type': 'push', 'event': 'overdue_coupon'})

        remind_coupon_expire_alert.delay(user_id=user_id)


def _remind_coupon_expire_according_to_value():
    today = datetime.datetime.now().date()
    # 先走一边一天的逻辑
    delta_day = datetime.timedelta(days=1)
    tomorrow = today + delta_day
    remind_days = delta_day
    will_expire_coupons = CouponInfo.objects.filter(
        status=COUPON_STATUS.CLAIMED).filter(
        end_time__gte=(today + remind_days)).filter(
        end_time__lt=(tomorrow + remind_days)
    )
    user_ids = set(will_expire_coupons.values_list('user_id', flat=True))

    result = will_expire_coupons.values('user_id', 'coupon_id', 'coupon__name', 'coupon__benefit_type').annotate(
        max_value=Max('coupon__value')).values('user_id', 'max_value', 'coupon_id', 'coupon__name', 'coupon__benefit_type')

    res = []

    def get_user_max_coupon_info(user_id):
        user_id_coupon_infos = filter(lambda x: x["user_id"] == user_id, result)
        if user_id_coupon_infos:
            ret = max(user_id_coupon_infos, key=lambda x: x['max_value'])
        else:
            ret = {}
        return ret

    for user_id in user_ids:
        if not get_user_max_coupon_info(user_id):
            continue
        res.append(get_user_max_coupon_info(user_id))

    extra = {
        'type': PUSH_INFO_TYPE.GM_PROTOCOL,
        'pushUrl': 'gengmei://coupon_list',
    }

    for dic in res:
        user_id = dic.get('user_id')
        value = dic.get('max_value')
        benefit_type = dic.get('coupon__benefit_type')
        name = dic.get('coupon__name')
        if benefit_type == BENEFIT_TYPE.DISCOUNT:
            alert = u"您有一张{}1天后过期，美丽不等人，快来使用吧".format(name)
        else:
            alert = u"您有一张价值{}元的优惠券1天后过期，美丽不等人，快来使用吧".format(value)
        automate_push_task(user_ids=[user_id],
                           push_type=AUTOMATED_PUSH.US_COUPON_EXPIRED,
                           extra=extra,
                           alert=alert,
                           )
        remind_coupon_expire_alert.delay(user_id=user_id)
    # 再走一遍7天的
    delta_day = datetime.timedelta(days=7)
    tomorrow = today + delta_day
    remind_days = delta_day
    will_expire_coupons = CouponInfo.objects.filter(
        status=COUPON_STATUS.CLAIMED).filter(
        end_time__gte=(today + remind_days)).filter(
        end_time__lt=(tomorrow + remind_days)
    ).exclude(user_id__in=user_ids)

    result = will_expire_coupons.values('user_id').annotate(max_value=Max('coupon__value')).values('user_id', 'max_value')
    result = will_expire_coupons.values('user_id', 'coupon_id', 'coupon__name', 'coupon__benefit_type').annotate(
        max_value=Max('coupon__value')).values('user_id', 'max_value', 'coupon_id', 'coupon__name',
                                               'coupon__benefit_type')

    res = []
    extra = {
        'type': PUSH_INFO_TYPE.GM_PROTOCOL,
        'pushUrl': 'gengmei://coupon_list',
    }

    for user_id in user_ids:
        if not get_user_max_coupon_info(user_id):
            continue
        res.append(get_user_max_coupon_info(user_id))

    for dic in res:
        user_id = dic.get('user_id')
        value = dic.get('max_value')
        benefit_type = dic.get('coupon__benefit_type')
        name = dic.get('coupon__name')
        if benefit_type == BENEFIT_TYPE.DISCOUNT:
            alert = u"您有一张{}1天后过期，美丽不等人，快来使用吧".format(name)
        else:
            alert = u"您有一张价值{}元的优惠券7天后过期，快来使用吧~".format(value)
        automate_push_task(user_ids=[user_id],
                           push_type=AUTOMATED_PUSH.US_COUPON_EXPIRED,
                           extra=extra,
                           alert=alert,
                           )
        remind_coupon_expire_alert.delay(user_id=user_id)


@shared_task()
def record_doctor_coupon_info():
    # 1. 找到所有在领取时间的医生礼包
    # 2. 通过礼包得到医生id集合
    # 3. 根据批次每一个医生生成一个详情页可领取礼包请求，然后记录数据
    now = datetime.datetime.now()
    coupon_data_batch = now.strftime('%Y%m%d%H0000')

    all_gift_ids = list(CouponGift.objects.filter(
        start_time__lte=now, end_time__gte=now, gift_type=COUPON_GIFT_TYPES.DOCTOR
    ).values_list('id', flat=True))

    coupon_ids = set(GiftToCoupon.objects.filter(
        coupon_gift_id__in=all_gift_ids, coupon_enable=True
    ).values_list('coupon_id', flat=True))

    doctor_ids = set(Coupon.objects.filter(id__in=coupon_ids).values_list('doctor_id', flat=True).distinct())

    for doctor_id in doctor_ids:
        try_record_doctor_coupon_info.delay(doctor_id, coupon_data_batch)


@shared_task()
def try_record_doctor_coupon_info(doctor_id, coupon_data_batch):
    already_done = DoctorCouponInfoRecord.objects.filter(doctor_id=doctor_id,
                                                         coupon_data_batch=coupon_data_batch).exists()

    if not already_done:
        gift_ids, _, gift_id_to_left_count = get_coupon_gift_id_for_doctor_detail(user=None, doctor_id=doctor_id)
        coupon_id_to_gift_id = {}

        for c in list(GiftToCoupon.objects.filter(
                coupon_gift_id__in=gift_ids, coupon_enable=True
        ).values_list('coupon_id', 'coupon_gift_id')):
            coupon_id_to_gift_id[c[0]] = c[1]

        coupon_ids = coupon_id_to_gift_id.keys()
        coupon_data = list(Coupon.objects.filter(
            id__in=coupon_ids
        ).values_list('id', 'has_threshold', 'prepay_threshold', 'value'))

        doctor = Doctor.objects.get(id=doctor_id)

        business_partener_id = None
        business_partener_user_name = None
        business_group_id = None
        business_group_name = None

        if doctor.business_partener:
            business_partener_id = doctor.business_partener.id
            business_partener_user_name = doctor.business_partener.username

        if doctor.business_group:
            from themis.models import Team
            business_group_id = doctor.business_group
            team = Team.objects.filter(id=doctor.business_group).values_list('id', 'name').first()
            if team:
                business_group_name = team[1]

        to_insert_coupon_record = []

        for coupon_id, has_threshold, prepay_threshold, value in coupon_data:
            if coupon_id in coupon_id_to_gift_id and coupon_id_to_gift_id[coupon_id] in gift_id_to_left_count:
                r = DoctorCouponInfoRecord()
                r.doctor_id = doctor_id
                r.coupon_data_batch = coupon_data_batch
                r.coupon_id = coupon_id
                r.threshold = prepay_threshold if has_threshold else 0
                r.value = value
                r.left_number = gift_id_to_left_count[coupon_id_to_gift_id[coupon_id]]

                r.business_partener_id = business_partener_id
                r.business_partener_user_name = business_partener_user_name
                r.business_group_id = business_group_id
                r.business_group_name = business_group_name

                to_insert_coupon_record.append(r)

        if to_insert_coupon_record:
            DoctorCouponInfoRecord.objects.bulk_create(to_insert_coupon_record)

# todo 功能废弃
@shared_task
def remind_user_reservation():
    # USER-336 http://jira.gengmei.cc/browse/USER-336
    now = datetime.datetime.now()
    tomorrow_start = now + datetime.timedelta(days=1)
    tomorrow_end = now + datetime.timedelta(days=2)
    tomorrow_reservation = Reservation.objects.filter(
        status=RESERVATION_STATUS.ACCEPTED,
        date__range=(
            tomorrow_start.date(),
            tomorrow_end.date()
        ),
    )

    for reservation in tomorrow_reservation:
        # 推送+通知
        # title = u'预约提醒'
        d = reservation.date
        doctor_name = reservation.schedule.doctor.name
        dt_str = u'{}年{}月{}日{}'.format(
            d.year, d.month,
            d.day, d.strftime('%H:%M')
        )
        type_str = RESERVATION_TYPE.getDesc(reservation.reservation_type)
        hospital = reservation.schedule.address.desc
        content = (
            u'尊敬的更美用户您好，您已经预约了' +
            u'{} {}医生的{}，'.format(dt_str, doctor_name, type_str) +
            u'敬请准时到达{}完成预约。如因其他安排取消预约，'.format(hospital) +
            u'请前往更美app我的订单中取消相关预约。'
        )
        gm_protocol = GmProtocol()
        extra = {
            'type': PUSH_INFO_TYPE.GM_PROTOCOL,
            'pushUrl': gm_protocol.get_my_reservation_detail(reservation.id),
        }

        try:
            user = reservation.user

            # 推送
            platform = [PLATFORM_CHOICES.ANDROID, PLATFORM_CHOICES.IPHONE]
            kwargs = {
                'platform': platform,
                'user_id': user.id,
                'extra': extra,
                'alert': content,
            }
            from api.tasks.push_task import allocate_push_task_one
            allocate_push_task_one(**kwargs)

            # 通知
            notification_tool.send_notification(
                uid=user.id, title=u'预约提醒',
                content=content, url=extra['pushUrl']
            )
        except:
            logging_exception()


@shared_task
@thread_local(DB_FOR_READ_OVERRIDE=settings.SLAVE_DB_NAME)
def generate_native_oversea_cities():
    data = native_and_oversea_area()
    citycache.set(NATIVE_OVERSEA_KEY, json.dumps(data))


@shared_task
@thread_local(DB_FOR_READ_OVERRIDE=settings.SLAVE_DB_NAME)
def generate_native_oversea_cities_v2():
    data = native_and_oversea_area(exclude_gat=True)
    citycache.set(NATIVE_OVERSEA_KEY_v2, json.dumps(data))


@shared_task
@thread_local(DB_FOR_READ_OVERRIDE=settings.SLAVE_DB_NAME)
def generate_common_filters():
    """
    生成通用筛选器的tag和area列表
    """
    area = get_all_areas()
    filter_cache.set(COMMON_AREA_CACHE_KEY, json.dumps(area))

    areas = get_all_areas_v1()
    filter_cache.set(COMMON_AREA_CACHE_KEY_V1, json.dumps(areas))

    for with_hot in [True, False]:
        for with_level_3 in [True, False]:
            cache_key = COMMON_TAG_CACHE_KEY.format(with_hot, with_level_3)
            tags = get_all_tags(with_hot=with_hot, with_level_3=with_level_3)
            tags = json.dumps(tags)
            filter_cache.set(cache_key, tags)


@shared_task
@thread_local(DB_FOR_READ_OVERRIDE=settings.SLAVE_DB_NAME)
def generate_special_filters():
    """
    生成专题筛选器的tag和area列表
    """
    filter_end_time = datetime.datetime.now()
    end_time_filter = (
        (Q(end_time__gte=filter_end_time) | Q(end_time__isnull=True)) &
        Q(is_online=True)
    )
    specials = Special.objects.filter(end_time_filter)
    for special in specials:
        build_special_tag_filter(special.id)
        build_special_area_filter(special.id, exclude_gat=True)


@shared_task
@thread_local(DB_FOR_READ_OVERRIDE=settings.SLAVE_DB_NAME)
def generate_service_filters():
    """
    生成美购筛选器的tag和area列表
    """
    services = Service.objects.filter(is_online=True)
    areas = generate_area_filter_by_service(services)
    areas = json.dumps(areas)
    filter_cache.set(SERVICE_AREA_CACHE_KEY, areas)

    areas = generate_service_area_filter(services)
    areas = json.dumps(areas)
    filter_cache.set(SERVICE_AREA_CACHE_KEY_V1, areas)

    areas = generate_service_area_filter_v2(services)
    areas = json.dumps(areas)
    filter_cache.set(SERVICE_AREA_CACHE_KEY_V2, areas)

    tags = generate_tag_filter_by_service(services)
    tags = json.dumps(tags)
    filter_cache.set(SERVICE_TAG_CACHE_KEY, tags)


@shared_task
@thread_local(DB_FOR_READ_OVERRIDE=settings.SLAVE_DB_NAME)
def generate_doctor_filters():
    """
    生成医生筛选器的area列表
    """
    doctors = Doctor.objects.filter(is_online=True)
    areas = generate_area_filter_by_doctors(doctors)
    areas = json.dumps(areas)
    filter_cache.set(DOCTOR_AREA_CACHE_KEY, areas)

    areas = generate_doctor_area_filter(doctors)
    areas = json.dumps(areas)
    filter_cache.set(DOCTOR_AREA_CACHE_KEY_V1, areas)

    areas = generate_doctor_area_filter_v2(doctors)
    areas = json.dumps(areas)
    filter_cache.set(DOCTOR_AREA_CACHE_KEY_V2, areas)


@shared_task
def get_phone_service():
    """
    录音接口
    """
    p4s = Phone400Service()
    today = datetime.date.today()
    yesterday = add_day(today, -1)
    for bgn_date in [yesterday, today]:
        end_time = datetime.datetime(
            bgn_date.year,
            bgn_date.month,
            bgn_date.day,
            23,
            59,
            59,
        )

        for _, client in p4s._instances.iteritems():
            for record in client.fetch_phone_records_iter(bgn_date, end_time):
                try:
                    client.save_record(record)
                except:
                    logging_exception()

                    # phone_service = PhoneService()
                    # results = phone_service.get_status_report()
                    # if results.get('result'):
                    #     for result in results['result']:
                    #         if isinstance(result, dict):
                    #             audio_url = phone_service.get_audio_file(result['ssid'])
                    #             if isinstance(audio_url, dict):
                    #                 if audio_url.get('result') != 'null':
                    #                     audio_url = audio_url['result']
                    #                 else:
                    #                     audio_url = ''

                    #             if audio_url:
                    #                 qiniu_url = get_download_from_400_upload_to_qiniu(audio_url)
                    #             else:
                    #                 qiniu_url = ''

                    #             phone_service_record = PhoneServiceRecord()
                    #             phone_service_record.call_phone = result["call_number"]
                    #             phone_service_record.call_status = result["call_status"]
                    #             phone_service_record.call_time = result["insert_time"]
                    #             phone_service_record.transfer_phone = result["transfer_number"]
                    #             phone_service_record.dest_phone = result["dest_number"]
                    #             phone_service_record.audio_url = qiniu_url
                    #             try:
                    #                 doctor = Doctor.objects.get(phone_ext=result["transfer_number"])
                    #                 phone_service_record.doctor = doctor
                    #             except:
                    #                 pass
                    #             phone_service_record.save()


@shared_task
def get_hosiptal_service_num():
    """
    统计医院底下所有医生对应的美购数量
    """
    hospitals = Hospital.objects.filter()
    for hospital in hospitals:
        service_num = 0
        for doctor in hospital.doctor_hospital.filter(is_online=True):
            service_num = service_num + doctor.services.filter(is_online=True).count()

        cache_info = {
            'service_num': service_num,
        }
        hospital_doctor_service_cache.set(hospital.id, json.dumps(cache_info))


@shared_task
def seckill_notify():
    """
    秒杀提醒任务
    """
    now = datetime.datetime.now()

    # 搜索的时间应该是 [ now + minutes - 10, now + minutes ], 记得时间从小到大
    start_minutes = [10, 60, 180]

    all_special_id_and_minutes = []
    all_special_id_set = set()

    for minutes in start_minutes:
        start = now + datetime.timedelta(minutes=minutes - 10)
        end = now + datetime.timedelta(minutes=minutes)

        push_specials = Special.objects.filter(
            is_online=True,
            is_seckill=True,
            start_time__range=(start, end)
        ).values_list('id', flat=True)

        for sp_id in push_specials:
            if sp_id not in all_special_id_set:
                all_special_id_and_minutes.append((sp_id, minutes))
                all_special_id_set.add(sp_id)

    for special_id, minutes in all_special_id_and_minutes:
        service_ids = list(SpecialItem.objects.filter(special_id=special_id).values_list('service_id', flat=True)
                           .distinct())

        services = Service.objects.filter(id__in=service_ids)

        for service in services:
            unsent_push_notification = []

            if service:
                unsent_q = Q(push_sent=False)
                filter_q = Q(special_id=special_id, service_id=service.id)
                unsent_push_notification = SeckillNotify.objects.filter(filter_q & unsent_q)

            for unsent in unsent_push_notification:

                time_desc = u"{}小时".format(minutes / 60) if minutes % 60 == 0 else u"{}分钟".format(minutes)
                content = u'您关注的“{}”秒杀项目还有{}即将开抢，想先人一步变更美，快点这里摆好开抢姿势！'.format(
                    service.name, time_desc
                )

                gm_protocol = GmProtocol()
                protocol = gm_protocol.seckill

                if not unsent.push_sent:
                    kwargs = {
                        'user_ids': [unsent.person.user.id],
                        'push_type': AUTOMATED_PUSH.PAYMENT_REMINDER,
                        'extra': {
                            "type": PUSH_INFO_TYPE.GM_PROTOCOL,
                            "pushUrl": protocol,
                        },
                        'alert': content,
                    }

                    automate_push_task(**kwargs)
                    unsent.push_sent = True
                    try:
                        print('seckill notify push sent uid:{}'
                              'service_id: {}'.format(
                            unsent.person.user.id, service.id
                        ))
                    except:
                        pass

                unsent.save()


@shared_task
def set_child_tags():
    query = Q(tag_type=TAG_TYPE.BODY_PART) | Q(tag_type=TAG_TYPE.BODY_PART_SUB_ITEM)
    query &= Q(is_online=True)
    tags = Tag.objects.filter(query)
    for tag in tags:
        if tag.tag_type == TAG_TYPE.BODY_PART:
            child_tags = tag.online_child_tags()
            tag_ids = []
            for child_tag in child_tags:
                tag_ids.extend(child_tag.related_tags)
        elif tag.tag_type == TAG_TYPE.BODY_PART_SUB_ITEM:
            tag_ids = tag.related_tags
        else:
            tag_ids = []

        cache_info = {
            'tag_ids': tag_ids,
        }
        tag_child_tags_cache.set(str(tag.id), json.dumps(cache_info))


def tran_day_to_num(today, order_day):
    diff_day = (today - order_day).days + 1
    if 0 < diff_day <= 1:
        return 1
    elif diff_day <= 2:
        return 2
    elif diff_day <= 3:
        return 3
    elif 4 <= diff_day <= 7:
        return 4
    elif 8 <= diff_day <= 30:
        return 5
    elif 31 <= diff_day <= 90:
        return 6
    else:
        return 0


@shared_task
def get_day_smart_rank():
    day = datetime.datetime.today()
    day = datetime.datetime(day.year, day.month, day.day, 0, 0, 0)

    file_dir = settings.LOG_DIR + 'smart_rank/'
    if not os.path.exists(file_dir):
        os.mkdir(file_dir)

    file_name = file_dir + day.strftime('%y-%m-%d') + '.log'
    fp = codecs.open(file_name, 'wr', 'utf-8')

    services = Service.objects.all()
    total_w = 0
    for _, val in settings.MAP_VALIDATE_ORDER_W.items():
        total_w = total_w + val

    max_gap_num = 2 * settings.GAP_NUM + 1

    max_price_dict = {}
    for service in services:
        tmp_price_dict = {}
        orders = service.order_set.filter(validated=True)
        for order in orders:
            order_day = order.validate_time
            if not order_day:
                continue

            map_w_key = tran_day_to_num(day, order_day)
            if map_w_key in settings.MAP_VALIDATE_ORDER_W:
                if map_w_key in tmp_price_dict:
                    tmp_price_dict[map_w_key] = tmp_price_dict[map_w_key] + float(order.service_price)
                else:
                    tmp_price_dict[map_w_key] = float(order.service_price)

                map_w_key = map_w_key + settings.GAP_NUM
                if map_w_key in tmp_price_dict:
                    tmp_price_dict[map_w_key] = tmp_price_dict[map_w_key] + float(order.discount)
                else:
                    tmp_price_dict[map_w_key] = float(order.discount)

        for key, val in tmp_price_dict.items():
            if key in max_price_dict:
                if tmp_price_dict[key] > max_price_dict[key]:
                    max_price_dict[key] = tmp_price_dict[key]
            else:
                max_price_dict[key] = tmp_price_dict[key]

    for service in services:
        orders = service.order_set.filter(validated=True)
        order_price_dict = {}
        for order in orders:
            order_day = order.validate_time
            if not order_day:
                continue

            map_w_key = tran_day_to_num(day, order_day)
            if map_w_key in settings.MAP_VALIDATE_ORDER_W:
                if map_w_key in order_price_dict:
                    order_price_dict[map_w_key] = order_price_dict[map_w_key] + float(order.service_price)
                else:
                    order_price_dict[map_w_key] = float(order.service_price)

                map_w_key = map_w_key + settings.GAP_NUM
                if map_w_key in order_price_dict:
                    order_price_dict[map_w_key] = order_price_dict[map_w_key] + float(order.discount)
                else:
                    order_price_dict[map_w_key] = float(order.discount)

        if not order_price_dict:
            continue

        price_r = {}
        total_rank = 0
        for i in range(1, max_gap_num):
            if i in order_price_dict:
                service_price_total = order_price_dict.get(i, 0)
                max_price = max_price_dict.get(i, 0)
                if max_price:
                    price_r[i] = service_price_total / max_price
                else:
                    price_r[i] = 0
            else:
                price_r[i] = 0

            total_rank = total_rank + price_r[i] * settings.MAP_VALIDATE_ORDER_W[i]

        smart_rank = total_rank / total_w
        json_data = collections.OrderedDict()
        if order_price_dict and smart_rank:
            for i in range(1, settings.GAP_NUM + 1):
                key = "price_r_{index}".format(index=i)
                val = price_r[i]
                json_data[key] = val

                key = "discount_r_{index}".format(index=i)
                val = price_r[i + settings.GAP_NUM]
                json_data[key] = val

            json_data = json.dumps(json_data)
            try:
                rank = SmartRank.objects.get(service=service)
            except SmartRank.DoesNotExist:
                rank = SmartRank()

            rank.service = service
            rank.smart_rank = smart_rank
            rank.price_data = json_data
            rank.save()

            service_name = service.name
            log_data = u"{id} {name} {rank} {data}\n".format(
                id=service.id,
                name=service_name,
                rank=smart_rank,
                data=json_data,
            )
            fp.write(log_data)
    fp.close()


@shared_task
@thread_local(DB_FOR_READ_OVERRIDE=settings.SLAVE_DB_NAME)
def set_famous_doctor_info():
    # todo 已无调用，不再维护 180314
    """

    :return:
    """
    doctors = Doctor.objects.filter(is_online=True)
    famous_doctor_client = FamousDoctorClient()
    today = datetime.datetime.today()

    def set_single_doctor(doctor):
        is_doctor = doctor.doctor_type == DOCTOR_TYPE.DOCTOR

        try:
            hospital = doctor.hospital.name
        except:
            hospital = None

        online_time = doctor.online_time if doctor.online_time != datetime.datetime.fromtimestamp(
            0) else doctor.created_at

        try:
            days = (today - online_time).days
        except TypeError:
            days = 746

        order_case = Order.objects.filter(service__doctor=doctor, status=ORDER_STATUS.PAID).count()
        problem_case = TopicReply.objects.filter(
            doctor_id=doctor.id,
            problem__private_status='0',
            problem__is_online=True,
            problem__flag='n'
        ).exclude(problem__topic_type=TOPIC_TYPE.OLD_DIARY).count()
        case_num = order_case * 10 + problem_case * 2

        doctor_info = {
            "doctor_id": doctor.id,
            "is_hospital": not is_doctor,
            "name": doctor.name if is_doctor else doctor.hospital.name,
            "portrait": doctor.portrait if is_doctor else doctor.hospital.portrait,
            "description": doctor.introduction,
            "days_in_gengmei": days,
            "hospital": hospital if is_doctor else '',
            "case_num": case_num,
            "title": DOCTOR_TITLE_TYPE.getDesc(doctor.title) if doctor.title != DOCTOR_TITLE_TYPE.EMPTY else '',
        }
        doctor_id = u'{}'.format(doctor.id)
        famous_doctor_client.set(doctor_id, json.dumps(doctor_info))

    for doctor in doctors:
        try:
            set_single_doctor(doctor)
        except Exception:
            pass


@shared_task
def cache_wiki_list():
    wikis = []
    online_struct = WikiStruct.objects.filter(
        wiki__is_online=True
    ).order_by(
        'struct_level', 'sort', 'parent_id'
    ).values(
        'id', 'struct_level', 'parent_id', 'sort',
        'wiki__id', 'wiki__item_name'
    )

    def filter_level(level, parent=None):
        def _filter(x):
            return (
                isinstance(x, dict) and
                x.get('struct_level') == level and
                (x.get('parent_id', '') == parent or not parent)
            )

        return _filter

    level_1 = filter_level(WIKI_LEVEL.ONE)

    def base_info(wiki):
        return {
            'wiki_id': wiki['wiki__id'],
            'wiki_name': wiki['wiki__item_name'],
        }

    advers = WikiAdver.objects.filter(is_online=True).values(
        'itemwiki__id', 'url_type', 'download_link', 'image_header'
    )
    advers_info = {
        adver['itemwiki__id']: {
            'url_type': adver['url_type'],
            'download_link': adver['download_link'],
            'image': adver['image_header']
        }
        for adver in advers
        }

    for first_level in filter(level_1, online_struct):
        first_level_info = base_info(first_level)
        first_level_info.update(
            {
                'sub_wikis': [],
                'banner': {
                    'url_type': '',
                    'image': '',
                    'download_link': '',
                }
            }
        )
        level_2 = filter_level(WIKI_LEVEL.TWO, first_level['id'])

        for second_level in filter(level_2, online_struct):
            second_level_info = base_info(second_level)
            second_level_info.update({'sub_wikis': []})

            level_3 = filter_level(WIKI_LEVEL.THREE,
                                   second_level['id'])

            for third_level in filter(level_3, online_struct):
                second_level_info['sub_wikis'].append(
                    base_info(third_level)
                )

            first_level_info['sub_wikis'].append(second_level_info)

        wiki_id = first_level['wiki__id']
        if advers_info.get(wiki_id):
            first_level_info['banner'].update(advers_info[wiki_id])

        wikis.append(first_level_info)

    if wikis:
        itemwiki_cache.set(settings.ITEMWIKI_LIST_CACHE,
                           json.dumps(wikis))

    return wikis


@shared_task
def get_tag_top_sale_service():
    tags = Tag.objects.filter(
        is_online=True,
        tag_type__in=[TAG_TYPE.BODY_PART, TAG_TYPE.BODY_PART_SUB_ITEM, TAG_TYPE.ITEM_WIKI]
    )

    for tag in tags:
        tag_id = tag.id
        filters = {'tag_ids': [tag_id]}
        _set_tags_service_cache(filters, tag_id=tag_id)

    # 没有筛选项目tag的
    _set_tags_service_cache({})


def _set_tags_service_cache(filters, tag_id=None):
    DAYS_NUM = [1, 7, 30]
    TOP_NUM = 5
    services = filter_service(
        offset=0,
        size=9999,
        sort_type=SERVICE_ORDER_TYPE.ORDER_HIGHEST_SALES,
        filters=filters
    )
    service_ids = services['service_ids']

    for days in DAYS_NUM:
        start_time = datetime.datetime.now() - datetime.timedelta(days=days)
        orders = Order.objects.filter(pay_time__gte=start_time)
        orders = orders.values('service').annotate(count=Count('service_id')).order_by('-count')
        top_services = []
        top_num = 0
        for service in orders:
            if top_num >= TOP_NUM:
                break
            if service['service'] in service_ids:
                top_services.append(service['service'])
                top_num = top_num + 1

        if tag_id:
            tag_key = 'tag_service:{tag_id}:{days}'.format(tag_id=tag_id, days=days)
        else:
            tag_key = 'no_tag_service:{days}'.format(days=days)

        cache_info = {
            "tag_service": top_services
        }
        cache_info = json.dumps(cache_info)
        tag_top_sale_cache.set(tag_key, cache_info)


@shared_task
def consume_qqcoupon():
    """
    :return:
    """
    # 筛选未通知已消费的卡券记录
    unconsume_coupons = QQCouponRecord.objects.filter(consume_notify=False)
    if not unconsume_coupons:
        return

    phone_numbers = unconsume_coupons.values_list('phone', flat=True)
    record_phone_map = {record: record.phone for record in unconsume_coupons}
    persons = Person.objects.filter(phone__in=phone_numbers)
    phone_user_id_map = {person[1]: person[0] for person in persons.values_list('user_id', 'phone')}

    if not phone_user_id_map:
        return

    for record, phone in record_phone_map.iteritems():
        user_id = phone_user_id_map[phone]
        gift_channel = record.gift_channel
        gift_id = record.gift

        try:
            consumed = can_qq_coupon_verificate(user_id, gift_channel, gift_id)
        except:
            logging_exception()
            continue

        if not consumed:
            continue

        try:
            init_kwargs = {
                'code': record.card_code,
                'card_id': record.card_id,
                'signature': '',
                'attach': record.attach,
                'redirect_uri': '',
            }

            connect = QQConnect(**init_kwargs)
        except:
            logging_exception()

        should_notify = connect.notify_consumed(record.openid, record.attach, record.access_token, record.card_code)
        if should_notify:
            record.consume_notify = True
            record.save()


@shared_task
def set_service_home_city():
    city_ids = set(Service.objects.filter(is_online=True).values_list(
        'doctor__hospital__city', flat=True))
    city_ids = list(city_ids)

    cache_info = {
        "service_home_city_ids": city_ids
    }
    key = 'service_home_city_key'
    cache_info = json.dumps(cache_info)
    service_home_city_cache.set(key, cache_info)


def _get_orders(start_time, end_time, need_filter_commented=False):
    """
    获取订单
    :param start_time:
    :param end_time:
    :return:
    """
    orders = Order.objects.filter(validate_time__range=[start_time, end_time])
    order_ids = set(orders.values_list('id', flat=True))
    if need_filter_commented:
        created_order_ids = set(ServiceComment.objects.filter(order_id__in=order_ids).values_list('order_id', flat=True))
        order_ids = order_ids - created_order_ids

    # exclude order ids in api_ordercategory
    orders_should_ignore = OrderCategory.objects.filter(
        order_id__in=order_ids, category=ORDER_CATEGORY.FARMING
    ).values_list('order_id', flat=True)
    order_ids = order_ids - set(orders_should_ignore)

    orders = Order.objects.filter(pk__in=order_ids)
    return orders


@shared_task
def generate_default_service_comment():
    """
    超时生成美购的默认评价
    :return:
    """
    now = datetime.datetime.today()
    start_time = now - datetime.timedelta(days=61)
    end_time = start_time + datetime.timedelta(days=1)

    default_kwargs = {
        'rating': 5,
        'content': u'用户默认给了好评',
        'default_rating': True,
        'operation_effect': 5,
        'doctor_attitude': 5,
        'hospital_env': 5,
        'images': [],
        'comment_options': []
    }
    orders = _get_orders(start_time, end_time, need_filter_commented=True)
    for order in orders:
        default_kwargs['order'] = order
        default_kwargs['service'] = order.service
        ServiceComment.create_comment(**default_kwargs)
        # update v 7.7.25 产品要求生成默认评价也自动创建日记本
        if not order.diary_id():
            order.create_diary()


@shared_task
def order_creation_diary_retried():
    """
    订单创建日记本重试接口, 查询 60-61天(48小时内)的数据
    :return:
    """
    now = datetime.datetime.today()
    start_time = now - datetime.timedelta(days=62)
    end_time = start_time + datetime.timedelta(days=2)
    orders = _get_orders(start_time, end_time)

    for order in orders:
        if order.has_service_comment() and not order.diary_id():  # 订单有评价并且未生成日记本，则走轮询机制自动创建
            order.create_diary()


@shared_task
def calc_service_avg_score():
    tags = Tag.objects.all()
    for tag in tags.iterator():
        scores = Service.objects.filter(project_type=tag).exclude(Q(rating=0) |
                                                                  Q(service_type=SERVICE_SELL_TYPE.ONEPURCHASE)). \
            aggregate(rating_avg=Avg('rating'))
        print(scores)
        rating = scores['rating_avg'] if scores['rating_avg'] else 0
        service_score_tag_avg.setex(str(tag.id), 60 * 60 * 24 * 2, rating)


@shared_task
def calc_all_service_doctor_hospital_score():
    calc_service_score()
    calc_service_avg_score()
    calc_doctor_score()
    calc_hospital_score()


def calc_service_score():
    ss = Service.objects.filter(is_online=True)
    for s in ss.iterator():
        try:
            comments = s.servicecomment_set.filter(is_effect=True,
                                                   is_online=True).aggregate(op_avg=Avg('operation_effect'),
                                                                             doc_avg=Avg('doctor_attitude'),
                                                                             hos_avg=Avg('hospital_env'),
                                                                             rating_avg=Avg('rating'))
            s.hospital_env_rating = round(comments['hos_avg'], 1) if comments['hos_avg'] else 0
            s.operation_effect_rating = round(comments['op_avg'], 1) if comments['op_avg'] else 0
            s.doctor_attitude_rating = round(comments['doc_avg'], 1) if comments['doc_avg'] else 0
            s.rating = round(comments['rating_avg'], 1) if comments['rating_avg'] else 0
            s.save(update_fields=['hospital_env_rating', 'operation_effect_rating',
                                  'doctor_attitude_rating', 'rating'])
        except:
            logging_exception()
            continue


def calc_doctor_score():
    doctors = Doctor.objects.all()
    for doctor in doctors.iterator():
        try:
            scores = doctor.services.exclude(Q(rating=0) | Q(hospital_env_rating=0)
                                             | Q(doctor_attitude_rating=0)
                                             | Q(service_type=SERVICE_SELL_TYPE.ONEPURCHASE)
                                             | Q(operation_effect_rating=0)).aggregate(
                op_avg=Avg('operation_effect_rating'),
                doc_avg=Avg('doctor_attitude_rating'),
                hos_avg=Avg('hospital_env_rating'),
                rating_avg=Avg('rating'))
            # update at 7655,向下取整
            avg_rate = round(scores['rating_avg'] - 0.05, 1) if scores['rating_avg'] else 0.0
            avg_effect_rating = round(scores['rating_avg'] - 0.05, 1) if scores['rating_avg'] else 0.0
            avg_attitude_rating = round(scores['rating_avg'] - 0.05, 1) if scores['rating_avg'] else 0.0
            avg_env_rating = round(scores['rating_avg'] - 0.05, 1) if scores['rating_avg'] else 0.0

            if not doctor.rate == avg_rate or not doctor.effect_rating == avg_effect_rating or \
                    not doctor.attitude_rating == avg_attitude_rating or not doctor.env_rating == avg_env_rating:
                doctor.rate = avg_rate
                doctor.effect_rating = avg_effect_rating
                doctor.attitude_rating = avg_attitude_rating
                doctor.env_rating = avg_env_rating
                doctor.save(update_fields=['rate', 'effect_rating',
                                           'attitude_rating', 'env_rating'])
        except:
            logging_exception()
            continue


def calc_hospital_score():
    hospitals = Hospital.objects.all()
    for hospital in hospitals.iterator():
        try:
            doctors = Doctor.objects.filter(hospital=hospital, is_online=True)
            scores = Service.objects.filter(is_online=True,
                                            doctor__in=doctors).exclude(Q(rating=0)
                                                                        | Q(hospital_env_rating=0)
                                                                        | Q(doctor_attitude_rating=0)
                                                                        | Q(service_type=SERVICE_SELL_TYPE.ONEPURCHASE)
                                                                        | Q(operation_effect_rating=0)).aggregate(
                op_avg=Avg('operation_effect_rating'),
                doc_avg=Avg('doctor_attitude_rating'),
                hos_avg=Avg('hospital_env_rating'),
                rating_avg=Avg('rating'))
            hospital.rate = round(scores['rating_avg'], 1) if scores['rating_avg'] else 0
            hospital.effect_rating = round(scores['op_avg'], 1) if scores['op_avg'] else 0
            hospital.attitude_rating = round(scores['doc_avg'], 1) if scores['doc_avg'] else 0
            hospital.env_rating = round(scores['hos_avg'], 1) if scores['hos_avg'] else 0
            hospital.save(update_fields=['rate', 'effect_rating', 'attitude_rating', 'env_rating'])
        except:
            logging_exception()
            continue


@shared_task()
def add_fake_fans_service():
    '''
    涨粉(虚拟)每日轮询
    :return:
    '''
    rpc_client = get_rpc_remote_invoker()
    ppl_get_fans = {}  # 最后需要增加粉丝的用户，及其增加的粉丝数量
    fans_type_dict = AddFansConfig.get()
    # 获取各类型下是否有符合条件的用户
    yesterday_time = datetime.datetime.today() - datetime.timedelta(days=1)
    beginning_of_day = datetime.datetime.combine(yesterday_time, datetime.time.min)
    end_of_day = datetime.datetime.combine(yesterday_time, datetime.time.max)

    # 回答（设为推荐, is_recommend=True)）
    answers = rpc_client['qa/answer/lst_for_task'](type_q='answer_recommend', ids=None,
                                                   begin_day=beginning_of_day.strftime('%Y-%m-%d %H:%M:%S'),
                                                   end_day=end_of_day.strftime('%Y-%m-%d %H:%M:%S')).unwrap()
    _get_dif_type_fans(fans_type_dict.get('qa'), 'qa', answers, ppl_get_fans)
    # 达人登录
    is_logging = UserExtra.objects.using(settings.SLAVE_DB_NAME).filter(last_login__range=[beginning_of_day, end_of_day],
                                          membership_level=MEMBERSHIP_LEVEL.STAR).values_list('user_id', flat=True)
    _get_dif_type_fans(fans_type_dict.get('talent'), 'talent', is_logging, ppl_get_fans)
    # 内容等级
    diarys = rpc_client['topic/task/add_fake_fans'](start_time=beginning_of_day.strftime('%Y-%m-%d %H:%M:%S'),
                                                    end_time=end_of_day.strftime('%Y-%m-%d %H:%M:%S'),
                                                    q_type='diary').unwrap()
    _get_dif_type_fans(fans_type_dict, 'diary', diarys, ppl_get_fans)
    # 收到评论
    user_get_reply = rpc_client['topic/task/add_fake_fans'](start_time=beginning_of_day.strftime('%Y-%m-%d %H:%M:%S'),
                                                            end_time=end_of_day.strftime('%Y-%m-%d %H:%M:%S'),
                                                            q_type='topicreply').unwrap()
    _get_dif_type_fans(fans_type_dict.get('topic'), 'topic', list(user_get_reply), ppl_get_fans)
    for uid, fans_num in ppl_get_fans.iteritems():
        user_extras = _get_sleep_user(fans_num)  # 获取沉睡用户
        for id in user_extras:
            id = UserExtra.objects.get(id=id).user_id
            SocialInfo.add_to_fans(uid, id, is_virtual_fan=True)


def _get_dif_type_fans(fans_dict, type_str, user_list, ppl_get_fans):
    '''
    根据不同策略类型来增加不同的粉丝数
    _get_dif_type_fans(fans_dict,'qa',answers,ppl_get_fans)
    :return:
    '''
    if not type_str[:5] == 'diary':
        for item in user_list:
            if ppl_get_fans.get(item):
                ppl_get_fans[item] += random.randint(int(fans_dict.get(type_str + '_min')),
                                                     int(fans_dict.get(type_str + '_max')))
            else:
                ppl_get_fans[item] = random.randint(int(fans_dict.get(type_str + '_min')),
                                                    int(fans_dict.get(type_str + '_max')))
    else:
        # 需要按照不同的日记的评审结果，给相应不同等级的粉丝数量
        for item in user_list:
            if item.get('content_level') == DIARY_CONTENT_LEVEL.EXCELLENT[0]:
                type_str_sub = type_str + '_a'
            elif item.get('content_level') == DIARY_CONTENT_LEVEL.FINE[0]:
                type_str_sub = type_str + '_b'
            else:
                type_str_sub = type_str + '_c'
            fans_dict_sub = fans_dict.get(type_str_sub)
            if ppl_get_fans.get(item.get('user_id')):
                ppl_get_fans[item.get('user_id')] += random.randint(int(fans_dict_sub.get(type_str_sub + '_min')),
                                                                    int(fans_dict_sub.get(type_str_sub + '_max')))
            else:
                ppl_get_fans[item.get('user_id')] = random.randint(int(fans_dict_sub.get(type_str_sub + '_min')),
                                                                   int(fans_dict_sub.get(type_str_sub + '_max')))


_MAX_SLEEP_PERSON = 5000  # redis里现存沉睡用户数量，详见user_related_tasks.py


def _get_sleep_user(repeat_times):
    '''
    获取马甲用户ID的列表
    :param repeat_times: 获取的数量
    :return:
    '''
    user_extras = []

    index_seq = random.sample(xrange(_MAX_SLEEP_PERSON), repeat_times)

    for i in index_seq:
        user_extras.append(sleep_user_cache.get('sleep_user_index:' + str(i)))
    return user_extras


def _remind_user_to_validate_after_order_create(day, ex_user_ids):
    today = datetime.datetime.now()
    pay_time = today - datetime.timedelta(days=day)
    sql_extra = "and o.user_id not in ({})".format(','.join(map(str, ex_user_ids))) if ex_user_ids else ''
    queryset = Order.objects.raw("""
        SELECT o.id, o.user_id, SUM(if(o.status = '1', 1, 0)) as order_count FROM api_order o
        WHERE o.status = '1' and o.pay_time between "{}" and "{}" {}
        GROUP BY o.user_id HAVING order_count > 0
    """.format(pay_time.strftime('%Y-%m-%d 00:00:00'), pay_time.strftime('%Y-%m-%d 23:59:59'), sql_extra))

    gm_protocol = GmProtocol()
    extra = {
        'type': PUSH_INFO_TYPE.GM_PROTOCOL,
        'pushUrl': gm_protocol.get_order_list(),
    }
    for user in queryset:
        alert = "您有{order_count}个美购还未使用呢！抓紧时间快去变美吧!".format(order_count=user.order_count)
        if user.order_count == 1:
            # 单订单
            try:
                order = Order.objects.get(user__pk=user.user_id, status='1')
                alert = "您的购买的{service_name}还未使用呢！抓紧时间快去变美吧！".format(service_name=order.service.name)
            except (Order.DoesNotExist, Order.MultipleObjectsReturned):
                pass

        from api.tasks.push_task import automate_push_task
        automate_push_task(user_ids=[user.user_id], push_type=AUTOMATED_PUSH.UNVERIFIED_US_PURCHASER, alert=alert, extra=extra)
    return [o.user_id for o in queryset]


@shared_task
@thread_local(DB_FOR_READ_OVERRIDE=settings.SLAVE_DB_NAME)
def remind_user_to_validate_after_order_create():
    """
    支付后未验证Push提醒：每天一次，找支付完成后，7天/30内未验证/验证过期前7天，进行推送
    优先级：过期前7天＞7天内未验证＞30天内未验证。
    # http://wiki.wanmeizhensuo.com/pages/viewpage.action?pageId=4424213
    :return:
    """
    push_days = [83, 7, 30]
    ex_user_ids = []
    for day in push_days:
        push_user_ids = _remind_user_to_validate_after_order_create(day, ex_user_ids)
        ex_user_ids.extend(push_user_ids)


@shared_task
@thread_local(DB_FOR_READ_OVERRIDE=settings.SLAVE_DB_NAME)
def remind_user_to_write_diary_after_order_validate():
    """
    验证成功未写日记 (每天运行一次)
    http://wiki.wanmeizhensuo.com/pages/viewpage.action?pageId=4424213
    :return:
    """
    r = get_rpc_remote_invoker()
    order_ids = r['diary/order_ids'](days=30).unwrap()

    # 在30天内验证，并且没有写日记的订单 按验证时间倒序排 优先发送最近的 避免被限流规则干掉
    queryset = Order.objects.filter(pk__in=order_ids, validate_time__gt=datetime.datetime.now() - timedelta(days=30)) \
        .order_by('-validate_time')

    tpl_with_cashback = {
        # 5h Push为已有逻辑，不考虑在内
        '5h': "您做的{service_name}项目怎么样啦？快给大家分享一下吧！还能返现金呢！",
        '1d': "您做的{service_name}项目怎么样啦？术后注意恢复，别忘了把变美分享出来，还能返你现金哦~",
        '3d': "您做的{service_name}项目怎么样啦？快给大家分享一下吧！还能拿返现呢！",
        "7d": "您做的{service_name}项目已经恢复7天咯，快来给小伙伴们传授经验～还能拿返现哦！",
        "15d": "您做的{service_name}项目已经15天咯，恢复的怎么样？快给大家分享一下吧！还能拿返现呢！",
        "21d": "您做的{service_name}项目怎么样啦？快给大家分享一下吧！还能拿返现呢！",
        "30d": "您做的{service_name}项目怎么样啦？术后1个月效果最明显，快给大家分享你的变美历程，还能拿返现哦！",
    }

    tpl_without_cashback = {
        # 5h Push为已有逻辑，不考虑在内
        '5h': "您做的{service_name}项目怎么样啦？快给大家分享一下吧！",
        '1d': "您做的{service_name}项目怎么样啦？术后注意恢复，别忘了把变美分享出来~",
        '3d': "您做的{service_name}项目怎么样啦？快给大家分享一下吧！",
        "7d": "您做的{service_name}项目已经恢复7天咯，快来给小伙伴们传授经验～",
        "15d": "您做的{service_name}项目已经15天咯，恢复的怎么样？快给大家分享一下吧！",
        "21d": "您做的{service_name}项目怎么样啦？快给大家分享一下吧！",
        "30d": "您做的{service_name}项目怎么样啦？术后1个月效果最明显，快给大家分享你的变美历程！",
    }

    gm_protocol = GmProtocol()
    extra = {
        'type': PUSH_INFO_TYPE.GM_PROTOCOL,
        'pushUrl': gm_protocol.get_select_diary()
    }
    from api.tasks.push_task import automate_push_task
    for order in queryset:
        if order.share_get_cashback:
            tpl = tpl_with_cashback
        else:
            tpl = tpl_without_cashback

        days = (datetime.datetime.now() - order.validate_time).days
        if not special_push_limit(user_id=order.user_id, push_type="verify_write_diary"):
            continue
        if days == 1:
            alert = tpl['1d'].format(service_name=order.service.name)
            automate_push_task(
                user_ids=[order.user_id],
                push_type=AUTOMATED_PUSH.PAYMENT_DIARY_UPDATE,
                extra=extra,
                alert=alert,
            )
            # todo: 短信Push
        elif days in (3, 7, 15, 21, 30):
            alert = tpl[str(days) + 'd'].format(service_name=order.service.name)
            automate_push_task(
                user_ids=[order.user_id],
                push_type=AUTOMATED_PUSH.PAYMENT_DIARY_UPDATE,
                extra=extra,
                alert=alert,
            )


@shared_task
def calc_doctor_and_hospital_service_tags_distribution():
    """刷新每个医生名下所有美购二级TAG的美购个数  每天定时跑的任务
        TODO: 有很大的性能方面问题， 因为定时任务， 所以修改的优先级不算很高
    """

    start = 0
    step = 1000
    tomorrow = datetime.datetime.now() + timedelta(days=1)

    while True:
        doctor_ids = Doctor.objects.values_list('id', flat=True).filter(is_online=True)[start: start + step]

        start += step
        if not doctor_ids:
            break
        for doctor_id in doctor_ids:
            service_tag = {}
            qs = Q(doctor_id=doctor_id) & Q(is_online=True)
            qs &= Q(end_time=None) | Q(end_time__gte=tomorrow)
            services = Service.objects.filter(qs).prefetch_related('tags')
            for service in services:
                items2 = list(filter(lambda tag: tag.tag_type == TAG_TYPE.BODY_PART_SUB_ITEM and tag.is_online is True,
                                     service.tags.all()))

                already_add_items2_ids = []
                for item in items2:
                    service_tag.setdefault(item.id, {'id': str(item.id), 'name': item.name, 'count': 0})['count'] += 1
                    already_add_items2_ids.append(item.id)

                subitems = list(filter(lambda tag: tag.tag_type == TAG_TYPE.ITEM_WIKI and tag.is_online is True,
                                       service.tags.all()))
                items3 = TagService.get_closure_tags(tag_ids=[subitem.id for subitem in subitems])
                for item in filter(lambda tag: tag.tag_type == TAG_TYPE.BODY_PART_SUB_ITEM, items3):
                    if item.id in already_add_items2_ids:
                        continue
                    service_tag.setdefault(item.id, {'id': str(item.id), 'name': item.name, 'count': 0})['count'] += 1

            service_tag = sorted(service_tag.values(), key=lambda x: (-x['count'], len(x['id']), x['id']))
            if len(services):
                service_tag.insert(0, {"id": "all", "name": "全部", "count": len(services)})
            doctor_cache.set(doctor_id, json.dumps(service_tag))

    start = 0
    step = 1000

    while True:
        hospitals = Hospital.objects.all()[start: start + step]

        start += step
        if not hospitals:
            break
        for hospital in hospitals:
            service_tag = {}
            doctor_ids = hospital.doctor_hospital.filter(is_online=True).values_list('id', flat=True)
            for doctor_id in doctor_ids:
                tags_json = doctor_cache.get(doctor_id)
                if not tags_json:
                    continue

                tags_distribution_list = json.loads(tags_json)
                for tags_distribution in tags_distribution_list:
                    if service_tag.get(tags_distribution['id']):
                        service_tag[tags_distribution['id']]['count'] += tags_distribution['count']
                    else:
                        service_tag[tags_distribution['id']] = tags_distribution
            service_tag = sorted(service_tag.values(), key=lambda x: (-x['count'], len(x['id']), x['id']))
            hospital_cache.set(hospital.id, json.dumps(service_tag))


@shared_task
def cache_recommends_college_data():
    """
    缓存推荐大学的数据，以字典形式存储，{count: [college_id, college_id]}
    :return:
    """
    colleges = College.objects.filter(is_online=True).annotate(c=Count("userextratocollege")).values_list("id", "c")

    pre_dic = {}
    for info in colleges:
        college_id, count = info
        if count not in pre_dic:
            pre_dic[count] = [college_id, ]
        pre_dic[count].append(college_id)

    pre_dic = {k: json.dumps(list(set(v))) for k, v in pre_dic.items()}  # 转成字符串
    recommend_college_cache.delete("recommend_colleges")
    recommend_college_cache.hmset("recommend_colleges", pre_dic)  # 写入缓存


@shared_task
def momo_stat_log(info_list, stat_log_type):
    """
        momo 数据打点
    setting 中配置项 MOMO_STAT_LOG_TEST_FLAG 用于标记打点地址
    True 测试地址
    False 线上地址

    打点位置：
        预支付成功，线下验证，退款成功
    :param info_list: list 数据结构：[{"momo_user_id": xx, "product_id": xx}, ]
    :param stat_log_type: 陌陌打点类型 type类型 枚举值
    :return:
    """
    need_order_param = [
        MOMO_STAT_LOG_TYPE.PRE_PAY_SUCCESS,
        MOMO_STAT_LOG_TYPE.ORDER_VERIFIED_SUCCESS,
        MOMO_STAT_LOG_TYPE.ORDER_REFUND_SUCCESS
    ]
    need_money_param = need_order_param + [MOMO_STAT_LOG_TYPE.SERVICE_PAY_CLICK, ]

    need_hospital_id = [MOMO_STAT_LOG_TYPE.CONVERSATION_FIRST_REPLY]

    def _user_sign(**kwargs):
        """
        用户签名
        :param kwargs:
        :return:
        """
        md5 = hashlib.md5()
        order_dict = collections.OrderedDict()
        sign_key = kwargs.pop("sign_key", "")

        for k, v in sorted(kwargs.items(), key=lambda d: d[0]):
            if v:
                order_dict[k] = v

        _data = u"&".join(["{k}={v}".format(k=k, v=v) for k, v in order_dict.items()])
        str_data = u"{0}{1}".format(_data, sign_key)
        md5.update(str_data.encode("utf-8"))
        return md5.hexdigest()

    params_list = []
    for info in list(filter(None, info_list)):  # 过滤误传空数据的情况
        params = {
            "source_type": "gengmei",
            "timestamp": str(int(time.time())),
            "user_id": info.get("momo_user_id", ""),
            "type": stat_log_type,
            "product_id": info.get("product_id", 0),
        }
        if stat_log_type in need_money_param:
            params["money"] = float(info.get("money", 0)) or ""  # 若钱数为 0 则置成空串不参与签名

        if stat_log_type in need_order_param:
            params["order_id"] = int(info.get("order_id", 0))

        if stat_log_type in need_hospital_id:
            params["hospital_id"] = info.get("hospital_id", "")

        params["sign"] = _user_sign(
            sign_key="ebcca4740cbf2cb5",
            **params
        )
        params_list.append(params)

    res_log_list = []
    for momo_log_params in params_list:
        # 陌陌打点
        url = "http://ad-api.immomo.com/gengmei/stat/log"
        if settings.MOMO_STAT_LOG_TEST_FLAG:
            url = "http://test.ad-api.immomo.com/gengmei/stat/log"
        response = requests.get(url=url, params=momo_log_params)
        response_info = json.loads(response.text)
        log_dict = {
            "url": url,
            "params": momo_log_params,
            "response_code": response.status_code,
            "response_coding": response.encoding,
            "response_info": response_info,
            "time_log": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        }

        if response.status_code == 200 and response_info.get("ec", 0) == 200:
            momo_stat_logger.info(json.dumps(log_dict, ensure_ascii=False))
        else:
            momo_stat_logger.error(json.dumps(log_dict, ensure_ascii=False))

        res_log_list.append(log_dict)

    return res_log_list


@shared_task
@thread_local(DB_FOR_READ_OVERRIDE=settings.SLAVE_DB_NAME)
def cache_latest_n_special_skus():
    special_ids = Special.get_latest_n_special_ids(2)
    if not special_ids:
        return

    from api.manager import service_info_manager
    import uuid
    task_id = uuid.uuid4().hex
    log_prefix = "[cache_latest_n_special_skus_job:{}]".format(task_id)

    for sp_id in special_ids:
        sku_ids_in_sp = list(SpecialItem.objects.filter(special_id=sp_id).values_list(
            'serviceitem_id', flat=True
        ).order_by('serviceitem_id'))

        cache_skus_logger.info(
            "{}all_sku_ids. special_id:{}, sku_ids:{}".format(log_prefix, sp_id, json.dumps(sku_ids_in_sp)))

        step = 20
        grouped_list = [sku_ids_in_sp[i: i + step] for i in range(0, len(sku_ids_in_sp), step)]

        for sku_ids in grouped_list:
            if sku_ids:
                try:
                    cache_skus_logger.info(
                        "{}ready_to_pre_write. special_id:{}, sku_ids:{}".format(log_prefix, sp_id,
                                                                                 json.dumps(sku_ids)))

                    sku_objs = ServiceItem.objects.filter(id__in=sku_ids, is_delete=False).values_list('id',
                                                                                                       'service_id')
                    skus_info = [{"service_id": sid, "sku_id": siid} for siid, sid in sku_objs]

                    cache_skus_logger.info(
                        "{}valid_skus_info. special_id:{}, skus_info:{}".format(log_prefix, sp_id,
                                                                                json.dumps(skus_info)))

                    if len(skus_info) == 0:
                        continue

                    seckill_sku_info_from_db_list = service_info_manager.get_seckill_sku_info_from_db(sp_id, skus_info)
                    seckill_sku_info_from_db = {}
                    for sidb in seckill_sku_info_from_db_list:
                        sku_id_str = str(sidb["service_item_id"])
                        seckill_sku_info_from_db[sku_id_str] = sidb

                    for sinfo in skus_info:
                        sku_id_str = str(sinfo['sku_id'])
                        seckill_sku_info = {}  # 如果SPU被下线，那么在上面的list里面是没有对应结果的，这个时候要set空的dict进去
                        if sku_id_str in seckill_sku_info_from_db:
                            seckill_sku_info = seckill_sku_info_from_db[sku_id_str]

                        service_info_manager.set_seckill_sku_info_to_pre_write_cache(sp_id, sku_id_str,
                                                                                     seckill_sku_info)

                        cache_skus_logger.info(
                            "{}set_pre_write_cache_succues. special_id:{}, sku_id:{}, has_seckill_info:{}".format(
                                log_prefix, sp_id, sku_id_str, bool(seckill_sku_info)))

                except:
                    cache_skus_logger.error(
                        "{}pre_write_error!!. special_id:{}, sku_ids:{}".format(log_prefix, sp_id, json.dumps(sku_ids)))
                    logging_exception()


@shared_task
def send_live_order_stats(payload):
    """ 触发 直播-订单转化 统计. """

    info_logger.info('order stats: %s' % payload)

    rpc_client = get_rpc_remote_invoker()

    res = rpc_client["live/action/stats"](payload=payload).unwrap()
