logJobStart($jobId, $queueLockKey); $this->execute(); $this->handleJobSuccess($job, $queueLockKey); return true; } catch (\Exception $e) { return $this->handleJobError($e, $job, $queueLockKey); } } /** * 执行任务 * @throws \Exception */ public function execute() { try { // 获取所有工作台 $workbenches = Workbench::where(['status' => 1, 'type' => 3, 'isDel' => 0,'id' => 256])->order('id desc')->select(); foreach ($workbenches as $workbench) { // 获取工作台配置 $config = WorkbenchGroupPush::where('workbenchId', $workbench->id)->find(); if (!$config) { continue; } //判断是否推送 $isPush = $this->isPush($workbench, $config); if (empty($isPush)) { continue; } // 获取内容库 $contentLibrary = $this->getContentLibrary($workbench, $config); if (empty($contentLibrary)) { continue; } // 处理内容发送 $this->sendMsgToGroup($workbench, $config, $contentLibrary); } } catch (\Exception $e) { Log::error("消息群发任务异常: " . $e->getMessage()); throw $e; } } // 发送消息(支持群推送和好友推送) public function sendMsgToGroup($workbench, $config, $msgConf) { // 消息拼接 msgType(1:文本 3:图片 43:视频 47:动图表情包(gif、其他表情包) 49:小程序/其他:图文、文件) // 当前,type 为文本、图片、动图表情包的时候,content为string, 其他情况为对象 {type: 'file/link/...', url: '', title: '', thunmbPath: '', desc: ''} $targetType = isset($config['targetType']) ? intval($config['targetType']) : 1; // 默认1=群推送 $toAccountId = ''; $username = Env::get('api.username', ''); $password = Env::get('api.password', ''); if (!empty($username) || !empty($password)) { $toAccountId = Db::name('users')->where('account', $username)->value('s2_accountId'); } // 建立WebSocket $wsController = new WebSocketController(['userName' => $username, 'password' => $password, 'accountId' => $toAccountId]); if ($targetType == 1) { // 群推送 $this->sendToGroups($workbench, $config, $msgConf, $wsController); } else { // 好友推送 $this->sendToFriends($workbench, $config, $msgConf, $wsController); } } /** * 发送群消息 */ protected function sendToGroups($workbench, $config, $msgConf, $wsController) { $groups = json_decode($config['groups'], true); if (empty($groups)) { return false; } $groupsData = Db::name('wechat_group')->whereIn('id', $groups)->field('id,wechatAccountId,chatroomId,companyId,ownerWechatId')->select(); if (empty($groupsData)) { return false; } foreach ($msgConf as $content) { $sendData = []; $sqlData = []; foreach ($groupsData as $group) { // msgType(1:文本 3:图片 43:视频 47:动图表情包(gif、其他表情包) 49:小程序/其他:图文、文件) $sqlData[] = [ 'workbenchId' => $workbench['id'], 'contentId' => $content['id'], 'groupId' => $group['id'], 'friendId' => null, 'targetType' => 1, 'wechatAccountId' => $group['wechatAccountId'], 'createTime' => time() ]; // 构建发送数据 $sendData = $this->buildSendData($content, $config, $group['wechatAccountId'], $group['id'], 'group'); if (empty($sendData)) { continue; } //发送消息 foreach ($sendData as $send) { $wsController->sendCommunity($send); } //插入发送记录 Db::name('workbench_group_push_item')->insertAll($sqlData); } } } /** * 发送好友消息 */ protected function sendToFriends($workbench, $config, $msgConf, $wsController) { $friends = json_decode($config['friends'], true); $devices = json_decode($config['devices'] ?? '[]', true); // 如果好友列表为空,则根据设备查询所有好友 if (empty($friends)) { if (empty($devices)) { // 如果没有选择设备,则无法推送 Log::warning('好友推送:未选择设备,无法推送全部好友'); return false; } // 根据设备查询所有好友 $friendsData = Db::table('s2_company_account') ->alias('ca') ->join(['s2_wechat_account' => 'wa'], 'ca.id = wa.deviceAccountId') ->join(['s2_wechat_friend' => 'wf'], 'wf.wechatAccountId = wa.id') ->where([ 'ca.status' => 0, 'wf.isDeleted' => 0, 'wa.deviceAlive' => 1, 'wa.wechatAlive' => 1 ]) ->whereIn('wa.currentDeviceId', $devices) ->field('wf.id,wf.wechatAccountId,wf.wechatId,wf.ownerWechatId') ->group('wf.id') ->select(); } else { // 查询指定的好友信息 $friendsData = Db::table('s2_wechat_friend') ->whereIn('id', $friends) ->where('isDeleted', 0) ->field('id,wechatAccountId,wechatId,ownerWechatId') ->select(); } if (empty($friendsData)) { return false; } // 获取所有已推送的好友ID列表(去重,不限制时间范围,用于过滤避免重复推送) $sentFriendIds = Db::name('workbench_group_push_item') ->where('workbenchId', $workbench->id) ->where('targetType', 2) ->column('friendId'); $sentFriendIds = array_filter($sentFriendIds); // 过滤null值 $sentFriendIds = array_unique($sentFriendIds); // 去重 // 获取今日已推送的好友ID列表(用于计算今日推送人数) $today = date('Y-m-d'); $todayStartTimestamp = strtotime($today . ' 00:00:00'); $todayEndTimestamp = strtotime($today . ' 23:59:59'); $todaySentFriendIds = Db::name('workbench_group_push_item') ->where('workbenchId', $workbench->id) ->where('targetType', 2) ->whereTime('createTime', 'between', [$todayStartTimestamp, $todayEndTimestamp]) ->column('friendId'); $todaySentFriendIds = array_filter($todaySentFriendIds); // 过滤null值 $todaySentFriendIds = array_unique($todaySentFriendIds); // 去重 // 过滤掉所有已推送的好友(不限制时间范围,避免重复推送) $friendsData = array_filter($friendsData, function($friend) use ($sentFriendIds) { return !in_array($friend['id'], $sentFriendIds); }); if (empty($friendsData)) { Log::info('好友推送:所有好友都已推送过'); return false; } // 重新索引数组 $friendsData = array_values($friendsData); // 计算剩余可推送人数(基于今日推送人数) $todaySentCount = count($todaySentFriendIds); $maxPerDay = intval($config['maxPerDay']); $remainingCount = $maxPerDay - $todaySentCount; if ($remainingCount <= 0) { Log::info('好友推送:今日推送人数已达上限'); return false; } // 限制本次推送人数(不超过剩余可推送人数) $friendsData = array_slice($friendsData, 0, $remainingCount); // 批量处理:每批最多500人 $batchSize = 500; $batches = array_chunk($friendsData, $batchSize); foreach ($msgConf as $content) { foreach ($batches as $batchIndex => $batch) { $sqlData = []; foreach ($batch as $friend) { // 构建发送数据 $sendData = $this->buildSendData($content, $config, $friend['wechatAccountId'], $friend['id'], 'friend'); if (empty($sendData)) { continue; } // 发送个人消息 foreach ($sendData as $send) { if ($send['msgType'] == 49){ $sendContent = json_encode($send['content'], 256); } else { $sendContent = $send['content']; } $wsController->sendPersonal([ 'wechatFriendId' => $friend['id'], 'wechatAccountId' => $friend['wechatAccountId'], 'msgType' => $send['msgType'], 'content' => $sendContent, ]); } // 准备插入发送记录 $sqlData[] = [ 'workbenchId' => $workbench['id'], 'contentId' => $content['id'], 'groupId' => null, 'friendId' => $friend['id'], 'targetType' => 2, 'wechatAccountId' => $friend['wechatAccountId'], 'createTime' => time() ]; } // 批量插入发送记录 if (!empty($sqlData)) { Db::name('workbench_group_push_item')->insertAll($sqlData); Log::info("好友推送:第" . ($batchIndex + 1) . "批,推送了" . count($sqlData) . "个好友"); } // 如果不是最后一批,等待一下再处理下一批(避免一次性推送太多) if ($batchIndex < count($batches) - 1) { sleep(1); // 等待1秒 } } } } /** * 构建发送数据 */ protected function buildSendData($content, $config, $wechatAccountId, $targetId, $type = 'group') { $sendData = []; // 内容处理 if (!empty($content['content'])) { // 京东转链 if (!empty($config['promotionSiteId'])) { $WorkbenchController = new WorkbenchController(); $jdLink = $WorkbenchController->changeLink($content['content'], $config['promotionSiteId']); $jdLink = json_decode($jdLink, true); if ($jdLink['code'] == 200) { $content['content'] = $jdLink['data']; } } if ($type == 'group') { $sendData[] = [ 'content' => $content['content'], 'msgType' => 1, 'wechatAccountId' => $wechatAccountId, 'wechatChatroomId' => $targetId, ]; } else { $sendData[] = [ 'content' => $content['content'], 'msgType' => 1, ]; } } // 根据内容类型处理 switch ($content['contentType']) { case 1: // 图片解析 $imgs = json_decode($content['resUrls'], true); if (!empty($imgs)) { foreach ($imgs as $img) { if ($type == 'group') { $sendData[] = [ 'content' => $img, 'msgType' => 3, 'wechatAccountId' => $wechatAccountId, 'wechatChatroomId' => $targetId, ]; } else { $sendData[] = [ 'content' => $img, 'msgType' => 3, ]; } } } break; case 2: // 链接解析 $url = json_decode($content['urls'], true); if (!empty($url[0])) { $url = $url[0]; $linkContent = [ 'desc' => $url['desc'], 'thumbPath' => $url['image'], 'title' => $url['desc'], 'type' => 'link', 'url' => $url['url'], ]; if ($type == 'group') { $sendData[] = [ 'content' => $linkContent, 'msgType' => 49, 'wechatAccountId' => $wechatAccountId, 'wechatChatroomId' => $targetId, ]; } else { $sendData[] = [ 'content' => $linkContent, 'msgType' => 49, ]; } } break; case 3: // 视频解析 $video = json_decode($content['resUrls'], true); if (!empty($video)) { $video = $video[0]; } if ($type == 'group') { $sendData[] = [ 'content' => $video, 'msgType' => 43, 'wechatAccountId' => $wechatAccountId, 'wechatChatroomId' => $targetId, ]; } else { $sendData[] = [ 'content' => $video, 'msgType' => 43, ]; } break; } return $sendData; } /** * 记录发送历史 * @param Workbench $workbench * @param array $devices * @param array $contentLibrary */ protected function recordSendHistory($workbench, $devices, $contentLibrary) { $now = time(); $data = []; foreach ($devices as $device) { $data = [ 'workbenchId' => $workbench->id, 'deviceId' => $device['deviceId'], 'contentId' => $contentLibrary['id'], 'wechatAccountId' => $device['wechatAccountId'], 'createTime' => $now, ]; Db::name('workbench_group_push_item')->insert($data); } } /** * 获取设备列表 * @param Workbench $workbench 工作台 * @param WorkbenchGroupPush $config 配置 * @return array|bool */ protected function isPush($workbench, $config) { // 检查发送间隔(新逻辑:根据startTime、endTime、maxPerDay动态计算) $today = date('Y-m-d'); $startTimestamp = strtotime($today . ' ' . $config['startTime'] . ':00'); $endTimestamp = strtotime($today . ' ' . $config['endTime'] . ':00'); // 如果时间不符,则跳过 if (($startTimestamp > time() || $endTimestamp < time()) && empty($config['pushType'])) { return false; } $totalSeconds = $endTimestamp - $startTimestamp; if ($totalSeconds <= 0 || empty($config['maxPerDay'])) { return false; } $targetType = isset($config['targetType']) ? intval($config['targetType']) : 1; // 默认1=群推送 if ($targetType == 2) { // 好友推送:maxPerDay表示每日推送人数 // 查询今日已推送的好友ID列表(去重,仅统计今日) $sentFriendIds = Db::name('workbench_group_push_item') ->where('workbenchId', $workbench->id) ->where('targetType', 2) ->whereTime('createTime', 'between', [$startTimestamp, $endTimestamp]) ->column('friendId'); $sentFriendIds = array_filter($sentFriendIds); // 过滤null值 $count = count(array_unique($sentFriendIds)); // 去重后统计今日推送人数 if ($count >= $config['maxPerDay']) { return false; } // 计算本次同步的最早允许时间(按人数计算间隔) $interval = floor($totalSeconds / $config['maxPerDay']); $nextSyncTime = $startTimestamp + $count * $interval; if (time() < $nextSyncTime) { return false; } } else { // 群推送:maxPerDay表示每日推送次数 $interval = floor($totalSeconds / $config['maxPerDay']); // 查询今日已同步次数 $count = Db::name('workbench_group_push_item') ->where('workbenchId', $workbench->id) ->where('targetType', 1) ->whereTime('createTime', 'between', [$startTimestamp, $endTimestamp]) ->count(); if ($count >= $config['maxPerDay']) { return false; } // 计算本次同步的最早允许时间 $nextSyncTime = $startTimestamp + $count * $interval; if (time() < $nextSyncTime) { return false; } } return true; } /** * 获取内容库 * @param Workbench $workbench 工作台 * @param WorkbenchGroupPush $config 配置 * @return array|bool */ protected function getContentLibrary($workbench, $config) { $contentids = json_decode($config['contentLibraries'], true); if (empty($contentids)) { return false; } $targetType = isset($config['targetType']) ? intval($config['targetType']) : 1; // 默认1=群推送 if ($config['pushType'] == 1) { $limit = 10; } else { $limit = 1; } //推送顺序 if ($config['pushOrder'] == 1) { $order = 'ci.sendTime desc, ci.id asc'; } else { $order = 'ci.sendTime desc, ci.id desc'; } // 基础查询,根据targetType过滤记录 $query = Db::name('content_library')->alias('cl') ->join('content_item ci', 'ci.libraryId = cl.id') ->join('workbench_group_push_item wgpi', 'wgpi.contentId = ci.id and wgpi.workbenchId = ' . $workbench->id . ' and wgpi.targetType = ' . $targetType, 'left') ->where(['cl.isDel' => 0, 'ci.isDel' => 0]) ->where('ci.sendTime <= ' . (time() + 60)) ->whereIn('cl.id', $contentids) ->field([ 'ci.id', 'ci.libraryId', 'ci.contentType', 'ci.title', 'ci.content', 'ci.resUrls', 'ci.urls', 'ci.comment', 'ci.sendTime' ]); // 复制 query $query2 = clone $query; $query3 = clone $query; // 根据isLoop处理不同的发送逻辑 if ($config['isLoop'] == 1) { // 可以循环发送(只有群推送时才能为1) // 1. 优先获取未发送的内容 $unsentContent = $query->where('wgpi.id', 'null') ->order($order) ->limit(0, $limit) ->select(); if (!empty($unsentContent)) { return $unsentContent; } $lastSendData = Db::name('workbench_group_push_item') ->where('workbenchId', $workbench->id) ->where('targetType', $targetType) ->order('id desc') ->find(); $fastSendData = Db::name('workbench_group_push_item') ->where('workbenchId', $workbench->id) ->where('targetType', $targetType) ->order('id asc') ->find(); if (empty($lastSendData) || empty($fastSendData)) { return []; } $sentContent = $query2->where('wgpi.contentId', '<', $lastSendData['contentId'])->order('wgpi.id ASC')->group('wgpi.contentId')->limit(0, $limit)->select(); if (empty($sentContent)) { $sentContent = $query3->where('wgpi.contentId', '=', $fastSendData['contentId'])->order('wgpi.id ASC')->group('wgpi.contentId')->limit(0, $limit)->select(); } return $sentContent; } else { // 不能循环发送,只获取未发送的内容(好友推送时isLoop=0) $list = $query->where('wgpi.id', 'null') ->order($order) ->limit(0, $limit) ->select(); return $list; } } /** * 记录任务开始 * @param string $jobId * @param string $queueLockKey */ protected function logJobStart($jobId, $queueLockKey) { Log::info('开始处理工作台消息群发任务: ' . json_encode([ 'jobId' => $jobId, 'queueLockKey' => $queueLockKey ])); } /** * 处理任务成功 * @param Job $job * @param string $queueLockKey */ protected function handleJobSuccess($job, $queueLockKey) { $job->delete(); Cache::rm($queueLockKey); Log::info('工作台消息群发任务执行成功'); } /** * 处理任务错误 * @param \Exception $e * @param Job $job * @param string $queueLockKey * @return bool */ protected function handleJobError(\Exception $e, $job, $queueLockKey) { Log::error('工作台消息群发任务异常:' . $e->getMessage()); if (!empty($queueLockKey)) { Cache::rm($queueLockKey); Log::info("由于异常释放队列锁: {$queueLockKey}"); } if ($job->attempts() > self::MAX_RETRY_ATTEMPTS) { $job->delete(); } else { $job->release(Config::get('queue.failed_delay', 10)); } return false; } }