server submit

@@ -89,6 +89,7 @@ class SyncToEsCmd {
                         }
                         }
                         // 再次验证是否是自动回复
                         // 再次验证是否是自动回复
                         if($this->folders[$data['folder_id']] == '收件箱' && !$data['is_auto']){
                         if($this->folders[$data['folder_id']] == '收件箱' && !$data['is_auto']){
+                            echo "xx";
                             $filter = redis()->get('ai_email_filter_lists',[]);
                             $filter = redis()->get('ai_email_filter_lists',[]);
                             $filter = is_array($filter) ? $filter : [];
                             $filter = is_array($filter) ? $filter : [];
                             foreach ($filter as $f){
                             foreach ($filter as $f){
@@ -99,7 +100,8 @@ class SyncToEsCmd {
                                 }elseif ($t==1){
                                 }elseif ($t==1){
                                     $haystack = $data['from'];
                                     $haystack = $data['from'];
                                 }
                                 }
-                                if(stripos($haystack,$str)!==false){
+                                if($haystack && $str && stripos($haystack,$str)!==false){
+                                    echo 'auto';
                                     $data['is_auto'] = 1;
                                     $data['is_auto'] = 1;
                                     break;
                                     break;
                                 }
                                 }

+<?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');
+        }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()->lPop('sync_to_es'.S_V2);
+            $code = 500;
+            if($id){
+                $doc_id = '';
+                try {
+                    $data = $db->throw()->first(\Model\listsSql::first('`id` = '.$id));
+                }catch (Throwable $e){
+                    redis()->rPush('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']]);
+                        // postid ai邮箱要用 这个是查询黑格
+                        list($data['postid'],$data['source']) = $this->getPostid($data['email_id']);
+                    }catch (Throwable $e){
+                        redis()->rPush('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);
+                }
+
+            }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_mail:'.$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}' and `deleted_at` is null 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;
+
+
+
+
+
+
+
+
+