mailbox = $options['mailbox']; $this->username = $options['username']; $this->password = $options['password']; $this->limit = $options['limit']; $this->per_page = $options['per_page']; $this->batch_size = $options['batch_size']; } } /** * 将任意字符串清洗为有效 UTF-8,移除或替换无效字节 */ protected function sanitizeUtf8(string $text): string { return iconv('UTF-8', 'UTF-8//IGNORE', $text); } /** * 记录错误到日志 */ protected function logError(string $msg): void { $line = '[' . date('Y-m-d H:i:s') . '] ' . $msg . "\n"; file_put_contents( runtime_path('logs/email_errors.log'), $line, FILE_APPEND | LOCK_EX ); } /** * 按编码解码并保留 "\r\n" */ protected function getMailBody($inbox, int $id, bool $isUid = false): string { $flags = $isUid ? FT_UID : 0; $structure = imap_fetchstructure($inbox, $id, $flags); if (!empty($structure->parts)) { // HTML 部分 foreach ($structure->parts as $i => $part) { if (isset($part->subtype) && strtolower($part->subtype) === 'html') { $data = imap_fetchbody($inbox, $id, (string)($i+1), $flags); return $this->decodeBody($data, $part->encoding); } } // 文本部分 foreach ($structure->parts as $i => $part) { if (isset($part->subtype) && strtolower($part->subtype) === 'plain') { $data = imap_fetchbody($inbox, $id, (string)($i+1), $flags); return $this->decodeBody($data, $part->encoding); } } // 原始 body return imap_body($inbox, $id, $flags); } return imap_body($inbox, $id, $flags); } /** * 按编码解码 */ protected function decodeBody(string $body, int $encoding): string { switch ($encoding) { case 3: return imap_base64($body); case 4: return imap_qprint($body); default: return $body; } } /** * 解析 HTML,找不到

Preview

则返回 null, * 否则提取所有 Key: Value 为小写蛇形 key=>value 数组 */ protected function parseHtml(string $html): ?array { $doc = new \DOMDocument(); libxml_use_internal_errors(true); $doc->loadHTML('' . $html); libxml_clear_errors(); $xpath = new \DOMXPath($doc); // 类型检查 if ($xpath->query("//h2[text()='Preview']")->length === 0) { return null; } $nodes = $xpath->query("//div[span]"); $data = []; /** @var \DOMElement $node */ foreach ($nodes as $node) { $span = $xpath->query("span", $node)->item(0); if (! $span) { continue; } // 原始键名 $key = trim(str_replace(':', '', $span->textContent)); // 转成小写蛇形 $col = strtolower(str_replace(' ', '_', $key)); // 获取值 $a = $xpath->query(".//a", $node)->item(0); if ($a && $a->hasAttribute('href')) { $value = trim($a->getAttribute('href')); } else { $clone = $node->cloneNode(true); $spanClone = $xpath->query(".//span", $clone)->item(0); if ($spanClone) { $spanClone->parentNode->removeChild($spanClone); } $value = trim($clone->textContent); } $data[$col] = $value; } return $data; } /** * 拉取邮件uid */ public function uids() { try { $retryTimes = 3; $retryDelay = 10; $inbox = false; $attempt = 0; $maxUid = Db::table('email_uids')->max('uid'); while ($attempt < $retryTimes) { $attempt++; flg("尝试第 {$attempt} 次连接 IMAP...", false); $inbox = @imap_open($this->mailbox, $this->username, $this->password); if ($inbox) { break; } $error = imap_last_error(); flg("第 {$attempt} 次失败:{$error}"); // 若出现 too many login failures 立即暂停一段时间 if (stripos($error, 'Too many login failures') !== false) { flg("检测到登录失败次数过多,暂停 60 秒避免封禁"); sleep(60); } else { sleep($retryDelay); } } if (!$inbox) { flg("最终仍然连接失败:" . imap_last_error() . ""); return false; } // 拉取 UID 并入库 flg("开始拉取 UID 并入库 …", false); $uids = imap_search($inbox, 'ALL', SE_UID) ?: []; $uids = array_filter($uids, function($uid) use ($maxUid) { return $uid > $maxUid; }); sort($uids); $now = time(); $count = 0; foreach ($uids as $uid) { try { Db::table('email_uids')->insert([ 'uid' => $uid, 'fetched_at' => $now, 'email' => $this->username ]); $count++; flg("UID:".$uid." 已写入数据库", false); } catch (\Throwable $e) { flg("{$e->getMessage()}"); } } imap_close($inbox); flg("✔ 获取到新询盘邮件 {$count} 条 ...", false); return true; } catch (\Throwable $th) { flg(dpr($th)); return false; } } /** * 获取邮件详情 */ public function details() { try { $inbox = @imap_open($this->mailbox, $this->username, $this->password); if (!$inbox) { flg("IMAP 连接失败:". imap_last_error() .""); return false; } // 依次拉取详情并写入 emails 表 flg("依次拉取详情并写入 emails 表 …", false); while (true) { $pending = Db::table('email_uids') ->where('status', 0) ->orderBy('uid', 'desc') ->limit(2) ->get(); if (count($pending) === 0) { flg("邮件UID的详情,已拉取完毕。", false); break; } flg("处理下一批 " . count($pending) . " 条 …", false); foreach ($pending as $row) { $uid = $row->uid; // 包裹整行处理在 try/catch 中,任何异常都不会中断循环 try { flg("---- 开始处理 UID={$uid} ----", false); // 1. fetch overview $ov = @imap_fetch_overview($inbox, $uid, FT_UID)[0] ?? null; if (!$ov) { $err = imap_last_error() ?: '未知错误'; flg("UID={$uid} overview 失败:{$err}"); continue; } $subject = mb_convert_encoding($ov->subject ?? '', 'UTF-8', 'GBK,UTF-8'); $from = $ov->from ?? ''; $date = $ov->date ?? ''; flg("UID={$uid} overview 成功,subject={$subject}", false); // 2. fetch body $body = $this->getMailBody($inbox, $uid, true); flg("UID={$uid} body 长度:" . strlen($body), false); // 3. sanitize & insert $body = $this->sanitizeUtf8($body); $now = time(); try { Db::table('emails')->insert([ 'uid' => $uid, 'subject' => $subject, 'from' => $from, 'date' => $date, 'body' => $body, 'created_at' => $now, 'updated_at' => $now, ]); flg("UID={$uid} 写入 emails 成功", false); } catch (\Throwable $e) { flg("UID={$uid} 写入失败:{$e->getMessage()}"); continue; } // 4. 标记已处理 try { Db::table('email_uids') ->where('uid', $uid) ->update([ 'status' => 1, 'processed_at' => $now, ]); flg("UID={$uid} 标记为已处理", false); } catch (\Throwable $e) { flg("UID={$uid} 标记状态失败:{$e->getMessage()}", false); // 即使更新 status 失败,也继续下一条 } } catch (\Throwable $e) { // 捕获任何非预期错误 flg("UID={$uid} 处理异常:{$e->getMessage()}"); continue; } } // 批次结束后稍作休眠,防止服务器过载 sleep(1); } imap_close($inbox); flg("✔ 新邮件详情拉取完成 ...", false); return true; } catch (\Throwable $th) { flg(dpr($th)); return false; } } /** * 邮件清洗处理 */ public function process() { try { $page = 0; // 明确定义 parsed_contacts 表里要插入的所有列(顺序无所谓,但要全部包含) $fields = [ 'email_id', 'date', 'email', 'name', 'tel', 'company', 'country', 'form_url', 'message', 'ip', 'iplocation', 'ilang', 'mailto', 'navigator_language', 'navigator_platform', 'screen_width', 'screen_height', 'ucountry', ]; $maxEmailId = Db::table('email_parsed_result')->max('email_id'); while (true) { $rows = Db::table('emails') ->where('status', 0) ->where('id', '>', $maxEmailId) ->orderBy('id', 'asc') ->limit($this->batch_size) ->offset($page * $this->batch_size) ->get(); if ($rows->isEmpty()) { break; } $toInsert = []; foreach ($rows as $row) { $parsed = $this->parseHtml($row->body); if ($parsed === null) { flg(" ↳ [{$row->id}] skip: not a contact-form HTML", false); continue; } // 严格按 $fields 顺序填充每条记录,缺失的字段补 null $record = []; foreach ($fields as $col) { if ($col === 'email_id') { $record[$col] = $row->id; } else { $record[$col] = $parsed[$col] ?? null; } } $toInsert[] = $record; } if (!empty($toInsert)) { Db::table('email_parsed_result')->insert($toInsert); $emailId = array_column($toInsert, 'email_id'); Db::table('emails')->whereIn('id', $emailId)->update(['status' => 1]); flg("Inserted batch #" . ($page + 1) . ", count: " . count($toInsert), false); } $page++; } flg("✔ 邮件清洗完毕,等待分配 ...", false); return true; } catch (\Throwable $th) { flg(dpr($th)); return false; } } public function fenpai() { try { // 配置参数 $batchSize = 100; // 每批处理量 $delaySeconds = 2; // 每封邮件间隔秒数 $throttleAfter = 50; // 每发送多少封后增加休息 $throttleSeconds = 5; // 批量休息秒数 // 固定录单人 $enterUserId = 75; $enterUser = User::where('id', $enterUserId)->first(); // 固定渠道 $channel = 12; $channelInfo = InquiryChannel::where('id', $channel)->first(); // 初始化 UID 轮询器 $uidRotator = new \app\common\utils\RedisUidRotator('email_parsed_result_20250804'); $uidRotator->setUids([64, 63, 93, 108, 132, 145, 150, 143, 156, 157, 158], false); 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)->where('status', 1)->first(); if (!$recipientInfo) { flg("用户ID=" . $assignedUid . " 已离职 跳过自动分配,需要手动移除 UID"); $assignedUid = $uidRotator->next(); $recipientInfo = User::where('id', $assignedUid)->where('status', 1)->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_id = $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)); // 9.2 修改消息发送 $data = [ 'id' => $inquiryLogResult->id ]; \Webman\RedisQueue\Redis::send('inquiry-to-crm-new', $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; } } }