添加对应115文件流读取的支持

chivehao · chivehao · commit aceceef7ad36 · 2025-12-28T21:00:04.000+08:00
diff --git a/CHANGELOG.MD b/CHANGELOG.MD
@@ -2,6 +2,10 @@
 
 更新日志文档，版本顺序从新到旧，最新版本在最前(上)面。
 
+# 1.0.7
+
+- 添加对应115文件流读取的支持
+
 # 1.0.6
 
 - 修复问题
diff --git a/build.gradle b/build.gradle
@@ -14,7 +14,7 @@ ext {
     springContext = '6.0.3'
     pf4j = '3.8.0'
     lombok = '1.18.24'
-    libFile = 'lib/api-1.0.4.jar'
+    libFile = 'lib/api-1.0.5.jar'
 }
 
 
diff --git a/gradle.properties b/gradle.properties
@@ -1,3 +1,3 @@
 group=run.ikaros.plugin
 description=A pan115 plugin for ikaros.
-version=1.0.6
+version=1.0.7
diff --git a/lib/api-1.0.5-sources.jar b/lib/api-1.0.5-sources.jar
diff --git a/lib/api-1.0.5.jar b/lib/api-1.0.5.jar
diff --git a/src/main/java/run/ikaros/plugin/pan115/Pan115AttachmentDriverFetcher.java b/src/main/java/run/ikaros/plugin/pan115/Pan115AttachmentDriverFetcher.java
@@ -2,21 +2,18 @@
 
 import lombok.extern.slf4j.Slf4j;
 import org.pf4j.Extension;
-import org.springframework.stereotype.Component;
+import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.util.Assert;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
 import run.ikaros.api.core.attachment.Attachment;
 import run.ikaros.api.core.attachment.AttachmentDriver;
 import run.ikaros.api.core.attachment.AttachmentDriverFetcher;
 import run.ikaros.api.core.attachment.AttachmentDriverOperate;
 import run.ikaros.api.infra.utils.StringUtils;
-import run.ikaros.api.wrap.PagingWrap;
 import run.ikaros.api.store.enums.AttachmentDriverType;
 import run.ikaros.api.store.enums.AttachmentType;
 import run.ikaros.plugin.pan115.model.Pan115Attachment;
-import run.ikaros.plugin.pan115.model.Pan115Folder;
 import run.ikaros.plugin.pan115.model.Pan115Path;
 import run.ikaros.plugin.pan115.repository.Pan115Repository;
 import run.ikaros.plugin.pan115.utils.FileUtils;
@@ -165,6 +162,16 @@ public Mono<String> parseDownloadUrl(Attachment attachment) {
         }
     }
 
