fnvtk/cunkebao_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

群发

Browse Source
This commit is contained in:
wong
2025-12-10 17:58:08 +08:00
parent 55fe2b46df
commit 198e0434b3
10 changed files with 1346 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

431
Server/application/job/WorkbenchGroupCreateJob.php
View File

@@ -7,6 +7,7 @@ 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;
@@ -16,6 +17,7 @@ use think\facade\Config;
use app\api\controller\MomentsController as Moments;
use Workerman\Lib\Timer;
use app\api\controller\WechatController;
use think\Queue;
/**
* 工作台群创建任务
@@ -50,6 +52,23 @@ class WorkbenchGroupCreateJob
}
}
/**
* 成员类型常量
*/
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
@@ -57,7 +76,7 @@ class WorkbenchGroupCreateJob
public function execute()
{
try {
// 获取所有工作台
// 1. 查询启用了建群功能的数据
$workbenches = Workbench::where(['status' => 1, 'type' => 4, 'isDel' => 0])->order('id desc')->select();
foreach ($workbenches as $workbench) {
// 获取工作台配置
@@ -65,158 +84,356 @@ class WorkbenchGroupCreateJob
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 = Db::name('device_wechat_login')->alias('dwl')
->join(['s2_wechat_account' => 'a'], 'dwl.wechatId = a.wechatId')
->whereIn('dwl.deviceId', $config['devices'])
->group('a.id')
->column('a.wechatId');
if (empty($groupMember)) {
$groupMember = [];
$wechatId = Db::name('device_wechat_login')
->whereIn('deviceId',$config['devices'])
->order('id desc')
->value('wechatId');
if (empty($wechatId)) {
continue;
}
$groupMemberWechatId = Db::table('s2_wechat_friend')
->where('ownerWechatId', $groupMember[0])
->whereIn('wechatId', $groupMember)
->column('id,wechatId');
$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;
}
$groupMemberId = array_keys($groupMemberWechatId);
//流量池用户
// 获取流量池用户
$poolItem = Db::name('traffic_source_package_item')
->whereIn('packageId', $config['poolGroups'])
->group('identifier')
->column('identifier');
if (empty($poolItem)) {
continue;
}
//群用户
// 获取已入群的用户（排除已成功入群的）
$groupUser = Db::name('workbench_group_create_item')
->where('workbenchId', $workbench->id)
->where('status', 'in', [self::STATUS_SUCCESS, self::STATUS_ADMIN_FRIEND_ADDED])
->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)) {
$user = [];
foreach ($batchRows as $row) {
$user[] = $row;
}
$addGroupUser[] = $user;
$addGroupUser[] = $batchRows;
}
}
foreach ($addGroupUser as $key => $val) {
//判断第一组用户是否满足创建群的条件
$friendIds = Db::name('wechat_friendship')->alias('f')
->join(['s2_wechat_account' => 'a'], 'f.ownerWechatId=a.wechatId')
->where('f.companyId', $workbench->companyId)
->whereIn('f.wechatId', $val)
->group('f.wechatId')
->column('f.id,f.wechatId,a.id as wechatAccountId');
// 初始化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]);
// 整理数组：按wechatAccountId分组，值为对应的id数组
$groupedFriends = [];
$wechatAccountIds = [];
$wechatIds = [];
foreach ($friendIds as $friend) {
$wechatAccountId = $friend['wechatAccountId'];
if (!in_array($wechatAccountId, $wechatAccountIds)) {
$wechatAccountIds[] = $wechatAccountId;
}
$friendId = $friend['id'];
if (!isset($groupedFriends[$wechatAccountId])) {
$groupedFriends[$wechatAccountId] = [];
}
$groupedFriends[$wechatAccountId][] = $friendId;
$wechatIds[$friendId] = $friend['wechatId'];
}
//==================== 群相关功能开始 ===========================
$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]);
//$webSocket = new WebSocketController(['userName' => 'wz_03', 'password' => 'key123456', 'accountId' => 5015]);
//拉人进群 $webSocket->CmdChatroomInvite(['wechatChatroomId' => 830794, 'wechatFriendIds' => [21168549]]);
//修改群名称 $webSocket->CmdChatroomModifyInfo(['wechatChatroomId' => 830794, 'wechatAccountId' => 300745,'chatroomName' => 'test111']);
//修改群公告 $webSocket->CmdChatroomModifyInfo(['wechatChatroomId' => 830794, 'wechatAccountId' => 300745,'announce' => 'test111']);
//建群 $webSocket->CmdChatroomCreate(['chatroomName' => '聊天测试群', 'wechatFriendIds' => [17453051,17453058],'wechatAccountId' => 300745]);
foreach ($groupedFriends as $wechatAccountId => $friendId) {
//列出所有群
$group = '';
$groupMemberNum = 0;
$groupIds = Db::name('workbench_group_create_item')->where(['workbenchId' => $workbench->id])->group('groupId')->column('groupId');
if (!empty($groupIds)) {
//最新创建的群
$group = Db::name('wechat_group')->where(['wechatAccountId' => $wechatAccountId])->whereIn('id', $groupIds)->order('createTime DESC')->find();
//群用户数量
if (!empty($group)) {
$groupMemberNum = Db::name('wechat_group_member')->where('groupId', $group['id'])->count();
}
}
//拉群或者建群
$wechatFriendIds = array_merge($friendId, $groupMemberId);
if ($groupMemberNum == 0 || (count($wechatFriendIds) + $groupMemberNum) >= $groupRandNum) {
if (count($groupIds) > 0) {
$chatroomName = $config['groupNameTemplate'] . count($groupIds) + 1 . '群';
} else {
$chatroomName = $config['groupNameTemplate'];
}
$webSocket->CmdChatroomCreate(['chatroomName' => $chatroomName, 'wechatFriendIds' => $wechatFriendIds,'wechatAccountId' => $wechatAccountId]);
} else {
$webSocket->CmdChatroomInvite(['wechatChatroomId' => $group['id'], 'wechatFriendIds' => $wechatFriendIds]);
}
$installData = [];
//记录进群人员
foreach ($wechatFriendIds as $v) {
$installData[] = [
'workbenchId' => $workbench->id,
'friendId' => $v,
'wechatId' => !empty($wechatIds[$v]) ? $wechatIds[$v] : $groupMemberWechatId[$v],
'groupId' => 0,
'wechatAccountId' => $wechatAccountId,
'createTime' => time(),
];
}
Db::name('workbench_group_create_item')->insertAll($installData);
}
// 遍历每批用户
foreach ($addGroupUser as $batchUsers) {
$this->processBatchUsers($workbench, $config, $batchUsers, $groupMemberId, $groupMemberWechatId, $groupRandNum, $webSocket);
}
}
} catch (\Exception $e) {
Log::error("消息群发任务异常: " . $e->getMessage());
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'];
}
}
}
exit_data($adminWechatIds);
// 3. 从流量池用户中筛选出是群主好友的用户（按微信账号分组）
$ownerFriendIdsByAccount = [];
$wechatIds = [];
// 获取群主的好友关系（从流量池中筛选）
$ownerFriends = Db::name('wechat_friendship')->alias('f')
->join(['s2_wechat_account' => 'a'], 'f.ownerWechatId=a.wechatId')
->where('f.companyId', $workbench->companyId)
->whereIn('f.wechatId', $batchUsers)
->whereIn('f.ownerWechatId', $groupOwnerWechatIds)
->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::name('wechat_friendship')->alias('f')
->join(['s2_wechat_account' => 'a'], 'f.ownerWechatId=a.wechatId')
->where('f.companyId', $workbench->companyId)
->whereIn('f.wechatId', $batchUsers)
->whereIn('f.ownerWechatId', $adminWechatIds)
->where('a.id', $wechatAccountId)
->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) {
Log::info("管理员好友拉入成功。工作台ID: {$workbench->id}, 群ID: {$chatroomId}, 拉入数量: " . count($adminFriendIdsToInvite));
} else {
Log::warning("管理员好友拉入失败。工作台ID: {$workbench->id}, 群ID: {$chatroomId}");
}
}
/**
* 获取设备列表