1、需求背景

app需要推送、如下线通知、折扣通知等

2、实现方式

采取将推送任务放在任务表中，然后在管理平台构建推送（异步）任务，读取推送任务表，进行实时推送

3、表设计

推送内容表

class PushContentInfo(db.Model):
    __tablename__ = 'tb_push_content_info'

    id = db.Column(db.Integer, autoincrement=True, primary_key=True)
    content_title = db.Column(db.String(64), nullable=False, info={'label': '内容标题'})
    push_title = db.Column(db.String(64), nullable=False, info={'label': '推送标题'})
    push_content = db.Column(db.String(255), nullable=False, info={'label': '推送内容'})
    redirect_type = db.Column(db.Integer, nullable=False, info={'label': '跳转位置'})
    redirect_url = db.Column(db.String(255), nullable=False, info={'label': '跳转url'})
    push_target = db.Column(db.String(64), nullable=False, info={'label': '推送目标'})
    message_save_hours = db.Column(db.Integer, nullable=False, default=0, info={'label': '消息离线保存时间'})
    add_time = db.Column(db.DateTime, nullable=False, default=db.func.now(), info={'label': '创建时间'})
    update_time = db.Column(db.DateTime, nullable=False, default=db.func.now(), info={'label': '更新时间'})

推送任务表

class PushNotifyRecord(db.Model):
    __tablename__ = 'tb_push_notify_record'

    id = db.Column(db.Integer, autoincrement=True, primary_key=True)
    uid = db.Column(db.String(64), nullable=False, info={'label': '用户id'})
    # device_code = db.Column(db.String(64), nullable=False, info={'label': '设备码'})
    notify_app_type = db.Column(db.String(32), nullable=False, default='all', info={'label': '推送目标'})
    push_content_id = db.Column(db.Integer, nullable=False, default=0, info={'label': '推送目标'})
    task_status = db.Column(db.Integer, nullable=False, default=0, info={'label': '规则状态'})
    add_time = db.Column(db.DateTime, nullable=False, default=db.func.now(), info={'label': '创建时间'})
    update_time = db.Column(db.DateTime, nullable=False, default=db.func.now(), info={'label': '更新时间'})

注意：友盟推送只需要用户id，因为友盟可以别名推送、给每个用户绑定唯一的别名(使用uid进行绑定)
谷歌推送需要单独的设备码，由设备提供，每个设备具备单个设备码，如果需要但用户多设备推送，需要进行用户和设备的关联

异步任务:

@app.task()
def scan_for_push_to_user():
    """
    定时通知记录表,用于通知用户
    :return:
    """
    # 获取待推送记录
    print 'start scan_for_push_to_user'
    rs = redis.StrictRedis.from_url(settings.REDIS_URL)
    ret = rs.set('qms_push_user_lock', 1, nx=True, ex=20 * 60)
    if not ret:
        logger.info('[scan_for_push_to_user] being locked')
        return
    push_task_dict = {}
    need_push_records = PushNotifyRecord.objects.filter(task_status=0)
    for item_record in need_push_records:
        push_task_dict.setdefault(item_record.notify_app_type + '-' + str(item_record.push_content_id), []).append(
            item_record
        )

    for item_task in push_task_dict:
        list_length = len(push_task_dict[item_task])
        logger.info('[scan_for_push_to_user] task %s push member count %s', item_task, list_length)
        limit = 1000
        step = list_length / limit + 1
        push_app_type, push_content_id = item_task.split('-')
        for i in range(step):
            start = i * limit
            end = start + limit
            if start >= list_length:
                break
            task_id_list = [x.id for x in push_task_dict[item_task][start:end]]
            uid_list = [x.uid for x in push_task_dict[item_task][start:end]]
            ret = PushNotifyRecord.objects.filter(id__in=task_id_list, task_status=0).update(task_status=1)
            if ret != len(task_id_list):
                break
            push_to_user.delay(task_id_list, uid_list, push_app_type, push_content_id)
        # 开始友盟推送
        logger.info('u_push------start')
        push_content_info = PushContentInfo.objects.get(id=push_content_id)
        go_url = push_content_info.redirect_url
        task_id_list = [x.id for x in push_task_dict[item_task]]
        # 获取到需要单独设备发送的数据
        push_notify_wait = PushNotifyRecord.objects.filter(
            id__in=task_id_list,
            task_status=1
        ).all()
        task_code_list = []
        task_uid_list = []
        for i in push_notify_wait:
            if i.device_code:
                task_code_list.append(i.uid + ':' + i.device_code)
            else:
                task_uid_list.append(i.uid)
        if push_app_type == 'android' or push_app_type == 'all':
            android_push.u_group_push(task_code_list,
                                      push_content_info.push_title,
                                      push_content_info.push_content,
                                      'qy_uid_device_android',
                                      go_url, redirect_type=push_content_info.redirect_type)
            android_push.u_group_push(task_uid_list,
                                      push_content_info.push_title,
                                      push_content_info.push_content,
                                      'qy_android',
                                      go_url, redirect_type=push_content_info.redirect_type)
        if push_app_type == 'ios' or push_app_type == 'all':
            ios_push.u_group_push(task_code_list,
                                  push_content_info.push_title,
                                  push_content_info.push_content,
                                  'qy_uid_device_ios',
                                  go_url, redirect_type=push_content_info.redirect_type,
                                  production_mode=settings.U_PUSH_PRODUCTION_MODE)
            ios_push.u_group_push(task_uid_list,
                                  push_content_info.push_title,
                                  push_content_info.push_content,
                                  'qy_ios',
                                  go_url, redirect_type=push_content_info.redirect_type,
                                  production_mode=settings.U_PUSH_PRODUCTION_MODE)
        PushNotifyRecord.objects.filter(id__in=task_id_list, task_status=1).update(task_status=2)
    rs.delete('qms_push_user_lock')


