yzk_wechat_event/apps/jqr/tasks.py

from celery import shared_task
from datetime import datetime, date, timedelta
import logging

from django.db.models import Count, Sum, When, Case, Value

from apps.jqr.choices import JqrWechatbizuserinfoDeleteTypeChoices, JqrAddTypeChoices
from apps.qc.choices import QcWechatbizeventAddcontactIsDeleteChoices, QcQrcodesEditLogOperateTypeChoices, \
    QcQrcodesEditLogTypeChoices
from apps.qc.models import QcWechatbizeventAddcontact, QcQrcodes, QcWechatbizuserinfo, QcQrcodesEditLog, QcCorpinfo
from apps.jqr.models import JqrExternalFollowUser, JqrExternalUser
from apps.qc.utils import generate_qrcode_by_qrcode
from apps.warning.models import QcWarningRule
from libs.wechat import WechatWorkerUtil

logger = logging.getLogger('apps')


@shared_task(name='save_add_contact', queue='contact')
def save_add_contact(data, corpinfo, *args, **kwargs):
    state = data.get('state')
    userid = data.get('userid')
    if state and state.startswith('mg') and '_' in state:
        [_, _, qrcodeid] = state.split('_')
        data['qrcodeid'] = qrcodeid
        # TODO 异步执行
        check_qrcode(qrcodeid, userid, corpinfo)
    externaluserid = data.get('externaluserid')
    corpid = data.get('corpid')
    QcWechatbizeventAddcontact.objects.update_or_create(
        userid=userid,
        externaluserid=externaluserid,
        corpid=corpid,
        defaults={
            **data,
            'isdelete': QcWechatbizeventAddcontactIsDeleteChoices.NOT_DELETE,
            'deletetime': None,
            'ctime': datetime.now(),
        }
    )
    edit_add_contact.delay(data, corpinfo, *args, **kwargs)


@shared_task(name='edit_add_contact', queue='contact')
def edit_add_contact(data, corpinfo, *args, **kwargs):
    # 更新 关系表
    externaluserid = data.get('externaluserid')
    corpid = corpinfo.get('corpid')
    appsecret = corpinfo.get('appsecret')
    wechat_worker = WechatWorkerUtil(corpid, appsecret)
    cursor = ''
    while cursor is not None:
        success, data = wechat_worker.get_external_contact(externaluserid, cursor=cursor)
        if not success:
            logger.error(f'获取外部联系人信息失败，{data}')
            return
        (external_contact, follow_user, cursor) = data
        JqrExternalUser.objects.update_or_create(
            corpid=corpid,
            external_userid=externaluserid,
            defaults={
                **external_contact,
                'addtype': JqrAddTypeChoices.EVENT_CALLBACK,
                'deletetype': None,
                'dtime': None
            }
        )
        for follow_info in follow_user:
            userid = follow_info.get('userid')
            follow_info['tags'] = [tag.get('tag_id') for tag in follow_info.get('tags', [])]
            JqrExternalFollowUser.objects.update_or_create(
                userid=userid,
                corpid=corpid,
                external_userid=externaluserid,
                defaults={
                    **follow_info,
                    'addtype': JqrAddTypeChoices.EVENT_CALLBACK,
                    'deletetype': None,
                    'dtime': None
                }
            )


@shared_task(name='delete_add_contact', queue='contact')
def delete_add_contact(data, corpinfo, *args, **kwargs):
    userid = data.get('userid')
    externaluserid = data.get('externaluserid')
    corpid = data.get('corpid')
    agentid = data.get('agentid')
    # 更新add_contact 表删除字段
    QcWechatbizeventAddcontact.objects.filter(userid=userid, externaluserid=externaluserid, corpid=corpid,
                                              agentid=agentid).update(
        isdelete=QcWechatbizeventAddcontactIsDeleteChoices.DELETED,
        deletetime=datetime.now()
    )
    JqrExternalFollowUser.objects.filter(userid=userid, external_userid=externaluserid, corpid=corpid).update(
        deletetype=JqrWechatbizuserinfoDeleteTypeChoices.EVENT_CALLBACK,
        dtime=datetime.now()
    )


