Merge branch 'priority' into dev/test

app/Console/Commands/ChatGroup/GroupBasicInfo.php

@@ -84,12 +84,13 @@ class GroupBasicInfo extends Command
     private function getGroupInfo($corpid, $userIds, $source)
     {
         # owner_filter参数处理
-        $userIdList = DjUser::select(['user_id'])->where('corpid', $corpid)->where('enable', 1)
-            ->where(function($query) use($userIds) {
-                if($userIds) $query->whereIn('user_id', $userIds);
-            })
-            ->where('status', 1)->orderBy('id')
-            ->distinct()->pluck('user_id')->toArray();
+//        $userIdList = DjUser::select(['user_id'])->where('corpid', $corpid)->where('enable', 1)
+//            ->where(function($query) use($userIds) {
+//                if($userIds) $query->whereIn('user_id', $userIds);
+//            })
+//            ->where('status', 1)->orderBy('id')
+//            ->distinct()->pluck('user_id')->toArray();
+        $userIdList = DjUser::getActiveUserListById($corpid, $userIds);
 
         $userCount = count($userIdList);
         # 需要截取的段数

app/Console/Commands/CorpInformation/BatchMarkTagDeal.php

@@ -0,0 +1,219 @@
+<?php
+
+namespace App\Console\Commands\CorpInformation;
+
+use App\Log;
+use App\Models\AuthorizeCorp;
+use App\Models\BatchMarkTag;
+use App\Models\BatchMarkTagRecord;
+use App\Models\CustomerDetails;
+use App\Models\DjUser;
+use App\Models\Tag;
+use App\RedisModel;
+use App\Service\TagService;
+use App\Support\EmailQueue;
+use App\Support\qyApi\QyCommon;
+use Illuminate\Console\Command;
+
+class BatchMarkTagDeal extends Command
+{
+    protected $signature = 'BatchMarkTagDeal';
+    protected $description = '客户批量打标签处理';
+
+    protected $emailSender = '猎羽-批量打标签';
+    protected $emailReceiver = ['song.shen@kuxuan-inc.com'];
+    protected $logName = 'BatchMarkTagDeal';
+
+    public function handle()
+    {
+        \DB::connection()->disableQueryLog();
+
+        $beginTime = time();
+        $this->info(date('m-d H:i:s') . ' 开始整理');
+
+        while(true) {
+            $result = $this->taskDeal();
+            if(!$result) sleep(1);
+            $now = time();
+            // 超过10分钟，主动停止循环
+            if ($now - $beginTime > 600) {
+                break;
+            }
+
+            sleep(1);
+        }
+
+        $this->info(date('Y-m-d H:i:s') . ' 整理结束');
+    }
+
+    private function taskDeal()
+    {
+        # 取出数据
+        $redisVal = RedisModel::rPop(BatchMarkTagRecord::BATCH_MARK_TAG_RECORD);
+        if(empty($redisVal))
+            return false;
+        try {
+            $this->createTask($redisVal);
+        } catch (\Exception $e) {
+            EmailQueue::rPush('客户批量打标签处理任务出现异常', $e->getTraceAsString(), $this->emailReceiver, $this->emailSender);
+            Log::logError('程序异常', [
+                'line'  => $e->getLine(), 'msg'   => $e->getMessage(), 'param' => $redisVal
+            ], $this->logName);
+
+            return false;
+        }
+
+        return true;
+    }
+
+    private function createTask($redisVal) {
+        $redisData = json_decode($redisVal, 1);
+        if(empty($redisData) || empty($redisData['task_id']) || empty($redisData['record_id'])) {
+            Log::logError('队列数据异常', [$redisData], $this->logName);
+            EmailQueue::rPush('处理客户批量打标签 - 队里数据异常', $redisVal, $this->emailReceiver, $this->emailSender);
+            return false;
+        }
+
+        $taskId = $redisData['task_id'];
+        $recordId = $redisData['record_id'];
+        $taskInfo = BatchMarkTag::getTaskInfoById($redisData['task_id']);
+        if(empty($taskInfo)) {
+            Log::logError('根据任务ID查询任务信息失败', ['task_id' => $taskId], $this->logName);
+            EmailQueue::rPush('处理客户批量打标签 - 根据任务ID查询任务信息失败', $redisVal, $this->emailReceiver, $this->emailSender);
+            $status = -1;
+            BatchMarkTagRecord::updateData($recordId, ['status' => $status]);
+            return false;
+        }
+
+        $taskInfo = json_decode(json_encode($taskInfo), 1);
+
+        $recordInfo = BatchMarkTagRecord::getInfoById($recordId);
+        if(empty($recordInfo)) {
+            Log::logError('根据记录ID查询记录信息失败', ['record_id' => $recordId], $this->logName);
+            EmailQueue::rPush('处理客户批量打标签 - 根据记录ID查询记录信息失败', $redisVal, $this->emailReceiver, $this->emailSender);
+            $status = -1;
+            BatchMarkTagRecord::updateData($recordId, ['status' => $status]);
+            return false;
+        }
+
+        $recordInfo = json_decode(json_encode($recordInfo), 1);
+
+        $this->dealTask($taskInfo, $recordInfo);
+    }
+
+    private function dealTask($taskInfo, $recordInfo) {
+        $externalUserIdList = explode(',', $recordInfo['external_userid']);
+        if(empty($externalUserIdList)) {
+            Log::logError('根据记录ID查询记录信息失败', ['record_id' => $recordInfo['id']], $this->logName);
+            EmailQueue::rPush('处理客户批量打标签 - 根据记录ID查询记录信息失败', 'task_id:'.$taskInfo['id'].'; record_id:'.$recordInfo['id'], $this->emailReceiver, $this->emailSender);
+            $status = -1;
+            BatchMarkTagRecord::updateData($recordInfo['id'], ['status' => $status]);
+            return false;
+        }
+
+        $tagList = json_decode($taskInfo['task_list'], 1);
+        // 标签转化
+        $newTagList = Tag::query()->where('corpid', $taskInfo['corpid'])->where('enable', 1)
+            ->whereIn('tag_md5', $tagList)->get();
+        $newTagList = $newTagList->isNotEmpty() ? array_column($newTagList->toArray(), 'tag_id') : [];
+
+        $update = ['exec_success' => 0, 'exec_fail' => 0];
+        foreach ($externalUserIdList as $externalUserId) {
+            if(1 == $taskInfo['type']) {
+                $newAddTagIdList = $newTagList;
+                $newRemoveTagIdList = [];
+            } else if(2 == $taskInfo['type']) {
+                $newAddTagIdList = [];
+                $newRemoveTagIdList = $newTagList;
+            }
+
+            $res = $this->markTag($taskInfo['corpid'], $recordInfo['user_id'], $externalUserId, $newAddTagIdList, $newRemoveTagIdList);
+            if($res) {
+                $update['exec_success']++;
+            } else {
+                $update['exec_fail']++;
+            }
+        }
+
+        # 验证总数
+        if($update['success'] + $update['exec_fail'] != $recordInfo['exec_total']) {
+            Log::logError('程序异常', ['record_id' => $recordInfo['id'], 'stat' => $update], $this->logName);
+            EmailQueue::rPush('客户批量打标签 - 处理数据总数与待处理数据总数不一致', json_encode(['record_id' => $recordInfo['id'], 'stat' => $update]), $this->emailReceiver, $this->emailSender);
+        }
+
+        # 更新任务状态
+        if($update['fail'] == $recordInfo['exec_total']) {
+            $update['status'] = -1;
+        } else {
+            $update['status'] = 1;
+        }
+        $res = BatchMarkTagRecord::updateData($recordInfo['id'], $update);
+        if(!$res) {
+            Log::logError('数据更新失败', ['record_id' => $recordInfo['id'], 'update' => $update], $this->logName);
+        }
+
+        # 判断总体任务是否执行完成
+        $this->chargeTaskStatus($taskInfo['id']);
+    }
+
+    private function markTag($corpid, $userId, $externalUserId, $newAddTagIdList, $newRemoveTagIdList, $retry = 0) {
+        $responseData = QyCommon::mark_tag($corpid, $userId, $externalUserId, $newAddTagIdList, $newRemoveTagIdList);
+
+        if(isset($responseData['errcode']) && $responseData['errcode'] == 0) {
+            // 编辑成功了
+            TagService::updateLocalCustomerTagSecond($corpid, $userId, $externalUserId, $newAddTagIdList, $newRemoveTagIdList);
+
+            $state = $data['state'] ?? null;
+            if (!is_null($state)) {
+                // 客户详情10s同步一遍
+                RedisModel::lPush(CustomerDetails::CUSTOMER_DATA_SYNC_RDS, json_encode([
+                    'corpid' => $corpid,
+                    'user_id' => $userId,
+                    'external_userid' => $externalUserId,
+                    'exec_time' => time() + 10
+                ]));
+            }
+
+            return true;
+        } else {
+           if( $responseData['errcode'] != 84061 ) { # 无法处理的状态码：没有好友关系
+                // 判断重试次数
+                if($retry <= 5) {
+                    $retry++;
+                    return $this->markTag($corpid, $userId, $externalUserId, $newAddTagIdList, $newRemoveTagIdList, $retry);
+                } else {
+                    $params = ['corpid' => $corpid, 'user_id' => $userId, 'external_userid' => $externalUserId, 'add_tag_id_list' => $newAddTagIdList, 'remove_tag_id_list' => $newRemoveTagIdList];
+                    // 还是一直失败，则不再请求，日志记录一下
+                    Log::logError('请求企微打标签接口3次重试均失败', [
+                        'param' => $params,
+                        'response' => $responseData,
+                    ], $this->logName);
+                    EmailQueue::rPush('请求企微打标签接口3次重试均失败', json_encode(['params' => $params, 'res' => $responseData], 256), $this->emailReceiver, $this->emailSender);
+                    return false;
+                }
+            } else {
+               //todo 隔1小时报警一次
+
+               return false;
+           }
+        }
+    }
+
+    private function chargeTaskStatus($taskId) {
+        $countData = BatchMarkTagRecord::recordCount($taskId);
+        $total = 0; $exec = 0;
+        if($countData->isNotEmpty()) {
+            foreach ($countData as $countInfo) {
+                if(0 == $countInfo->status) {
+                    $exec += $countInfo->count;
+                }
+
+                $total += $countInfo->count;
+            }
+        }
+
+        if($exec == 0) {
+            BatchMarkTag::updateStatus($taskId, 3);
+        }
+    }
+}

app/Console/Commands/CorpInformation/BatchMarkTagNew.php

@@ -75,40 +75,28 @@ class BatchMarkTagNew extends Command
         }
 
         $taskInfo = json_decode(json_encode($taskInfo), 1);
