采集

app/Cache/Contracts/HashRedisInterface.php

@@ -1,20 +0,0 @@
-<?php
-
-namespace App\Cache\Contracts;
-
-interface SetRedisInterface
-{
-    public function count();
-
-    public function add(string ...$member);
-
-    public function rem(string ...$member);
-
-    public function isMember(string $member);
-
-    public function randMember($count = 1);
-
-    public function all();
-
-    public function delete();
-}

app/Cache/Contracts/StreamRedisInterface.php

@@ -1,20 +0,0 @@
-<?php
-
-namespace App\Cache\Contracts;
-
-interface SetRedisInterface
-{
-    public function count();
-
-    public function add(string ...$member);
-
-    public function rem(string ...$member);
-
-    public function isMember(string $member);
-
-    public function randMember($count = 1);
-
-    public function all();
-
-    public function delete();
-}

app/Controller/CollectorController.php

@@ -839,5 +839,5 @@ class CollectorController extends AbstractController
         // var_dump("=====222",$result);
         return Result::success($result);
     }
-   
+
 }

app/Controller/SseController.php

@@ -0,0 +1,66 @@
+<?php
+declare(strict_types=1);
+
+namespace App\Controller;
+
+use Hyperf\HttpServer\Contract\RequestInterface;
+use Hyperf\HttpServer\Contract\ResponseInterface;
+use Psr\Http\Message\ServerRequestInterface;
+
+class SseController extends AbstractController
+{
+    public function stream(RequestInterface $request, ResponseInterface $response)
+    {
+          $params = [
+              "model"=>"glm-4",
+              "messages"=>[
+                  [
+                      "role"=>"system",
+                      "content"=>"你是一个乐于解答各种问题的助手，你的任务是为用户提供专业、准确、有见地的建议。"//$requireData['message']
+                  ],
+                  [
+                      "role"=>"user",
+                      "content"=>"我是一个初级程序员,如何快速的提升自己"//$requireData['message']
+                  ]
+              ],
+              "stream"=>true
+          ];
+         $apiUrl = "https://open.bigmodel.cn/api/paas/v4/chat/completions";
+         $result= $this->callApi($apiUrl, 'POST', $params, ['Authorization: Bearer be1856920c54ac537b530d69bc2eda73.gOO2BMq9NXavzEMq']);
+          $arr =  preg_split('/\n\s*\n/', $result);
+        // 开始流式传输
+        $this->container->get(\Swoole\Http\Response::class)->status(200);
+        foreach ($arr as $val){
+            $response->write("$val\n\n");
+            usleep(1000000); // 暂停1秒
+        }
+        return $response;
+    }
+
+
+    function callApi($url, $method = 'GET', $data = null, $headers = [])
+    {
+        $ch = curl_init();
+
+        // 配置 CURL
+        curl_setopt($ch, CURLOPT_URL, $url);
+        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
+        curl_setopt($ch, CURLOPT_CUSTOMREQUEST, $method);
+        // 如果是 POST/PUT 请求，附加数据
+        if ($data && ($method === 'POST' || $method === 'PUT')) {
+            curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($data));
+            $headers[] = 'Content-Type: application/json';
+        }
+        // 附加头部信息
+        curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);
+        // 执行请求
+        $response = curl_exec($ch);
+        // 错误处理
+        if (curl_errno($ch)) {
+            $response = json_encode(['error' => curl_error($ch)]);
+        }
+        curl_close($ch);
+        return $response;
+    }
+
+}

app/Job/GatherExampleJob.php

@@ -1,39 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace App\Job;
-
-use Hyperf\AsyncQueue\Job;
-use App\Controller\CollectorController;
-/**
- * @Job(name="default")
- */
-class GatherExampleJob extends Job
-{
-    public $params;
-
-    /**
-     * 任务执行失败后的重试次数，即最大执行次数为 $maxAttempts+1 次
-     */
-    protected int $maxAttempts = 10;
-
-    public function __construct($params)
-    {
-        // 这里最好是普通数据，不要使用携带 IO 的对象，比如 PDO 对象
-        $this->params = $params;
-    }
-
-
-    public function handle()
-    {
-        try {
-            var_dump("消费者");
-            $collector = new CollectorController();
-            $result = $collector->goSendCrawler($this->params);
-            var_dump("消费消息队列：",$this->params,$result);
-        }catch (\Exception $e){
-            var_dump($e->getMessage());
-        }
-    }
-}

app/Job/ImportExampleJob.php

