多文档解析

package nexus.io.mosskb.service.kb;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.zip.*;
import javax.xml.XMLConstants;
import javax.xml.parsers.DocumentBuilderFactory;
import org.w3c.dom.*;
import org.apache.pdfbox.pdmodel.PDDocument;
import org.apache.pdfbox.text.PDFTextStripper;
import nexus.io.gitee.*;
import nexus.io.db.activerecord.Db;
import nexus.io.tio.utils.crypto.Md5Utils;

/** Detect the actual document structure before selecting text extraction or OCR. */
public class DocumentParsingService {

  /** 一个页面提取到的内容。 */
  private record PageResult(int page, String text, boolean viaOcr, boolean failed) {}

  /**
   * 解析结果。
   *
   * @param strategy    解析策略
   * @param text        正文
   * @param pages       物理页数，非分页格式为 0
   * @param ocrPages    走 OCR 的页数
   * @param failedPages 尝试多次后仍然失败的页数
   */
  public record Parsed(String strategy, String text, int pages, int ocrPages, int failedPages) {
    /** 兼容只关心策略、正文和页数的调用方。 */
    public Parsed(String strategy, String text, int pages) {
      this(strategy, text, pages, 0, 0);
    }
  }

  /** 解析进度：已完成页数与总页数。 */
  public record Progress(int completed, int total) {
  }

  public Parsed parse(byte[] data, String filename) throws Exception {
    return parse(data, filename, null);
  }

