where('status',1)->first(); //$mailboxModel['last_fetch_at'] = time(); //$mailboxModel->save(); $mailbox = $this->mailboxSplice($mailboxModel); // $INBOX= $mailboxModel['imap_host'].':'.$mailboxModel['imap_port'].$mailboxModel['imap_flags'].$mailboxModel['mailbox']; // $mailbox = '{'.$INBOX.'}'.'INBOX'; //$username = $mailbox['username']; //$password = $mailbox['password']; $this->pullUid($mailboxModel, $mailbox); } /** * 拼接INBOX * @param $mailboxModel * @return string */ function mailboxSplice($mailboxModel){ $INBOX= $mailboxModel['imap_host'].':'.$mailboxModel['imap_port'].$mailboxModel['imap_flags']; $mailbox = '{'.$INBOX.'}'.'INBOX'; return $mailbox; } /** * 拉取邮件uid */ function pullUid($mailboxModel, $mailbox) { try { $username = $mailboxModel['username']; $password = $mailboxModel['password']; $lastUid = $mailboxModel['last_uid']; $retryTimes = 3; $retryDelay = 10; $inbox = false; $attempt = 0; while ($attempt < $retryTimes) { $attempt++; flg("尝试第 {$attempt} 次连接 IMAP...", false,'mailbox'); $inbox = @imap_open($mailbox, $username, $password); if ($inbox) { break; } $error = imap_last_error(); flg("{$username}第 {$attempt} 次失败:{$error}", false,'mailbox'); // 若出现 too many login failures 立即暂停一段时间 if (stripos($error, 'Too many login failures') !== false) { flg("检测到登录失败次数过多,暂停 60 秒避免封禁", false,'mailbox'); sleep(60); } else { sleep($retryDelay); } } if (!$inbox) { flg("{$username}最终仍然连接失败:" . imap_last_error() . "", false,'mailbox'); return false; } // 拉取 UID 并入库 flg("开始拉取 UID 并入库 …", false,'mailbox'); $uids = imap_search($inbox, 'ALL', SE_UID) ?: []; $uids = array_filter($uids, function($uid) use ($lastUid) { return $uid > $lastUid; }); sort($uids); $now = time(); $count = 0; foreach ($uids as $uid) { try { Db::table('email_pull_uids')->insert([ 'uid' => $uid, 'fetched_at' => $now, 'email' => $username ]); $count++; flg("{$username} UID:".$uid." 已写入数据库", false,'mailbox'); } catch (\Throwable $e) { flg("{$username} {$e->getMessage()}", false,'mailbox'); } } imap_close($inbox); flg("✔ {$username}获取到新询盘邮件 {$count} 条 ...", false,'mailbox'); if(empty($uids)){ $mailboxModel->last_uid = max($uids); } $mailboxModel->last_fetch_at = time(); $mailboxModel->save(); return true; } catch (\Throwable $th) { flg(dpr($th), false,'mailbox'); return false; } } /** * 多邮箱处理:拉取详情 */ function pullDetails() { $emails = EmailPullUids::where('status', 0) ->select('email') ->distinct() ->get(); foreach ($emails as $email) { $mailboxModel = EmailMailbox::where('username', $email->email)->first(); $mailbox = $this->mailboxSplice($mailboxModel); $this->details($mailboxModel, $mailbox); } } /** * 获取邮件详情 */ public function details($mailboxModel, $mailbox) { try { $username = $mailboxModel['username']; $password = $mailboxModel['password']; $inbox = @imap_open($mailbox, $username, $password); if (!$inbox) { flg("{$username} IMAP 连接失败:". imap_last_error() ."", false,'mailbox'); return false; } // 依次拉取详情并写入 emails 表 flg("依次拉取详情并写入 email_pull 表 …", false,'mailbox'); while (true) { $pending = Db::table('email_pull_uids') ->where('email', $username) ->where('status', 0) ->orderBy('uid', 'desc') ->limit(2) ->get(); if (count($pending) === 0) { flg("邮件UID的详情,已拉取完毕。", false,'mailbox'); break; } flg("处理下一批 " . count($pending) . " 条 …", false,'mailbox'); foreach ($pending as $row) { $uid = $row->uid; // 包裹整行处理在 try/catch 中,任何异常都不会中断循环 try { flg("---- 开始处理 UID={$uid} ----", false,'mailbox'); // 1. fetch overview $ov = @imap_fetch_overview($inbox, $uid, FT_UID)[0] ?? null; if (!$ov) { $err = imap_last_error() ?: '未知错误'; flg("UID={$uid} overview 失败:{$err}", false,'mailbox'); 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,'mailbox'); // 2. fetch body $body = $this->getMailBody($inbox, $uid, true); flg("UID={$uid} body 长度:" . strlen($body), false,'mailbox'); // 3. sanitize & insert $body = $this->sanitizeUtf8($body); $now = time(); try { Db::table('email_pull')->insert([ 'email' => $username, 'uid' => $uid, 'subject' => $subject, 'from' => $from, 'date' => $date, 'body' => $body, 'created_at' => $now, 'updated_at' => $now, ]); flg($username." UID={$uid} 写入 emails 成功", false,'mailbox'); } catch (\Throwable $e) { flg($username." UID={$uid} 写入失败:{$e->getMessage()}", false,'mailbox'); continue; } // 4. 标记已处理 try { Db::table('email_pull_uids') ->where('uid', $uid) ->update([ 'status' => 1, 'processed_at' => $now, ]); flg($username." UID={$uid} 标记为已处理", false,'mailbox'); } catch (\Throwable $e) { flg($username." UID={$uid} 标记状态失败:{$e->getMessage()}", false,'mailbox'); // 即使更新 status 失败,也继续下一条 } } catch (\Throwable $e) { // 捕获任何非预期错误 flg($username." UID={$uid} 处理异常:{$e->getMessage()}", false,'mailbox'); continue; } } // 批次结束后稍作休眠,防止服务器过载 sleep(1); } imap_close($inbox); flg($username." ✔ 新邮件详情拉取完成 ...", false,'mailbox'); return true; } catch (\Throwable $th) { flg(dpr($th), false,'mailbox'); return false; } } /** * 将邮件按编码解码并保留 "\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); } /** * 将任意字符串清洗为有效 UTF-8,移除或替换无效字节 */ protected function sanitizeUtf8(string $text): string { return iconv('UTF-8', 'UTF-8//IGNORE', $text); } /** * 按编码解码 */ 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; } } /** * 多邮箱处理:清洗数据 */ function processMailbox() { $emails = EmailPull::where('status', 0) ->select('email') ->distinct() ->get(); foreach ($emails as $email) { $mailboxModel = EmailMailbox::where('username', $email->email)->first(); $this->process($mailboxModel); } } function process($mailboxModel) { switch ($mailboxModel['username']) { case 'paika@paikacar.com': $this->paika($mailboxModel); break; case 'hesu@hesucar.com': $this->hesu($mailboxModel); break; } } /** * 清洗询盘邮件 * @param $mailbox * @return void */ function paika($mailbox) { $batchSize = 100; // 每次处理 100 条 $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', ]; while (true) { $rows = EmailPull::where('email', $mailbox['username']) ->orderBy('id', 'asc') ->limit($batchSize) ->offset($page * $batchSize) ->get(); if ($rows->isEmpty()) { break; } foreach ($rows as $row) { // $parsed = $this->parseHtml($row->body); $parsed = $this->parsePaikaInquiry($row->body, $row->date); if ($parsed === null) { $row->status = 2; //其他邮件 $row->save(); flg("{$mailbox['username']} {$row->id} skip: not a contact-form HTML", false,'mailbox'); continue; } $row->status = 1; //已处理 $row->save(); // 严格按 $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)) { $emailId = array_column($toInsert, 'email_id'); $mailto = array_column($toInsert, 'mailto')[0]; $email = Db::table('email_pull_parsed')->whereIn('email_id', $emailId) // ->where('mailto', $mailto) ->first(); if(!empty($email)){ continue; } Db::table('email_pull_parsed')->insert($toInsert); } $page++; } } /** * 解析https://paikacar.com/wp-admin询盘邮件 * DOM 解析为数组 */ public function parsePaikaInquiry(string $html, $date) { libxml_use_internal_errors(true); $dom = new DOMDocument(); //$dom->loadHTML($html); $dom->loadHTML('' . $html); libxml_clear_errors(); $xpath = new DOMXPath($dom); if ( !str_contains($html, 'Product Inquiry') || !str_contains($html, 'Product Name') || !str_contains($html, 'Your Name') ) { // 认为是询盘 return null; } $data = []; // 找到所有 tr $rows = $xpath->query('//table//tr'); foreach ($rows as $row) { $th = $xpath->query('.//th//p', $row)->item(0); $td = $xpath->query('.//td//p', $row)->item(0); if (!$th || !$td) { continue; } $key = trim(str_replace(':', '', $th->textContent)); $value = trim($td->textContent); // 特殊字段:Product Name 是链接 if ($key === 'Product Name') { $a = $xpath->query('.//a', $td)->item(0); if ($a) { $data['product_name'] = trim($a->textContent); $data['product_url'] = $a->getAttribute('href'); continue; } } $data[$key] = $value; } $data['date'] = $date; return $this->mapInquiryFields($data); } /** * paika.@paikacar.com字段处理 */ function mapInquiryFields(array $raw): array { $result = [ 'product_name' => $raw['product_name'] ?? '', 'form_url' => $raw['product_url'] ?? '', 'quantity' => $this->normalizeValue($raw['Quantity'] ?? ''), //数量 'name' => $raw['Your Name'] ?? '', 'user_type' => $raw['User type'] ?? '', 'email' => $raw['E-mail'] ?? '', 'tel' => $this->normalizePhone($raw['Phone'] ?? ''), 'whatsapp' => $raw['WhatsApp'] ?? '', // 'skype' => $raw['Skype'] ?? '', 'company' => $raw['Company'] ?? '', 'country' => $this->normalizeCountry($raw['Country'] ?? ''), 'ip' => $this->extractIp($raw['IP'] ?? ''), 'content' => $raw['Content'] ?? '', 'date' => date('Y-m-d H:i:s',strtotime($raw['date'])), ]; //只有ipv4分析ip所属地区 if (filter_var($result['ip'], FILTER_VALIDATE_IP, FILTER_FLAG_IPV4)) { $result['iplocation'] = empty($result['ip'])?ip2region($result['ip']).'(仅供参考)':''; } // $message = trim( // (!empty($result['product_name']) ? "意向车型:{$result['product_name']}" : '') . // (!empty($result['quantity']) ? (!empty($result['product_name']) ? ', ' : '') . "数量:{$result['quantity']}" : '') // ) ?: ''; $message[] = !empty($result['product_name']) ? "意向车型:{$result['product_name']}" : ''; $message[] = !empty($result['quantity']) ? "数量:{$result['quantity']}" : ''; $message[] = !empty($result['user_type']) ? "用户类型:{$result['user_type']}" : ''; $message[] = "

