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, 'This e-mail was sent from X a contact form')
) {
// 认为是询盘
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 = 75;
$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;
}
}
}