  public Parsed parse(byte[] data, String filename, java.util.function.Consumer<Progress> progress) throws Exception {
    if (data == null || data.length == 0) {
      throw new IllegalArgumentException("文件为空");
    }
    if (data.length > 100 * 1024 * 1024) {
      throw new IllegalArgumentException("文件不能超过100MB");
    }
    String name = filename.toLowerCase(Locale.ROOT);
    Parsed result;
    if (new String(data, 0, Math.min(5, data.length), StandardCharsets.US_ASCII).equals("%PDF-")) {
      // PDF：逐页判断
    } else if (data.length > 4 && data[0] == 'P' && data[1] == 'K' && name.endsWith(".docx")) {
      // DOCX：结构提取，必要时补图片 OCR
    } else if (name.endsWith(".xlsx") || name.endsWith(".xls")) {
      // 表格：按工作表提取
    } else if (name.endsWith(".doc")) {
      // 旧版 Word
    } else if (name.endsWith(".zip")) {
      // 受限 ZIP 文档集合
    } else if (name.matches(".*\\.(txt|md|log|csv|json|html|htm)$")) {
      // 文本与 HTML
    } else if (javax.imageio.ImageIO.read(new ByteArrayInputStream(data)) != null) {
      // 图片：直接 OCR
      result = new Parsed("image-ocr", ocr(data, filename), 1, 1, 0);
    } else {
      throw new IllegalArgumentException("不支持的文件格式，请转换为PDF、DOCX、TXT或图片后上传");
    }
    if (result.text().isBlank()) {
      // 全部走 OCR 的文档如果每一页都失败，原因是远程服务不可用，不是文档本身没有内容。
      if (result.ocrPages() > 0 && result.ocrPages() == result.failedPages()) {
        throw new IOException("扫描页OCR识别失败，请稍后重试");
      }
      throw new IllegalArgumentException("未提取到内容，请检查文档或使用扫描PDF");
    }
    return result;
  }
}

  /** 并发提取的页面数：OCR 是网络调用，页与页之间并发即可，不需要按 CPU 核数拉满。 */
  private static final int OCR_PAGE_PARALLELISM = Math.max(2, Math.min(4, Runtime.getRuntime().availableProcessors()));

  /** 逐页提取 PDF：文本层优先，扫描页走 OCR，页面之间并发处理但按原页序输出。 */
  private Parsed parsePdf(byte[] data, java.util.function.Consumer<Progress> progress) throws Exception {
    try (PDDocument pdf = PDDocument.load(data)) {
      if (pdf.isEncrypted()) {
        throw new IllegalArgumentException("请先解密PDF文件");
      }
      int totalPages = pdf.getNumberOfPages();
      PDFTextStripper stripper = new PDFTextStripper();
      stripper.setSortByPosition(true);
      String sourceHash = Md5Utils.md5Hex(data);
      PageResult[] results = new PageResult[totalPages];
      AtomicInteger completed = new AtomicInteger();
      ForkJoinPool pool = new ForkJoinPool(OCR_PAGE_PARALLELISM);
      try {
        List<ForkJoinTask<?>> tasks = new ArrayList<>(totalPages);
        for (int page = 1; page <= totalPages; page++) {
          int pageNumber = page;
          tasks.add(pool.submit(() -> {
            try {
              results[pageNumber - 1] = extractPdfPage(pdf, stripper, sourceHash, pageNumber);
            } catch (IOException e) {
              throw new UncheckedIOException("解析第 " + pageNumber + " 页失败", e);
            }
            if (progress != null) {
              progress.accept(new Progress(completed.incrementAndGet(), totalPages));
            }
          }));
        }
        for (ForkJoinTask<?> task : tasks) {
          task.join();
        }
      } finally {
        pool.shutdown();
      }
      StringBuilder text = new StringBuilder();
      int ocrPages = 0;
      int nativePages = 0;
      int failedPages = 0;
      for (PageResult pageResult : results) {
        if (pageResult.viaOcr()) {
          ocrPages++;
        } else {
          nativePages++;
        }
        if (pageResult.failed()) {
          failedPages++;
        }
        if (!pageResult.text().isBlank()) {
          text.append("\n\n> Page ").append(pageResult.page()).append("\n\n").append(pageResult.text());
        }
      }
      String suffix = failedPages == 0 ? "" : "-partial";
      String strategy;
      if (ocrPages == 0) {
        strategy = "pdf-text";
      } else if (nativePages == 0) {
        strategy = "pdf-ocr" + suffix;
      } else {
        strategy = "pdf-mixed" + suffix;
      }
      return new Parsed(strategy, text.toString(), totalPages, ocrPages, failedPages);
    }
  }

  /** 提取单页：先尝试文本层，文本不足时把该页单独保存后交给 OCR。 */
  private PageResult extractPdfPage(PDDocument pdf, PDFTextStripper stripper, String sourceHash, int pageNumber)
      throws IOException {
    String page;
    // PDFTextStripper 与 PDDocument 都不是线程安全的，逐页提取时串行访问。
    synchronized (pdf) {
      stripper.setStartPage(pageNumber);
      stripper.setEndPage(pageNumber);
      page = stripper.getText(pdf);
    }
    if (page.replaceAll("\\s", "").length() >= 40 && !page.contains("\uFFFD")) {
      return new PageResult(pageNumber, page, false, false);
    }
    byte[] pageData;
    try (PDDocument single = new PDDocument(); ByteArrayOutputStream out = new ByteArrayOutputStream()) {
      synchronized (pdf) {
        single.importPage(pdf.getPage(pageNumber - 1));
      }
      single.save(out);
      pageData = out.toByteArray();
    }
    // OCR 结果为空既可能是空白页，也可能是服务没能识别，两种都按空页处理并保证页码占位。
    boolean[] failed = new boolean[1];
    String[] recognized = new String[1];
    boolean ok = withOcrRetry(() -> ocrPdfPage(sourceHash, pageNumber, pageData), recognized, failed,
        sourceHash + ":page:" + pageNumber + ":" + OCR_CACHE_VERSION);
    return new PageResult(pageNumber, ok ? recognized[0] : "", true, failed[0]);
  }

  protected String ocr(byte[] data, String filename) throws Exception {
    return ocrWithCache(data, filename, Md5Utils.md5Hex(data) + ":" + OCR_CACHE_VERSION);
  }

  protected String ocrPdfPage(String sourceHash, int page, byte[] pageData) throws Exception {
    // Serializing a PDF page can generate a fresh internal document ID. Cache by
    // original file content and page number instead of the transient PDF bytes.
    return ocrWithCache(pageData, "page.pdf", sourceHash + ":page:" + page + ":" + OCR_CACHE_VERSION);
  }

  private String ocrWithCache(byte[] data, String filename, String hash) throws Exception {
    String cached = readOcrCache(hash);
    if (cached != null && !cached.isBlank()) {
      return textOnly(cached);
    }
    synchronized (OCR_LOCK_STRIPES[Math.floorMod(hash.hashCode(), OCR_LOCKS)]) {
      // 同一份内容的第一个请求可能已经写好缓存，这里再查一次就能省掉重复的付费调用。
      cached = readOcrCache(hash);
      if (cached != null && !cached.isBlank()) {
        return textOnly(cached);
      }
      String markdown = recognize(data, filename);
      if (markdown != null && !markdown.isBlank()) {
        writeOcrCache(hash, markdown);
      }
      return markdown;
    }
  }

  protected String readOcrCache(String hash) {
    return Db.queryStr("select content from moss_kb_document_markdown_cache where id=?", hash);
  }

  protected void writeOcrCache(String hash, String markdown) {
    Db.update("insert into moss_kb_document_markdown_cache(id,content) values(?,?) "
        + "on conflict(id) do update set content=excluded.content", hash, markdown);
  }

  protected String recognize(byte[] data, String filename) throws Exception {
    GiteeClient client = new GiteeClient();
    GiteeDocumentParseRequest request = new GiteeDocumentParseRequest();
    request.setModel(OCR_MODEL);
    request.setPrompt(GiteePromptConst.pdf_to_markdown_prompt);
    request.setInclude_image(false);
    request.setInclude_image_base64(false);
    GiteeTaskResponse task = client.parseDocument(data, filename, request);
    long deadline = System.nanoTime() + java.util.concurrent.TimeUnit.MINUTES.toNanos(8);
    while (!Arrays.asList("success", "succeeded", "completed")
        .contains(String.valueOf(task.getStatus()).toLowerCase(Locale.ROOT))) {
      if (task.getStatus() == null || task.getTask_id() == null) {
        throw new IOException("OCR服务未返回有效任务状态");
      }
      if (Arrays.asList("failed", "failure", "cancelled", "canceled")
          .contains(task.getStatus().toLowerCase(Locale.ROOT))) {
        throw new IOException("OCR任务失败: " + task.getTask_id());
      }
      if (System.nanoTime() > deadline) {
        throw new IOException("OCR任务超时: " + task.getTask_id());
      }
      Thread.sleep(2000);
      task = client.getTask(task.getTask_id());
    }
    // 扫描页可能是空白页或整页图片，服务会正常返回空正文，这里按空内容处理，由调用方决定是否记录。
    String markdown = textOnly(GiteeSimpleMarkdownUtils.toMarkdown(task.getOutput(), null));
    if (markdown == null || markdown.isBlank()) {
      log.info("OCR returned no text for {}", filename);
      return "";
    }
    return markdown;
  }

  private String textOnly(String markdown) {
    if (markdown == null) {
      return "";
    }
    return markdown.replaceAll("(?is)<img\\b[^>]*>", "")
        .replaceAll("(?is)<div[^>]*>\\s*</div>", "").replaceAll("!\\[[^\\]]*\\]\\([^)]*\\)", "");
  }

  /** 单个页面 OCR 的最大尝试次数，用于兜底偶发网络与服务端错误。 */
  private static final int OCR_MAX_ATTEMPTS = 3;

  /**
   * OCR 重试：失败时按 1s、2s 退避重试，最后一次仍失败就记为失败页。
   *
   * <p>“未识别到内容”是正常返回，不是异常，因此不在这里重试。
   *
   * @return 是否拿到结果
   */
  private boolean withOcrRetry(OcrCall call, String[] result, boolean[] failed, String describe) {
    for (int attempt = 1; attempt <= OCR_MAX_ATTEMPTS; attempt++) {
      try {
        result[0] = call.recognize();
        return true;
      } catch (Exception e) {
        log.warn("OCR attempt {}/{} failed for {}: {}", attempt, OCR_MAX_ATTEMPTS, describe, e.getMessage());
        if (attempt < OCR_MAX_ATTEMPTS) {
          try {
            Thread.sleep(1000L * attempt);
          } catch (InterruptedException interrupted) {
            Thread.currentThread().interrupt();
            failed[0] = true;
            return false;
          }
        }
      }
    }
    log.warn("OCR gave up for {}", describe);
    failed[0] = true;
    return false;
  }

  /** 一次 OCR 调用，用于把重试逻辑包在可覆盖的 OCR 方法外面。 */
  private interface OcrCall {
    String recognize() throws Exception;
  }

  private static final int DOCX_XML_LIMIT = 20 * 1024 * 1024;

  private String docx(byte[] data) throws Exception {
    try (ZipInputStream zip = new ZipInputStream(new ByteArrayInputStream(data))) {
      ZipEntry entry;
      while ((entry = zip.getNextEntry()) != null) {
        if (!entry.getName().equals("word/document.xml")) {
          continue;
        }
        byte[] xml = zip.readNBytes(DOCX_XML_LIMIT + 1);
        if (xml.length > DOCX_XML_LIMIT) {
          throw new IllegalArgumentException("DOCX内容过大");
        }
        DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
        factory.setNamespaceAware(true);
        factory.setFeature("http://apache.org/xml/features/disallow-doctype-decl", true);
        factory.setAttribute(XMLConstants.ACCESS_EXTERNAL_DTD, "");
        factory.setAttribute(XMLConstants.ACCESS_EXTERNAL_SCHEMA, "");
        Document doc = factory.newDocumentBuilder().parse(new ByteArrayInputStream(xml));
        StringBuilder out = new StringBuilder();
        Node body = doc.getElementsByTagNameNS("*", "body").item(0);
        for (Node node = body.getFirstChild(); node != null; node = node.getNextSibling()) {
          if ("p".equals(node.getLocalName())) {
            out.append(text(node)).append("\n\n");
          }
          if ("tbl".equals(node.getLocalName())) {
            NodeList rows = ((Element) node).getElementsByTagNameNS("*", "tr");
            for (int r = 0; r < rows.getLength(); r++) {
              NodeList cells = ((Element) rows.item(r)).getElementsByTagNameNS("*", "tc");
              out.append('|');
              for (int c = 0; c < cells.getLength(); c++) {
                out.append(text(cells.item(c)).replace("|", "\\|")).append('|');
              }
              out.append('\n');
              if (r == 0) {
                out.append('|');
                for (int c = 0; c < cells.getLength(); c++) {
                  out.append("---|");
                }
                out.append('\n');
              }
            }
            out.append('\n');
          }
        }
        return out.toString();
      }
    }
    throw new IllegalArgumentException("DOCX缺少正文");
  }

  private String text(Node node) {
    StringBuilder out = new StringBuilder();
    NodeList texts = ((Element) node).getElementsByTagNameNS("*", "t");
    for (int i = 0; i < texts.getLength(); i++) {
      out.append(texts.item(i).getTextContent());
    }
    return out.toString();
  }

    } else if (data.length > 4 && data[0] == 'P' && data[1] == 'K' && name.endsWith(".docx")) {
      String extracted = docx(data);
      if (extracted.replaceAll("\\s", "").length() < MIN_NATIVE_CHARS) {
        String images = docxImages(data);
        result = new Parsed(images.isBlank() ? "docx-structure" : "docx-ocr", extracted + images, 0);
      } else {
        result = new Parsed("docx-structure", extracted, 0);
      }
    }

  private String spreadsheet(byte[] data) throws Exception {
    try (org.apache.poi.ss.usermodel.Workbook workbook =
        org.apache.poi.ss.usermodel.WorkbookFactory.create(new ByteArrayInputStream(data))) {
      org.apache.poi.ss.usermodel.DataFormatter formatter =
          new org.apache.poi.ss.usermodel.DataFormatter(Locale.ROOT);
      formatter.setUseCachedValuesForFormulaCells(true);
      StringBuilder out = new StringBuilder();
      int rows = 0;
      for (org.apache.poi.ss.usermodel.Sheet sheet : workbook) {
        out.append("\n## ").append(sheet.getSheetName()).append("\n\n");
        for (org.apache.poi.ss.usermodel.Row row : sheet) {
          if (++rows > 100000 || row.getLastCellNum() > 500 || out.length() > 20 * 1024 * 1024) {
            throw new IllegalArgumentException("表格内容过大，请拆分后上传");
          }
          for (int i = 0; i < row.getLastCellNum(); i++) {
            out.append('|').append(formatter.formatCellValue(row.getCell(i))
                .replace("|", "\\|").replace('\n', ' '));
          }
          out.append("|\n");
        }
      }
      return out.toString();
    }
  }

    } else if (name.endsWith(".doc")) {
      try (org.apache.poi.hwpf.HWPFDocument word =
              new org.apache.poi.hwpf.HWPFDocument(new ByteArrayInputStream(data));
          org.apache.poi.hwpf.extractor.WordExtractor extractor =
              new org.apache.poi.hwpf.extractor.WordExtractor(word)) {
        result = new Parsed("doc-text", extractor.getText(), 0);
      }
    }

  private String archive(byte[] data) throws Exception {
    StringBuilder out = new StringBuilder();
    try (ZipInputStream zip = new ZipInputStream(new ByteArrayInputStream(data))) {
      ZipEntry entry;
      int count = 0;
      long total = 0;
      while ((entry = zip.getNextEntry()) != null) {
        if (entry.isDirectory()) {
          continue;
        }
        String name = entry.getName().replace('\\', '/');
        if (name.startsWith("__MACOSX/") || name.endsWith(".DS_Store")) {
          continue;
        }
        if (name.toLowerCase(Locale.ROOT).endsWith(".zip") || ++count > 50) {
          throw new IllegalArgumentException("ZIP最多包含50个文件，不支持嵌套ZIP");
        }
        byte[] bytes = zip.readNBytes(100 * 1024 * 1024 + 1);
        total += bytes.length;
        if (total > 100 * 1024 * 1024) {
          throw new IllegalArgumentException("ZIP解压后的内容不能超过100MB");
        }
        out.append("\n\n# ").append(name).append("\n\n").append(parse(bytes, name).text());
      }
    }
    return out.toString();
  }

    } else if (name.matches(".*\\.(txt|md|log|csv|json|html|htm)$")) {
      String text;
      try {
        text = StandardCharsets.UTF_8.newDecoder()
            .decode(java.nio.ByteBuffer.wrap(data)).toString();
      } catch (java.nio.charset.CharacterCodingException e) {
        text = new String(data, java.nio.charset.Charset.forName("GB18030"));
      }
      if (name.matches(".*\\.(html|htm)$")) {
        text = org.jsoup.Jsoup.parse(text).wholeText();
      }
      result = new Parsed("text", text, 0);
    }

