513 lines
20 KiB
PHP
513 lines
20 KiB
PHP
<?php
|
||
|
||
namespace app\job;
|
||
|
||
use app\api\controller\WebSocketController;
|
||
use app\cunkebao\model\Workbench;
|
||
use app\cunkebao\model\WorkbenchGroupCreate;
|
||
use app\api\model\WechatFriendModel as WechatFriend;
|
||
use app\api\model\WechatMomentsModel as WechatMoments;
|
||
use app\common\model\DeviceWechatLogin as DeviceWechatLoginModel;
|
||
use think\facade\Log;
|
||
use think\facade\Env;
|
||
use think\Db;
|
||
use think\queue\Job;
|
||
use think\facade\Cache;
|
||
use think\facade\Config;
|
||
use app\api\controller\MomentsController as Moments;
|
||
use Workerman\Lib\Timer;
|
||
use app\api\controller\WechatController;
|
||
use think\Queue;
|
||
|
||
/**
|
||
* 工作台群创建任务
|
||
* Class WorkbenchGroupCreateJob
|
||
* @package app\job
|
||
*/
|
||
class WorkbenchGroupCreateJob
|
||
{
|
||
|
||
/**
|
||
* 最大重试次数
|
||
*/
|
||
const MAX_RETRY_ATTEMPTS = 3;
|
||
|
||
/**
|
||
* 队列任务处理
|
||
* @param Job $job 队列任务
|
||
* @param array $data 任务数据
|
||
* @return bool
|
||
*/
|
||
public function fire(Job $job, $data)
|
||
{
|
||
$jobId = $data['jobId'] ?? '';
|
||
$queueLockKey = $data['queueLockKey'] ?? '';
|
||
try {
|
||
$this->logJobStart($jobId, $queueLockKey);
|
||
$this->execute();
|
||
$this->handleJobSuccess($job, $queueLockKey);
|
||
return true;
|
||
} catch (\Exception $e) {
|
||
return $this->handleJobError($e, $job, $queueLockKey);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 成员类型常量
|
||
*/
|
||
const MEMBER_TYPE_OWNER = 1; // 群主成员
|
||
const MEMBER_TYPE_ADMIN = 2; // 管理员
|
||
const MEMBER_TYPE_OWNER_FRIEND = 3; // 群主好友
|
||
const MEMBER_TYPE_ADMIN_FRIEND = 4; // 管理员好友
|
||
|
||
/**
|
||
* 状态常量
|
||
*/
|
||
const STATUS_PENDING = 0; // 待创建
|
||
const STATUS_CREATING = 1; // 创建中
|
||
const STATUS_SUCCESS = 2; // 创建成功
|
||
const STATUS_FAILED = 3; // 创建失败
|
||
const STATUS_ADMIN_FRIEND_ADDED = 4; // 管理员好友已拉入
|
||
|
||
/**
|
||
* 执行任务
|
||
* @throws \Exception
|
||
*/
|
||
public function execute()
|
||
{
|
||
try {
|
||
// 1. 查询启用了建群功能的数据
|
||
$workbenches = Workbench::where(['status' => 1, 'type' => 4, 'isDel' => 0,'id' => 315])->order('id desc')->select();
|
||
foreach ($workbenches as $workbench) {
|
||
// 获取工作台配置
|
||
$config = WorkbenchGroupCreate::where('workbenchId', $workbench->id)->find();
|
||
if (!$config) {
|
||
continue;
|
||
}
|
||
|
||
// 解析配置
|
||
$config['poolGroups'] = json_decode($config['poolGroups'], true);
|
||
$config['devices'] = json_decode($config['devices'], true);
|
||
$config['admins'] = json_decode($config['admins'] ?? '[]', true) ?: [];
|
||
|
||
if (empty($config['poolGroups']) || empty($config['devices'])) {
|
||
continue;
|
||
}
|
||
$groupMember = [];
|
||
$wechatId = Db::name('device_wechat_login')
|
||
->whereIn('deviceId',$config['devices'])
|
||
->order('id desc')
|
||
->value('wechatId');
|
||
if (empty($wechatId)) {
|
||
continue;
|
||
}
|
||
$groupMember[] = $wechatId;
|
||
// 获取群主好友ID映射（所有群主的好友）
|
||
$groupMemberWechatId = [];
|
||
$groupMemberId = [];
|
||
|
||
foreach ($groupMember as $ownerWechatId) {
|
||
$friends = Db::table('s2_wechat_friend')
|
||
->where('ownerWechatId', $ownerWechatId)
|
||
->whereIn('wechatId', $groupMember)
|
||
->field('id,wechatId')
|
||
->select();
|
||
|
||
foreach ($friends as $friend) {
|
||
if (!isset($groupMemberWechatId[$friend['id']])) {
|
||
$groupMemberWechatId[$friend['id']] = $friend['wechatId'];
|
||
$groupMemberId[] = $friend['id'];
|
||
}
|
||
}
|
||
}
|
||
if (empty($groupMemberWechatId)) {
|
||
continue;
|
||
}
|
||
|
||
// 获取流量池用户
|
||
$poolItem = Db::name('traffic_source_package_item')
|
||
->whereIn('packageId', $config['poolGroups'])
|
||
->group('identifier')
|
||
->column('identifier');
|
||
|
||
|
||
if (empty($poolItem)) {
|
||
continue;
|
||
}
|
||
|
||
// 获取已入群的用户（排除已成功入群的）111
|
||
$groupUser = Db::name('workbench_group_create_item')
|
||
->where('workbenchId', $workbench->id)
|
||
->where('status', 'in', [self::STATUS_SUCCESS, self::STATUS_ADMIN_FRIEND_ADDED, self::STATUS_CREATING])
|
||
->whereIn('wechatId', $poolItem)
|
||
->group('wechatId')
|
||
->column('wechatId');
|
||
// 待入群的用户
|
||
$joinUser = array_diff($poolItem, $groupUser);
|
||
if (empty($joinUser)) {
|
||
continue;
|
||
}
|
||
// 计算随机群人数（不包含管理员，只减去群主成员数）
|
||
$groupRandNum = mt_rand($config['groupSizeMin'], $config['groupSizeMax']) - count($groupMember);
|
||
|
||
// 分批处理待入群用户
|
||
$addGroupUser = [];
|
||
$totalRows = count($joinUser);
|
||
for ($i = 0; $i < $totalRows; $i += $groupRandNum) {
|
||
$batchRows = array_slice($joinUser, $i, $groupRandNum);
|
||
if (!empty($batchRows)) {
|
||
$addGroupUser[] = $batchRows;
|
||
}
|
||
}
|
||
// 初始化WebSocket
|
||
$toAccountId = '';
|
||
$username = Env::get('api.username2', '');
|
||
$password = Env::get('api.password2', '');
|
||
if (!empty($username) || !empty($password)) {
|
||
$toAccountId = Db::name('users')->where('account', $username)->value('s2_accountId');
|
||
}
|
||
$webSocket = new WebSocketController(['userName' => $username, 'password' => $password, 'accountId' => $toAccountId]);
|
||
|
||
// 遍历每批用户
|
||
foreach ($addGroupUser as $batchUsers) {
|
||
$this->processBatchUsers($workbench, $config, $batchUsers, $groupMemberId, $groupMemberWechatId, $groupRandNum, $webSocket);
|
||
}
|
||
}
|
||
} catch (\Exception $e) {
|
||
Log::error("工作台建群任务异常: " . $e->getMessage());
|
||
throw $e;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 处理一批用户
|
||
* @param Workbench $workbench 工作台
|
||
* @param array $config 配置
|
||
* @param array $batchUsers 批次用户（微信ID数组，来自流量池）
|
||
* @param array $groupMemberId 群主成员ID数组
|
||
* @param array $groupMemberWechatId 群主成员微信ID映射
|
||
* @param int $groupRandNum 随机群人数（不包含管理员）
|
||
* @param WebSocketController $webSocket WebSocket实例
|
||
*/
|
||
protected function processBatchUsers($workbench, $config, $batchUsers, $groupMemberId, $groupMemberWechatId, $groupRandNum, $webSocket)
|
||
{
|
||
// 1. 获取群主微信ID列表（用于验证管理员）
|
||
// 从群主成员的好友记录中提取所有群主的微信ID（ownerWechatId）
|
||
$groupOwnerWechatIds = [];
|
||
foreach ($groupMemberId as $memberId) {
|
||
$member = Db::table('s2_wechat_friend')->where('id', $memberId)->find();
|
||
if ($member && !in_array($member['ownerWechatId'], $groupOwnerWechatIds)) {
|
||
$groupOwnerWechatIds[] = $member['ownerWechatId'];
|
||
}
|
||
}
|
||
|
||
|
||
// 如果从好友表获取不到，使用群主成员微信ID列表（作为备用）
|
||
if (empty($groupOwnerWechatIds)) {
|
||
$groupOwnerWechatIds = array_values(array_unique($groupMemberWechatId));
|
||
}
|
||
// 2. 验证并获取管理员好友ID（管理员必须是群主的好友）
|
||
$adminFriendIds = [];
|
||
$adminWechatIds = [];
|
||
if (!empty($config['admins'])) {
|
||
$adminFriends = Db::table('s2_wechat_friend')
|
||
->where('id', 'in', $config['admins'])
|
||
->field('id,wechatId,ownerWechatId')
|
||
->select();
|
||
|
||
foreach ($adminFriends as $adminFriend) {
|
||
// 验证：管理员必须是群主的好友
|
||
if (in_array($adminFriend['ownerWechatId'], $groupOwnerWechatIds)) {
|
||
$adminFriendIds[] = $adminFriend['id'];
|
||
$adminWechatIds[$adminFriend['id']] = $adminFriend['wechatId'];
|
||
}
|
||
}
|
||
}
|
||
|
||
|
||
// 3. 从流量池用户中筛选出是群主好友的用户（按微信账号分组）
|
||
$ownerFriendIdsByAccount = [];
|
||
$wechatIds = [];
|
||
|
||
// 获取群主的好友关系（从流量池中筛选）
|
||
$ownerFriends = Db::table('s2_wechat_friend')->alias('f')
|
||
->join(['s2_wechat_account' => 'a'], 'f.wechatAccountId=a.id')
|
||
->whereIn('f.wechatId', $batchUsers)
|
||
->whereIn('a.wechatId', $groupOwnerWechatIds)
|
||
->where('f.isDeleted', 0)
|
||
->field('f.id,f.wechatId,a.id as wechatAccountId')
|
||
->select();
|
||
|
||
if (empty($ownerFriends)) {
|
||
Log::warning("未找到群主的好友，跳过。工作台ID: {$workbench->id}");
|
||
return;
|
||
}
|
||
|
||
// 按微信账号分组群主好友
|
||
foreach ($ownerFriends as $friend) {
|
||
$wechatAccountId = $friend['wechatAccountId'];
|
||
if (!isset($ownerFriendIdsByAccount[$wechatAccountId])) {
|
||
$ownerFriendIdsByAccount[$wechatAccountId] = [];
|
||
}
|
||
$ownerFriendIdsByAccount[$wechatAccountId][] = $friend['id'];
|
||
$wechatIds[$friend['id']] = $friend['wechatId'];
|
||
}
|
||
|
||
// 4. 遍历每个微信账号，创建群
|
||
foreach ($ownerFriendIdsByAccount as $wechatAccountId => $ownerFriendIds) {
|
||
// 4.1 获取当前账号的管理员好友ID
|
||
$currentAdminFriendIds = [];
|
||
$accountWechatId = Db::table('s2_wechat_account')->where('id', $wechatAccountId)->value('wechatId');
|
||
foreach ($adminFriendIds as $adminFriendId) {
|
||
$adminFriend = Db::table('s2_wechat_friend')->where('id', $adminFriendId)->find();
|
||
if ($adminFriend && $adminFriend['ownerWechatId'] == $accountWechatId) {
|
||
$currentAdminFriendIds[] = $adminFriendId;
|
||
$wechatIds[$adminFriendId] = $adminWechatIds[$adminFriendId];
|
||
}
|
||
}
|
||
|
||
// 4.2 获取当前账号的群主成员ID
|
||
$currentGroupMemberIds = [];
|
||
foreach ($groupMemberId as $memberId) {
|
||
$member = Db::table('s2_wechat_friend')->where('id', $memberId)->find();
|
||
if ($member && $member['ownerWechatId'] == $accountWechatId) {
|
||
$currentGroupMemberIds[] = $memberId;
|
||
if (!isset($wechatIds[$memberId])) {
|
||
$wechatIds[$memberId] = $groupMemberWechatId[$memberId] ?? '';
|
||
}
|
||
}
|
||
}
|
||
|
||
// 4.3 限制群主好友数量（按随机群人数）
|
||
$limitedOwnerFriendIds = array_slice($ownerFriendIds, 0, $groupRandNum);
|
||
|
||
// 4.4 创建群：管理员 + 群主成员 + 群主好友（从流量池筛选）
|
||
$createFriendIds = array_merge($currentAdminFriendIds, $currentGroupMemberIds, $limitedOwnerFriendIds);
|
||
|
||
if (count($createFriendIds) < 2) {
|
||
Log::warning("建群好友数量不足，跳过。工作台ID: {$workbench->id}, 微信账号ID: {$wechatAccountId}");
|
||
continue;
|
||
}
|
||
|
||
// 4.5 生成群名称
|
||
$existingGroupCount = Db::name('workbench_group_create_item')
|
||
->where('workbenchId', $workbench->id)
|
||
->where('wechatAccountId', $wechatAccountId)
|
||
->where('status', self::STATUS_SUCCESS)
|
||
->group('groupId')
|
||
->count();
|
||
|
||
$chatroomName = $existingGroupCount > 0
|
||
? $config['groupNameTemplate'] . ($existingGroupCount + 1) . '群'
|
||
: $config['groupNameTemplate'];
|
||
|
||
// 4.6 调用建群接口
|
||
$createTime = time();
|
||
$createResult = $webSocket->CmdChatroomCreate([
|
||
'chatroomName' => $chatroomName,
|
||
'wechatFriendIds' => $createFriendIds,
|
||
'wechatAccountId' => $wechatAccountId
|
||
]);
|
||
|
||
$createResultData = json_decode($createResult, true);
|
||
|
||
// 4.7 解析建群结果，获取群ID
|
||
$chatroomId = 0;
|
||
if (!empty($createResultData) && isset($createResultData['code']) && $createResultData['code'] == 200) {
|
||
// 尝试从返回数据中获取群ID（根据实际API返回格式调整）
|
||
if (isset($createResultData['data']['chatroomId'])) {
|
||
$chatroomId = $createResultData['data']['chatroomId'];
|
||
} elseif (isset($createResultData['data']['id'])) {
|
||
$chatroomId = $createResultData['data']['id'];
|
||
}
|
||
}
|
||
|
||
// 4.8 记录创建请求
|
||
$installData = [];
|
||
foreach ($createFriendIds as $friendId) {
|
||
$memberType = in_array($friendId, $currentAdminFriendIds)
|
||
? self::MEMBER_TYPE_ADMIN
|
||
: (in_array($friendId, $currentGroupMemberIds) ? self::MEMBER_TYPE_OWNER : self::MEMBER_TYPE_OWNER_FRIEND);
|
||
|
||
$installData[] = [
|
||
'workbenchId' => $workbench->id,
|
||
'friendId' => $friendId,
|
||
'wechatId' => $wechatIds[$friendId] ?? ($groupMemberWechatId[$friendId] ?? ''),
|
||
'groupId' => $chatroomId,
|
||
'wechatAccountId' => $wechatAccountId,
|
||
'status' => $chatroomId > 0 ? self::STATUS_SUCCESS : self::STATUS_CREATING,
|
||
'memberType' => $memberType,
|
||
'retryCount' => 0,
|
||
'chatroomId' => $chatroomId > 0 ? $chatroomId : null,
|
||
'createTime' => $createTime,
|
||
];
|
||
}
|
||
Db::name('workbench_group_create_item')->insertAll($installData);
|
||
|
||
// 5. 如果群创建成功，拉管理员的好友进群
|
||
if ($chatroomId > 0 && !empty($currentAdminFriendIds)) {
|
||
$this->inviteAdminFriends($workbench, $config, $batchUsers, $currentAdminFriendIds, $chatroomId, $wechatAccountId, $wechatIds, $createTime, $webSocket);
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 拉管理员的好友进群
|
||
* @param Workbench $workbench 工作台
|
||
* @param array $config 配置
|
||
* @param array $batchUsers 批次用户（流量池微信ID数组）
|
||
* @param array $adminFriendIds 管理员好友ID数组
|
||
* @param int $chatroomId 群ID
|
||
* @param int $wechatAccountId 微信账号ID
|
||
* @param array $wechatIds 好友ID到微信ID的映射
|
||
* @param int $createTime 创建时间
|
||
* @param WebSocketController $webSocket WebSocket实例
|
||
*/
|
||
protected function inviteAdminFriends($workbench, $config, $batchUsers, $adminFriendIds, $chatroomId, $wechatAccountId, $wechatIds, $createTime, $webSocket)
|
||
{
|
||
// 获取管理员的微信ID列表
|
||
$adminWechatIds = [];
|
||
foreach ($adminFriendIds as $adminFriendId) {
|
||
if (isset($wechatIds[$adminFriendId])) {
|
||
$adminWechatIds[] = $wechatIds[$adminFriendId];
|
||
}
|
||
}
|
||
|
||
if (empty($adminWechatIds)) {
|
||
return;
|
||
}
|
||
|
||
// 从流量池用户中筛选出是管理员好友的用户
|
||
$adminFriendsFromPool = Db::table('s2_wechat_friend')->alias('f')
|
||
->join(['s2_wechat_account' => 'a'], 'f.wechatAccountId=a.id')
|
||
->whereIn('f.wechatId', $batchUsers)
|
||
->whereIn('a.wechatId', $adminWechatIds)
|
||
->where('a.id', $wechatAccountId)
|
||
->where('f.isDeleted', 0)
|
||
->field('f.id,f.wechatId')
|
||
->select();
|
||
|
||
if (empty($adminFriendsFromPool)) {
|
||
Log::info("未找到管理员的好友，跳过拉人。工作台ID: {$workbench->id}, 群ID: {$chatroomId}");
|
||
return;
|
||
}
|
||
|
||
// 提取好友ID列表
|
||
$adminFriendIdsToInvite = [];
|
||
foreach ($adminFriendsFromPool as $friend) {
|
||
$adminFriendIdsToInvite[] = $friend['id'];
|
||
$wechatIds[$friend['id']] = $friend['wechatId'];
|
||
}
|
||
|
||
// 调用拉人接口
|
||
$inviteResult = $webSocket->CmdChatroomInvite([
|
||
'wechatChatroomId' => $chatroomId,
|
||
'wechatFriendIds' => $adminFriendIdsToInvite
|
||
]);
|
||
|
||
$inviteResultData = json_decode($inviteResult, true);
|
||
$inviteSuccess = !empty($inviteResultData) && isset($inviteResultData['code']) && $inviteResultData['code'] == 200;
|
||
|
||
// 记录管理员好友拉入状态
|
||
$adminFriendData = [];
|
||
foreach ($adminFriendIdsToInvite as $friendId) {
|
||
$adminFriendData[] = [
|
||
'workbenchId' => $workbench->id,
|
||
'friendId' => $friendId,
|
||
'wechatId' => $wechatIds[$friendId] ?? '',
|
||
'groupId' => $chatroomId,
|
||
'wechatAccountId' => $wechatAccountId,
|
||
'status' => $inviteSuccess ? self::STATUS_ADMIN_FRIEND_ADDED : self::STATUS_FAILED,
|
||
'memberType' => self::MEMBER_TYPE_ADMIN_FRIEND,
|
||
'retryCount' => 0,
|
||
'chatroomId' => $chatroomId,
|
||
'createTime' => $createTime,
|
||
];
|
||
}
|
||
Db::name('workbench_group_create_item')->insertAll($adminFriendData);
|
||
|
||
if ($inviteSuccess) {
|
||
// 去除成功日志，减少日志空间消耗
|
||
} else {
|
||
Log::warning("管理员好友拉入失败。工作台ID: {$workbench->id}, 群ID: {$chatroomId}");
|
||
}
|
||
}
|
||
|
||
|
||
/**
|
||
* 获取设备列表
|
||
* @param Workbench $workbench 工作台
|
||
* @param WorkbenchGroupPush $config 配置
|
||
* @return array|bool
|
||
*/
|
||
protected function isCreate($workbench, $config, $groupIds = [])
|
||
{
|
||
// 检查发送间隔（新逻辑：根据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($groupIds)) {
|
||
return false;
|
||
}
|
||
|
||
// 查询今日建群数量
|
||
$count = Db::name('wechat_group')
|
||
->whereIn('id', $groupIds)
|
||
->whereTime('createTime', 'between', [$startTimestamp, $endTimestamp])
|
||
->count();
|
||
if ($count >= $config['maxGroupsPerDay']) {
|
||
return false;
|
||
}
|
||
return true;
|
||
}
|
||
|
||
|
||
/**
|
||
* 记录任务开始
|
||
* @param string $jobId
|
||
* @param string $queueLockKey
|
||
*/
|
||
protected function logJobStart($jobId, $queueLockKey)
|
||
{
|
||
// 去除开始日志，减少日志空间消耗
|
||
}
|
||
|
||
/**
|
||
* 处理任务成功
|
||
* @param Job $job
|
||
* @param string $queueLockKey
|
||
*/
|
||
protected function handleJobSuccess($job, $queueLockKey)
|
||
{
|
||
$job->delete();
|
||
Cache::rm($queueLockKey);
|
||
// 去除成功日志，减少日志空间消耗
|
||
}
|
||
|
||
/**
|
||
* 处理任务错误
|
||
* @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;
|
||
}
|
||
} |