diff --git a/app/command/TbCommand.php b/app/command/TbCommand.php index fa0a33a..265d24c 100644 --- a/app/command/TbCommand.php +++ b/app/command/TbCommand.php @@ -29,7 +29,7 @@ class TbCommand extends Command // } $m = new Emails(); - $m->process(); + $m->fenpai(); return self::SUCCESS; } diff --git a/app/common/logic/AuthLogic.php b/app/common/logic/AuthLogic.php index b2a74fc..59155dc 100644 --- a/app/common/logic/AuthLogic.php +++ b/app/common/logic/AuthLogic.php @@ -93,7 +93,7 @@ class AuthLogic ->orWhere('email', $email); }) ->first(); - + $user->isLogin = true; if (!$user) { // 获取编号 $userSn = $this->getAvailableUserSn(); diff --git a/app/common/model/EmailParsedResult.php b/app/common/model/EmailParsedResult.php new file mode 100644 index 0000000..de1a1be --- /dev/null +++ b/app/common/model/EmailParsedResult.php @@ -0,0 +1,49 @@ +key = "uid_rotator:{$rotatorName}:index"; + $this->setKey = "uid_rotator:{$rotatorName}:uids"; + $this->lockKey = "uid_rotator:{$rotatorName}:lock"; + $this->expireSeconds = $expireSeconds; + } + + public function setUids(array $uids, bool $resetIndex = true): bool + { + // 删除旧数据 + Redis::del($this->setKey); + + if (!empty($uids)) { + // 添加新UID集合 + foreach ($uids as $uid) { + Redis::sAdd($this->setKey, $uid); + } + + // 重置索引 + if ($resetIndex) { + Redis::set($this->key, 0); + } + } + + $this->touchExpire(); + return true; + } + + public function next(): ?int + { + $lockAcquired = false; + try { + // 获取锁(改用原生命令) + $lockAcquired = $this->acquireLock(5); + if (!$lockAcquired) { + throw new \RuntimeException("获取锁失败"); + } + + // 获取当前索引 + $currentIndex = (int)Redis::get($this->key); + + // 获取所有UID + $uids = Redis::sMembers($this->setKey); + if (empty($uids)) { + return null; + } + + // 计算位置 + $uid = $uids[$currentIndex % count($uids)]; + $nextIndex = $currentIndex + 1; + + // 更新索引 + Redis::set($this->key, $nextIndex); + $this->touchExpire(); + + return is_numeric($uid) ? (int)$uid : $uid; + } catch (\Throwable $e) { + Log::info("UID分配错误: " . $e->getMessage()); + return null; + } finally { + if ($lockAcquired) { + Redis::del($this->lockKey); + } + } + } + + private function acquireLock(int $timeout = 5): bool + { + $start = time(); + $lockTimeout = $timeout + 1; // 锁过期时间比等待时间长 + + while (time() - $start < $timeout) { + // 使用SETNX实现锁(分两步) + if (Redis::setNx($this->lockKey, 1)) { + // 设置过期时间 + Redis::expire($this->lockKey, $lockTimeout); + return true; + } + + // 检查锁是否已过期但未被删除(处理死锁) + if (Redis::ttl($this->lockKey) < 0) { + Redis::del($this->lockKey); + } + + usleep(100000); // 100ms重试间隔 + } + + return false; + } + + private function touchExpire(): void + { + Redis::expire($this->key, $this->expireSeconds); + Redis::expire($this->setKey, $this->expireSeconds); + } + + public function getPosition(): int + { + return (int)Redis::get($this->key); + } + + public function resetIndex(int $position = 0): bool + { + return (bool)Redis::set($this->key, max(0, $position)); + } + + public function getUids(): array + { + return Redis::sMembers($this->setKey) ?: []; + } + + public function addUids($uids): int + { + $added = 0; + $uids = is_array($uids) ? $uids : [$uids]; + + foreach ($uids as $uid) { + $added += Redis::sAdd($this->setKey, $uid); + } + + $this->touchExpire(); + return $added; + } + + public function removeUids($uids): int + { + $removed = 0; + $uids = is_array($uids) ? $uids : [$uids]; + + foreach ($uids as $uid) { + $removed += Redis::sRem($this->setKey, $uid); + } + + return $removed; + } + + public function count(): int + { + return (int)Redis::sCard($this->setKey); + } + + public function destroy(): void + { + Redis::del([$this->key, $this->setKey, $this->lockKey]); + } +} \ No newline at end of file diff --git a/app/common/utils/UidRotator.php b/app/common/utils/UidRotator.php new file mode 100644 index 0000000..7e0746c --- /dev/null +++ b/app/common/utils/UidRotator.php @@ -0,0 +1,178 @@ +first(); + + // 固定渠道 + $channel = 8; + $channelInfo = InquiryChannel::where('id', $channel)->first(); + + // 初始化 UID 轮询器 + $uidRotator = new \app\common\utils\RedisUidRotator('email_parsed_result_20250804'); + $uidRotator->setUids([64, 63, 83, 93, 108, 118, 131, 132, 145, 150, 143, 156, 157, 158]); + + EmailParsedResult::where('status', 0) + ->chunkById($batchSize, function ($records) use ($delaySeconds, $throttleAfter, $throttleSeconds, $enterUser, $channelInfo, $uidRotator) { + $count = 0; + + foreach ($records as $record) { + try { + // 获取国家信息 + $countryName = $record->iplocation ? removeParentheses($record->iplocation) : '未知'; + $countryInfo = Country::where('cn_addr', $countryName)->first(); + + // 获取下一个分配的UID + $assignedUid = $uidRotator->next(); + + // 获取分配的用户信息 + $recipientInfo = User::where('id', $assignedUid)->first(); + + // 生成录单 + $log = [ + 'order_no' => orderNum(), + 'enter_user_id' => $enterUser->id, + 'recipient' => $assignedUid, + 'quality' => 2, + 'channel' => $channelInfo->id, + 'email' => $record->email, + 'tel' => $record->tel, + 'whatsapp' => '', + 'country' => $countryInfo->id, + 'contacts' => $record->name, + 'content' => $record->message, + 'r_time' => time(), + 't_time' => time(), + 'status' => 2, + 'sync_crm' => 1 + ]; + $inquiryLogResult = InquiryLog::create($log); + if (!$inquiryLogResult) { + throw new BusinessException('录单失败~'); + } + + // 同步客户 + $crmCustomer = new CrmCustomer(); + $crmCustomer->sn = $inquiryLogResult->order_no; + $crmCustomer->corp_abbr = $recipientInfo->corp_id; + $crmCustomer->user_id = $assignedUid; + $crmCustomer->type = 1; + $crmCustomer->channel_id = $inquiryLogResult->channel; + $crmCustomer->country_id = $inquiryLogResult->country; + $crmCustomer->magnitude = $inquiryLogResult->quality; + $crmCustomer->email = $inquiryLogResult->email; + $crmCustomer->phone = $inquiryLogResult->tel; + $crmCustomer->WhatsApp = $inquiryLogResult->whatsapp; + $crmCustomer->contacts = $inquiryLogResult->contacts; + $crmCustomer->desc_collect = $inquiryLogResult->content; + $crmCustomer->enter_user_id = $inquiryLogResult->enter_user_id; + if (!$crmCustomer->save()) { + throw new BusinessException('同步失败~'); + } + + // 邮件模板 + $data = [ + 'order_no' => $inquiryLogResult->order_no, + 'enter_user_id' => $enterUser->name, + 'recipient' => $recipientInfo->name, + 'quality' => (new \app\inquiry\service\InquiryLogService())->qualityName($inquiryLogResult->quality), + 'channel' => $channelInfo->name, + 'email' => $inquiryLogResult->email && $inquiryLogResult->email != '' ? $inquiryLogResult->email : ' - ', + 'tel' => $inquiryLogResult->tel && $inquiryLogResult->tel != '' ? $inquiryLogResult->tel : ' - ', + 'whatsapp' => $inquiryLogResult->whatsapp && $inquiryLogResult->whatsapp != '' ? $inquiryLogResult->whatsapp : ' - ', + 'country' => $countryName, + 'contacts' => $inquiryLogResult->contacts && $inquiryLogResult->contacts != '' ? $inquiryLogResult->contacts : ' - ', + 'content' => $inquiryLogResult->content && $inquiryLogResult->content != '' ? $inquiryLogResult->content : ' - ', + 'r_time' => $inquiryLogResult->r_time ? date('Y-m-d H:i:s', $inquiryLogResult->r_time) : ' - ', + 't_time' => $inquiryLogResult->t_time ? date('Y-m-d H:i:s', $inquiryLogResult->t_time) : ' - ' + ]; + \app\common\utils\email\MailSend::send($recipientInfo->email, '【XYH】新的客户线索', replaceTpl($data)); + + // 更新状态 + $record->update(['status' => 1]); + flg('====== 编号:'.$inquiryLogResult->order_no.' 分配给:'.$recipientInfo->name.' email:'.$recipientInfo->email, false); + + // 每封邮件后延迟 + if ($delaySeconds > 0) { + sleep($delaySeconds); + } + + // 每发送一定数量后额外休息 + if (++$count % $throttleAfter === 0) { + sleep($throttleSeconds); + flg("已发送 {$count} 封邮件,休息 {$throttleSeconds} 秒", false); + } + + } catch (\Exception $e) { + flg("邮件发送失败 ID:{$record->id} - " . $e->getMessage()); + // 更新状态 + $record->update(['status' => 3]); + // 失败后也加入延迟,避免快速连续失败 + sleep($delaySeconds); + } + } + }); + } catch (\Throwable $th) { + flg(dpr($th)); + return false; + } + } } \ No newline at end of file diff --git a/app/controller/IndexController.php b/app/controller/IndexController.php index d9ac96c..6dc874b 100644 --- a/app/controller/IndexController.php +++ b/app/controller/IndexController.php @@ -80,7 +80,7 @@ class IndexController // ], 'model' => [ 'module' => 'common', - 'name' => 'inquiry_channel' + 'name' => 'email_parsed_result' ], // 'service' => [ // 'module' => 'inquiry', @@ -94,7 +94,7 @@ class IndexController // 'module' => 'inquiry', // 'name' => 'country' // ], - 'table' => 'inquiry_channel' + 'table' => 'email_parsed_result' ]; $result = (new Generate($options))->doGenerate(); diff --git a/app/functions.php b/app/functions.php index 45af92c..f69f082 100644 --- a/app/functions.php +++ b/app/functions.php @@ -194,4 +194,30 @@ if (!function_exists('InquiryFormat')) { $content .= '内容:'.$data['content']; return $content; } +} + +/** + * 移除字符串中的括号及其内部内容(仅移除第一个匹配项) + * + * @param string $str 要处理的字符串 + * @return string 处理后的字符串 + */ +if (!function_exists('removeParentheses')) { + function removeParentheses(string $str): string + { + // 匹配模式:半角括号或全角括号及其内部内容 + $pattern = '/ + (?: # 非捕获分组开始 + \( # 匹配半角左括号 + [^)]* # 匹配除半角右括号外的任意字符 + \) # 匹配半角右括号 + )|(?: # 或 + \( # 匹配全角左括号 + [^)]* # 匹配除全角右括号外的任意字符 + \) # 匹配全角右括号 + ) # 非捕获分组结束 + /xu'; // x: 忽略空白 u: Unicode模式 + + return trim(preg_replace($pattern, '', $str, 1)); + } } \ No newline at end of file diff --git a/app/process/Emails.php b/app/process/Emails.php index 7886fff..f9641ea 100644 --- a/app/process/Emails.php +++ b/app/process/Emails.php @@ -2,8 +2,6 @@ namespace app\process; -use app\crm\model\Customer; -use support\Log; use Workerman\Crontab\Crontab; use app\common\utils\email\Emails as EmailsUtils; @@ -12,23 +10,47 @@ class Emails public function onWorkerStart() { - // 拉取新邮件 - new Crontab('0 */30 * * * *', function(){ + // // 拉取新邮件 + // new Crontab('0 */30 * * * *', function(){ + // $email = new EmailsUtils(); + // $email->uids(); + // }); + + // // 邮件详情 + // new Crontab('0 */35 * * * *', function(){ + // $email = new EmailsUtils(); + // $email->details(); + // }); + + // // 邮件处理 + // new Crontab('0 */40 * * * *', function(){ + // $email = new EmailsUtils(); + // $email->process(); + // }); + + // 拉取新邮件 - 每3小时的第0分钟0秒执行 + new Crontab('0 0 */3 * * *', function(){ $email = new EmailsUtils(); $email->uids(); }); - // 邮件详情 - new Crontab('0 */35 * * * *', function(){ + // 邮件详情 - 每3小时的第10分钟0秒执行 + new Crontab('0 10 */3 * * *', function(){ $email = new EmailsUtils(); $email->details(); }); - // 邮件处理 - new Crontab('0 */40 * * * *', function(){ + // 邮件处理 - 每3小时的第20分钟0秒执行 + new Crontab('0 20 */3 * * *', function(){ $email = new EmailsUtils(); $email->process(); }); + + // 邮件分派 - 每3小时的第30分钟0秒执行 + new Crontab('0 30 */3 * * *', function(){ + $email = new EmailsUtils(); + $email->fenpai(); + }); } } \ No newline at end of file diff --git a/app/queue/redis/EmailMessage.php b/app/queue/redis/EmailMessage.php index 8fbeb22..7291868 100644 --- a/app/queue/redis/EmailMessage.php +++ b/app/queue/redis/EmailMessage.php @@ -20,9 +20,9 @@ class EmailMessage implements Consumer try { // 询单信息 $inquiryLogService = new InquiryLogService(); - $data = $inquiryLogService->getContent($data['id']); + $content = $inquiryLogService->getContent($data['id']); - MailSend::send($data['email'], '【XYH】新的客户线索', replaceTpl($data)); + MailSend::send($data['email'], '【XYH】新的客户线索', replaceTpl($content)); } catch (\Throwable $th) { throw $th; } diff --git a/app/queue/redis/InquiryToCrm.php b/app/queue/redis/InquiryToCrm.php index 966d4a3..8e1ae51 100644 --- a/app/queue/redis/InquiryToCrm.php +++ b/app/queue/redis/InquiryToCrm.php @@ -9,6 +9,7 @@ use Webman\RedisQueue\Consumer; use app\common\model\User; use app\common\utils\FeishuMessage; use support\exception\BusinessException; +use support\Log; class InquiryToCrm implements Consumer { @@ -52,7 +53,7 @@ class InquiryToCrm implements Consumer $crmCustomer->type = 1; $crmCustomer->channel_id = $info->channel; $crmCustomer->country_id = $info->country; - $crmCustomer->magnitude = $info->quality - 1; + $crmCustomer->magnitude = $info->quality; $crmCustomer->email = $info->email; $crmCustomer->phone = $info->tel; $crmCustomer->WhatsApp = $info->whatsapp; @@ -70,11 +71,9 @@ class InquiryToCrm implements Consumer if ($uinfo && $uinfo->email) { $options = [ 'id' => $info->id, - 'email' => $uinfo->email, - // 'category' => 'inquiry', - // 'scene' => $scene, - // 'receive_type' => $receiveType + 'email' => $uinfo->email ]; + // Log::info('发送邮件消息:' . json_encode($options)); \Webman\RedisQueue\Redis::send('email-message', $options); } }