sse接入修改 上传文档修改

takai-admin/src/main/java/com/takai/web/controller/takaiai/TakaiAiController.java

@@ -3,9 +3,11 @@ package com.takai.web.controller.takaiai;
 import com.alibaba.fastjson2.JSONObject;
 import com.github.pagehelper.Page;
 import com.github.pagehelper.PageHelper;
+import com.takai.ai.domain.TakaiSysOss;
 import com.takai.ai.domain.dto.*;
 import com.takai.ai.domain.entity.*;
 import com.takai.ai.service.ITakaiAiService;
+import com.takai.ai.utils.MinioUtil;
 import com.takai.bigmodel.domain.entity.PageParams;
 import com.takai.common.annotation.Log;
 import com.takai.common.constant.HttpStatus;
@@ -58,6 +60,9 @@ public class TakaiAiController extends BaseController {
     @Autowired
     private ISysProjectService sysProjectService;
 
+    @Autowired
+    private MinioUtil minioUtil;
+
     /**
      * 对话聊天
      *
@@ -331,6 +336,18 @@ public class TakaiAiController extends BaseController {
         return success(i);
     }
 
+    /**
+     * 删除知识
+     *
+     * @param documentId
+     * @return
+     */
+    @DeleteMapping("/delUnfinishedDocument/{documentId}")
+    public R<Integer> delUnfinishedDocument(@PathVariable String documentId) {
+        int i = takaiAisService.delUnfinishedDocument(documentId);
+        return R.ok(i);
+    }
+
     /**
      * 上传知识文件
      *
@@ -343,8 +360,21 @@ public class TakaiAiController extends BaseController {
     public AjaxResult uploadDocument(@RequestParam("files") MultipartFile[] files,
                                      @PathVariable String knowledgeId) throws Exception {
         try {
+            // 1. 先同步上传文件到MinIO（IO操作，建议后续也异步，这里先保留）
+            List<TakaiSysOss> ossList;
+            try {
+                ossList = minioUtil.uploadMultiple(files);
+                if (ossList == null || ossList.isEmpty()) {
+                    log.error("MinIO上传文件为空，直接返回");
+                    return error("MinIO上传文件为空");
+                }
+            } catch (Exception e) {
+                log.error("MinIO上传文件失败", e);
+                return error("MinIO上传文件失败");
+            }
+            String updateBy = SecurityUtils.getLoginUser() == null ? "" : SecurityUtils.getLoginUser().getUserId();
             // 调用异步方法
-            CompletableFuture<Integer> future = takaiAisService.uploadDocument(files, knowledgeId);
+            CompletableFuture<Integer> future = takaiAisService.uploadDocument(files, knowledgeId,ossList,updateBy);
             // 处理结果（非阻塞）
             future.whenComplete((successCount, ex) -> {
                 if (ex != null) {
@@ -356,7 +386,7 @@ public class TakaiAiController extends BaseController {
             });
             return success(1);
         } catch (Exception e) {
-            return error("0");
+            return error("上传文档失败");
         }
     }
 

takai-admin/src/main/resources/application.yml

@@ -166,7 +166,7 @@ deepseek:
   #base url
   baseurl: http://10.168.100.17:6666
   #baseurl: http://10.1.27.4:18079
-#  baseurl: http://xia0miduo.gicp.net:6001
+#  baseurl: http://xia0miduo.gicp.net:6666
 #  baseurl: http://192.168.3.209:18078
   #创建知识库
   createKnowledge: /rag/create_collection
@@ -196,6 +196,8 @@ deepseek:
   webSearch: /web/search
   #deepseek单独聊天
   baseChat: http://10.1.27.4:18078
+  #取消上传
+  cancelTask: /rag/cancel_task
   #提示词
   prompt: 你是总结和提问大师。你只根据用户的对话记录，根据对话记录生成可能会问的问题，不要杜撰问题。问题的答案必须来自上面的对话记录。 你必须遵守以下要求：1. 不要输出用户问过的问题；2. 你需要输出3个问题供用户选择。3. 你只需要输出问题，不需要解释，不需要提问。4. 你的问题可以是空的，但你不能杜撰问题。5. 问题需要站在使用这个应用的人的视角提出，因此你要注意提问的语气和人称代词。不要用您这个字。6.生成的问题，必须能从提供的对话记录中找到答案。你一定要按照以下格式输出：{'问题':['xxx','xxx','xxx']}
 jk:

takai-ai/src/main/java/com/takai/ai/service/ITakaiAiService.java

@@ -1,6 +1,7 @@
 package com.takai.ai.service;
 
 import com.alibaba.fastjson2.JSONObject;
+import com.takai.ai.domain.TakaiSysOss;
 import com.takai.ai.domain.dto.*;
 import com.takai.ai.domain.entity.*;
 import com.takai.common.core.domain.entity.SysDictData;
@@ -44,7 +45,7 @@ public interface ITakaiAiService
 
     int delKnowledge(String knowledgeId);
 
-    CompletableFuture<Integer> uploadDocument(MultipartFile[] files, String knowledgeId) throws Exception;
+    CompletableFuture<Integer> uploadDocument(MultipartFile[] files, String knowledgeId,List<TakaiSysOss> ossList,String updateBy) throws Exception;
 
     void updateDocumentByPython(JkDocumentPythonParams parms);
 
@@ -56,6 +57,9 @@ public interface ITakaiAiService
 
     int delDocument(String documentId);
 
+    int delUnfinishedDocument(String documentId);
+
+
     void uploadUrl(TakaiDocumentObject object);
 
     int createApplication(TakaiApplicationParams params, boolean isAdmin);

takai-ai/src/main/java/com/takai/ai/service/impl/TakaiAiServiceImpl.java

@@ -668,92 +668,86 @@ public class TakaiAiServiceImpl implements ITakaiAiService {
         return 0;
     }
 
-    @Override
+
     /**
      * 异步上传文档，analysisFile最大并发5个
      * @return 成功处理的文件数量
      */
-    public CompletableFuture<Integer> uploadDocument(MultipartFile[] files, String knowledgeId) {
+    @Override
+    public CompletableFuture<Integer> uploadDocument(MultipartFile[] files, String knowledgeId,List<TakaiSysOss> ossList,String updateBy) {
+        // 2. 循环处理每个文件
+        List<TakaiDocument> takaiDocuments = new ArrayList<>();
+        for (int i = 0; i < ossList.size(); i++) {
+            TakaiSysOss oss = ossList.get(i);
+            // 生成文档ID
+            SnowflakeDigitGenerator documentIdGenerator = new SnowflakeDigitGenerator(i, i);
+            long documentId = documentIdGenerator.nextId();
+            String docId = "a" + documentId;
+            // 构建上传参数
+            UploadDocumentParams param = new UploadDocumentParams();
+            param.setDocument_id(docId);
+            param.setName(oss.getOriginalName());
+            param.setUrl(oss.getUrl());
+            List<UploadDocumentParams> params = new ArrayList<>();
+            params.add(param);
+            // 3. 同步保存OSS到数据库
+            takaiSysOssMapper.insertSysOss(oss);
+            // 4. 保存文档信息到数据库
+            TakaiDocument document = TakaiDocument.builder()
+                    .documentId(docId)
+                    .knowledgeId(knowledgeId)
+                    .customSeparator(String.format("[\"%s\"", "\\n") + "]")
+                    .sentenceSize("300")
+                    .name(oss.getOriginalName())
+                    .url(oss.getUrl())
+                    .status("0")
+                    .build();
+            document.setCreateBy(updateBy);
+            int insertResult = takaiDocumentMapper.insertDocument(document);
+            takaiDocuments.add(document);
+            // 5. 保存文档设置
+            if (insertResult > 0) {
+                SnowflakeDigitGenerator settingsIdGenerator = new SnowflakeDigitGenerator(1, 1);
+                long settingsId = settingsIdGenerator.nextId();
+                TakaiDocumentSettings dSettings = new TakaiDocumentSettings();
+                dSettings.setId(String.valueOf(settingsId));
+                dSettings.setKnowledgeId(knowledgeId);
+                dSettings.setDocumentId(docId);
+                dSettings.setSetSlice("0");   // 默认 按标题段落切片
+                dSettings.setSetAnalyze("1"); // 默认 图片转换成标识符
+                dSettings.setSetTable("0");   // 默认 table转图片
+                takaiDocumentSettingsMapper.insertDocumentSettings(dSettings);
+            }
+        }
         // 整体逻辑异步执行（默认使用ForkJoinPool，也可指定自定义线程池）
         return CompletableFuture.supplyAsync(() -> {
             // 记录成功处理的文件数
             AtomicInteger successCount = new AtomicInteger(0);
             // 收集analysisFile的异步任务，用于等待全部完成
             List<CompletableFuture<Void>> analysisTasks = new ArrayList<>();
-
-            // 1. 先同步上传文件到MinIO（IO操作，建议后续也异步，这里先保留）
-            List<TakaiSysOss> ossList;
-            try {
-                ossList = minioUtil.uploadMultiple(files);
-                if (ossList == null || ossList.isEmpty()) {
-                    log.error("MinIO上传文件为空，直接返回");
-                    return 0;
-                }
-            } catch (Exception e) {
-                log.error("MinIO上传文件失败", e);
-                return 0;
-            }
-
             // 2. 循环处理每个文件
-            String updateBy = SecurityUtils.getLoginUser() == null ? "" : SecurityUtils.getLoginUser().getUserId();
-            for (int i = 0; i < ossList.size(); i++) {
-                TakaiSysOss oss = ossList.get(i);
+            for (int i = 0; i < takaiDocuments.size(); i++) {
+                TakaiDocument document = takaiDocuments.get(i);
                 try {
-                    // 生成文档ID
-                    SnowflakeDigitGenerator documentIdGenerator = new SnowflakeDigitGenerator(i, i);
-                    long documentId = documentIdGenerator.nextId();
-                    String docId = "a" + documentId;
-
                     // 构建上传参数
                     UploadDocumentParams param = new UploadDocumentParams();
-                    param.setDocument_id(docId);
-                    param.setName(oss.getOriginalName());
-                    param.setUrl(oss.getUrl());
+                    param.setDocument_id(document.getDocumentId());
+                    param.setName(document.getName());
+                    param.setUrl(document.getUrl());
                     List<UploadDocumentParams> params = new ArrayList<>();
                     params.add(param);
-
-                    // 3. 同步保存OSS到数据库
-                    takaiSysOssMapper.insertSysOss(oss);
-
-                    // 4. 保存文档信息到数据库
-                    TakaiDocument document = TakaiDocument.builder()
-                            .documentId(docId)
-                            .knowledgeId(knowledgeId)
-                            .customSeparator(String.format("[\"%s\"", "\\n") + "]")
-                            .sentenceSize("300")
-                            .name(oss.getOriginalName())
-                            .url(oss.getUrl())
-                            .status("0")
-                            .build();
-                    document.setCreateBy(updateBy);
-                    int insertResult = takaiDocumentMapper.insertDocument(document);
-
-                    // 5. 保存文档设置
-                    if (insertResult > 0) {
-                        SnowflakeDigitGenerator settingsIdGenerator = new SnowflakeDigitGenerator(1, 1);
-                        long settingsId = settingsIdGenerator.nextId();
-                        TakaiDocumentSettings dSettings = new TakaiDocumentSettings();
-                        dSettings.setId(String.valueOf(settingsId));
-                        dSettings.setKnowledgeId(knowledgeId);
-                        dSettings.setDocumentId(docId);
-                        dSettings.setSetSlice("0");   // 默认 按标题段落切片
-                        dSettings.setSetAnalyze("1"); // 默认 图片转换成标识符
-                        dSettings.setSetTable("0");   // 默认 table转图片
-                        takaiDocumentSettingsMapper.insertDocumentSettings(dSettings);
-                    }
-
                     // 6. 异步调用analysisFile（限制5个并发）
                     TakaiDocumentSettings settings = new TakaiDocumentSettings();
                     settings.setKnowledgeId(knowledgeId);
+                    settings.setCreateBy(updateBy);
                     // 注意：捕获循环变量，避免闭包陷阱
                     List<UploadDocumentParams> finalParams = new ArrayList<>(params);
                     CompletableFuture<Void> analysisTask = CompletableFuture.runAsync(() -> {
                         try {
                             // 执行文件解析逻辑
                             analysisFile(finalParams, settings, "upload");
-                            log.info("文件解析完成：{}", oss.getOriginalName());
                         } catch (Exception e) {
-                            log.error("文件解析失败：{}", oss.getOriginalName(), e);
+                            log.error("文件解析失败：{}", e);
                         }
                     }, analysisFileExecutor); // 使用5线程池执行
 
@@ -762,7 +756,7 @@ public class TakaiAiServiceImpl implements ITakaiAiService {
                     successCount.incrementAndGet();
 
                 } catch (Exception e) {
-                    log.error("处理文件失败：{}", oss.getOriginalName(), e);
+                    log.error("处理文件失败：{}", e);
                     // 单个文件失败不影响其他文件，继续处理
                     continue;
                 }
@@ -939,6 +933,62 @@ public class TakaiAiServiceImpl implements ITakaiAiService {
         return 0;
     }
 
+    @Override
+    public int delUnfinishedDocument(String documentId) {
+        TakaiDocument vo = takaiDocumentMapper.selectTargetDocument(TakaiDocument.builder().documentId(documentId).build());
+        if(vo != null){
+            String url = deepseekConfig.getBaseurl() + deepseekConfig.getCancelTask() + "/" + documentId ;
+            JSONObject json = new JSONObject();
+            //重新设置knowledgeIds为数组
+            json.put("documentId",documentId);
+
+            RequestBody requestBody = FormBody.create(MediaType.parse("application/json; charset=utf-8"), json.toJSONString());
+            Request request = buildPostRequest(url,requestBody);
+            OkHttpClient client = buildOkHttpClient();
+            try {
+                Response response = client.newCall(request).execute();
+                if (response.isSuccessful()) {
+                    String body = response.body().string();
+                    logger.info("删除知识文件调用python接口返回结果：{}", body);
+                    JSONObject obj = JSON.parseObject(body);
+                    Integer code = obj.getInteger("code");
+                    if (code == 200) {
+                        List<TakaiMediaReplacement> list = takaiMediaReplacementMapper.selectMediaList(TakaiMediaReplacement.builder().documentId(documentId).build());
+
+                        // 删除图片
+                        takaiMediaReplacementMapper.deleteMedia(documentId);
+
+                        // 删除document文件
+                        int i = takaiDocumentMapper.delDocument(documentId);
+
+                        if(i > 0){
+                            //更新知识库
+                            updateKnowledgeSize(vo.getKnowledgeId());
+                            // 删除minio文件
+                            try {
+                                logger.info("删除minio文件");
+                                minioUtil.remove("deepseek-doc", "/"+vo.getName());
+                                if(list != null && list.size() > 0){
+                                    for (int k=1; k<=list.size(); k++ ) {
+                                        logger.info("删除minio图片{}", k);
+                                        minioUtil.remove("deepseek-doc" , "/pdf/"+vo.getKnowledgeId()+"/"+documentId+"/【示意图序号_"+documentId+"_"+k+"】.jpg");
+                                    }
+                                }
+                            } catch (Exception e) {
+                                logger.error("删除minion文件失败", e.getMessage());
+                                throw new RuntimeException(e);
+                            }
+                            return i;
+                        }
+                    }
+                }
+            } catch (IOException e) {
+                logger.error("删除知识文件调用python接口失败", e.getMessage());
+            }
+        }
+        return 0;
+    }
+
     @Override
     public void uploadUrl(TakaiDocumentObject object) {
 
@@ -1239,6 +1289,7 @@ public class TakaiAiServiceImpl implements ITakaiAiService {
         settings.setSliceValue(documentSettings.getSliceValue());
         settings.setSetAnalyze(documentSettings.getSetAnalyze());
         settings.setSetTable(documentSettings.getSetTable());
+        settings.setCreateBy(SecurityUtils.getUserId());
         TakaiDocument document = takaiDocumentMapper.selectTargetDocument(TakaiDocument.builder().documentId(documentId).build());
         List<UploadDocumentParams> listParams = new ArrayList<>();
         if (document != null) {
@@ -1787,6 +1838,7 @@ public class TakaiAiServiceImpl implements ITakaiAiService {
         JSONObject json = new JSONObject();
         json.put("knowledge_id", settings.getKnowledgeId());
         json.put("docs", result);
+        json.put("userId",settings.getCreateBy());
         if ("upload".equals(flag)) {
             json.put("set_slice", "0");
             json.put("slice_value", null);

takai-ai/src/main/resources/mapper/takaiai/TakaiDocumentMapper.xml

@@ -85,7 +85,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
  			<if test="remark != null and remark != ''">#{remark},</if>
  			<if test="createBy != null and createBy != ''">#{createBy},</if>
 		    <if test="sliceTotal != null and sliceTotal != ''">#{sliceTotal},</if>
-			<if test="status != null and status != ''">status,</if>
+			<if test="status != null and status != ''">#{status},</if>
  			sysdate()
  		)
 	</insert>

takai-common/src/main/java/com/takai/common/config/DeepseekConfig.java

@@ -44,4 +44,6 @@ public class DeepseekConfig {
 
     private boolean contentBase64Encode;
 
+    private String cancelTask;
+
 }

takai-common/src/main/java/com/takai/common/core/controller/SseController.java

@@ -33,13 +33,14 @@ public class SseController implements DisposableBean {
      * 建立 SSE 连接
      */
     @GetMapping(value = "/resource/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
-    public SseEmitter connect() {
-        LoginUser loginUser = SecurityUtils.getLoginUser();
-        if (loginUser == null) {
-            return null;
-        }
-        String tokenValue = loginUser.getToken();
-        String userId = loginUser.getUserId();
+    public SseEmitter connect(String userId,String tokenValue) {
+//        LoginUser loginUser = SecurityUtils.getLoginUser();
+//        if (loginUser == null) {
+//            return null;
+//        }
+//        SecurityUtils.getLoginUser().
+//        String tokenValue = loginUser.getToken();
+//        String userId = loginUser.getUserId();
         return sseEmitterManager.connect(Long.parseLong(userId), tokenValue);
     }
 

takai-framework/src/main/java/com/takai/framework/config/SecurityConfig.java

@@ -116,7 +116,7 @@ public class SecurityConfig extends WebSecurityConfigurerAdapter
                         "/checkToken/**", "/**/presets/**", "/**/index/**", "/**/createApplaction/**", "/**/createKnowledge/**",
                         "/**/updateKnowledge/**", "/**/detailKnowledge/**", "/**/delKnowledge/**", "/**/knowledgeList/**",
                         "/**//bigmodel/api/**", "/**/tool/gen/**","/**/deepseek/api/chat","/**/deepseek/api/free/**",
-                        "/admin/login", "/**/deepseek/getInfo/**", "/**/system/info/**", "/**/system/audit/config/**","/resource/**",
+                        "/admin/login", "/**/deepseek/getInfo/**", "/**/system/info/**", "/**/system/audit/config/**","/resource/**","/deepseek/api/updateDocumentByPython",
                         "/**/system/project/**").permitAll()
                 // 静态资源，可匿名访问
                 .antMatchers(HttpMethod.GET, "/", "/*.html", "/**/*.html", "/**/*.css", "/**/*.js", "/profile/**").permitAll()