4dkankan-utils/4dkankan-utils-fyun-s3/src/main/java/com/fdkankan/fyun/s3/S3FileService.java

package com.fdkankan.fyun.s3;

import cn.hutool.core.collection.CollUtil;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.HttpMethod;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.amazonaws.services.s3.model.*;
import com.fdkankan.fyun.constant.FYunTypeEnum;
import com.fdkankan.fyun.face.AbstractFYunFileService;
import com.fdkankan.fyun.util.FileInfoVo;
import com.fdkankan.fyun.util.MD5Checksum;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.io.*;
import java.math.BigDecimal;
import java.net.URL;
import java.util.*;
import java.util.stream.Collectors;

@Component
@ConditionalOnProperty(name = "fyun.type",havingValue = "aws")
public class S3FileService extends AbstractFYunFileService {

    private Logger log = LoggerFactory.getLogger(this.getClass().getName());

    public static AmazonS3 amazonS3(){
        BasicAWSCredentials awsCreds = new BasicAWSCredentials("AKIAWCV5QFZ3ZNELKYUY", "epS5ghyR4LJ7rxk/qJO9ZYh6m9Oz6g5haKDu4yws");
        ClientConfiguration clientConfiguration = new ClientConfiguration();
        clientConfiguration.setMaxErrorRetry(4);
        clientConfiguration.setMaxConnections(100);
        AwsClientBuilder.EndpointConfiguration endpointConfiguration = new AwsClientBuilder.EndpointConfiguration("test-4dkankan.s3.eu-west-2.amazonaws.com",Regions.EU_WEST_2.getName());
        return AmazonS3ClientBuilder.standard()
                .withCredentials(new AWSStaticCredentialsProvider(awsCreds))
                .withEndpointConfiguration(endpointConfiguration)
                .withClientConfiguration(clientConfiguration)
//                .withRegion(Regions.EU_WEST_2).
                        .build();
    }

    @Autowired
    private AmazonS3 s3;

    @Override
    public String uploadFile(String bucket, byte[] data, String remoteFilePath){
        try {
            ObjectMetadata metadata = new ObjectMetadata();
            PutObjectRequest request = new PutObjectRequest(bucket, remoteFilePath, new ByteArrayInputStream(data), metadata);
            request.withCannedAcl(CannedAccessControlList.PublicRead);
            s3.putObject(request);
        } catch (Exception e) {
            log.error("s3上传文件失败", e);
        }
        return null;
    }

    @Override
    public String uploadFile(String bucket, String filePath, String remoteFilePath){
        return uploadFile(bucket, filePath, remoteFilePath,null);
    }

