diff --git a/.gitignore b/.gitignore new file mode 100644 index 000000000..4af24f061 --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +.idea/ +Cunkebao/.next/ +Store_vue/node_modules/ +*.zip +Cunkebao/.specstory/ +*.cursorindexingignore +Server/.specstory/ +Store_vue/.specstory/ +Store_vue/unpackage/ +Store_vue/.vscode/ +SuperAdmin/.specstory/ diff --git a/Server/application/api/controller/AccountController.php b/Server/application/api/controller/AccountController.php index 303452927..f239be006 100644 --- a/Server/application/api/controller/AccountController.php +++ b/Server/application/api/controller/AccountController.php @@ -576,6 +576,76 @@ class AccountController extends BaseController + public function accountModify($data = []) + { + // 获取授权token + $authorization = $this->authorization; + if (empty($authorization)) { + return errorJson('缺少授权信息'); + } + + + $id = !empty($data['id']) ? $data['id'] : ''; + if (empty($id)) { + return errorJson('账号ID不能为空'); + } + + $account = CompanyAccountModel::where('id', $id)->find(); + + + + if (empty($account)) { + return errorJson('账号不存在'); + } + $privilegeIds = json_decode($account->privilegeIds,true); + $privilegeIds = !empty($privilegeIds) ? $privilegeIds : [1001,1002,1004,1023,1406,20003,20021,20022,20023,20032,20041,20049,20054,20055,20060,20100,20102,20107,20055]; + + // 构建请求参数 + $params = [ + 'accountType' => !empty($data['accountType']) ? $data['accountType'] : $account->accountType, + 'alive' => !empty($data['alive']) ? $data['alive'] : $account->alive, + 'avatar' => !empty($data['avatar']) ? $data['avatar'] : $account->avatar, + 'createTime' => !empty($data['createTime']) ? $data['createTime'] : $account->createTime, + 'creator' => !empty($data['creator']) ? $data['creator'] : $account->creator, + 'creatorRealName' => !empty($data['creatorRealName']) ? $data['creatorRealName'] : $account->creatorRealName, + 'creatorUserName' => !empty($data['creatorUserName']) ? $data['creatorUserName'] : $account->creatorUserName, + 'departmentId' => !empty($data['departmentId']) ? $data['departmentId'] : $account->departmentId, + 'departmentIdArr' => !empty($data['departmentIdArr']) ? $data['departmentIdArr'] : [914,$account->departmentId], + 'departmentName' => !empty($data['departmentName']) ? $data['departmentName'] : $account->departmentName, + 'hasXiakeAccount' => !empty($data['hasXiakeAccount']) ? $data['hasXiakeAccount'] : false, + 'id' => !empty($data['id']) ? $data['id'] : $account->id, + 'memo' => !empty($data['memo']) ? $data['memo'] : $account->memo, + 'nickname' => !empty($data['nickname']) ? $data['nickname'] : $account->nickname, + 'privilegeIds' => !empty($data['privilegeIds']) ? $data['privilegeIds'] : $privilegeIds, + 'realName' => !empty($data['realName']) ? $data['realName'] : $account->realName, + 'status' => !empty($data['status']) ? $data['status'] : $account->status, + 'tenantId' => !empty($data['tenantId']) ? $data['tenantId'] : $account->tenantId, + 'userName' => !empty($data['userName']) ? $data['userName'] : $account->userName, + ]; + // 设置请求头 + $headerData = ['client:system']; + $header = setHeader($headerData, $authorization, 'json'); + + // 发送请求修改部门 + $result = requestCurl($this->baseUrl . 'api/account/modify', $params, 'PUT', $header, 'json'); + $response = handleApiResponse($result); + + + if(empty($response)){ + $newData = [ + 'nickname' => $params['nickname'], + 'avatar' => $params['avatar'], + ]; + CompanyAccountModel::where('id', $id)->update($newData); + return json_encode(['code' => 200, 'msg' => '账号修改成功']); + }else{ + return json_encode(['code' => 500, 'msg' => $response]); + } + } + + + + /************************ 私有辅助方法 ************************/ diff --git a/Server/application/api/controller/WebSocketController.php b/Server/application/api/controller/WebSocketController.php index 5090a5789..24d67fa6f 100644 --- a/Server/application/api/controller/WebSocketController.php +++ b/Server/application/api/controller/WebSocketController.php @@ -9,12 +9,11 @@ use think\Db; use think\facade\Log; use WebSocket\Client; use think\facade\Env; -use app\api\model\WechatFriendModel as WechatFriend; +use app\api\model\WechatFriendModel as WechatFriend; use app\api\model\WechatMomentsModel as WechatMoments; use think\facade\Cache; - class WebSocketController extends BaseController { protected $authorized; @@ -27,7 +26,7 @@ class WebSocketController extends BaseController /************************************ * 初始化相关功能 ************************************/ - + /** * 构造函数 - 初始化WebSocket连接 * @param array $userData 用户数据 @@ -44,39 +43,39 @@ class WebSocketController extends BaseController */ protected function initConnection($userData = []) { - if(!empty($userData) && count($userData)){ + if (!empty($userData) && count($userData)) { if (empty($userData['userName']) || empty($userData['password'])) { - return json_encode(['code'=>400,'msg'=>'参数缺失']); + return json_encode(['code' => 400, 'msg' => '参数缺失']); } // 检查缓存中是否存在有效的token $cacheKey = 'websocket_token_' . $userData['userName']; - $cachedToken = Cache::get($cacheKey); - + $cachedToken = Cache::get($cacheKey); + if ($cachedToken) { $this->authorized = $cachedToken; $this->accountId = $userData['accountId']; } else { - $params = [ - 'grant_type' => 'password', - 'username' => $userData['userName'], - 'password' => $userData['password'] - ]; + $params = [ + 'grant_type' => 'password', + 'username' => $userData['userName'], + 'password' => $userData['password'] + ]; - // 调用登录接口获取token - $headerData = ['client:kefu-client']; - $header = setHeader($headerData, '', 'plain'); - $result = requestCurl('https://kf.quwanzhi.com:9991/token', $params, 'POST', $header); - $result_array = handleApiResponse($result); + // 调用登录接口获取token + $headerData = ['client:kefu-client']; + $header = setHeader($headerData, '', 'plain'); + $result = requestCurl('https://kf.quwanzhi.com:9991/token', $params, 'POST', $header); + $result_array = handleApiResponse($result); - if (isset($result_array['access_token']) && !empty($result_array['access_token'])) { - $this->authorized = $result_array['access_token']; - $this->accountId = $userData['accountId']; - - // 将token存入缓存,有效期5分钟 - Cache::set($cacheKey, $this->authorized, 300); - } else { - return json_encode(['code'=>400,'msg'=>'获取系统授权信息失败']); + if (isset($result_array['access_token']) && !empty($result_array['access_token'])) { + $this->authorized = $result_array['access_token']; + $this->accountId = $userData['accountId']; + + // 将token存入缓存,有效期5分钟 + Cache::set($cacheKey, $this->authorized, 300); + } else { + return json_encode(['code' => 400, 'msg' => '获取系统授权信息失败']); } } } else { @@ -85,7 +84,7 @@ class WebSocketController extends BaseController } if (empty($this->authorized) || empty($this->accountId)) { - return json_encode(['code'=>400,'msg'=>'缺失关键参数']); + return json_encode(['code' => 400, 'msg' => '缺失关键参数']); } $this->connect(); @@ -97,40 +96,40 @@ class WebSocketController extends BaseController protected function connect() { try { - //证书 - $context = stream_context_create(); - stream_context_set_option($context, 'ssl', 'verify_peer', false); - stream_context_set_option($context, 'ssl', 'verify_peer_name', false); - - //开启WS链接 - $result = [ - "accessToken" => $this->authorized, - "accountId" => $this->accountId, - "client" => "kefu-client", - "cmdType" => "CmdSignIn", - "seq" => 1, - ]; + //证书 + $context = stream_context_create(); + stream_context_set_option($context, 'ssl', 'verify_peer', false); + stream_context_set_option($context, 'ssl', 'verify_peer_name', false); - $content = json_encode($result); - $this->client = new Client("wss://kf.quwanzhi.com:9993", - [ - 'filter' => ['text', 'binary', 'ping', 'pong', 'close','receive', 'send'], - 'context' => $context, - 'headers' => [ - 'Sec-WebSocket-Protocol' => 'soap', - 'origin' => 'localhost', - ], - 'timeout' => 86400, - ] - ); - - $this->client->send($content); + //开启WS链接 + $result = [ + "accessToken" => $this->authorized, + "accountId" => $this->accountId, + "client" => "kefu-client", + "cmdType" => "CmdSignIn", + "seq" => 1, + ]; + + $content = json_encode($result); + $this->client = new Client("wss://kf.quwanzhi.com:9993", + [ + 'filter' => ['text', 'binary', 'ping', 'pong', 'close', 'receive', 'send'], + 'context' => $context, + 'headers' => [ + 'Sec-WebSocket-Protocol' => 'soap', + 'origin' => 'localhost', + ], + 'timeout' => 86400, + ] + ); + + $this->client->send($content); $this->isConnected = true; $this->lastHeartbeatTime = time(); - + // 启动心跳检测 //$this->startHeartbeat(); - + } catch (\Exception $e) { Log::error("WebSocket连接失败:" . $e->getMessage()); $this->isConnected = false; @@ -143,7 +142,7 @@ class WebSocketController extends BaseController protected function startHeartbeat() { // 使用定时器发送心跳 - \Swoole\Timer::tick($this->heartbeatInterval * 1000, function() { + \Swoole\Timer::tick($this->heartbeatInterval * 1000, function () { if ($this->isConnected) { $this->sendHeartbeat(); } @@ -160,10 +159,10 @@ class WebSocketController extends BaseController "cmdType" => "CmdHeartbeat", "seq" => time() ]; - + $this->client->send(json_encode($heartbeat)); $this->lastHeartbeatTime = time(); - + } catch (\Exception $e) { Log::error("发送心跳包失败:" . $e->getMessage()); $this->reconnect(); @@ -204,7 +203,7 @@ class WebSocketController extends BaseController protected function sendMessage($data) { $this->checkConnection(); - + try { $this->client->send(json_encode($data)); $response = $this->client->receive(); @@ -235,9 +234,9 @@ class WebSocketController extends BaseController $currentPage = 1; // 当前页码 $allMoments = []; // 存储所有朋友圈数据 - //过滤消息 + //过滤消息 if (empty($wechatAccountId)) { - return json_encode(['code'=>400,'msg'=>'指定账号不能为空']); + return json_encode(['code' => 400, 'msg' => '指定账号不能为空']); } try { @@ -263,21 +262,21 @@ class WebSocketController extends BaseController sleep(10); continue; } - + // 检查返回结果 if (!isset($message['result']) || empty($message['result']) || !is_array($message['result'])) { break; } - + // 检查是否遇到旧数据 $hasOldData = false; foreach ($message['result'] as $moment) { $momentId = WechatMoments::where('snsId', $moment['snsId']) ->where('wechatAccountId', $wechatAccountId) ->value('id'); - + if (!empty($momentId)) { $hasOldData = true; break; @@ -330,31 +329,31 @@ class WebSocketController extends BaseController return json_encode($result); } catch (\Exception $e) { - return json_encode(['code'=>500,'msg'=>$e->getMessage()]); + return json_encode(['code' => 500, 'msg' => $e->getMessage()]); } } - /** + /** * 朋友圈点赞 * @return \think\response\Json */ public function momentInteract($data = []) { - + $snsId = !empty($data['snsId']) ? $data['snsId'] : ''; $wechatAccountId = !empty($data['wechatAccountId']) ? $data['wechatAccountId'] : ''; $wechatFriendId = !empty($data['wechatFriendId']) ? $data['wechatFriendId'] : 0; //过滤消息 - if (empty($snsId)) { - return json_encode(['code'=>400,'msg'=>'snsId不能为空']); + if (empty($snsId)) { + return json_encode(['code' => 400, 'msg' => 'snsId不能为空']); } - if (empty($wechatAccountId)) { - return json_encode(['code'=>400,'msg'=>'微信id不能为空']); + if (empty($wechatAccountId)) { + return json_encode(['code' => 400, 'msg' => '微信id不能为空']); } - - try { + + try { $result = [ "cmdType" => "CmdMomentInteract", "momentInteractType" => 1, @@ -362,16 +361,16 @@ class WebSocketController extends BaseController "snsId" => $snsId, "wechatAccountId" => $wechatAccountId, "wechatFriendId" => $wechatFriendId, - ]; + ]; $message = $this->sendMessage($result); - return json_encode(['code'=>200,'msg'=>'点赞成功','data'=>$message]); - } catch (\Exception $e) { - return json_encode(['code'=>500,'msg'=>$e->getMessage()]); + return json_encode(['code' => 200, 'msg' => '点赞成功', 'data' => $message]); + } catch (\Exception $e) { + return json_encode(['code' => 500, 'msg' => $e->getMessage()]); } } - /** + /** * 朋友圈取消点赞 * @return \think\response\Json */ @@ -381,36 +380,36 @@ class WebSocketController extends BaseController $data = $this->request->param(); if (empty($data)) { - return json_encode(['code'=>400,'msg'=>'参数缺失']); + return json_encode(['code' => 400, 'msg' => '参数缺失']); } //过滤消息 if (empty($data['snsId'])) { - return json_encode(['code'=>400,'msg'=>'snsId不能为空']); + return json_encode(['code' => 400, 'msg' => 'snsId不能为空']); } if (empty($data['wechatAccountId'])) { - return json_encode(['code'=>400,'msg'=>'微信id不能为空']); + return json_encode(['code' => 400, 'msg' => '微信id不能为空']); } - + try { - $result = [ - "CommentId2" => '', - "CommentTime" => 0, - "cmdType" => "CmdMomentCancelInteract", - "optType" => 1, - "seq" => time(), - "snsId" => $data['snsId'], - "wechatAccountId" => $data['wechatAccountId'], - "wechatFriendId" => 0, - ]; + $result = [ + "CommentId2" => '', + "CommentTime" => 0, + "cmdType" => "CmdMomentCancelInteract", + "optType" => 1, + "seq" => time(), + "snsId" => $data['snsId'], + "wechatAccountId" => $data['wechatAccountId'], + "wechatFriendId" => 0, + ]; $message = $this->sendMessage($result); - return json_encode(['code'=>200,'msg'=>'取消点赞成功','data'=>$message]); + return json_encode(['code' => 200, 'msg' => '取消点赞成功', 'data' => $message]); } catch (\Exception $e) { - return json_encode(['code'=>500,'msg'=>$e->getMessage()]); + return json_encode(['code' => 500, 'msg' => $e->getMessage()]); } } else { - return json_encode(['code'=>400,'msg'=>'非法请求']); + return json_encode(['code' => 400, 'msg' => '非法请求']); } } @@ -462,19 +461,19 @@ class WebSocketController extends BaseController // 发送请求 $this->client->send(json_encode($params)); - + // 接收响应 $response = $this->client->receive(); $message = json_decode($response, true); - if(empty($message)){ - return json_encode(['code'=>500,'msg'=>'获取朋友圈资源链接失败']); + if (empty($message)) { + return json_encode(['code' => 500, 'msg' => '获取朋友圈资源链接失败']); } - if($message['cmdType'] == 'CmdDownloadMomentImagesResult' && is_array($message['urls']) && count($message['urls']) > 0){ - $urls = json_encode($message['urls'],256); - Db::table('s2_wechat_moments')->where('snsId',$data['snsId'])->update(['resUrls'=>$urls]); + if ($message['cmdType'] == 'CmdDownloadMomentImagesResult' && is_array($message['urls']) && count($message['urls']) > 0) { + $urls = json_encode($message['urls'], 256); + Db::table('s2_wechat_moments')->where('snsId', $data['snsId'])->update(['resUrls' => $urls]); } - return json_encode(['code'=>200,'msg'=>'获取朋友圈资源链接成功','data'=>$message]); + return json_encode(['code' => 200, 'msg' => '获取朋友圈资源链接成功', 'data' => $message]); } catch (\Exception $e) { // 记录错误日志 Log::error('获取朋友圈资源链接异常:' . $e->getMessage()); @@ -507,19 +506,18 @@ class WebSocketController extends BaseController return false; } - + try { foreach ($momentList as $moment) { // 提取momentEntity中的数据 $momentEntity = $moment['momentEntity'] ?? []; - + // 检查朋友圈数据是否已存在 $momentId = WechatMoments::where('snsId', $moment['snsId']) ->where('wechatAccountId', $wechatAccountId) ->value('id'); - $dataToSave = [ 'commentList' => json_encode($moment['commentList'] ?? [], 256), 'createTime' => $moment['createTime'] ?? 0, @@ -543,7 +541,7 @@ class WebSocketController extends BaseController // 如果已存在,则更新数据 Db::table('s2_wechat_moments')->where('id', $momentId)->update($dataToSave); } else { - if(empty($wechatFriendId)){ + if (empty($wechatFriendId)) { $wechatFriendId = WechatFriend::where('wechatAccountId', $wechatAccountId)->where('wechatId', $momentEntity['userName'])->value('id'); } // 如果不存在,则插入新数据 @@ -588,7 +586,7 @@ class WebSocketController extends BaseController if (empty($wechatFriendId)) { return json_encode(['code' => 400, 'msg' => '好友ID不能为空']); } - + if (empty($wechatAccountId)) { return json_encode(['code' => 400, 'msg' => '微信账号ID不能为空']); } @@ -609,7 +607,7 @@ class WebSocketController extends BaseController // 发送请求并获取响应 $message = $this->sendMessage($params); - + // 记录日志 Log::info('修改好友标签:' . json_encode($params, 256)); Log::info('修改好友标签结果:' . json_encode($message, 256)); @@ -619,7 +617,7 @@ class WebSocketController extends BaseController } catch (\Exception $e) { // 记录错误日志 Log::error('修改好友标签失败:' . $e->getMessage()); - + // 返回错误响应 return json_encode(['code' => 500, 'msg' => '修改标签失败:' . $e->getMessage()]); } @@ -653,24 +651,23 @@ class WebSocketController extends BaseController // 消息拼接 msgType(1:文本 3:图片 43:视频 47:动图表情包(gif、其他表情包) 49:小程序/其他:图文、文件) // 当前,type 为文本、图片、动图表情包的时候,content为string, 其他情况为对象 {type: 'file/link/...', url: '', title: '', thunmbPath: '', desc: ''} $params = [ - "cmdType" => "CmdSendMessage", - "content" => $dataArray['content'], - "msgSubType" => 0, - "msgType" => $dataArray['msgType'], - "seq" => time(), - "wechatAccountId" => $dataArray['wechatAccountId'], - "wechatChatroomId" => 0, - "wechatFriendId" => $dataArray['wechatFriendId'], - ]; + "cmdType" => "CmdSendMessage", + "content" => $dataArray['content'], + "msgSubType" => 0, + "msgType" => $dataArray['msgType'], + "seq" => time(), + "wechatAccountId" => $dataArray['wechatAccountId'], + "wechatChatroomId" => 0, + "wechatFriendId" => $dataArray['wechatFriendId'], + ]; // 发送请求 $this->client->send(json_encode($params)); // 接收响应 $response = $this->client->receive(); $message = json_decode($response, true); - - if(!empty($message)){ - return json_encode(['code'=>500,'msg'=>'信息发送成功','data'=>$message]); + if (!empty($message)) { + return json_encode(['code' => 200, 'msg' => '信息发送成功', 'data' => $message]); } } @@ -678,65 +675,55 @@ class WebSocketController extends BaseController * 发送群消息 * @return \think\response\Json */ - public function sendCommunity() + public function sendCommunity($dataArray = []) { - if ($this->request->isPost()) { - $data = $this->request->post(); - if (empty($data)) { - return json_encode(['code'=>400,'msg'=>'参数缺失']); - } - $dataArray = $data; - if (!is_array($dataArray)) { - return json_encode(['code'=>400,'msg'=>'数据格式错误']); - } - - //过滤消息 - if (empty($dataArray['content'])) { - return json_encode(['code'=>400,'msg'=>'内容缺失']); - } - if (empty($dataArray['wechatAccountId'])) { - return json_encode(['code'=>400,'msg'=>'微信id不能为空']); - } - - if (empty($dataArray['msgType'])) { - return json_encode(['code'=>400,'msg'=>'类型缺失']); - } - if (empty($dataArray['wechatChatroomId'])) { - return json_encode(['code'=>400,'msg'=>'群id不能为空']); - } - - $msg = '消息成功发送'; - $message = []; - try { - //消息拼接 msgType(1:文本 3:图片 43:视频 47:动图表情包 49:小程序) - $result = [ - "cmdType" => "CmdSendMessage", - "content" => htmlspecialchars_decode($dataArray['content']), - "msgSubType" => 0, - "msgType" => $dataArray['msgType'], - "seq" => time(), - "wechatAccountId" => $dataArray['wechatAccountId'], - "wechatChatroomId" => $dataArray['wechatChatroomId'], - "wechatFriendId" => 0, - ]; - - $result = json_encode($result); - $this->client->send($result); - $message = $this->client->receive(); - //关闭WS链接 - $this->client->close(); - //Log::write('WS群消息发送'); - //Log::write($message); - $message = json_decode($message, 1); - } catch (\Exception $e) { - $msg = $e->getMessage(); - } - return json_encode(['code'=>200,'msg'=>$msg,'data'=>$message]); - - } else { - return json_encode(['code'=>400,'msg'=>'非法请求']); - //return errorJson('非法请求'); + if (!is_array($dataArray)) { + return json_encode(['code' => 400, 'msg' => '数据格式错误']); } + + //过滤消息 + if (empty($dataArray['content'])) { + return json_encode(['code' => 400, 'msg' => '内容缺失']); + } + if (empty($dataArray['wechatAccountId'])) { + return json_encode(['code' => 400, 'msg' => '微信id不能为空']); + } + + if (empty($dataArray['msgType'])) { + return json_encode(['code' => 400, 'msg' => '类型缺失']); + } + if (empty($dataArray['wechatChatroomId'])) { + return json_encode(['code' => 400, 'msg' => '群id不能为空']); + } + + $message = []; + try { + //消息拼接 msgType(1:文本 3:图片 43:视频 47:动图表情包 49:小程序) + $params = [ + "cmdType" => "CmdSendMessage", + "content" => htmlspecialchars_decode($dataArray['content']), + "msgSubType" => 0, + "msgType" => $dataArray['msgType'], + "seq" => time(), + "wechatAccountId" => $dataArray['wechatAccountId'], + "wechatChatroomId" => $dataArray['wechatChatroomId'], + "wechatFriendId" => 0, + ]; + + // 发送请求 + $this->client->send(json_encode($params)); + // 接收响应 + $response = $this->client->receive(); + $message = json_decode($response, true); + if (!empty($message)) { + return json_encode(['code' => 200, 'msg' => '信息发送成功', 'data' => $message]); + } + } catch (\Exception $e) { + $msg = $e->getMessage(); + return json_encode(['code' => 400, 'msg' => $msg, 'data' => $message]); + } + + } /** @@ -747,26 +734,26 @@ class WebSocketController extends BaseController public function sendCommunitys($data = []) { if (empty($data)) { - return json_encode(['code'=>400,'msg'=>'参数缺失']); + return json_encode(['code' => 400, 'msg' => '参数缺失']); } $dataArray = $data; if (!is_array($dataArray)) { - return json_encode(['code'=>400,'msg'=>'数据格式错误']); + return json_encode(['code' => 400, 'msg' => '数据格式错误']); } //过滤消息 if (empty($dataArray['content'])) { - return json_encode(['code'=>400,'msg'=>'内容缺失']); + return json_encode(['code' => 400, 'msg' => '内容缺失']); } if (empty($dataArray['wechatAccountId'])) { - return json_encode(['code'=>400,'msg'=>'微信id不能为空']); + return json_encode(['code' => 400, 'msg' => '微信id不能为空']); } if (empty($dataArray['msgType'])) { - return json_encode(['code'=>400,'msg'=>'类型缺失']); + return json_encode(['code' => 400, 'msg' => '类型缺失']); } if (empty($dataArray['wechatChatroomId'])) { - return json_encode(['code'=>400,'msg'=>'群id不能为空']); + return json_encode(['code' => 400, 'msg' => '群id不能为空']); } $msg = '消息成功发送'; @@ -796,11 +783,10 @@ class WebSocketController extends BaseController $msg = $e->getMessage(); } - return json_encode(['code'=>200,'msg'=>$msg,'data'=>$message]); + return json_encode(['code' => 200, 'msg' => $msg, 'data' => $message]); } - /** * 邀请好友入群 * @param array $data 请求参数 @@ -843,11 +829,11 @@ class WebSocketController extends BaseController Log::info('邀请好友入群请求:' . json_encode($params, 256)); $message = $this->sendMessage($params); - return json_encode(['code'=>200,'msg'=>'邀请成功','data'=>$message]); + return json_encode(['code' => 200, 'msg' => '邀请成功', 'data' => $message]); } catch (\Exception $e) { // 记录错误日志 Log::error('邀请好友入群异常:' . $e->getMessage()); - // 返回错误响应 + // 返回错误响应 return json_encode(['code' => 500, 'msg' => '邀请好友入群异常:' . $e->getMessage()]); } } diff --git a/Server/application/command.php b/Server/application/command.php index 570c76a4c..17904a8a7 100644 --- a/Server/application/command.php +++ b/Server/application/command.php @@ -32,5 +32,6 @@ return [ 'sync:wechatData' => 'app\command\SyncWechatDataToCkbTask', // 同步微信数据到存客宝 'sync:allFriends' => 'app\command\SyncAllFriendsCommand', // 同步所有在线好友 'workbench:trafficDistribute' => 'app\command\WorkbenchTrafficDistributeCommand', // 工作台流量分发任务 + 'workbench:groupPush' => 'app\command\WorkbenchGroupPushCommand', // 工作台群组同步任务 'switch:friends' => 'app\command\SwitchFriendsCommand', ]; diff --git a/Server/application/command/SyncWechatDataToCkbTask.php b/Server/application/command/SyncWechatDataToCkbTask.php index 4e1bc4824..420040d9c 100644 --- a/Server/application/command/SyncWechatDataToCkbTask.php +++ b/Server/application/command/SyncWechatDataToCkbTask.php @@ -51,6 +51,8 @@ class SyncWechatDataToCkbTask extends Command $this->syncWechatDeviceLoginLog($ChuKeBaoAdapter); $this->syncWechatDevice($ChuKeBaoAdapter); $this->syncWechatCustomer($ChuKeBaoAdapter); + $this->syncWechatGroup($ChuKeBaoAdapter); + $this->syncWechatGroupCustomer($ChuKeBaoAdapter); $this->syncWechatFriendToTrafficPoolBatch($ChuKeBaoAdapter); $this->syncTrafficSourceUser($ChuKeBaoAdapter); $this->syncTrafficSourceGroup($ChuKeBaoAdapter); @@ -108,5 +110,14 @@ class SyncWechatDataToCkbTask extends Command return $ChuKeBaoAdapter->syncTrafficSourceGroup(); } + protected function syncWechatGroup(ChuKeBaoAdapter $ChuKeBaoAdapter) + { + return $ChuKeBaoAdapter->syncWechatGroup(); + } + protected function syncWechatGroupCustomer(ChuKeBaoAdapter $ChuKeBaoAdapter) + { + return $ChuKeBaoAdapter->syncWechatGroupCustomer(); + } + } \ No newline at end of file diff --git a/Server/application/command/WorkbenchGroupPushCommand.php b/Server/application/command/WorkbenchGroupPushCommand.php new file mode 100644 index 000000000..59c70ddc0 --- /dev/null +++ b/Server/application/command/WorkbenchGroupPushCommand.php @@ -0,0 +1,76 @@ +setName('workbench:groupPush') + ->setDescription('工作台群发同步任务队列') + ->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID,用于区分不同实例', date('YmdHis') . rand(1000, 9999)); + } + + protected function execute(Input $input, Output $output) + { + $output->writeln('开始处理工作台群发同步任务...'); + + try { + // 获取任务ID + $jobId = $input->getOption('jobId'); + + $output->writeln('任务ID: ' . $jobId); + + // 检查队列是否已经在运行 + $queueLockKey = "queue_lock:{$this->queueName}"; + Cache::rm($queueLockKey); + if (Cache::get($queueLockKey)) { + $output->writeln("队列 {$this->queueName} 已经在运行中,跳过执行"); + Log::warning("队列 {$this->queueName} 已经在运行中,跳过执行"); + return false; + } + + // 设置队列运行锁,有效期1小时 + Cache::set($queueLockKey, $jobId, 3600); + $output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时"); + + // 将任务添加到队列 + $this->addToQueue($jobId, $queueLockKey); + + $output->writeln('工作台群发同步任务已添加到队列'); + } catch (\Exception $e) { + Log::error('工作台群发同步任务添加失败:' . $e->getMessage()); + $output->writeln('工作台群发同步任务添加失败:' . $e->getMessage()); + return false; + } + + return true; + } + + /** + * 添加任务到队列 + * @param string $jobId 任务ID + * @param string $queueLockKey 队列锁键名 + */ + public function addToQueue($jobId = '', $queueLockKey = '') + { + $data = [ + 'jobId' => $jobId, + 'queueLockKey' => $queueLockKey + ]; + + // 添加到队列,设置任务名为 workbench_groupPush + Queue::push(WorkbenchGroupPushJob::class, $data, $this->queueName); + } +} \ No newline at end of file diff --git a/Server/application/common/TaskServer.php b/Server/application/common/TaskServer.php index fb3b57d2b..5a6d2aa4f 100644 --- a/Server/application/common/TaskServer.php +++ b/Server/application/common/TaskServer.php @@ -11,7 +11,7 @@ use WeChatDeviceApi\Adapters\ChuKeBao\Adapter as ChuKeBaoAdapter; class TaskServer extends Server { - const PROCESS_COUNT = 4; + const PROCESS_COUNT = 5; protected $socket = 'text://0.0.0.0:2980'; @@ -50,15 +50,23 @@ class TaskServer extends Server Log::info('Workerman进程:' . $current_worker_id); + // 在一个进程里处理获客任务新是数据 + if ($current_worker_id == 4) { + Timer::add(60, function () use($adapter) { + $adapter->handleCustomerTaskNewUser(); + }); + } + + // 在一个进程里处理获客任务添加后的相关逻辑 - if ($current_worker_id == self::PROCESS_COUNT - 1) { + if ($current_worker_id == 3) { Timer::add(60, function () use($adapter) { $adapter->handleCustomerTaskWithStatusIsCreated(); }); } // 3个进程处理获客新任务 - if ($current_worker_id < self::PROCESS_COUNT - 1) { + if ($current_worker_id < 3) { Timer::add(1, function () use ($current_worker_id, $process_count_for_status_0, $adapter) { $adapter->handleCustomerTaskWithStatusIsNew($current_worker_id, $process_count_for_status_0); }); diff --git a/Server/application/cunkebao/config/route.php b/Server/application/cunkebao/config/route.php index 8f7777f8e..2a45e9f6d 100644 --- a/Server/application/cunkebao/config/route.php +++ b/Server/application/cunkebao/config/route.php @@ -7,6 +7,14 @@ use think\facade\Route; // 定义RESTful风格的API路由 Route::group('v1/', function () { + + Route::group('user', function () { + Route::put('editUserInfo', 'app\cunkebao\controller\BaseController@editUserInfo'); + Route::put('editPassWord', 'app\cunkebao\controller\BaseController@editPassWord'); + }); + + + // 设备管理相关 Route::group('devices', function () { Route::put('refresh', 'app\cunkebao\controller\device\RefreshDeviceDetailV1Controller@index'); @@ -86,6 +94,9 @@ Route::group('v1/', function () { Route::get('device-labels', 'app\cunkebao\controller\WorkbenchController@getDeviceLabels'); // 获取设备微信好友标签统计 Route::get('group-list', 'app\cunkebao\controller\WorkbenchController@getGroupList'); // 获取群列表 Route::get('account-list', 'app\cunkebao\controller\WorkbenchController@getAccountList'); // 获取账号列表 + + Route::get('getJdSocialMedia', 'app\cunkebao\controller\WorkbenchController@getJdSocialMedia'); // 获取京东联盟导购媒体 + Route::get('getJdPromotionSite', 'app\cunkebao\controller\WorkbenchController@getJdPromotionSite'); // 获取京东联盟广告位 }); // 内容库相关 diff --git a/Server/application/cunkebao/controller/BaseController.php b/Server/application/cunkebao/controller/BaseController.php index 961497efa..26bd79dba 100644 --- a/Server/application/cunkebao/controller/BaseController.php +++ b/Server/application/cunkebao/controller/BaseController.php @@ -2,8 +2,11 @@ namespace app\cunkebao\controller; +use app\api\controller\AccountController; use app\common\service\ClassTableService; +use library\ResponseHelper; use think\Controller; +use think\Db; /** * 设备管理控制器 @@ -58,4 +61,96 @@ class BaseController extends Controller return $column ? $user[$column] : $user; } + + + public function editUserInfo() + { + $userId = $this->request->param('userId', ''); + $nickname = $this->request->param('nickname', ''); + $avatar = $this->request->param('avatar', ''); + $phone = $this->request->param('phone', ''); + $companyId = $this->getUserInfo('companyId'); + if (empty($userId)) { + return ResponseHelper::error('用户id不能为空'); + } + + if (empty($nickname) && empty($avatar) && empty($phone)) { + return ResponseHelper::error('修改的用户信息不能为空'); + } + + $user = Db::name('users')->where(['id' => $userId, 'companyId' => $companyId])->find(); + if (empty($user)) { + return ResponseHelper::error('用户不存在'); + } + + $user2 = Db::name('users')->where(['phone' => $phone])->find(); + if (!empty($user2) && $user2['id'] != $userId) { + return ResponseHelper::error('修改的手机号已存在'); + } + + $data = [ + 'id' => $user['s2_accountId'], + ]; + + if (!empty($nickname)) { + $data['nickname'] = $nickname; + } + if (!empty($avatar)) { + $data['avatar'] = $avatar; + } + if (!empty($phone)) { + $data['phone'] = $phone; + } + + $AccountControllel = new AccountController(); + $res = $AccountControllel->accountModify($data); + $res = json_decode($res, true); + if ($res['code'] == 200) { + unset($data['id']); + if (!empty($nickname)) { + $data['username'] = $nickname; + unset($data['nickname']); + } + Db::name('users')->where(['id' => $userId, 'companyId' => $companyId])->update($data); + return ResponseHelper::success('更新成功'); + } else { + return ResponseHelper::error($res['msg']); + } + } + + + public function editPassWord() + { + $userId = $this->request->param('userId', ''); + $passWord = $this->request->param('passWord', ''); + $companyId = $this->getUserInfo('companyId'); + if (empty($userId)) { + return ResponseHelper::error('用户id不能为空'); + } + + if (empty($passWord)) { + return ResponseHelper::error('密码不能为空'); + } + + $user = Db::name('users')->where(['id' => $userId, 'companyId' => $companyId])->find(); + if (empty($user)) { + return ResponseHelper::error('用户不存在'); + } + if ($user['passwordMd5'] == md5($passWord)) { + return ResponseHelper::error('新密码与旧密码一致'); + } + + $data = [ + 'passwordMd5' => md5($passWord), + 'passwordLocal' => localEncrypt($passWord), + 'updateTime' => time() + ]; + + $res = Db::name('users')->where(['id' => $userId, 'companyId' => $companyId])->update($data); + if (!empty($res)) { + return ResponseHelper::success('密码修改成功'); + } else { + return ResponseHelper::error('密码修改失败'); + } + } } \ No newline at end of file diff --git a/Server/application/cunkebao/controller/WorkbenchController.php b/Server/application/cunkebao/controller/WorkbenchController.php index 5bf682a6c..95ec91a9a 100644 --- a/Server/application/cunkebao/controller/WorkbenchController.php +++ b/Server/application/cunkebao/controller/WorkbenchController.php @@ -1500,4 +1500,38 @@ class WorkbenchController extends Controller } + /** + * 获取京东联盟导购媒体 + * @return \think\response\Json + * @throws \think\db\exception\DataNotFoundException + * @throws \think\db\exception\ModelNotFoundException + * @throws \think\exception\DbException + */ + public function getJdSocialMedia() + { + $data = Db::name('jd_social_media')->order('id DESC')->select(); + return json(['code' => 200, 'msg' => '获取成功', 'data' => $data]); + } + + /** + * 获取京东联盟广告位 + * @return \think\response\Json + * @throws \think\db\exception\DataNotFoundException + * @throws \think\db\exception\ModelNotFoundException + * @throws \think\exception\DbException + */ + public function getJdPromotionSite() + { + $id = $this->request->param('id', ''); + if (empty($id)) { + return json(['code' => 500, 'msg' => '参数缺失']); + } + + $data = Db::name('jd_promotion_site')->where('jdSocialMediaId',$id)->order('id DESC')->select(); + return json(['code' => 200, 'msg' => '获取成功', 'data' => $data]); + } + + + + } \ No newline at end of file diff --git a/Server/application/cunkebao/controller/plan/PostCreateAddFriendPlanV1Controller.php b/Server/application/cunkebao/controller/plan/PostCreateAddFriendPlanV1Controller.php index a6d0cec2f..3429f4b7b 100644 --- a/Server/application/cunkebao/controller/plan/PostCreateAddFriendPlanV1Controller.php +++ b/Server/application/cunkebao/controller/plan/PostCreateAddFriendPlanV1Controller.php @@ -127,7 +127,6 @@ class PostCreateAddFriendPlanV1Controller extends Controller ]; - try { Db::startTrans(); // 插入数据 @@ -263,6 +262,12 @@ class PostCreateAddFriendPlanV1Controller extends Controller } } + //群获客 + if($params['sceneId'] == 7){ + + } + + Db::commit(); return ResponseHelper::success(['planId' => $planId], '添加计划任务成功'); diff --git a/Server/application/cunkebao/controller/traffic/GetPotentialListWithInCompanyV1Controller.php b/Server/application/cunkebao/controller/traffic/GetPotentialListWithInCompanyV1Controller.php index 9a68d3c74..ec35f5031 100644 --- a/Server/application/cunkebao/controller/traffic/GetPotentialListWithInCompanyV1Controller.php +++ b/Server/application/cunkebao/controller/traffic/GetPotentialListWithInCompanyV1Controller.php @@ -270,7 +270,7 @@ class GetPotentialListWithInCompanyV1Controller extends BaseController $percentage = number_format(($taskNum / $passNum) * 100, 2); $total['percentage'] = $percentage; } - + $data['total'] = $total; @@ -322,14 +322,20 @@ class GetPotentialListWithInCompanyV1Controller extends BaseController public function getUserTags() { $userId = $this->request->param('userId', ''); + $companyId = $this->getUserInfo('companyId'); if (empty($userId)) { return json_encode(['code' => 500, 'msg' => '用户id不能为空']); } $data = Db::name('traffic_pool')->alias('tp') - ->join(['s2_wechat_friend' => 'wf'], 'tp.wechatId=wf.wechatId', 'left') + ->join('wechat_friendship f', 'tp.wechatId=f.wechatId AND f.companyId='.$companyId, 'left') + ->join(['s2_wechat_friend' => 'wf'], 'f.wechatId=wf.wechatId', 'left') ->where(['tp.id' => $userId]) ->order('tp.createTime desc') ->column('wf.id,wf.labels,wf.siteLabels'); + if (empty($data)) { + return ResponseHelper::success(['wechat' => [], 'siteLabels' => []]); + } + $tags = []; $siteLabels = []; @@ -350,5 +356,31 @@ class GetPotentialListWithInCompanyV1Controller extends BaseController return ResponseHelper::success(['wechat' => $tags, 'siteLabels' => $siteLabels]); } + /* public function editUserTags() + { + $userId = $this->request->param('userId', ''); + if (empty($userId)) { + return json_encode(['code' => 500, 'msg' => '用户id不能为空']); + } + $tags = $this->request->param('tags', []); + $tags = $this->request->param('tags', []); + $isWechat = $this->request->param('isWechat', false); + $companyId = $this->getUserInfo('companyId'); + + $friend = Db::name('traffic_pool')->alias('tp') + ->join('wechat_friendship f', 'tp.wechatId=f.wechatId AND f.companyId='.$companyId, 'left') + ->join(['s2_wechat_friend' => 'wf'], 'f.wechatId=wf.wechatId', 'left') + ->where(['tp.id' => $userId]) + ->order('tp.createTime desc') + ->column('wf.id,wf.accountId,wf.labels,wf.siteLabels'); + if (empty($data)) { + return ResponseHelper::error('该用户不存在'); + } + + + }*/ + + + } \ No newline at end of file diff --git a/Server/application/job/WorkbenchGroupPushJob.php b/Server/application/job/WorkbenchGroupPushJob.php new file mode 100644 index 000000000..cf9e77c2c --- /dev/null +++ b/Server/application/job/WorkbenchGroupPushJob.php @@ -0,0 +1,401 @@ +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])->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: ''} + // $result = [ + // "content" => $dataArray['content'], + // "msgSubType" => 0, + // "msgType" => $dataArray['msgType'], + // "seq" => time(), + // "wechatAccountId" => $dataArray['wechatAccountId'], + // "wechatChatroomId" => 0, + // "wechatFriendId" => $dataArray['wechatFriendId'], + // ]; + + + $groups = json_decode($config['groups'], true); + $groupsData = Db::name('wechat_group')->whereIn('id', $groups)->field('id,wechatAccountId,chatroomId,companyId,ownerWechatId')->select(); + if (empty($groupsData)) { + return false; + } + + $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]); + foreach ($msgConf as $content) { + $sendData = []; + $sqlData = []; + + foreach ($groupsData as $groups) { + // msgType(1:文本 3:图片 43:视频 47:动图表情包(gif、其他表情包) 49:小程序/其他:图文、文件) + $sqlData[] = [ + 'workbenchId' => $workbench['id'], + 'contentId' => $content['id'], + 'groupId' => $groups['id'], + 'wechatAccountId' => $groups['wechatAccountId'], + 'createTime' => time() + ]; + + //内容 + if (!empty($content['content'])) { + $sendData[] = [ + 'content' => $content['content'], + 'msgType' => 1, + 'wechatAccountId' => $groups['wechatAccountId'], + 'wechatChatroomId' => $groups['id'], + ]; + } + + switch ($content['contentType']) { + case 1: + //图片解析 + $imgs = json_decode($content['resUrls'], true); + if (!empty($imgs)) { + foreach ($imgs as $img) { + $sendData[] = [ + 'content' => $img, + 'msgType' => 3, + 'wechatAccountId' => $groups['wechatAccountId'], + 'wechatChatroomId' => $groups['id'], + ]; + } + } + break; + case 2: + //链接解析 + $url = json_decode($content['urls'], true); + if (!empty($url[0])) { + $url = $url[0]; + $sendData[] = [ + 'content' => [ + 'desc' => '', + 'thumbPath' => $url['image'], + 'title' => $url['desc'], + 'type' => 'link', + 'url' => $url['url'], + ], + 'msgType' => 49, + 'wechatAccountId' => $groups['wechatAccountId'], + 'wechatChatroomId' => $groups['id'], + ]; + } + + break; + case 3: + //视频解析 + $video = json_decode($content['urls'], true); + if (!empty($video)) { + $video = $video[0]; + } + $sendData[] = [ + 'content' => $video, + 'msgType' => 43, + 'wechatAccountId' => $groups['wechatAccountId'], + 'wechatChatroomId' => $groups['id'], + ]; + break; + } + + if (empty($sendData)) { + continue; + } + + //发送消息 + foreach ($sendData as $send) { + $wsController->sendCommunity($send); + } + //插入发送记录 + Db::name('workbench_group_push_item')->insertAll($sqlData); + } + } + } + + + /** + * 记录发送历史 + * @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; + } + $interval = floor($totalSeconds / $config['maxPerDay']); + + + // 查询今日已同步次数 + $count = Db::name('workbench_group_push_item') + ->where('workbenchId', $workbench->id) + ->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; + } + + 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'; + } + + // 基础查询 + $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, '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; + // 根据accountType处理不同的发送逻辑 + if ($config['isLoop'] == 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)->order('id desc')->find(); + $fastSendData = Db::name('workbench_group_push_item')->where('workbenchId', $workbench->id)->order('id asc')->find(); + + $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 { + // 不能循环发送,只获取未发送的内容 + $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; + } +} \ No newline at end of file diff --git a/Server/crontab_tasks.md b/Server/crontab_tasks.md index 00aadff30..f6852e586 100644 --- a/Server/crontab_tasks.md +++ b/Server/crontab_tasks.md @@ -57,6 +57,9 @@ # 同步微信数据到存客宝 0 9 * * * cd /www/wwwroot/mckb_quwanzhi_com/Server && php think sync:wechatData >> /www/wwwroot/mckb_quwanzhi_com/Server/runtime/log/sync_wechat_data.log 2>&1 +# 工作台群发消息 +*/2 * * * * cd /www/wwwroot/mckb_quwanzhi_com/Server && php think workbench:groupPush >> /www/wwwroot/mckb_quwanzhi_com/Server/runtime/log/workbench_groupPush.log 2>&1 + # 工作台流量分发 0 9 * * * cd /www/wwwroot/mckb_quwanzhi_com/Server && php think workbench:trafficDistribute >> /www/wwwroot/mckb_quwanzhi_com/Server/runtime/log/traffic_distribute.log 2>&1 diff --git a/Server/extend/WeChatDeviceApi/Adapters/ChuKeBao/Adapter.php b/Server/extend/WeChatDeviceApi/Adapters/ChuKeBao/Adapter.php index 2dbded4d2..03d7c51d9 100644 --- a/Server/extend/WeChatDeviceApi/Adapters/ChuKeBao/Adapter.php +++ b/Server/extend/WeChatDeviceApi/Adapters/ChuKeBao/Adapter.php @@ -6,6 +6,7 @@ use think\facade\Cache; use think\facade\Env; use WeChatDeviceApi\Contracts\WeChatServiceInterface; use WeChatDeviceApi\Exceptions\ApiException; + // 如果有 Client.php // use WeChatDeviceApi\Adapters\ChuKeBao\Client as ChuKeBaoApiClient; use GuzzleHttp\Client; @@ -23,6 +24,7 @@ use Workerman\Lib\Timer; class Adapter implements WeChatServiceInterface { protected $config; + // protected $apiClient; // 如果使用 VendorAApiClient public function __construct(array $config = []) @@ -154,8 +156,8 @@ class Adapter implements WeChatServiceInterface public function handleCustomerTaskWithStatusIsNew(int $current_worker_id, int $process_count_for_status_0) { $task = Db::name('customer_acquisition_task') - ->where(['status' => 1,'deleteTime' => 0]) -// ->whereRaw("id % $process_count_for_status_0 = {$current_worker_id}") + ->where(['status' => 1, 'deleteTime' => 0]) + ->whereRaw("id % $process_count_for_status_0 = {$current_worker_id}") ->order('id desc') ->select(); @@ -168,11 +170,11 @@ class Adapter implements WeChatServiceInterface $reqConf = json_decode($item['reqConf'], true); $device = $reqConf['device'] ?? []; $deviceCount = count($device); - if ($deviceCount <= 0){ + if ($deviceCount <= 0) { continue; } $tasks = Db::name('task_customer') - ->where(['status'=> 0,'task_id'=>$item['id']]) + ->where(['status' => 0, 'task_id' => $item['id']]) ->order('id DESC') ->limit($deviceCount) ->select(); @@ -198,7 +200,7 @@ class Adapter implements WeChatServiceInterface foreach ($wechatIdAccountIdMap as $accountId => $wechatId) { // 是否已经是好友的判断,如果已经是好友,直接break; 但状态还是维持1,让另外一个进程处理发消息的逻辑 $wechatTags = json_decode($task['tags'], true); - $isFriend = $this->checkIfIsWeChatFriendByPhone($wechatId, $task['phone'],$task['siteTags']); + $isFriend = $this->checkIfIsWeChatFriendByPhone($wechatId, $task['phone'], $task['siteTags']); if (!empty($isFriend)) { $friendAddTaskCreated = true; $task['processed_wechat_ids'] = $task['processed_wechat_ids'] . ',' . $wechatId; // 处理失败任务用,用于过滤已处理的微信号 @@ -220,13 +222,13 @@ class Adapter implements WeChatServiceInterface // 采取乐观尝试的策略,假设第一个可以添加的人可以添加成功的; 回头再另外一个任务进程去判断 // 创建好友添加任务, 对接触客宝 - $tags = array_merge($task_info['tagConf']['customTags'],$task_info['tagConf']['scenarioTags']); - if (!empty($wechatTags)){ - $tags = array_merge($tags,$wechatTags); + $tags = array_merge($task_info['tagConf']['customTags'], $task_info['tagConf']['scenarioTags']); + if (!empty($wechatTags)) { + $tags = array_merge($tags, $wechatTags); } $tags = array_unique($tags); $tags = array_values($tags); - $conf = array_merge($task_info['reqConf'], ['task_name' => $task_info['name'],'tags' => $tags]); + $conf = array_merge($task_info['reqConf'], ['task_name' => $task_info['name'], 'tags' => $tags]); $this->createFriendAddTask($accountId, $task['phone'], $conf); @@ -249,13 +251,12 @@ class Adapter implements WeChatServiceInterface } } - // 处理添加中的获客任务, only run in workerman process! public function handleCustomerTaskWithStatusIsCreated() { $tasks = Db::name('task_customer') - ->whereIn('status', [1,2]) + ->whereIn('status', [1, 2]) ->where('updateTime', '>=', (time() - 86400 * 3)) ->limit(50) ->order('updateTime DESC') @@ -280,20 +281,19 @@ class Adapter implements WeChatServiceInterface $weChatIds = explode(',', $task['processed_wechat_ids']); $passedWeChatId = ''; - foreach ($weChatIds as $wechatId) { - // 先是否是好友,如果不是好友,先查询执行状态,看是否还能以及需要换账号继续添加,还是直接更新状态为3 // 如果添加成功,先更新为2,然后去发消息(先判断有无消息设置,发消息的log记录?) - $isFriend = $this->checkIfIsWeChatFriendByPhone($wechatId, $task['phone']); - if ($isFriend) { - $passedWeChatId = $wechatId; - break; + if (!empty($wechatId)) { + $isFriend = $this->checkIfIsWeChatFriendByPhone($wechatId, $task['phone']); + if ($isFriend) { + $passedWeChatId = $wechatId; + break; + } } } - if ($passedWeChatId && !empty($task_info['msgConf'])) { Db::name('task_customer') @@ -334,6 +334,104 @@ class Adapter implements WeChatServiceInterface } } + + public function handleCustomerTaskNewUser() + { + $task = Db::name('customer_acquisition_task') + ->where(['status' => 1, 'deleteTime' => 0]) + ->whereIn('sceneId', [7]) + ->order('id desc') + ->select(); + + if (empty($task)) { + return false; + } + + foreach ($task as $item) { + $sceneConf = json_decode($item['sceneConf'], true); + //群获客 + if ($item['sceneId'] == 7) { + if (!empty($sceneConf['groupSelected']) && is_array($sceneConf['groupSelected'])) { + $rows = Db::name('wechat_group_member')->alias('gm') + ->join('wechat_account wa', 'gm.identifier = wa.wechatId') + ->where('gm.companyId', $item['companyId']) + ->whereIn('gm.groupId', $sceneConf['groupSelected']) + ->group('gm.identifier') + ->column('wa.id,wa.wechatId,wa.alias,wa.phone'); + + + // 1000条为一组进行批量处理 + $batchSize = 1000; + $totalRows = count($rows); + + for ($i = 0; $i < $totalRows; $i += $batchSize) { + $batchRows = array_slice($rows, $i, $batchSize); + + if (!empty($batchRows)) { + // 1. 提取当前批次的phone + $phones = []; + foreach ($batchRows as $row) { + if (!empty($row['phone'])) { + $phone = !empty($row['phone']); + } elseif (!empty($row['alias'])) { + $phone = $row['alias']; + } else { + $phone = $row['wechatId']; + } + if (!empty($phone)) { + $phones[] = $phone; + } + } + + // 2. 批量查询已存在的phone + $existingPhones = []; + if (!empty($phones)) { + $existing = Db::name('task_customer') + ->where('task_id', $item['id']) + ->where('phone', 'in', $phones) + ->field('phone') + ->select(); + $existingPhones = array_column($existing, 'phone'); + } + + // 3. 过滤出新数据,批量插入 + $newData = []; + foreach ($batchRows as $row) { + if (!empty($row['phone'])) { + $phone = !empty($row['phone']); + } elseif (!empty($row['alias'])) { + $phone = $row['alias']; + } else { + $phone = $row['wechatId']; + } + if (!empty($phone) && !in_array($phone, $existingPhones)) { + $newData[] = [ + 'task_id' => $item['id'], + 'name' => '', + 'source' => '场景获客_' . $item['name'], + 'phone' => $phone, + 'tags' => json_encode([], JSON_UNESCAPED_UNICODE), + 'siteTags' => json_encode([], JSON_UNESCAPED_UNICODE), + 'createTime' => time(), + ]; + } + } + + // 4. 批量插入新数据 + if (!empty($newData)) { + Db::name('task_customer')->insertAll($newData); + } + } + } + } + } + + + exit_data($sceneConf); + } + } + + // 发微信个人消息 public function sendMsgToFriend(int $friendId, int $wechatAccountId, array $msgConf) { @@ -352,7 +450,7 @@ class Adapter implements WeChatServiceInterface $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'); + $toAccountId = Db::name('users')->where('account', $username)->value('s2_accountId'); } // 建立WebSocket @@ -381,7 +479,7 @@ class Adapter implements WeChatServiceInterface case 'file': $msgType = 49; - + $detail = [ 'type' => 'file', 'title' => $content['content'][0]['name'], @@ -418,12 +516,12 @@ class Adapter implements WeChatServiceInterface } - if(empty($detail)){ + if (empty($detail)) { continue; } if ($gap) { - Timer::add($gap, function () use ($wsController, $friendId, $wechatAccountId, $msgType, $content,$detail) { + Timer::add($gap, function () use ($wsController, $friendId, $wechatAccountId, $msgType, $content, $detail) { $wsController->sendPersonal([ 'wechatFriendId' => $friendId, 'wechatAccountId' => $wechatAccountId, @@ -463,7 +561,8 @@ class Adapter implements WeChatServiceInterface } // 检查是否是好友关系 - public function checkIfIsWeChatFriendByPhone(string $wxId, string $phone,string $siteTags): bool + + public function checkIfIsWeChatFriendByPhone($wxId = '', $phone = '', $siteTags = '') { if (empty($wxId) || empty($phone)) { return false; @@ -472,7 +571,7 @@ class Adapter implements WeChatServiceInterface try { $friend = Db::table('s2_wechat_friend') ->where('ownerWechatId', $wxId) - ->where(['isPassed' => 1,'isDeleted' => 0]) + ->where(['isPassed' => 1, 'isDeleted' => 0]) ->where('phone|alias|wechatId', 'like', $phone . '%') ->order('createTime', 'desc') ->find(); @@ -480,17 +579,17 @@ class Adapter implements WeChatServiceInterface if (!empty($siteTags)) { $siteTags = json_decode($siteTags, true); $siteLabels = json_decode($friend['siteLabels'], true); - $tags = array_merge($siteTags,$siteLabels); + $tags = array_merge($siteTags, $siteLabels); $tags = array_unique($tags); $tags = array_values($tags); - if (empty($tags)){ + if (empty($tags)) { $tags = []; } - $tags = json_encode($tags,256); - Db::table('s2_wechat_friend')->where(['id' => $friend['id']])->update(['siteLabels' => $tags,'updateTime' => time()]); + $tags = json_encode($tags, 256); + Db::table('s2_wechat_friend')->where(['id' => $friend['id']])->update(['siteLabels' => $tags, 'updateTime' => time()]); } return true; - }else{ + } else { return false; } } catch (\Exception $e) { @@ -637,8 +736,7 @@ class Adapter implements WeChatServiceInterface } //强制请求添加好友的列表 $friendController = new FriendTaskController(); - $friendController->getlist(0,50); - + $friendController->getlist(0, 50); $record = $this->getLatestFriendTask($wechatId); @@ -751,9 +849,9 @@ class Adapter implements WeChatServiceInterface $friendController = new FriendTaskController(); $result = $friendController->addFriendTask($params); $result = json_decode($result, true); - if ($result['code'] == 200){ + if ($result['code'] == 200) { return $result; - }else{ + } else { $authorization = AuthService::getSystemAuthorization(false); return $this->addFriendTaskApi($wechatAccountId, $phone, $message, $remark, $labels, $authorization); } @@ -768,7 +866,7 @@ class Adapter implements WeChatServiceInterface return; } - switch ($conf['remarkType']){ + switch ($conf['remarkType']) { case 'phone': $remark = $phone . '-' . $conf['task_name']; break; @@ -776,7 +874,7 @@ class Adapter implements WeChatServiceInterface $remark = ''; break; case 'source': - $remark = $conf['task_name']; + $remark = $conf['task_name']; break; default: $remark = ''; @@ -955,7 +1053,7 @@ class Adapter implements WeChatServiceInterface * 大数据量分批处理版本 * 适用于数据源非常大的情况,避免一次性加载全部数据到内存 * 独立脚本执行,30min 同步一次 和 流量来源的更新一起 - * + * * @param int $batchSize 每批处理的数据量 * @return int 影响的行数 */ @@ -1009,7 +1107,7 @@ class Adapter implements WeChatServiceInterface /** * 同步/更新微信客服信息到ck_wechat_customer表 - * + * * @param int $batchSize 每批处理的数据量 * @return int 影响的行数 */ @@ -1151,7 +1249,7 @@ class Adapter implements WeChatServiceInterface /** * 计算客服权重 - * + * * @param array $basic 基础信息 * @param array $activity 活跃信息 * @param array $friendShip 好友关系信息 @@ -1213,7 +1311,7 @@ class Adapter implements WeChatServiceInterface /** * 同步设备信息到ck_device表 * 数据量不大,仅同步一次所有设备 - * + * * @return int 影响的行数 */ public function syncDevice() @@ -1318,4 +1416,76 @@ class Adapter implements WeChatServiceInterface } } while ($affected > 0); } + + public function syncWechatGroup() + { + $sql = "insert into ck_wechat_group(`id`,`wechatAccountId`,`chatroomId`,`name`,`avatar`,`companyId`,`ownerWechatId`,`createTime`,`updateTime`,`deleteTime`) + SELECT + g.id id, + g.wechatAccountId wechatAccountId, + g.chatroomId chatroomId, + g.nickname name, + g.chatroomAvatar avatar, + c.departmentId companyId, + g.wechatAccountWechatId ownerWechatId, + g.createTime createTime, + g.updateTime updateTime, + g.deleteTime deleteTime + FROM + s2_wechat_chatroom g + LEFT JOIN s2_company_account c ON g.accountId = c.id + ORDER BY g.id DESC + LIMIT ?, ? + ON DUPLICATE KEY UPDATE + chatroomId=VALUES(chatroomId), + companyId=VALUES(companyId), + ownerWechatId=VALUES(ownerWechatId)"; + + + $offset = 0; + $limit = 2000; + $usleepTime = 50000; + do { + $affected = Db::execute($sql, [$offset, $limit]); + $offset += $limit; + if ($affected > 0) { + usleep($usleepTime); + } + } while ($affected > 0); + } + + public function syncWechatGroupCustomer() + { + $sql = "insert into ck_wechat_group_member(`identifier`,`chatroomId`,`companyId`,`groupId`,`createTime`) + SELECT + m.wechatId identifier, + g.chatroomId chatroomId, + c.departmentId companyId, + g.id groupId, + m.createTime createTime + FROM + s2_wechat_chatroom_member m + LEFT JOIN s2_wechat_chatroom g ON g.chatroomId = m.chatroomId + LEFT JOIN s2_company_account c ON g.accountId = c.id + ORDER BY m.id DESC + LIMIT ?, ? + ON DUPLICATE KEY UPDATE + identifier=VALUES(identifier), + chatroomId=VALUES(chatroomId), + companyId=VALUES(companyId), + groupId=VALUES(groupId)"; + + $offset = 0; + $limit = 2000; + $usleepTime = 50000; + do { + $affected = Db::execute($sql, [$offset, $limit]); + $offset += $limit; + if ($affected > 0) { + usleep($usleepTime); + } + } while ($affected > 0); + } + + }