+        BatchMarkTag::updateStatus($taskId, 2);
 
-        # 客服许可(兼容新授权到系统的企微，此种企微的客服没有许可，但是会有75天的试用期)
-        $corpInfo = AuthorizeCorp::getCorpInfo($taskInfo['corpid']);
-        if(!empty($corpInfo)) {
-            $chargeActiveType = 1;
-            $activeUserList = [];
-            if (strtotime($corpInfo->created_at) <= strtotime('-75 days')) {
-                $chargeActiveType = 2;
-                $activeUserList = DjUser::getActiveUserListById($taskInfo['corpid'], []);
-                $activeUserList = json_decode(json_encode($activeUserList), 1);
-                if(empty($activeUserList)) {
-                    # 选择的客服列表全部没有许可，不去查询客户列表，直接更新状态
-                    $status = -1;
-                    BatchMarkTag::updateStatus($taskId, $status);
-                    Log::logError('所属企微的客服全部没有许可信息', ['task_id' => $taskId, 'corpid' => $taskInfo['corpid']], $this->logName);
-                    return false;
-                }
-            }
-
-            # 直接筛选出要处理的客户列表，方便生成记录表数据
-            if(0 == $taskInfo['select_all']) {
-                $this->dealSelectTask($taskInfo, $chargeActiveType, $activeUserList);
-            } else {
-                $this->dealFilterTask($taskInfo, $chargeActiveType, $activeUserList);
-            }
-        } else {
+        $activeUserList = DjUser::getActiveUserListById($taskInfo['corpid']);
+        if(empty($activeUserList)) {
+            # 选择的客服列表全部没有许可，不去查询客户列表，直接更新状态
             $status = -1;
             BatchMarkTag::updateStatus($taskId, $status);
-            Log::logError('企微信息异常', ['task_id' => $taskId], $this->logName);
-            EmailQueue::rPush('客户批量打标签-企微信息查询异常', 'task_id:'.$taskId, $this->emailReceiver, $this->emailSender);
+            Log::logError('所属企微的客服全部没有许可信息', ['task_id' => $taskId, 'corpid' => $taskInfo['corpid']], $this->logName);
+            return false;
         }
+
+
+        # 直接筛选出要处理的客户列表，方便生成记录表数据
+        if(0 == $taskInfo['select_all']) {
+            $this->dealSelectTask($taskInfo, $activeUserList);
+        } else {
+            $this->dealFilterTask($taskInfo, $activeUserList);
+        }
+
     }
 