package nexus.io.gitee;

public class GiteeDocumentParseRequest {
  private String model;
  private Boolean include_image;
  private Boolean include_image_base64;
  private Integer end_pages;
  private String output_format;
  private String prompt;
  // getter / setter 省略
}

  public GiteeTaskResponse parseDocument(byte[] data, String filename, GiteeDocumentParseRequest requestEntity) {
    String suffix = FilenameUtils.getSuffix(filename);
    String contentType = ContentTypeUtils.getContentType(suffix);
    RequestBody fileBody = RequestBody.create(data, MediaType.get(contentType));
    return parseDocument(fileBody, filename, requestEntity);
  }

  public GiteeTaskResponse getTask(String taskId) {
    String url = baseUrl + "/api/v1/task/" + taskId;
    Request request = new Request.Builder().url(url)
        .addHeader("Authorization", "Bearer " + apiKey).get().build();
    return execute(request);
  }

package nexus.io.mosskb.service.kb;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;

import com.jfinal.kit.Kv;

import dev.langchain4j.data.document.Document;
import dev.langchain4j.data.document.DocumentSplitter;
import dev.langchain4j.data.document.splitter.DocumentSplitters;
import dev.langchain4j.data.segment.TextSegment;
import lombok.extern.slf4j.Slf4j;
import nexus.io.jfinal.aop.Aop;
import nexus.io.model.result.ResultVo;
import nexus.io.model.upload.UploadResult;
import nexus.io.mosskb.utils.ExecutorServiceUtils;
import nexus.io.openai.token.OpenAiTokenizer;

