周海龙 / mail-serve · 提交

转到一个项目

作者 邓超
提交 81ec2f01e5b1b89181ee4e9676906946949c2ad4 1 个父辈 c168941b

调优

内嵌 并排对比
正在显示 1 个修改的文件 包含 0 行增加 和 227 行删除
<?php
//error_reporting();
require_once "../vendor/autoload.php";
if(in_array('v2',$argv)){
define('S_V2','2');
}else{
define('S_V2','');
}
/**
* 把mysql的数据同步到es
* @author：dc
* @time 2025/3/4 10:19
* Class SyncToEsCmd
*/
class SyncToEsCmd {
public $isStop = false;
/**
* 文件夹
* @var array
*/
public $folders = [];
/**
* @var \Lib\Db
*/
public $fob_db;
public function handler(){
$handler = function ($signal){
_echo('收到进程信号 '. $signal);
// 可以处理其他程序
$this->isStop = true;
};
pcntl_signal(SIGTERM, $handler); // 这个是kill
pcntl_signal(SIGINT, $handler); // 这个是 ctrl+c
// pcntl_signal(SIGHUP, $handler);
if(S_V2){
$es = es('email_lists_copy');
}else{
$es = es();
}
$db = db();
$this->fob_db = fob_mysql();
$startTime = time();
// 查询gmail的所有邮件文件夹 ai邮件不同步这个文件夹
$fids = $db->throw()->all("select `id` from `folders` where `origin_folder` in ('[Gmail]/&UWiQ6JD1TvY-','[Gmail]/All Mail','[Gmail]/&YkBnCZCuTvY-')");
$fids = array_column($fids,'id','id');
while (1){
if(time()-43200 > $startTime){
break;
}
// 检查是否接收到信号
pcntl_signal_dispatch();
if($this->isStop) {
_echo('已退出进程');
break;
}
$id = redis()->rPop('sync_to_es'.S_V2);
$code = 500;
if($id){
$doc_id = '';
try {
$data = $db->throw()->first('select * from `lists_hot` where `id` = '.$id);
if(!$data){
$data = $db->throw()->first('select * from `lists` where `id` = '.$id);
}
}catch (Throwable $e){
redis()->lPush('sync_to_es'.S_V2,$id);
_echo('sync to es '.$e->getMessage());
break;
}
if($data){
// 是否是所有邮件文件夹 是就跳过
if($fids[$data['folder_id']]??0){
continue;
}
// 设置 进程 是否在运行
$data['is_auto'] = $db->count('select count(*) from `lists_auto` where `list_id` = '.$data['id']) ? 1 : 0;
try {
// 文件夹
if(empty($this->folders[$data['folder_id']])){
$this->folders[$data['folder_id']] = $db->throw()->value(\Model\folderSql::first($data['folder_id'],'folder'));
}
// 为文件夹打标 方便查询
$data['folder_as_int'] = folder2int($this->folders[$data['folder_id']]);
// 再次验证是否是自动回复
if($data['folder_as_int'] === 1 && !$data['is_auto']){
$filter = redis()->get('ai_email_filter_lists',[]);
$filter = is_array($filter) ? $filter : [];
foreach ($filter as $f){
list($t,$str) = $f;
$haystack = '';
if($t==2){
$haystack = $data['subject'];
}elseif ($t==1){
$haystack = $data['from'];
}
if($haystack && $str && stripos($haystack,$str)!==false){
$data['is_auto'] = 1;
break;
}
}
}
// 再次验证是否是预热邮件
// if()
// postid ai邮箱要用 这个是查询黑格
list($data['postid'],$data['source']) = $this->getPostid($data['email_id']);
}catch (Throwable $e){
redis()->lPush('sync_to_es'.S_V2,$id);
_echo('sync to es '.$e->getMessage());
break;
}
$data = $this->getEsData($data);
$doc_id = $data['email_id'].'_'.$data['folder_id'].'_'.$data['uid'];
$code = $es->save($doc_id,$data);
}
if($code!==200){
@file_put_contents(LOG_PATH.'/sync_es_fail'.S_V2.'.log',$id."\n",FILE_APPEND);
_echo('同步es: '.$doc_id.'===>'.$code);
}
// _echo('同步es: '.$doc_id.'===>'.$code);
// print_r($data);
// var_dump($id);
// exit;
}else{
sleep(1);
}
}
}
/**
* 查询项目id 和 邮件来源
* @param int $email_id 邮箱id
* @return array
* @author：dc
* @time 2025/5/29 11:47
*/
public function getPostid($email_id){
$data = redis()->getSet('fob_bind_mailx:'.$email_id,300,function ($email_id){
return $this->fob_db->throw()->first("select `post_id`,`source` from `e_mail_binds` where `email_id` = '{$email_id}' order by `id` desc limit 1");
},$email_id);
return [
$data['post_id']??0,
$data['source']??0,
];
}
public function getEsData($data){
if(!empty($data['id'])){
$data['uuid'] = $data['id'];
unset($data['id']);
}
$data['from'] = [
'email' => $data['from'],
'name' => $data['from_name']??''
];
unset($data['from_name']);
unset($data['date']);
if(!empty($data['to_name']) && strlen($data['to_name'])>10000){
$data['to_name'] = @json_decode($data['to_name'],true);
if(is_array($data['to_name'])){
$data['to_name'] = array_map(function ($v){
$v['name'] = '';
return $v;
},$data['to_name']);
$data['to_name'] = array_slice($data['to_name'],0,100);
$data['to_name'] = json_encode($data['to_name']);
}else{
$data['to_name'] = '';
}
}
$data['description'] = str_replace(["\n"],"",$data['description']);
// unset($data['to_name']);
if(!empty($data['created_at'])){
$data['created_at'] = date('Y-m-d\TH:i:s',strtotime($data['created_at']));
}
if(!empty($data['updated_at'])){
$data['updated_at'] = date('Y-m-d\TH:i:s',strtotime($data['updated_at']));
}
$data['references'] = empty($data['references']) ? '' : $data['references'];
return $data;
}
}
(new SyncToEsCmd())->handler();
return 1;