-    private function dealSelectTask($taskInfo, $chargeActiveType, $activeUserList) {
+    private function dealSelectTask($taskInfo, $activeUserList) {
         $customerList = json_decode($taskInfo['customer_list'], 1);
         $taskId = $taskInfo['id'];
         $corpid = $taskInfo['corpid'];
@@ -124,7 +112,7 @@ class BatchMarkTagNew extends Command
         }
 
         foreach ($userCustomerList as $userId => $childCustomerIdListArr) {
-            if(2 == $chargeActiveType && !in_array($userId, $activeUserList)) {
+            if(!in_array($userId, $activeUserList)) {
                 Log::logInfo('客服无许可，跳过', ['user_id' => $userId, 'customer_list' => $childCustomerIdListArr], $this->logName);
                 continue;
             }
@@ -148,7 +136,7 @@ class BatchMarkTagNew extends Command
                     if(!$result) {
                         Log::logError('记录写入失败', ['task_info' => $taskInfo], $this->logName);
                     } else {
-                        RedisModel::rPush(BatchMarkTagRecord::BATCH_MARK_TAG_RECORD, $result->id);
+                        RedisModel::rPush(BatchMarkTagRecord::BATCH_MARK_TAG_RECORD, json_encode(['task_id' => $taskId, 'record_id' => $result->id]));
                     }
                 }
             }
@@ -179,10 +167,11 @@ class BatchMarkTagNew extends Command
         return $data;
     }
 