@app.task()
def push_to_user(task_id_list, uid_list, push_app_type, push_content_id):
    if uid_list:
        logger.info('[g_push_to_user] receive task %s-%s', push_app_type, push_content_id)
        push_content_info = PushContentInfo.objects.get(id=push_content_id)
        if push_content_info.db_push_target != 'APP':
            return True
        # 获取到需要单独设备发送的数据
        push_notify_wait = PushNotifyRecord.objects.filter(
            id__in=task_id_list,
            task_status=1
        ).all()
        member_objs = GoogleMemberDeviceToken.objects.filter(uid__in=uid_list).all()
        member_dict = {}
        for i in member_objs:
            member_dict.setdefault(i.uid, []).append({'code': i.device_code, 'token': i.device_token})
        task_token_list = []
        for i in push_notify_wait:
            if push_app_type == 'android' or push_app_type == 'all':
                if i.device_code:
                    for j in member_dict[i.uid]:
                        if i.device_code == j['code']:
                            task_token_list.append(j['token'])
                else:
                    for j in member_dict[i.uid]:
                        task_token_list.append(j['token'])
        for i in task_token_list:
            android_push.g_token_push(
                i,
                push_content_info.push_title,
                push_content_info.push_content,
                push_content_info.message_save_hours * 60 * 60
            )
        PushNotifyRecord.objects.filter(id__in=task_id_list, task_status=1).update(task_status=2)


@app.task()
def expired_user_push():
    mc = MemberClient()
    start_time = (datetime.datetime.now() - datetime.timedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:00')
    end_time = (datetime.datetime.now() - datetime.timedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:59')
    success, expired_users = mc.query_expired_users(start_time, end_time)
    logger.info('response data: %s', expired_users)
    if not success:
        return True
    uids = expired_users.get('uid_list')
    if not uids:
        return True
    t_content_record, _ = PushContentInfo.objects.get_or_create(
        content_title='会员权益到期通知',
        push_title='会员权益到期通知',
        push_content='VIP体验权益已到期，新用户充值VIP尽享惊喜折扣，去看看吧~',
        redirect_url='sixfast://com.xg.push/main?type=qeeyou://personal_center',
    )
    v_content_record, _ = PushContentInfo.objects.get_or_create(
        content_title='会员权益到期通知',
        push_title='会员权益到期通知',
        push_content='您的VIP权益已到期，立即续费VIP，享受极速回国体验~',
        redirect_url='sixfast://com.xg.push/main?type=qeeyou://personal_center',
    )
    for member_id in expired_users.get('uid_list'):
        success, result = mc.query(member_id)
        if not success:
            continue
        duration_account = result.get('duration_account')
        max_time = datetime.datetime(year=1990, month=1, day=1)
        max_time_duration_type = 'OVIP'
        for account in duration_account:
            duration_time = datetime.datetime.strptime(account.get('duration_expire_at'), "%Y-%m-%d %H:%M:%S")
            if duration_time > max_time:
                max_time = duration_time
                max_time_duration_type = account.get('duration_type')
        if max_time > datetime.datetime.now():
            continue
        if max_time_duration_type in ("TVIP", ):
            push_record = PushNotifyRecord(
                uid=member_id,
                notify_app_type='all',
                push_content_id=t_content_record.id
            )

        else:
            push_record = PushNotifyRecord(
                uid=member_id,
                notify_app_type='all',
                push_content_id=v_content_record.id
            )
        push_record.save()
    return True