+    @Override
+    public Flux<DataBuffer> getSteam(Attachment attachment) {
+        Assert.notNull(attachment, "'attachment' must not null.");
+        AttachmentType type = attachment.getType();
+        if (AttachmentType.Driver_Directory.equals(type)) return Flux.empty();
+        Long driverId = attachment.getDriverId();
+        Mono<AttachmentDriver> checkoutMono = checkoutToken(driverId);
+        return checkoutMono.flatMapMany(driver -> pan115Repository.openUFileSteam(attachment.getUrl()));
+    }
+
     private void applyPan115Token(AttachmentDriver driver) {
         Assert.notNull(driver, "'driver' must not null.");
         pan115Repository.refreshToken(driver);
diff --git a/src/main/java/run/ikaros/plugin/pan115/repository/DefaultPan115Repository.java b/src/main/java/run/ikaros/plugin/pan115/repository/DefaultPan115Repository.java
@@ -3,16 +3,24 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.http.*;
-import org.springframework.http.client.ClientHttpRequestExecution;
-import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.http.client.ClientHttpRequest;
+import org.springframework.http.client.SimpleClientHttpRequestFactory;
 import org.springframework.stereotype.Component;
 import org.springframework.util.Assert;
 import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
 import org.springframework.web.client.HttpClientErrorException;
+import org.springframework.web.client.RequestCallback;
+import org.springframework.web.client.ResponseExtractor;
 import org.springframework.web.client.RestTemplate;
 import org.springframework.web.util.UriComponentsBuilder;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
+import reactor.core.scheduler.Schedulers;
 import run.ikaros.api.core.attachment.AttachmentDriver;
 import run.ikaros.plugin.pan115.Pan115Const;
 import run.ikaros.plugin.pan115.exception.Pan115RequestFailException;
@@ -22,18 +30,25 @@
 import run.ikaros.plugin.pan115.model.Pan115Result;
 import run.ikaros.plugin.pan115.utils.JsonUtils;
 
+import java.io.IOException;
+import java.io.InputStream;
 import java.net.URI;
-import java.net.URLEncoder;
+import java.net.URLDecoder;
 import java.nio.charset.StandardCharsets;
 import java.time.LocalDateTime;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static run.ikaros.plugin.pan115.Pan115Const.*;
 
 @Slf4j
 @Component
 public class DefaultPan115Repository implements Pan115Repository {
-    private final RestTemplate restTemplate = new RestTemplate();
+    private final RestTemplate restTemplate = new RestTemplate(new SimpleClientHttpRequestFactory());
+    private final DataBufferFactory dataBufferFactory = new DefaultDataBufferFactory();
     private final HttpHeaders headers = new HttpHeaders();
 
     @Override
@@ -238,8 +253,126 @@ public String openVideoPlay(String pickCode) {
         }
     }
 
-    @Scheduled(fixedDelay = 5000)  // 5秒
-    public void fixedDelayTask() {
-        log.info("Fixed delay task - " + new Date());
+    @Override
+    public Flux<DataBuffer> openUFileSteam(String pickCode) {
+        Assert.hasText(pickCode, "'pickCode' must has text.");
+        final String uFileDownUrl = openUFileDownUrl(pickCode);
+        return streamFileWithRestTemplate(uFileDownUrl);
+        // return Flux.create(sink -> {
+        //     Schedulers.boundedElastic().schedule(() -> {
+        //         try {
+        //             ResponseEntity<byte[]> response = restTemplate.getForEntity(
+        //                     uFileDownUrl, byte[].class);
+        //             if (response.getStatusCode() == HttpStatus.OK &&
+        //                     response.getBody() != null) {
+        //                 // 一次性发送所有数据（适合小文件）
+        //                 DataBuffer dataBuffer = dataBufferFactory.allocateBuffer(
+        //                         response.getBody().length);
+        //                 dataBuffer.write(response.getBody());
+        //                 sink.next(dataBuffer);
+        //                 sink.complete();
+        //             } else {
+        //                 sink.error(new RuntimeException(
+        //                         "请求失败，状态码: " + response.getStatusCode()));
+        //             }
+        //         } catch (Exception e) {
+        //             sink.error(e);
+        //         }
+        //     });
+        // });
+    }
+
+
+    /**
+     * 使用 RestTemplate 获取文件流，转换为 Flux<DataBuffer>
+     *
+     * @param fileUrl   文件URL
+     * @param chunkSize 分块大小（字节），默认为 8192
+     * @return Flux<DataBuffer>
+     */
+    public Flux<DataBuffer> streamFileWithRestTemplate(String fileUrl, int chunkSize) {
+        if (chunkSize <= 0) {
+            chunkSize = 8192; // 默认8KB
+        }
+
+        final int finalChunkSize = chunkSize;
+
+        return Flux.create(sink -> {
+            // 使用 boundedElastic 调度器执行阻塞操作
+            Schedulers.boundedElastic().schedule(() -> {
+                try {
+                    String newUrl = fileUrl;
+                    if (StringUtils.isNoneBlank(newUrl)) {
+                        newUrl = URLDecoder.decode(newUrl, StandardCharsets.UTF_8);
+                    }
+                    log.debug("Open stream for url: {}", newUrl);
+                    restTemplate.execute(newUrl, HttpMethod.GET, new RequestCallback() {
+                                @Override
+                                public void doWithRequest(ClientHttpRequest request) throws IOException {
+                                    request.getHeaders().addAll(headers);
+                                }
+                            },
+                            (ResponseExtractor<Void>) response -> {
+                                if (response.getStatusCode() != HttpStatus.OK) {
+                                    sink.error(new RuntimeException(
+                                            "请求失败，状态码: " + response.getStatusCode()));
+                                    return null;
+                                }
+
+                                try (InputStream inputStream = response.getBody()) {
+                                    byte[] buffer = new byte[finalChunkSize];
+                                    int bytesRead;
+                                    AtomicBoolean isCanceled = new AtomicBoolean(false);
+
+                                    sink.onCancel(() -> {
+                                        isCanceled.set(true);
+                                        try {
+                                            inputStream.close();
+                                        } catch (IOException e) {
+                                            // 忽略关闭异常
+                                        }
+                                    });
+
+                                    while (!isCanceled.get() &&
+                                            (bytesRead = inputStream.read(buffer)) != -1) {
+
+                                        // 创建 DataBuffer
+                                        DataBuffer dataBuffer = dataBufferFactory.allocateBuffer(bytesRead);
+                                        dataBuffer.write(buffer, 0, bytesRead);
+
+                                        // 发布数据块
+                                        sink.next(dataBuffer);
+
+                                        // 检查是否取消
+                                        if (sink.isCancelled()) {
+                                            log.debug("Close stream for url: {}", fileUrl);
+                                            break;
+                                        }
+                                    }
+
+                                    if (!isCanceled.get()) {
+                                        sink.complete();
+                                    }
+                                } catch (Exception e) {
+                                    if (!sink.isCancelled()) {
+                                        sink.error(e);
+                                    }
+                                }
+                                return null;
+                            });
+                } catch (Exception e) {
+                    if (!sink.isCancelled()) {
+                        sink.error(e);
+                    }
+                }
+            });
+        }, FluxSink.OverflowStrategy.BUFFER);
+    }
+
+    /**
+     * 简化版本：使用默认分块大小
+     */
+    public Flux<DataBuffer> streamFileWithRestTemplate(String fileUrl) {
+        return streamFileWithRestTemplate(fileUrl, 8192);
     }
 }
diff --git a/src/main/java/run/ikaros/plugin/pan115/repository/Pan115Repository.java b/src/main/java/run/ikaros/plugin/pan115/repository/Pan115Repository.java
@@ -1,5 +1,7 @@
 package run.ikaros.plugin.pan115.repository;
 
+import org.springframework.core.io.buffer.DataBuffer;
+import reactor.core.publisher.Flux;
 import run.ikaros.api.core.attachment.AttachmentDriver;
 import run.ikaros.plugin.pan115.model.Pan115Attachment;
 import run.ikaros.plugin.pan115.model.Pan115Folder;
@@ -22,5 +24,5 @@ List<Pan115Attachment> openUFileFiles(String cid, Integer limit, Integer cur,
 
     String openVideoPlay(String pickCode);
 
-//    void openVideoPlay(String pc);
+    Flux<DataBuffer> openUFileSteam(String pickCode);
 }
diff --git a/src/main/resources/plugin.yaml b/src/main/resources/plugin.yaml
@@ -4,8 +4,8 @@ name: PluginPan115
 # plugin entry class that extends BasePlugin
 clazz: run.ikaros.plugin.pan115.Pan115Plugin
 # plugin 'version' is a valid semantic version string (see semver.org).
-version: 1.0.6
-requires: ">=1.0.4"
+version: 1.0.7
+requires: ">=1.0.5"
 author:
   name: Ikaros OSS Team
   website: https://github.com/ikaros-dev

Original file line number	Diff line number	Diff line change
`@@ -14,7 +14,7 @@ ext {`
`14`	`14`	`springContext = '6.0.3'`
`15`	`15`	`pf4j = '3.8.0'`
`16`	`16`	`lombok = '1.18.24'`
`17`		`- libFile = 'lib/api-1.0.4.jar'`
	`17`	`+ libFile = 'lib/api-1.0.5.jar'`
`18`	`18`	`}`
`19`	`19`
`20`	`20`