@Slf4j
public class MossKbDocumentSplitService {

  public ResultVo split(byte[] data, UploadResult vo) throws IOException, InterruptedException, ExecutionException {
    String filename = vo.getName();
    DocumentParsingService.Parsed parsed;
    try {
      parsed = new DocumentParsingService().parse(data, filename);
    } catch (Exception e) {
      throw new IOException(e.getMessage(), e);
    }
    return ResultVo.ok(splitParsed(parsed, filename, vo));
  }

  /**
   * 异步拆分：立刻返回 task_id，解析与分段交给后台虚拟线程，前端轮询任务状态取结果。
   *
   * <p>大文档的解析要几分钟，占着请求线程等待既容易超时，也让前端只能干等。
   */
  public Long splitAsync(byte[] data, UploadResult vo, Long userId) {
    MossKbDocumentSplitTaskService taskService = Aop.get(MossKbDocumentSplitTaskService.class);
    Long taskId = taskService.start(userId, vo, data == null ? 0L : (long) data.length);
    ExecutorServiceUtils.getDocumentParseExecutor().execute(() -> runSplit(data, vo, taskId, taskService));
    return taskId;
  }

  /** 后台解析入口：把解析进度与最终分段写回任务表，异常统一落到任务失败原因。 */
  private void runSplit(byte[] data, UploadResult vo, Long taskId, MossKbDocumentSplitTaskService taskService) {
    String filename = vo.getName();
    try {
      DocumentParsingService.Parsed parsed = new DocumentParsingService().parse(data, filename,
          (progress) -> taskService.markProgress(taskId, progress.completed(), progress.total()));
      taskService.markTotal(taskId, parsed.pages());
      taskService.markProgress(taskId, parsed.pages(), parsed.pages());
      taskService.complete(taskId, splitParsed(parsed, filename, vo));
    } catch (Exception e) {
      taskService.fail(taskId, e.getMessage());
    }
  }