-    private function dealFilterTask($taskInfo, $chargeActiveType, $activeUserList)
+    private function dealFilterTask($taskInfo, $activeUserList)
     {
         $params = json_decode($taskInfo['params'], 1);
         $params['corpid'] = $taskInfo['corpid'];
+        $params['user_id_list'] = !is_array($params['user_id_list']) ? json_decode($params['user_id_list']) : $params['user_id_list'];
         $source = 1;
 
         $filterCustomerList = json_decode($taskInfo['filter_customer_list'], 1);
@@ -195,11 +184,7 @@ class BatchMarkTagNew extends Command
         # 先查询出所有需要执行的客服，再按照客服来查询客户列表
         $userIdList = $params['user_id_list'];
         if (empty($userIdList)) {
-            if (1 == $chargeActiveType) {
-                $userIdList = DjUser::query()->where('corpid', $taskInfo['corpid'])->where('enable', 1)->pluck('user_id')->toArray();
-            } else {
-                $userIdList = $activeUserList;
-            }
+            $userIdList = $activeUserList;
         } else {
             $userIdList = array_intersect($params['user_id_list'], $activeUserList);
         }
@@ -230,7 +215,7 @@ class BatchMarkTagNew extends Command
                         if (!$result) {
                             Log::logError('记录写入失败', ['task_info' => $taskInfo], $this->logName);
                         } else {
-                            RedisModel::rPush(BatchMarkTagRecord::BATCH_MARK_TAG_RECORD, $result->id);
+                            RedisModel::rPush(BatchMarkTagRecord::BATCH_MARK_TAG_RECORD, json_encode(['task_id' => $taskInfo['id'], 'record' => $result->id]));
                         }
                     }
                 }