    @Override
    public String uploadFile(String bucket, InputStream inputStream, String remoteFilePath) {
        try {
            s3.putObject(bucket, remoteFilePath, inputStream, null);
        } catch (Exception e) {
            log.error("文件流上传失败,目标路径：{}", remoteFilePath);
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public String uploadFile(String bucket, String filePath, String remoteFilePath,Map<String, String> headers){
        try {
            File file = new File(filePath);
            if (!file.exists()) {
                log.error("要上传的文件不存在：" + filePath);
                return null;
            }

            // 设置文件并设置公读
            ObjectMetadata metadata = new ObjectMetadata();
            if (filePath.contains(".jpg")) {
                metadata.setContentType("image/jpeg");
            }
            if (filePath.contains(".png")) {
                metadata.setContentType("image/png");
            }
            if(org.apache.commons.lang3.ObjectUtils.isNotEmpty(headers)){
                for (Map.Entry<String, String> header : headers.entrySet()) {
                    metadata.setHeader(header.getKey(),header.getValue());
                }
            }
            PutObjectRequest request = new PutObjectRequest(bucket, remoteFilePath, file);
            request.withCannedAcl(CannedAccessControlList.PublicRead);
            request.withMetadata(metadata);

            // 上传文件
            PutObjectResult putObjectResult = s3.putObject(request);
            if (StringUtils.isNotEmpty(putObjectResult.getETag())) {
                log.info("s3上传文件成功：" + remoteFilePath);
            }
        } catch (Exception e) {
            log.error("文件上传失败：{}",filePath);
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public String uploadFileByCommand(String bucket, String filePath, String remoteFilePath){
        try {
            String optType = new File(filePath).isDirectory() ? "folder" : "file";
            String command = String.format(fYunConstants.UPLOAD_SH, bucket, filePath, remoteFilePath, FYunTypeEnum.AWS.code(), optType);
            log.info("开始上传文件, ossPath:{}, srcPath:{}", remoteFilePath, filePath);
            callshell(command);
        } catch (Exception e) {
            log.error("上传文件失败, ossPath:{}, srcPath:{}", remoteFilePath, filePath);
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void copyFileBetweenBucketParallel(String sourceBucketName, String sourcePath, String targetBucketName, String targetPath) {
        try {
            List<String> files = listRemoteFiles(sourceBucketName, sourcePath);
            if (ObjectUtils.isEmpty(files)) {
                return;
            }
            files.parallelStream().forEach(file -> {
                CopyObjectRequest request = new CopyObjectRequest(sourceBucketName, file, targetBucketName, file.replace(sourcePath, targetPath));
                request.withCannedAccessControlList(CannedAccessControlList.PublicRead);
                s3.copyObject(request);
            });
        } catch (Exception e) {
            log.error("列举文件目录失败,key=" + sourcePath);
        }
    }

    @Override
    public void downloadFileByCommand(String bucket, String filePath, String remoteFilePath) {
        try {
            String optType = remoteFilePath.contains(".") ? "file" : "folder";
            String command = String.format(fYunConstants.DOWNLOAD_SH, bucket, remoteFilePath, filePath, FYunTypeEnum.AWS.code(), optType);
            log.info("开始下载文件, ossPath:{}, srcPath:{}", remoteFilePath, filePath);
            callshell(command);
        } catch (Exception e) {
            log.error("上传文件失败, ossPath:{}, srcPath:{}", remoteFilePath, filePath);
            e.printStackTrace();
        }
    }

    @Override
    public void downloadByCommand(String bucket, String filePath, String remoteFilePath, boolean isDir) {
        try {
            String optType = isDir ? "folder" : "file";
            String command = String.format(fYunConstants.DOWNLOAD_SH, bucket, remoteFilePath, filePath, FYunTypeEnum.AWS.code(), optType);
            log.info("开始下载文件, ossPath:{}, srcPath:{}", remoteFilePath, filePath);
            callshell(command);
        } catch (Exception e) {
            log.error("上传文件失败, ossPath:{}, srcPath:{}", remoteFilePath, filePath);
            e.printStackTrace();
        }
    }

    @Override
    public void deleteFile(String bucket, String remoteFilePath){
        if (remoteFilePath.startsWith("/")) {
            remoteFilePath = remoteFilePath.substring(1);
        }
        try {
            s3.deleteObject(bucket, remoteFilePath);
        } catch (Exception e) {
            log.error("s3删除文件失败,key=" + remoteFilePath, e);
        }
    }

    @Override
    public void deleteFolder(String bucket, String remoteFolderPath){
        try {
            if (!remoteFolderPath.endsWith("/")) {
                remoteFolderPath = remoteFolderPath + "/";
            }
            log.info("开始删除文件夹：{}", remoteFolderPath);
            int maxKeys = 1000;
            String nextMaker = null;

            ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
            listObjectsRequest.setBucketName(bucket);
            listObjectsRequest.setPrefix(remoteFolderPath);
            listObjectsRequest.setMaxKeys(maxKeys);

            ObjectListing objectListing;
            DeleteObjectsRequest multiObjectDeleteRequest = new DeleteObjectsRequest(bucket).withQuiet(false);
            do {
                listObjectsRequest.setMarker(nextMaker);
                objectListing = s3.listObjects(listObjectsRequest);
                List<S3ObjectSummary> objectSummaries = objectListing.getObjectSummaries();
                List<DeleteObjectsRequest.KeyVersion> keys = objectSummaries.stream().map(summary -> new DeleteObjectsRequest.KeyVersion(summary.getKey())).collect(Collectors.toList());
                if (!ObjectUtils.isEmpty(keys)) {
                    multiObjectDeleteRequest.setKeys(keys);
                    s3.deleteObjects(multiObjectDeleteRequest);
                }
                nextMaker = objectListing.getNextMarker();
            } while (objectListing.isTruncated());
        } catch (Exception e) {
            log.error("删除aws文件失败，path=" + remoteFolderPath, e);
        }
    }

    @Override
    public void uploadMulFiles(String bucket, Map<String, String> filepaths){
        try {
            for (Map.Entry<String, String> entry : filepaths.entrySet()) {
                uploadFile(bucket, entry.getKey(), entry.getValue(),null);
            }
        } catch (Exception e) {
            log.error("OSS批量上传文件失败！");
        }
    }

    @Override
    public List<String> listRemoteFiles(String bucket, String sourcePath) {
        List<String> keyList = new ArrayList<>();
        try {
            boolean flag = true;
            String nextMaker = null;
            ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
            listObjectsRequest.setBucketName(bucket);
            listObjectsRequest.setPrefix(sourcePath);
            listObjectsRequest.setMaxKeys(200);

            do {
                listObjectsRequest.setMarker(nextMaker);
                ObjectListing objectListing = s3.listObjects(listObjectsRequest);
                List<S3ObjectSummary> objectSummaries = objectListing.getObjectSummaries();
                List<String> collect = objectSummaries.stream().map(S3ObjectSummary::getKey).collect(Collectors.toList());
                if (CollUtil.isNotEmpty(collect)) {
                    keyList.addAll(collect);
                }
                nextMaker = objectListing.getNextMarker();
                flag = objectListing.isTruncated();
            } while (flag);
        } catch (Exception e) {
            log.error("获取文件列表失败，path=" + sourcePath, e);
            e.printStackTrace();
        }
        return keyList;
    }

    @Override
    public void copyFileBetweenBucket(String sourceBucketName, String sourcePath, String targetBucketName, String targetPath){
        try {
            List<String> files = listRemoteFiles(sourceBucketName, sourcePath);
            if (ObjectUtils.isEmpty(files)) {
                return;
            }
            files.stream().forEach(file -> {
				CopyObjectRequest request = new CopyObjectRequest(sourceBucketName, file, targetBucketName, file.replace(sourcePath, targetPath));
				request.withCannedAccessControlList(CannedAccessControlList.PublicRead);
				s3.copyObject(request);
            });
        } catch (Exception e) {
            log.error("列举文件目录失败,key=" + sourcePath);
        }
    }

    @Override
    public void copyFilesBetweenBucket(String sourceBucketName, String targetBucketName, Map<String, String> pathMap){
        if (ObjectUtils.isEmpty(pathMap)) {
            return;
        }
        try {
            for (Map.Entry<String, String> entry : pathMap.entrySet()) {
                copyFileBetweenBucket(sourceBucketName, entry.getKey(), targetBucketName, entry.getValue());
            }
        } catch (Exception e) {
            log.error("批量复制文件失败!");
        }
    }

    @Override
    public String getFileContent(String bucketName, String remoteFilePath){
        try (S3Object object = s3.getObject(bucketName,remoteFilePath);
             S3ObjectInputStream inputStream = object.getObjectContent()){
			StringBuilder content = new StringBuilder();
			try(BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))){
				while (true) {
					String line = reader.readLine();
					if (line == null) break;
					content.append(line);
				}
			} catch (IOException e) {
				log.error("读取aws文件流失败", e);
			}
			return content.toString();
        } catch (Exception e) {
            log.error("获取文件内容失败：{}", remoteFilePath, e);
            return null;
        }
    }

    @Override
    public boolean fileExist(String bucket, String objectName) {
        try {
            return s3.doesObjectExist(bucket, objectName);
        } catch (Exception e) {
            log.error("判断文件是否存在失败：{}", objectName);
            return false;
        }
    }

    @Override
    public void downloadFile(String bucket, String remoteFilePath, String localPath) {
        try {
            File localFile = new File(localPath);
            if (!localFile.getParentFile().exists()) {
                localFile.getParentFile().mkdirs();
            }
            if(localFile.isDirectory()){
                String fileName = remoteFilePath.substring(remoteFilePath.lastIndexOf("/")+1);
                log.info("未配置文件名，使用默认文件名：{}",fileName);
                localPath = localPath.concat(File.separator).concat(fileName);
            }
			GetObjectRequest request = new GetObjectRequest(bucket, remoteFilePath);
			s3.getObject(request,new File(localPath));
        } catch (Throwable throwable) {
            log.error("文件下载失败：{}", remoteFilePath);
            throwable.printStackTrace();
        }
    }

    @Override
    public URL getPresignedUrl(String bucket,String url) {
        java.util.Date expiration = new java.util.Date();
        long expTimeMillis = expiration.getTime();
        expTimeMillis += 1000 * 60 * 60 * 8;
        expiration.setTime(expTimeMillis);
        GeneratePresignedUrlRequest generatePresignedUrlRequest = new GeneratePresignedUrlRequest(bucket, url)
                .withMethod(HttpMethod.PUT).withExpiration(expiration);
        return s3.generatePresignedUrl(generatePresignedUrlRequest);
    }

    @Override
    public long getSubFileNums(String bucket, String url) {
        long totalFileNums = 0;
        try {
            boolean flag = true;
            String nextMaker = null;
            ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
            listObjectsRequest.setBucketName(bucket);
            listObjectsRequest.setPrefix(url);
            listObjectsRequest.setMaxKeys(200);

            do {
                listObjectsRequest.setMarker(nextMaker);
                ObjectListing objectListing = s3.listObjects(listObjectsRequest);
                List<S3ObjectSummary> objectSummaries = objectListing.getObjectSummaries();
                List<String> collect = objectSummaries.stream().map(S3ObjectSummary::getKey).collect(Collectors.toList());
                totalFileNums = totalFileNums + collect.size();
                nextMaker = objectListing.getNextMarker();
                flag = objectListing.isTruncated();
            } while (flag);
        } catch (Exception e) {
            log.error("获取文件数量失败，path=" + url, e);
            e.printStackTrace();
        }
        return totalFileNums;
    }

    @Override
    public void restoreFolder(String bucket, String folderName, Integer priority) {
        List<String> objectList = this.listRemoteFiles(bucket, folderName);
        if(CollUtil.isEmpty(objectList)){
            return;
        }
        objectList.parallelStream().forEach(objectName -> {
            this.restoreFile(bucket, objectName, priority);
        });
    }

    @Override
    public void restoreFile(String bucket, String objectName, Integer priority){
        ObjectMetadata objectMetadata = s3.getObjectMetadata(bucket, objectName);

        // 校验Object是否为归档类型Object。
        StorageClass storageClass = StorageClass.fromValue(objectMetadata.getStorageClass());
        if (storageClass == StorageClass.Glacier) {
            Tier restoreTier = null;
            switch (priority){
                case 1 :
                    restoreTier = Tier.Expedited;
                    break;
                case 2 :
                    restoreTier = Tier.Standard;
                    break;
                default:
                    restoreTier = Tier.Bulk;
            }
            RestoreObjectRequest requestRestore = new RestoreObjectRequest(bucket, objectName, 1);
            GlacierJobParameters glacierJobParameters = new GlacierJobParameters();
            glacierJobParameters.setTier(restoreTier);
            requestRestore.setGlacierJobParameters(glacierJobParameters);
            //开始解冻
            s3.restoreObjectV2(requestRestore);
        }
    }

    @Override
    public Boolean checkStore(String bucket, String url) {
        ObjectMetadata objectMetadata =  s3.getObjectMetadata(bucket, url);
        return !isRestoreCompleted(objectMetadata);
    }

    private boolean isRestoreCompleted(ObjectMetadata objectMetadata){
        Date restoreExpirationTime = objectMetadata.getRestoreExpirationTime();
        if(Objects.nonNull(restoreExpirationTime) && restoreExpirationTime.after(Calendar.getInstance().getTime())){
            return true;
        }
        return false;
    }

    @Override
    public void restoreFolder(String bucket, String url) {
        ObjectMetadata objectMetadata ;

        List<String> objectList = this.listRemoteFiles(bucket, url);
        if(CollUtil.isEmpty(objectList)){
            return;
        }
        for (String objectName : objectList) {
            objectMetadata = s3.getObjectMetadata(bucket, objectName);
            // 校验Object是否为归档类型Object。
            StorageClass storageClass = StorageClass.fromValue(objectMetadata.getStorageClass());
            if (storageClass == StorageClass.Glacier) {
                // 解冻Object。
                RestoreObjectRequest requestRestore = new RestoreObjectRequest(bucket, url, 1);
                s3.restoreObjectV2(requestRestore);
            }
        }
    }

    @Override
    public Integer getRestoreFolderProcess(String bucket, String url) {
        ObjectMetadata objectMetadata ;

        List<String> objectList = this.listRemoteFiles(bucket, url);
        if(CollUtil.isEmpty(objectList)){
            return 100;
        }
        List<String> restoreFileList = new ArrayList<>();
        for (String objectName : objectList) {
            objectMetadata = s3.getObjectMetadata(bucket, objectName);
            if(this.isRestoreCompleted(objectMetadata)){

                restoreFileList.add(objectName);
            }
        }
        if(objectList.size() <= restoreFileList.size() ){
            return 100;
        }
        BigDecimal rite = new BigDecimal(restoreFileList.size()).divide(new BigDecimal(objectList.size()),2,BigDecimal.ROUND_HALF_UP);
        BigDecimal multiply = rite.multiply(new BigDecimal(100));
        return multiply.intValue();
    }

    @Override
    public Long getSpace(String bucket, String key) {
        List<String> keyList = new ArrayList<>();
        Long total = 0L;
        boolean flag = true;
        String nextMaker = null;
        ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
        listObjectsRequest.setBucketName(bucket);
        listObjectsRequest.setPrefix(key);
        listObjectsRequest.setMaxKeys(200);

        do {
            listObjectsRequest.setMarker(nextMaker);
            ObjectListing objectListing = s3.listObjects(listObjectsRequest);
            List<S3ObjectSummary> objectSummaries = objectListing.getObjectSummaries();
            Long space = objectSummaries.stream().mapToLong(S3ObjectSummary::getSize).sum();
            total += space;
            nextMaker = objectListing.getNextMarker();
            flag = objectListing.isTruncated();
        } while (flag);

        return total;
    }

    @Override
    public void copyFileToArchive(String sourceBucketName, String sourcePath, String targetBucketName, String targetPath) {
        try {
            List<String> files = listRemoteFiles(sourceBucketName, sourcePath);
            if (ObjectUtils.isEmpty(files)) {
                return;
            }
            files.stream().forEach(file -> {
                CopyObjectRequest request = new CopyObjectRequest(sourceBucketName, file, targetBucketName, file.replace(sourcePath, targetPath));
                request.setStorageClass(StorageClass.Glacier);
                s3.copyObject(request);
            });
        } catch (Exception e) {
            log.error("复制文件或目录失败,key:" + sourcePath, e);
        }
    }

    public static void main(String[] args) {

        AmazonS3 amazonS3 = amazonS3();

        List<String> keyList = new ArrayList<>();
        boolean flag = true;
        String nextMaker = null;
        ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
        listObjectsRequest.setBucketName("test-4dkankan");
        listObjectsRequest.setPrefix("test/KJ-t-ekRwoMX8XP");
        listObjectsRequest.setMaxKeys(200);

        do {
            listObjectsRequest.setMarker(nextMaker);
            ObjectListing objectListing = amazonS3.listObjects(listObjectsRequest);
            List<S3ObjectSummary> objectSummaries = objectListing.getObjectSummaries();
            List<String> collect = objectSummaries.stream().map(S3ObjectSummary::getKey).collect(Collectors.toList());
            if (CollUtil.isNotEmpty(collect)) {
                keyList.addAll(collect);
            }
            nextMaker = objectListing.getNextMarker();
            flag = objectListing.isTruncated();
        } while (flag);

        keyList.stream().forEach(file -> {
            CopyObjectRequest request = new CopyObjectRequest("test-4dkankan", file, "test-4dkk-bak", file.replace("test/KJ-t-ekRwoMX8XP", "test/KJ-t-ekRwoMX8XP"));
            request.setStorageClass(StorageClass.DeepArchive);
            amazonS3.copyObject(request);
        });
    }

    @Override
    public FileInfoVo getFileInfo(String bucket, String key) {
        S3Object s3Object = null;
        S3ObjectInputStream objectInputStream = null;

        try {
            // 1. 获取文件元数据
            GetObjectMetadataRequest metadataRequest = new GetObjectMetadataRequest(bucket, key);
            ObjectMetadata metadata = s3.getObjectMetadata(metadataRequest);
            Date lastModified = metadata.getLastModified();
            String eTag = metadata.getETag();
            Long size = metadata.getContentLength();
            String md5 = eTag != null ? eTag.replace("\"", "") : "";

            GetObjectRequest getObjectRequest = new GetObjectRequest(bucket, key);
            s3Object = s3.getObject(getObjectRequest);
            objectInputStream = s3Object.getObjectContent();

            String sha1 = MD5Checksum.getSHA1(objectInputStream);

            return new FileInfoVo(
                    md5,
                    sha1.toUpperCase(),
                    lastModified.getTime(),
                    size
            );

        } catch (Exception e) {
            log.error("处理S3文件信息异常 - bucket:{}, key:{}", bucket, key, e);

        } finally {
            closeQuietly(objectInputStream);
            closeQuietly(s3Object);
        }
        return null;
    }
    private void closeQuietly(S3ObjectInputStream stream) {
        if (stream != null) {
            try {
                stream.close();
            } catch (IOException e) {
                log.warn("关闭S3对象流时发生异常", e);
            }
        }
    }

    private void closeQuietly(S3Object s3Object) {
        if (s3Object != null) {
            try {
                s3Object.close();
            } catch (IOException e) {
                log.warn("关闭S3对象时发生异常", e);
            }
        }
    }

}