  /** 把解析结果按分词器切成段落，并组装成前端提交分段需要的结构。 */
  private List<Kv> splitParsed(DocumentParsingService.Parsed parsed, String filename, UploadResult vo) {
    String markdown = parsed.text();
    List<TextSegment> segments = split(markdown);
    Kv fileSplitResult = Kv.by("name", filename).set("id", vo.getId());
    List<Kv> contents = new ArrayList<>();
    for (TextSegment textSegment : segments) {
      contents.add(Kv.by("title", "").set("content", textSegment.text()));
    }
    fileSplitResult.set("content", contents).set("parse_strategy", parsed.strategy()).set("page_count", parsed.pages());
    if (parsed.ocrPages() > 0) {
      fileSplitResult.set("ocr_page_count", parsed.ocrPages());
    }
    if (parsed.failedPages() > 0) {
      fileSplitResult.set("failed_page_count", parsed.failedPages());
    }
    return List.of(fileSplitResult);
  }

  public List<TextSegment> split(String markdown) {
    Document document = new Document(markdown);
    // 使用较大的块大小（2000）和相同的重叠（400）
    DocumentSplitter splitter = DocumentSplitters.recursive(2000, 400, new OpenAiTokenizer());
    List<TextSegment> segments = splitter.split(document);
    return segments;
  }
}