询盘来源:" . "paikacar.com

"; $message[] = "

FB 创建时间原始值:".$raw['date']."

"; $result['message'] = implode(',', $message); $result['ilang'] = 'en'; $result['mailto'] = 'paika.@paikacar.com'; unset($result['product_name']); unset($result['quantity']); return $result; } function normalizeValue(string $value): string { if (str_contains($value, '[your-')) { return ''; } return $value; } function normalizePhone(string $phone): string { return preg_replace('/[^\d+]/', '', $phone); } function extractIp(string $text): string { if (preg_match('/\b\d{1,3}(\.\d{1,3}){3}\b/', $text, $m)) { return $m[0]; } return ''; } function normalizeCountry(string $country): string { if (str_contains($country, 'Please select')) { return ''; } return $country; } /** * 清洗合塑询盘邮件 * @param $mailbox * @return void */ function hesu($mailbox) { $batchSize = 100; // 每次处理 100 条 $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', ]; while (true) { $rows = EmailPull::where('email', $mailbox['username']) ->where('status', 0) ->orderBy('id', 'asc') ->limit($batchSize) ->offset($page * $batchSize) ->get(); if ($rows->isEmpty()) { break; } $toInsert = []; foreach ($rows as $row) { // $parsed = $this->parseHtml($row->body); $parsed = $this->parseHesucarInquiry($row->body, $row->date); if ($parsed === null) { $row->status = 2; //其他邮件 $row->save(); flg($mailbox['username']." {$mailbox['username']} {$row->id} skip: not a contact-form HTML", false,'mailbox'); continue; } $row->status = 1; //已处理 $row->save(); // 严格按 $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)) { $emailId = array_column($toInsert, 'email_id'); $mailto = array_column($toInsert, 'mailto')[0]; $email = Db::table('email_pull_parsed')->whereIn('email_id', $emailId) //->where('mailto', $mailto) ->first(); if(!empty($email)){ continue; } Db::table('email_pull_parsed')->insert($toInsert); } $page++; } } /** * 解析https://hesucar.com询盘邮件 * DOM 解析为数组 */ public function parseHesucarInquiry(string $html, $date) { libxml_use_internal_errors(true); $dom = new DOMDocument(); //$dom->loadHTML($html); $dom->loadHTML('' . $html); libxml_clear_errors(); $xpath = new DOMXPath($dom); if ( !str_contains($html, 'HESUCAR') || !str_contains($html, 'Tel/Whatsapp') || !str_contains($html, 'Sent from') ) { // 认为是询盘 return null; } $data = [ 'product_url' => '', 'name' => '', 'phone' => '', 'message' => '', 'source' => 'hesucar_inquiry', 'date' => $date, ]; /** * 1️⃣ 产品链接(正文最前面的 a) */ $productLink = $xpath->query( '//td[contains(@class,"mcnTextContent")]//a[starts-with(@href,"http")]' )->item(0); if ($productLink) { $data['product_url'] = trim($productLink->getAttribute('href')); } /** * 2️⃣ Name / Tel / Message * 结构特点: * Label * 下一行 td 就是 value */ $labels = $xpath->query('//strong'); foreach ($labels as $labelNode) { $label = trim($labelNode->textContent); // 当前 strong 所在 td $td = $labelNode->parentNode; // 下一个 tr 的 td 才是值 $valueTd = $td->parentNode->nextSibling; while ($valueTd && $valueTd->nodeName !== 'tr') { $valueTd = $valueTd->nextSibling; } if (!$valueTd) { continue; } $value = trim($valueTd->textContent); switch ($label) { case 'Name': $data['name'] = $value; break; case 'Tel/Whatsapp': $data['phone'] = $this->normalizePhone($value); break; case 'Message': $data['message'] = $this->cleanMessage($value); break; } } // return $data; return $this->hesucarInquiryFields($data); } /** * hesucar.@hesucar.com字段处理 */ function hesucarInquiryFields(array $raw): array { $result = [ 'form_url' => $raw['product_url'] ?? '', 'name' => $raw['name'] ?? '', 'tel' => $this->normalizePhone($raw['phone'] ?? ''), 'whatsapp' => $raw['phone'] ?? '', 'phone' => $raw['phone'] ?? '', 'country' => '未知', 'message' => $raw['message'] ?? '', 'date' => date('Y-m-d H:i:s',strtotime($raw['date'])), ]; $message = []; $message[] = "