@shared_task(name='check_qrcode', queue='qrcode')
def check_qrcode(qrcodeid, userid, corpinfo):
    qrcode = QcQrcodes.objects.get(id=qrcodeid)
    users = qrcode.userids
    corpid = qrcode.corpid
    userinfo = QcWechatbizuserinfo.objects.get(corpid=corpid, userid=userid)
    today = date.today()
    # 账号当前活码的接粉情况
    count_info = QcWechatbizeventAddcontact.objects.filter(corpid=corpinfo.get('corpid'), userid=userid).values(
        'corpid',
        'userid').aggregate(
        total_count=Count('*'),
        today_count=Sum(
            Case(
                When(ctime__date=today, then=Value(1)),
                default=Value(0),
            )
        ),
        qrcode_count=Sum(
            Case(
                When(qrcodeid=qrcodeid, then=Value(1)),
                default=Value(0),
            )
        ),
        qrcode_today_count=Sum(
            Case(
                When(qrcodeid=qrcodeid, ctime__date=today, then=Value(1)),
                default=Value(0),
            )
        )
    )
    # 该账号总共加了多少人
    total_count = count_info.get('total_count', 0)
    # 该账号今天加了多少人
    today_count = count_info.get('today_count', 0)
    # 该账号当前活码加了多少人
    qrcode_count = count_info.get('qrcode_count')
    # 该账号今天当前活码加了多少人
    qrcode_today_count = count_info.get('qrcode_today_count')
    # 该用户所有活码每日新增好友上限
    limitaddusercount = userinfo.limitaddusercount
    # 该用户总加粉总上限
    totallimitaddusercount = userinfo.totallimitaddusercount
    # 该用户当前活码上限
    user_qrcode_limitcount = None
    for user in users:
        if user.get('userid') == userid:
            user_qrcode_limitcount = user.get('limitcount')
            break

    user_count = len(users)
    offline_user_count = len(
        [user for user in users if user.get('isonline') == QcQrcodesEditLogOperateTypeChoices.OFFLINE])
    # 该账号是否需要下线
    need_offline = False
    # 该账号今天当前活码加的人数 大于 该用户当前活码上限
    if user_qrcode_limitcount is not None and qrcode_today_count > user_qrcode_limitcount > 0:
        need_offline = True
    if total_count is not None and total_count > totallimitaddusercount > 0:
        need_offline = True
    if today_count is not None and today_count > limitaddusercount > 0:
        need_offline = True

    if need_offline:
        if user_count - offline_user_count > 1:
            for user in users:
                if user.get('userid') == userid:
                    user['isonline'] = QcQrcodesEditLogOperateTypeChoices.OFFLINE
                    user['endtime'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            qrcode.save()
            # 重新生成活码, 请求企微api更新活码
            generate_qrcode_by_qrcode(qrcode, corpinfo, is_update=True)
        # 生成活码修改记录
        QcQrcodesEditLog.objects.create(
            userid=userid,
            uid=qrcode.uid,
            corpid=qrcode.corpid,
            agentid=qrcode.agentid,
            qrcodeid=qrcode.pk,
            detail={
                'totallimit': totallimitaddusercount,
                'daliylimit': limitaddusercount,
                'qrcodelimit': user_qrcode_limitcount,
                'addcount': total_count,
                'qrcodeaddcount': qrcode_count,
                'qrcode_today_count': qrcode_today_count,
            },
            type=QcQrcodesEditLogTypeChoices.AUTO,
            operatetype=QcQrcodesEditLogOperateTypeChoices.OFFLINE,
        )


# 接粉号报警检测任务
@shared_task(name='check_follow_user', queue='check_follow_user')
def check_follow_user():
    batch_size = 1000
    warning_rules = QcWarningRule.objects.filter(is_on=True, is_delete=False)
    count = warning_rules.count()
    for i in range(0, count, batch_size):
        for warning_rule in warning_rules[i:i + batch_size]:
            users = warning_rule.userids
            users = [{"alias": "饿了么小霸王155", "corpid": "ww4450b72962300373", "userid": "mg12976",
                      "username": "[155]5G-8-1-汪琦"},
                     {"alias": "饿了么小霸王157", "corpid": "ww4450b72962300373", "userid": "mg04598",
                      "username": "[157]5G-3-2-汪潇"}]
            if users and len(users) > 0:
                corpid = users[0]["corpid"]
                userid_map = {user.get('userid'): user for user in users}
                warning_interval = warning_rule.warning_interval
                now = datetime.now()
                warning_interval_ago = now - timedelta(minutes=warning_interval)
                user_count_info = QcWechatbizeventAddcontact.objects.filter(corpid=corpid,
                                                                            userid__in=list(userid_map.keys())).values(
                    'corpid',
                    'userid').annotate(
                    warning_interval_count=Sum(
                        Case(
                            When(ctime__gte=warning_interval_ago, then=Value(1)),
                            default=Value(0),
                        )
                    ),
                ).values('userid', 'warning_interval_count')
                for user_count in user_count_info:
                    if user_count.get('warning_interval_count') < 1:
                        user = userid_map.get(user_count.get('userid'))
                        text = f"账号 {user.get('userid')} {user.get('username')} {user.get('alias')} 在{warning_interval}分钟内没进粉"
                        log_info = {
                            'corpid': corpid,
                            'userid': user_count.get('userid'),
                        }

            print(warning_rule.warning_setting)


# 活码上线
@shared_task(name='online_qrcode', queue='qrcode')
def online_qrcode():
    batch_size = 1000
    qrcodes = QcQrcodes.objects.all()
    for i in range(0, qrcodes.count(), batch_size):
        for qrcode in qrcodes[i:i + batch_size]:
            users = qrcode.userids
            need_generate_code = False
            # 获取下线的用户
            userids = []
            for user in users:
                if user.get('isonline') == QcQrcodesEditLogOperateTypeChoices.OFFLINE:
                    userids.append(user.get('userid'))
                    need_generate_code = True
                    user['isonline'] = QcQrcodesEditLogOperateTypeChoices.ONLINE
                    user['endtime'] = None
                    # 上线时间
                    user['starttime'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

            if need_generate_code:
                qrcode.save()
                corp = QcCorpinfo.objects.get(corpid=qrcode.corpid, agentid=qrcode.agentid)
                # 重新生成活码, 请求企微api更新活码
                generate_qrcode_by_qrcode(qrcode, corp, is_update=True)
                # 生成活码修改记录
                for userid in userids:
                    QcQrcodesEditLog.objects.create(
                        userid=userid,
                        uid=qrcode.uid,
                        corpid=qrcode.corpid,
                        agentid=qrcode.agentid,
                        qrcodeid=qrcode.pk,
                        detail={
                            'totallimit': 0,
                            'daliylimit': 0,
                            'qrcodelimit': 0,
                            'addcount': 0,
                            'qrcodeaddcount': 0,
                            'qrcode_today_count': 0,
                        },
                        type=QcQrcodesEditLogTypeChoices.AUTO,
                        operatetype=QcQrcodesEditLogOperateTypeChoices.ONLINE,
                    )