561 lines
22 KiB
PHP
561 lines
22 KiB
PHP
<?php
|
||
|
||
namespace app\common\utils\email;
|
||
|
||
use app\common\model\Country;
|
||
use app\common\model\CrmCustomer;
|
||
use support\Db;
|
||
use support\Log;
|
||
use support\exception\BusinessException;
|
||
use app\common\model\EmailParsedResult;
|
||
use app\common\model\InquiryChannel;
|
||
use app\common\model\InquiryLog;
|
||
use app\common\model\User;
|
||
|
||
class Emails
|
||
{
|
||
|
||
private $mailbox = '{imap.feishu.cn:993/imap/ssl}INBOX';
|
||
private $username = 'info@borderlesscar.com';
|
||
private $password = 'ErkhUA629aocm7Bc';
|
||
private $limit = 5;
|
||
private $per_page = 20;
|
||
private $batch_size = 10;
|
||
|
||
|
||
public function __construct($options = [])
|
||
{
|
||
if ($options) {
|
||
$this->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，找不到 <h2>Preview</h2> 则返回 null，
|
||
* 否则提取所有 <span>Key:</span> Value 为小写蛇形 key=>value 数组
|
||
*/
|
||
protected function parseHtml(string $html): ?array
|
||
{
|
||
$doc = new \DOMDocument();
|
||
libxml_use_internal_errors(true);
|
||
$doc->loadHTML('<?xml encoding="utf-8"?>' . $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("<comment>第 {$attempt} 次失败：{$error}</comment>");
|
||
|
||
// 若出现 too many login failures 立即暂停一段时间
|
||
if (stripos($error, 'Too many login failures') !== false) {
|
||
flg("<error>检测到登录失败次数过多，暂停 60 秒避免封禁</error>");
|
||
sleep(60);
|
||
} else {
|
||
sleep($retryDelay);
|
||
}
|
||
}
|
||
|
||
if (!$inbox) {
|
||
flg("<error>最终仍然连接失败：" . imap_last_error() . "</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("<comment>{$e->getMessage()}</comment>");
|
||
}
|
||
}
|
||
|
||
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("<error>IMAP 连接失败：". imap_last_error() ."</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();
|
||
|
||
// 从库实时获取
|
||
$users = \app\common\model\User::where('corp_id', 3)->where('status', 1)->orderBy('id')->pluck('id')->toArray();
|
||
|
||
// 初始化 UID 轮询器
|
||
$uidRotator = new \app\common\utils\RedisUidRotator('email_parsed_result_20250804');
|
||
$uidRotator->setUids($users, 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;
|
||
}
|
||
}
|
||
} |