询盘来源:" . $raw['product_url']."

"; $message[] = "

Tel/Whatsapp:" . $raw['phone']."

"; $message[] = "

FB 创建时间原始值:".$raw['date']."

"; $result['message'] = $result['message'].implode(',', $message); $result['ilang'] = 'en'; $result['mailto'] = 'hesucar.@hesucar.com'; return $result; } function cleanMessage(string $message): string { return trim(preg_replace("/\s+/", " ", $message)); } /** * 多邮箱处理:分配客户 */ function fenpaiMailbox() { $emails = EmailPullParsed::where('status', 0) ->select('email') ->distinct() ->get(); foreach ($emails as $email) { $mailboxModel = EmailMailbox::where('username', $email->mailto)->first(); $this->fenpai($mailboxModel); } } /** * 分配清洗好的数据 * @param $mailbox * @return false|void */ public function fenpai($mailbox) { try { // 配置参数 $batchSize = 100; // 每批处理量 $delaySeconds = 2; // 每封邮件间隔秒数 $throttleAfter = 50; // 每发送多少封后增加休息 $throttleSeconds = 5; // 批量休息秒数 // 固定录单人 $enterUserId = 186; $enterUser = User::where('id', $enterUserId)->first(); // 固定渠道 //$channel = 12; $channel = $mailbox['channel_id']; $channelInfo = InquiryChannel::where('id', $channel)->first(); // 从库实时获取 $corp_id = $mailbox['corp_id']; // $users = \app\common\model\User::where('corp_id', $corp_id)->where('status', 1)->orderBy('id')->pluck('id')->toArray(); if($corp_id == 1){ //合塑 $users = \app\common\model\UserGroupMembers::where('group_id', 10) ->orderBy('user_id')->pluck('user_id')->toArray(); }else if($corp_id == 3){ $users = \app\common\model\UserGroupMembers::where('group_id', 9) ->orderBy('user_id')->pluck('user_id')->toArray(); } // 移除不参与分配人员 $users = array_values(array_diff($users, [186])); // 初始化 UID 轮询器 $uidRotator = new \app\common\utils\RedisUidRotator('email_parsed_result_20250804'); $uidRotator->setUids($users, false); EmailPullParsed::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", false,'mailbox'); $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('同步失败~'); } // 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,'mailbox'); // 每封邮件后延迟 if ($delaySeconds > 0) { sleep($delaySeconds); } // 每发送一定数量后额外休息 if (++$count % $throttleAfter === 0) { sleep($throttleSeconds); flg("已发送 {$count} 封邮件,休息 {$throttleSeconds} 秒", false,'mailbox'); } } catch (\Exception $e) { flg("邮件发送失败 ID:{$record->id} - " . $e->getMessage(), false,'mailbox'); // 更新状态 $record->update(['status' => 3]); // 失败后也加入延迟,避免快速连续失败 sleep($delaySeconds); } } }); } catch (\Throwable $th) { flg(dpr($th), false,'mailbox'); return false; } } }