es_producer/app/JsonRpc/EsService.php

<?php
namespace App\JsonRpc;

use App\Tools\PublicData;
use App\Tools\Result;
use Hyperf\DbConnection\Db;
use Hyperf\RpcServer\Annotation\RpcService;
use Psr\Log\LoggerInterface;
use Psr\Http\Message\ServerRequestInterface;
use Hyperf\Context\Context;
use App\Service\ElasticsearchService;
use Hyperf\Elasticsearch\ClientBuilderFactory;

#[RpcService(name: "EsService", protocol: "jsonrpc-http", server: "jsonrpc-http")]
class EsService implements EsServiceInterface
{
    private LoggerInterface $logger;
    private ElasticsearchService $elasticsearchService; // 添加 ElasticsearchService 属性
    private ClientBuilderFactory $clientBuilderFactory;
    private $client; // 添加 Elasticsearch 客户端属性

    public function __construct(LoggerInterface $logger, ElasticsearchService $elasticsearchService , ClientBuilderFactory $clientBuilderFactory)
    {
        $this->logger = $logger;
        $this->elasticsearchService = $elasticsearchService; // 注入 ElasticsearchService 实例
        $this->clientBuilderFactory = $clientBuilderFactory;
        // 创建 Elasticsearch 客户端实例
        $builder = $this->clientBuilderFactory->create();
        $this->client = $builder->setHosts(['http://127.0.0.1:9200'])->build();

    }

    public function createIndex(array $data): array
    {// 获取当前请求的 ServerRequestInterface
        $request = Context::get(ServerRequestInterface::class);
        var_dump($request->getUri());
          // 记录请求数据
        $this->logger->info("Received request to createIndex with data: " . json_encode($data));

        //  如果在协程环境下创建，则会自动使用协程版的 Handler，非协程环境下无改变
        $builder = $this->clientBuilderFactory->create();
        $client = $builder->setHosts(['http://127.0.0.1:9200'])->build();
        $info = $client->info();
        $this->logger->info("es info: " . json_encode($info));
        // return Result::success(  $info);
        $params = [
            'index' => 'articles',
            'body' => [
                                'mappings' => [
                                    'properties' => [
                                        'id' => ['type' => 'long'],
                                        'catid' => ['type' => 'integer'],
                                        'level' => ['type' => 'integer'],
                                        'title' => ['type' => 'text'],
                                        'introduce' => ['type' => 'text'],
                                        'tag' => ['type' => 'keyword'],
                                        'keyword' => ['type' => 'text'],
                                        'author' => ['type' => 'keyword'],
                                        'copyfrom' => ['type' => 'text'],
                                        'fromurl' => ['type' => 'text'],
                                        'hits' => ['type' => 'integer'],
                                        'ip' => ['type' => 'keyword'],
                                        'status' => ['type' => 'integer'],
                                        'islink' => ['type' => 'integer'],
                                        'linkurl' => ['type' => 'text'],
                                        'imgurl' => ['type' => 'text'],
                                        'admin_user_id' => ['type' => 'integer'],
                                        'cat_arr_id' => ['type' => 'text'],
                                        'created_at' => ['type' => 'date', 'format' => 'yyyy-MM-dd HH:mm:ss||strict_date_optional_time||epoch_millis'],
                                        'updated_at' => ['type' => 'date', 'format' => 'yyyy-MM-dd HH:mm:ss||strict_date_optional_time||epoch_millis'],
                                        'is_original' => ['type' => 'integer'],
                                        'content' => ['type' => 'text'],
                                    ],
                                ],
                            ],
                        ];

        $response = $client->indices()->create($params);
        return Result::success(  $response);

        // try {
        //     // 调用 ElasticsearchService 的 createIndex 方法
        //     $response = $this->elasticsearchService->createIndex();
        //     $this->logger->info("Index created successfully: " . json_encode($response));
        //     return Result::success('创建成功', $response);
        // } catch (\Exception $e) {
        //     $this->logger->error("Failed to create index: " . $e->getMessage());
        //     return Result::error('创建失败', $e->getMessage());
        // }
    }

    public function associateData(array $data): array
    {
         
            // 使用注入的 ClientBuilderFactory 创建客户端
            $builder = $this->clientBuilderFactory->create();
            $client = $builder->setHosts(['http://127.0.0.1:9200'])->build();

            $indexName = $data['index'] ?? 'articles';

            // 获取所有文章和文章数据
            $articles = Db::table('article')->limit(2)->orderBy('id', 'desc')->get(); 
            $articleData = Db::table('article_data')->limit(2)->orderBy('article_id', 'desc')->get(); 

            // 将 article_data 转换为数组，以便快速查找
            $articleDataMap = [];
            foreach ($articleData as $data) {
                $articleDataMap[$data->article_id] = $data;
            }
            $this->logger->info("articleDataMap: " . json_encode($articleDataMap));

            $successCount = 0;
            $errorCount = 0;
            $errors = [];

            foreach ($articles as $article) {
                $data = $articleDataMap[$article->id] ?? null;
                if ($data) {
                    $params = [
                        'index' => $indexName,
                        'id' => $article->id,
                        'body' => [
                            'id' => $article->id,
                            'catid' => $article->catid,
                            'level' => $article->level,
                            'title' => $article->title,
                            'introduce' => $article->introduce,
                            'tag' => $article->tag,
                            'keyword' => $article->keyword,
                            'author' => $article->author,
                            'copyfrom' => $article->copyfrom,
                            'fromurl' => $article->fromurl,
                            'hits' => $article->hits,
                            'ip' => $article->ip,
                            'status' => $article->status,
                            'islink' => $article->islink,
                            'linkurl' => $article->linkurl,
                            'imgurl' => $article->imgurl,
                            'admin_user_id' => $article->admin_user_id,
                            'cat_arr_id' => $article->cat_arr_id,
                            'created_at' => $article->created_at,
                            'updated_at' => $article->updated_at,
                            'is_original' => $article->is_original,
                            'content' => $data->content,
                        ],
                    ];

                    try {
                        $response = $client->index($params);
                        $successCount++;
                        // 记录成功日志（可选，根据需要）
                        // $this->logger->info("Data associated successfully: " . json_encode($response));
                    } catch (\Exception $e) {
                        $errorCount++;
                        $errors[] = "Failed to associate data for article ID {$article->id}: " . $e->getMessage();
                        $this->logger->error("Failed to associate data for article ID {$article->id}: " . $e->getMessage());
                    }
                } else {
                    $errorCount++;
                    $errors[] = "No data found for article ID {$article->id}";
                    $this->logger->error("No data found for article ID {$article->id}");
                }
            }

            if ($errorCount > 0) {
                return Result::error("部分数据关联失败", [
                    'success_count' => $successCount,
                    'error_count' => $errorCount,
                    'errors' => $errors,
                ]);
            }

            return Result::success(  [
                'success_count' => $successCount,
            ]);
       
    }