package nexus.io.mosskb.controller;

import java.util.ArrayList;
import java.util.List;

import com.jfinal.kit.Kv;

import nexus.io.annotation.Get;
import nexus.io.annotation.Post;
import nexus.io.annotation.RequestPath;
import nexus.io.jfinal.aop.Aop;
import nexus.io.mosskb.service.SystemFileService;
import nexus.io.mosskb.service.kb.MossKbDocumentSplitService;
import nexus.io.mosskb.service.kb.MossKbDocumentSplitTaskService;
import nexus.io.mosskb.vo.MossKbDocumentSplitTaskVo;
import nexus.io.model.result.ResultVo;
import nexus.io.model.upload.UploadFile;
import nexus.io.model.upload.UploadResult;
import nexus.io.tio.boot.http.TioRequestContext;

@RequestPath("/api/dataset/document")
public class ApiDatasetDocumentController {

  /**
   * 分段预览：只负责上传与建任务，解析在后台线程逐页推进。
   *
   * <p>大文档解析要几分钟，请求线程不再等待，前端拿 task_id 轮询 {@link #splitTask(Long)} 取结果。
   */
  @Post("/split")
  public ResultVo split(nexus.io.tio.http.common.HttpRequest request) {
    Object[] files = request.getParams().get("file");
    if (files == null || files.length == 0) {
      return ResultVo.fail("请求体中未找到文件");
    }
    Long userId = TioRequestContext.getUserIdLong();
    List<Long> taskIds = new ArrayList<>();
    List<Kv> failures = new ArrayList<>();
    for (Object item : files) {
      if (!(item instanceof UploadFile)) {
        return ResultVo.fail("文件格式不正确");
      }
      UploadFile file = (UploadFile) item;
      UploadResult uploaded = Aop.get(SystemFileService.class).upload(file, "default", "default");
      if (uploaded == null) {
        failures.add(Kv.by("name", file.getName()).set("message", "文件保存失败"));
        continue;
      }
      try {
        taskIds.add(Aop.get(MossKbDocumentSplitService.class).splitAsync(file.getData(), uploaded, userId));
      } catch (Exception e) {
        failures.add(Kv.by("name", file.getName()).set("message", e.getMessage()));
      }
    }
    if (taskIds.isEmpty()) {
      return ResultVo.fail(failures.isEmpty() ? "没有可解析的文件" : String.valueOf(failures.get(0).get("message")));
    }
    return ResultVo.ok(Kv.by("task_id_list", taskIds).set("failures", failures));
  }