app/Console/Kernel.php

@@ -48,6 +48,7 @@ use App\Console\Commands\CleanData;
 use App\Console\Commands\ContactChangeRecycle;
 use App\Console\Commands\CorpAccessTokenRefresh;
 use App\Console\Commands\CorpDataSync;
+use App\Console\Commands\CorpInformation\BatchMarkTagDeal;
 use App\Console\Commands\CorpInformation\BatchMarkTagNew;
 use App\Console\Commands\CorpInformation\CustomerPayTagDeal;
 use App\Console\Commands\CorpInformation\LossContactCntStat;
@@ -479,7 +480,8 @@ class Kernel extends ConsoleKernel
         SyncYunXuanOrder::class,                // 同步云选联盟订单
         ECommerceUserActionsDataUpload::class,  // 电商直投订单数据回传
         OrganizationAccountGet::class,          // 查询组织下的广告主信息
-        BatchMarkTagNew::class,                 // 客户批量打标签（新）
+        BatchMarkTagNew::class,                 // 客户批量打标签任务分发（新）
+        BatchMarkTagDeal::class,                // 客户批量打标签任务处理
     ];
 
     /**

app/Models/BatchMarkTagRecord.php

@@ -24,4 +24,17 @@ class BatchMarkTagRecord extends Model
 
         return $model;
     }
+
+    public static function getInfoById($recordId) {
+        return self::query()->where('enable', 1)->where('id', $recordId)->first();
+    }
+
+    public static function updateData($recordId, $update) {
+        return self::query()->where('id', $recordId)->update($update);
+    }
+
+    public static function recordCount($taskId) {
+        $model = self::query()->where('enable', 1)->where('task_id', $taskId);
+        return $model->selectRaw('status, count(1) as count')->groupBy(['status'])->get();
+    }
 }

app/Models/DjUser.php

@@ -224,11 +224,22 @@ class DjUser extends Model
             ->where('corpid', $corpid)->get()->pluck('user_id');
     }
 
-    public static function getActiveUserListById($corpid, $userIdList)  {
-        return self::query()->where('corpid', $corpid)->where(function($query) use($userIdList){
-            if(!empty($userIdList)) $query->whereIn('user_id', $userIdList);
-        })->where('enable', 1)->where('status', 1)->where('is_active', 1)
-            ->pluck('user_id');
+    public static function getActiveUserListById($corpid, $userIdList = null)  {
+        $corpInfo = AuthorizeCorp::getCorpInfo($corpid);
+        # 客服许可(兼容新授权到系统的企微，此种企微的客服没有许可，但是会有75天的试用期)
+        if(!empty($corpInfo)) {
+            $model = self::query()->where('corpid', $corpid)->where(function ($query) use ($userIdList) {
+                if (!empty($userIdList)) $query->whereIn('user_id', $userIdList);
+            })->where('enable', 1)->where('status', 1);
+
+            if (strtotime($corpInfo->created_at) <= strtotime('-75 days')) {
+                $model = $model->where('is_active', 1);
+            }
+
+            return $model->pluck('user_id')->toArray();
+        } else {
+            return [];
+        }
     }
 
     public static function getActiveUserListByPage($startId, $limit, $corpid = null) {

app/Service/ChatGroup/ChatGroupService.php

@@ -848,9 +848,10 @@ class ChatGroupService
         Log::logInfo('start', [time()], 'syncCorpChatGroup');
         try{
             # owner_filter参数处理
-            $userIdList = DjUser::select(['user_id'])->where('corpid', $corpid)->where('enable', 1)
-                ->where('status', 1)->orderBy('id')
-                ->distinct()->pluck('user_id')->toArray();
+//            $userIdList = DjUser::select(['user_id'])->where('corpid', $corpid)->where('enable', 1)
+//                ->where('status', 1)->orderBy('id')
+//                ->distinct()->pluck('user_id')->toArray();
+            $userIdList = DjUser::getActiveUserListById($corpid);
 
             if(empty($userIdList)) {
                 return ['待同步成员列表为空', 2014];