@@ -1,40 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace App\Job;
-use Hyperf\AsyncQueue\Job;
-use App\Controller\CollectorController;
-class ImportExampleJob extends Job
-{
-
-    public $params;
-    /**
-     * 任务执行失败后的重试次数，即最大执行次数为 $maxAttempts+1 次
-     */
-    protected int $maxAttempts = 10;
-
-    public function __construct($params)
-    {
-        // 这里最好是普通数据，不要使用携带 IO 的对象，比如 PDO 对象
-        $this->params = $params;
-    }
-    public function handle()
-    {
-        // if(isset($data['admin_user_id'])){
- 
-        // }else{
-
-            try {
-                var_dump("============",$this->params);
-                $collector = new CollectorController();
-                $result = $collector->goaddArt($this->params);
-                var_dump("消费消息队列：",$this->params,$result);
-            }catch (\Exception $e){
-                var_dump($e->getMessage());
-            }
-            // var_dump($$this->params);
-        // }
-
-    }
-}

app/Middleware/CorsMiddleware.php

@@ -15,7 +15,8 @@ class CorsMiddleware implements MiddlewareInterface
     public function process(ServerRequestInterface $request, RequestHandlerInterface $handler): ResponseInterface
     {
         $response = Context::get(ResponseInterface::class);
-        $response = $response->withHeader('Access-Control-Allow-Origin', '*')
+        $response = $response
+            ->withHeader('Access-Control-Allow-Origin', '*')
             ->withHeader('Access-Control-Allow-Credentials', 'true')
             // Headers 可以根据实际情况进行改写。
             ->withHeader('Access-Control-Allow-Headers', '*');

app/Service/Server/StreamServer.php

@@ -0,0 +1,41 @@
+<?php
+
+
+namespace App\Service\Server;
+
+
+use Hyperf\Dispatcher\HttpDispatcher;
+use Hyperf\ExceptionHandler\ExceptionHandlerDispatcher;
+use Hyperf\HttpServer\ResponseEmitter;
+use Hyperf\HttpServer\Server;
+use Psr\Container\ContainerInterface;
+use Swoole\Http\Request;
+use Swoole\Http\Response;
+
+class StreamServer extends Server
+{
+    const STREAM_URL = [
+        '/chat/stream'
+    ];
+    public function __construct(ContainerInterface $container, HttpDispatcher $dispatcher, ExceptionHandlerDispatcher $exceptionHandlerDispatcher, ResponseEmitter $responseEmitter)
+    {
+        parent::__construct($container, $dispatcher, $exceptionHandlerDispatcher, $responseEmitter);
+    }
+
+    /**
+     * @param Request $request
+     * @param Response $response
+     */
+    public function onRequest($request, $response): void
+    {
+        $pathInfo = $request->server['path_info'];
+        if (in_array($pathInfo, self::STREAM_URL)) {
+            $response->header('Content-Type', 'text/event-stream');
+            $response->header('Access-Control-Allow-Origin', '*');
+            $response->header('Access-Control-Allow-Methods', 'GET');
+            $response->header('Cache-Control', 'no-cache');
+            $response->header('Connection', 'keep-alive');
+        }
+        parent::onRequest($request, $response);
+    }
+}

app/Tools/PublicData.php

@@ -127,4 +127,66 @@ class PublicData
         }
     }
 
+    /**
+     * POST 请求
+     * @param string $url
+     * @param array $param
+     * @param boolean $post_file 是否文件上传
+     * @return string content
+     */
+    public static function http_post_zp($url, $data, $options = [])
+    {
+
+        // 初始化CURL会话
+        $ch = curl_init($url);
+        var_dump("参数：",$data);
+        // 设置CURL选项
+        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true); // 将curl_exec()获取的信息以文件流的形式返回，而不是直接输出。
+        curl_setopt($ch, CURLOPT_HEADER, false); // 将curl_exec()获取的信息以文件流的形式返回，而不是直接输出。
+        curl_setopt($ch, CURLOPT_POST, true); // 发送一个常规的POST请求。
+        curl_setopt($ch, CURLOPT_POSTFIELDS, http_build_query($data)); // 要传递的POST数据，这里使用http_build_query将数组转换为URL编码的查询字符串。
+        $headers = [
+            'Authorization: Bearer be1856920c54ac537b530d69bc2eda73.gOO2BMq9NXavzEMq',
+            'Content-Type: application/json',
+//            'Custom-Header: customHeaderValue'
+        ];
+        curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);
+        // 如果有额外的CURL选项，则合并它们
+        if (!empty($options)) {
+            curl_setopt_array($ch, $options);
+        }
+
+        // 执行CURL会话并获取响应
+        $response = curl_exec($ch);
+        // 检查是否有CURL错误
+        if (curl_errno($ch)) {
+            $error_msg = curl_error($ch);
+            curl_close($ch);
+            throw new Exception("CURL Error: $error_msg");
+        }
+
+        // 获取HTTP状态码
+        $http_code = curl_getinfo($ch, CURLINFO_HTTP_CODE);
+
+        // 关闭CURL会话
+        curl_close($ch);
+
+        // 返回一个包含响应和HTTP状态码的数组
+        $responseBody = $response;
+        $headerEnd = strpos($response, "\r\n\r\n");
+
+        if ($headerEnd !== false) {
+            // 去除响应头，只保留响应体
+            $responseBody = substr($response, $headerEnd + 4); // +4 是因为 "\r\n\r\n" 有4个字符
+//            echo $responseBody; // 输出：This is the response body.
+        } else {
+            // 如果没有找到空行，可能响应格式不正确或没有响应头
+//            echo "No headers found in response.";
+        }
+        return [
+            'response' => $responseBody,
+            'http_code' => $http_code
+        ];
+    }
+
 }