  /**
   * 分段预览任务状态。
   *
   * <p>running 时 data 为空，前端继续轮询；success 时 data 是分段结果；failed 时按失败返回，前端直接提示。
   */
  @Get("/split/task/{taskId}")
  public ResultVo splitTask(Long taskId) {
    Long userId = TioRequestContext.getUserIdLong();
    MossKbDocumentSplitTaskVo task = Aop.get(MossKbDocumentSplitTaskService.class).get(userId, taskId);
    if (task == null) {
      return ResultVo.fail("任务不存在或无权访问");
    }
    if (MossKbDocumentSplitTaskService.STATUS_FAILED.equals(task.getStatus())) {
      return ResultVo.fail(task.getErrorMessage() == null ? "文档解析失败" : task.getErrorMessage());
    }
    if (MossKbDocumentSplitTaskService.STATUS_SUCCESS.equals(task.getStatus())) {
      return ResultVo.ok(task.getResult() == null ? new ArrayList<>() : task.getResult());
    }
    return ResultVo.ok();
  }

package nexus.io.mosskb.controller;

import java.util.List;

import nexus.io.annotation.Post;
import nexus.io.annotation.RequestPath;
import nexus.io.jfinal.aop.Aop;
import nexus.io.mosskb.service.kb.MossKbParagraphSplitService;
import nexus.io.mosskb.vo.ParagraphBatchVo;
import nexus.io.model.result.ResultVo;
import nexus.io.tio.boot.http.TioRequestContext;
import nexus.io.tio.http.common.HttpRequest;
import nexus.io.tio.utils.json.JsonUtils;

@RequestPath("/api/dataset")
public class ApiDatasetController {

  @Post("/{id}/document/_bach")
  public ResultVo batch(Long id, HttpRequest request) {
    String bodyString = request.getBodyString();
    Long userId = TioRequestContext.getUserIdLong();
    List<ParagraphBatchVo> list = JsonUtils.parseArray(bodyString, ParagraphBatchVo.class);
    return Aop.get(MossKbParagraphSplitService.class).batch(userId, id, list);
  }
}

package nexus.io.mosskb.vo;

import java.util.List;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;

@Data
@NoArgsConstructor
@AllArgsConstructor
@Accessors(chain = true)
public class ParagraphBatchVo {
  private Long id;
  private String name;
  private List<Paragraph> paragraphs;
}

package nexus.io.mosskb.service.kb;

import java.io.ByteArrayOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;
import org.apache.pdfbox.pdmodel.PDDocument;
import org.apache.pdfbox.pdmodel.PDPage;
import org.apache.pdfbox.pdmodel.PDPageContentStream;
import org.apache.pdfbox.pdmodel.font.PDType1Font;
import org.junit.Test;
import static org.junit.Assert.*;

public class DocumentParsingServiceTest {
  private static class Parser extends DocumentParsingService {
    int ocrCalls;
    @Override
    protected String ocr(byte[] data, String filename) {
      ocrCalls++;
      return "Scanned content " + ocrCalls;
    }
    @Override
    protected String ocrPdfPage(String sourceHash, int page, byte[] data) {
      return ocr(data, "page.pdf");
    }
  }

  @Test
  public void pdfUsesTextAndOcrPerPageInOriginalOrder() throws Exception {
    try (PDDocument pdf = new PDDocument(); ByteArrayOutputStream out = new ByteArrayOutputStream()) {
      PDPage first = new PDPage();
      pdf.addPage(first);
      try (PDPageContentStream canvas = new PDPageContentStream(pdf, first)) {
        canvas.beginText();
        canvas.setFont(PDType1Font.HELVETICA, 12);
        canvas.newLineAtOffset(40, 700);
        canvas.showText("Native administrative review document with sufficient readable text for extraction.");
        canvas.endText();
      }
      pdf.addPage(new PDPage());
      pdf.save(out);
      Parser parser = new Parser();
      DocumentParsingService.Parsed result = parser.parse(out.toByteArray(), "misleading.txt");
      assertEquals("pdf-mixed", result.strategy());
      assertEquals(2, result.pages());
      assertEquals(1, parser.ocrCalls);
      assertTrue(result.text().indexOf("Native administrative") < result.text().indexOf("Scanned content"));
    }
  }
}