    public function searchIndex(array $data): array
    {
        
        // 使用注入的 ClientBuilderFactory 创建客户端
        $builder = $this->clientBuilderFactory->create();
        $client = $builder->setHosts(['http://127.0.0.1:9200'])->build();

        $indexName = $data['index'] ?? 'articles';
        $query = $data['query'] ?? [];
        $from = $data['from'] ?? 0; // 从哪个文档开始，默认为 0
        $size = $data['size'] ?? 10; // 返回的文档数量，默认为 10

        $params = [
            'index' => $indexName,
            'body' => [
                'query' => [
                    'multi_match' => [
                        'query' => $query,
                        'fields' => ['*'] // 匹配所有字段
                    ]
                    ],
                    'from' => $from,
                    'size' => $size
            ],
        ];

        try {
            $response = $client->search($params);
            $this->logger->info("Search successful: " . json_encode($response));
            return Result::success(  $response);
        } catch (\Exception $e) {
            $this->logger->error("Failed to search index: " . $e->getMessage());
            return Result::error("查询索引失败", $e->getMessage());
        }
    }


    public function deleteIndex(array $data): array
    {
        $indexName = $data['index'] ?? 'articles';
        $builder = $this->clientBuilderFactory->create();
        $client = $builder->setHosts(['http://127.0.0.1:9200'])->build();
        try {
            $response = $client->indices()->delete(['index' => $indexName]);
            $this->logger->info("Index {$indexName} deleted successfully: " . json_encode($response));
            return Result::success(  $response);
        } catch (\Exception $e) {
            $this->logger->error("Failed to delete index {$indexName}: " . $e->getMessage());
            throw new \RuntimeException("删除索引失败: " . $e->getMessage(), 0, $e);
            return Result::error("删除索引失败".$e->getMessage());
        }
    }

    public function indexExists($index)
    {
        $indexName = $data['index'] ?? 'articles';
        $builder = $this->clientBuilderFactory->create();
        $client = $builder->setHosts(['http://127.0.0.1:9200'])->build();
        try {
            return $client->indices()->exists(['index' => $index]);
        } catch (\Exception $e) {
            $this->logger->error("Failed to check if index {$index} exists: " . $e->getMessage());
            throw new \RuntimeException("检查索引是否存在失败: " . $e->getMessage(), 0, $e);
        }
    }
    public function updateIndex(array $data): array
    { 
       
        $newMapping = $data['mapping'] ?? [];
    
        $builder = $this->clientBuilderFactory->create();
        $client = $builder->setHosts(['http://127.0.0.1:9200'])->build();
        $indexName = $data['index'] ?? 'articles';
        if (empty($newMapping)) {
            return Result::error("Mapping data is required", "No mapping data provided");
        }
    
        try {
            $response = $client->indices()->putMapping([
                'index' => $indexName,
                'body' => [
                    'properties' => $newMapping
                ]
            ]);
            $this->logger->info("Successfully updated index {$indexName}: " . json_encode($response));
            return Result::success($response);
        } catch (\Exception $e) {
            $this->logger->error("Failed to update index {$indexName}: " . $e->getMessage());
            throw new \RuntimeException("更新索引失败: " . $e->getMessage(), 0, $e);
        }
    }

    public function renameIndex(array $data): array
        {
            $oldIndexName = $data['old_index'] ?? 'articles_v2';
            $newIndexName = $data['new_index'] ?? 'articles';

            $builder = $this->clientBuilderFactory->create();
            $client = $builder->setHosts(['http://127.0.0.1:9200'])->build();
            $indexName = $data['index'] ?? 'articles';

            $params = [
                'body' => [
                    'actions' => [
                        [
                            'rename' => [
                                'old_index' => $oldIndexName,
                                'new_index' => $newIndexName
                            ]
                        ]
                    ]
                ]
            ];

            try {
                $response = $client->indices()->updateAliases($params);
                $this->logger->info("Index renamed successfully: " . json_encode($response));
                return Result::success( $response);
            } catch (\Exception $e) {
                $this->logger->error("Failed to rename index: " . $e->getMessage());
                return Result::error( $e->getMessage());
            }
        }
        public function deleteData(array $data): array
        {
            return Result::success( '成功');
        }
        public function updateData(array $data): array
        {
            return Result::success( '成功');
        }
}