composer.json

@@ -16,6 +16,7 @@
         "death_satan/hyperf-validate": "^3.71",
         "doctrine/annotations": "^2.0",
         "easyswoole/verifycode": "3.x",
+        "friendsofhyperf/openai-client": "^3.1",
         "hyperf/amqp": "^3.1",
         "hyperf/async-queue": "^3.1",
         "hyperf/cache": "~3.1.0",
@@ -45,6 +46,7 @@
         "hyperf/translation": "^3.1",
         "hyperf/validation": "^3.1",
         "hyperf/websocket-server": "^3.1",
+        "openai-php/client": "^0.10.3",
         "phper666/jwt-auth": "^4.0"
     },
     "require-dev": {
@@ -79,7 +81,10 @@
     "prefer-stable": true,
     "config": {
         "optimize-autoloader": true,
-        "sort-packages": true
+        "sort-packages": true,
+        "allow-plugins": {
+            "php-http/discovery": true
+        }
     },
     "extra": [],
     "scripts": {

config/api/collector.php

@@ -2,10 +2,14 @@
 
 declare(strict_types=1);
 
-use App\Controller\CollectorController;
+
 use App\Middleware\Auth\FooMiddleware;
 use Hyperf\HttpServer\Router\Router;
-
+use App\Controller\CollectorController;
+//智普测试接口
+Router::addGroup('/chat', function () {
+    Router::get('/stream', 'App\Controller\SseController@stream');
+});
 Router::addGroup(
     '/collector', function () {
         Router::get('/index', [CollectorController::class, 'index']);
@@ -46,9 +50,7 @@ Router::addGroup(
         Router::post('/addCatid', [CollectorController::class, 'addCatid']);
         //导入数据
         Router::get('/addArt', [CollectorController::class, 'addArt']);
-        
-        
-        
+
     },
     ['middleware' => [FooMiddleware::class]]
 );

config/autoload/redis.php

@@ -19,11 +19,39 @@ return [
         'db' => (int) env('REDIS_DB', 0),
         'pool' => [
             'min_connections' => 1,
-            'max_connections' => 10,
-            'connect_timeout' => 10.0,
-            'wait_timeout' => 3.0,
+            'max_connections' => 100,
+            'connect_timeout' => 100000000.0,
+            'wait_timeout' => 30000000.0,
             'heartbeat' => -1,
-            'max_idle_time' => (float) env('REDIS_MAX_IDLE_TIME', 60),
+            'max_idle_time' => (float) env('REDIS_MAX_IDLE_TIME', 6000000000),
         ],
+        'driver' => Hyperf\AsyncQueue\Driver\RedisDriver::class,
+        'redis' => [
+            'pool' => 'default'
+        ],
+        'channel' => '{queue:import}',
+        'timeout' => 20,
+        'retry_seconds' => 5,
+        'handle_timeout' => 10,
+        'processes' => 100,
+        'concurrent' => [
+            'limit' => 10000,
+        ],
+        'max_messages' => 0,
+    ],
+    'import' => [
+        'driver' => Hyperf\AsyncQueue\Driver\RedisDriver::class,
+        'redis' => [
+            'pool' => 'default'
+        ],
+        'channel' => '{queue:import}',
+        'timeout' => 20,
+        'retry_seconds' => 5,
+        'handle_timeout' => 10,
+        'processes' => 100,
+        'concurrent' => [
+            'limit' => 10000,
+        ],
+        'max_messages' => 0,
     ],
 ];

config/autoload/server.php

@@ -23,7 +23,8 @@ return [
             'port' => 9501,
             'sock_type' => SWOOLE_SOCK_TCP,
             'callbacks' => [
-                Event::ON_REQUEST => [Hyperf\HttpServer\Server::class, 'onRequest'],
+                Event::ON_REQUEST => [App\Service\Server\StreamServer::class, 'onRequest'],
+//                Event::ON_REQUEST => [Hyperf\HttpServer\Server::class, 'onRequest'],
             ],
         ],
         [