#!/usr/bin/env python # -*- coding: utf-8 -*- # 聚合推送 import datetime import json from utils.common import big_data_iter from .push_base import ( hour, minute, push_cache, push_logger, PushServiceBase, get_datetime_to_str, ) from .push_builder import PushEvents class AggregationPushService(PushServiceBase): """ 聚合推送 service """ _default_cache_name = "aggregation_push" ratelimit_unit_times = 2 push_handlers = PushEvents.aggregation_push_handlers # < ----------- 用户 缓存名 获取 begin ------------ > @classmethod def _get_aggregation_push_normal_time_cache_name(cls, push_types, times): """ 获取正常时间段内 需要聚合推送的name :param push_types: 推送类型 :param times: 字符串类型的格式化时间 :return: """ _cache_name = "{d_cache_name}_normal_types_{push_types}_times_{times}".format( d_cache_name=cls._default_cache_name, push_types=":".join(push_types), times=times ) return _cache_name @classmethod def _get_aggregation_push_abnormal_times_info(cls, push_types, now): """ 获取非正常时间段的缓存name + expire_time :param push_types: :return: """ if now.hour >= cls.end_push_hour: # 22 - 23 点 start_date = now.date() end_date = start_date + datetime.timedelta(days=1) elif now.hour < cls.start_push_hour: # 0 - 8 点 end_date = now.date() start_date = end_date - datetime.timedelta(days=1) _name = "{d_cache_name}_abnormal_types_{push_types}_times_{times}".format( d_cache_name=cls._default_cache_name, push_types=":".join(push_types), times="{}_{}".format( get_datetime_to_str(datetime.datetime.combine(start_date, datetime.time(hour=cls.end_push_hour))), get_datetime_to_str(datetime.datetime.combine(end_date, datetime.time(hour=cls.start_push_hour))) ) ) eta = datetime.datetime.combine(end_date, datetime.time(cls.start_push_hour)) return { "cache_name": _name, "eta": eta, } # < ----------- 用户 缓存名 获取 end ------------ > @classmethod def _create_cache(cls, cache_name): """ 缓存创建 :param cache_name: 缓存key :return: """ if not push_cache.exists(cache_name): push_cache.hincrby(cache_name, cls._default_cache_name) push_cache.expire(cache_name, hour * 26) # 设置超时时间 cls._trigger_task_switch(cache_name) # 触发异步任务开关 加锁 return True return False # < ----------- 用户 聚合推送 缓存创建 + 异步任务触发 begin ------------ > @staticmethod def _trigger_task_switch(cache_name): """ 触发异步任务开关 :return: """ _cache_name = "{}_task_switch".format(cache_name) # 任务触发开关，缓存名 区分下 return push_cache.set(_cache_name, 1, ex=60, nx=True) @classmethod def _trigger_task(cls, cache_name, eta, unit_time, **kwargs): """ 异步任务触发 只会触发有效时间内的任务 :param cache_name: 缓存名字 :param eta: 延迟时间 :return: """ sole_sign = kwargs.get("sole_sign", "") from communal.tasks import aggregation_push_trigger_task task_ = aggregation_push_trigger_task.apply_async( args=( cache_name, cls._default_cache_name, unit_time, sole_sign, ), eta=eta - datetime.timedelta(hours=8) # utc时间 ε=(´ο｀*)))唉 ) return task_.id @classmethod def get_cache_name_and_trigger_task(cls, push_types, now, unit_time=60 * 60, **kwargs): """ 获取缓存name + 触发任务 :return: """ sole_sign = kwargs.get("sole_sign", "") if cls.in_push_limit_time(now): # 在正常推送时间内 9 - 22小时 hour_str = (cls.current_times(now) or {}).get("hour_str", "") _cache_name = cls._get_aggregation_push_normal_time_cache_name(push_types, hour_str) eta = cls.expire_time(unit_time=unit_time, need_datetime=True) else: # 时间段问题 今日22时 - 次日8时内数据 由定时任务在9点触发 _abnormal_info = cls._get_aggregation_push_abnormal_times_info( push_types=push_types, now=now ) _cache_name = _abnormal_info.get("cache_name", "") eta = _abnormal_info.get("eta", None) _status = cls._create_cache(_cache_name) # 检测创建状态 # 业务逻辑触发异步任务 if _status and cls._trigger_task_switch(_cache_name): task_id = cls._trigger_task( _cache_name, eta=eta, unit_time=unit_time, sole_sign=sole_sign ) need_trigger_task = True else: task_id = "" need_trigger_task = False push_logger.info(json.dumps({ "aggregation_step": 4, "sole_sign": sole_sign, "subordinate": "aggregation_push", "resume": "Aggregation Push. get_cache_name_and_trigger_task", "push_types": push_types, "cache_name": _cache_name, "now": now.strftime("%Y%m%d %H:%M:%S"), "need_trigger_task": need_trigger_task, "Triggered task": { "task_id": task_id, "parameters": { "cache_name": _cache_name, "default_cache_key": cls._default_cache_name, "unit_time": unit_time, }, "eta": eta and eta.strftime("%Y%m%d %H:%M:%S") or eta, }, })) return _cache_name # < ----------- 用户 聚合推送 缓存创建 + 异步任务触发 end ------------ > # < ----------- 用户 聚合推送数据汇总 begin ------------ > @classmethod def aggregation_push_data(cls, user_ids, action_type, **kwargs): """ 聚合推送信息 把需要聚合的数据写到redis中 hash 操作 name 精确到小时的时间 key use_id value jsonstring {push_action_type: [source_id, source_id]} ---> {推送类型: [对应表索引id,]} --> eg: {回答点赞:[AnswerVote.id]} :param user_ids: :param action_type: :param kwargs: :return: """ _source_id = kwargs.get("source_id", 0) now = kwargs.pop("now", cls.get_now_datetime()) # 当前时间 sole_sign = kwargs.get("sole_sign", "") # 唯一标识 push_logger.info(json.dumps({ "aggregation_step": 2, "sole_sign": sole_sign, "resume": "Aggregation Push. Trigger parameters.", "now": now.strftime("%Y%m%d %H:%M:%S"), "user_ids": user_ids, "action_type": action_type, "other_params": kwargs, })) if all([user_ids, action_type, _source_id]): # 获取异步任务触发时间 delay_time = hour * 1 _push_types = [] for item in PushEvents.aggregation_push_rule.values(): aggregation_push_types = sorted(item.get("aggregation_push_types", [])) # 这里做了一次排序!!! if action_type in aggregation_push_types: delay_time = item.get("delay_times", hour * 1) _push_types = aggregation_push_types break push_logger.info(json.dumps({ "aggregation_step": 3, "sole_sign": sole_sign, "resume": "Aggregation Push. action_type conditions satisfied.", "action_type": action_type, "aggregation_push_types": _push_types, "delay_times": delay_time, })) # 获取缓存名 _cache_name = cls.get_cache_name_and_trigger_task( push_types=_push_types, now=now, unit_time=delay_time, sole_sign=sole_sign # 这是个日志的标识，没多大意义 ) # 聚合数据 处理 for user_ids in big_data_iter(user_ids, fetch_num=200): vs = push_cache.hmget(_cache_name, user_ids) write_to_cache_dic = {} for user_id, v in zip(user_ids, vs): if v: action_type_record_dic = json.loads(v) # 从缓存中获取 else: action_type_record_dic = {} if action_type not in action_type_record_dic: action_type_record_dic[action_type] = [] action_type_record_dic[action_type].append(_source_id) write_to_cache_dic[user_id] = json.dumps(action_type_record_dic) # 写入缓存 push_cache.hmset(_cache_name, write_to_cache_dic) push_logger.info(json.dumps({ "aggregation_step": 5, "sole_sign": sole_sign, "resume": "Aggregation Push. write in redis OK.", "cache_name": _cache_name, "action_type": action_type, "user_ids": user_ids, })) # < ----------- 用户 聚合推送数据汇总 end ------------ >