package com.persagy.dmp.storage.service.impl;

import cn.hutool.core.io.IoUtil;
import cn.hutool.core.util.StrUtil;
import com.persagy.dmp.common.constant.ResponseCode;
import com.persagy.dmp.common.exception.BusinessException;
import com.persagy.dmp.storage.service.IFileStorageService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.IOException;
import java.io.InputStream;

/**
 * MinIO存储服务实现类
 * @author Charlie Yu
 * @date 2021-05-15
 */
@Slf4j
@RequiredArgsConstructor
public class HdfsStorageServiceImpl implements IFileStorageService {

    private final Configuration configuration;
    /** 文件key默认存储路径  */
    private final static String BASE_FILE_PATH="/test/files";
    /** 对文件key进行hash散列计算可能的结果数量  */
    public static final int HASH_FILE_FOLDER_NUM = 1024;
    /**
     * 文件上传
     * @param bucketName 桶名
     * @param fileName 文件名 - 同名文件则覆盖
     * @param inputStream 输入流
     */
    @Override
    public void upload(String bucketName, String fileName, InputStream inputStream) {
        // 容错判断
        if(StrUtil.isBlank(fileName) || inputStream == null) {
            throw new BusinessException("上传文件参数有误！");
        }
        try (FileSystem fs = FileSystem.get(configuration)) {
            Path filePath = getFilePath(bucketName,fileName);
            if (fs.exists(filePath)) {
                throw new BusinessException(ResponseCode.B0300.getCode(),"File is exists");
            }
            try (FSDataOutputStream outputStream = fs.create(filePath)) {
                IoUtil.copy(inputStream, outputStream);
            }
        } catch (IOException e) {
            MinioExceptionHandler.handleException(e);
        }
    }

    /**
     * 文件下载
     * @param bucketName 桶名
     * @param fileName 文件名
     * @return
     */
    @Override
    public InputStream download(String bucketName, String fileName) {
        // 容错判断
        if(StrUtil.isBlank(fileName)) {
            throw new BusinessException("下载文件参数有误！");
        }
        try (FileSystem fs = FileSystem.get(configuration)) {
            Path filePath = getFilePath(bucketName,fileName);
            if (!fs.exists(filePath)) {
                throw new BusinessException(ResponseCode.B0300.getCode(),"File is not exists");
            }
            return fs.open(filePath);
        } catch (IOException e) {
            MinioExceptionHandler.handleException(e);
        }
        return null;
    }

    @Override
    public String fetchUrl(String bucketName, String fileName) {
        // 容错判断
        if(StrUtil.isBlank(fileName)) {
            throw new BusinessException("下载文件参数有误！");
        }
        // TODO
        return null;
    }
    /**
     * 文件是否存在
     * @param bucketName 桶名
     * @param fileName 文件名
     * @return
     */
    @Override
    public boolean exists(String bucketName, String fileName) {
        // 容错判断
        if(StrUtil.isBlank(fileName)) {
            throw new BusinessException("检查文件参数有误！");
        }
        try (FileSystem fs = FileSystem.get(configuration)) {
            Path filePath = getFilePath(bucketName,fileName);
            return fs.exists(filePath);
        } catch (IOException e) {
            MinioExceptionHandler.handleException(e);
        }
        return false;
    }
    /**
     * 删除文件
     * @param bucketName 桶名
     * @param fileName 文件名
     */
    @Override
    public void delete(String bucketName, String fileName) {
        // 容错判断
        if(StrUtil.isBlank(fileName)) {
            throw new BusinessException("删除文件参数有误！");
        }
        try (FileSystem fs = FileSystem.get(configuration)) {
            Path filePath = getFilePath(bucketName,fileName);
            if (!fs.exists(filePath)){
                return;
            }
            if (!fs.isFile(filePath)){
                return;
            }
            // 非递归删除
            fs.delete(filePath,false);
        } catch (IOException e) {
            MinioExceptionHandler.handleException(e);
        }
    }
    /**
     * 删除桶
     * @param bucketName 桶名
     */
    @Override
    public void deletePath(String bucketName) {
        // 容错判断 桶名为空时也不让删
        if(StrUtil.isBlank(bucketName)) {
            throw new BusinessException("删除文件参数有误！");
        }
        try (FileSystem fs = FileSystem.get(configuration)) {
            Path filePath = getFilePath(bucketName,null);
            if (!fs.exists(filePath)){
                return;
            }
            // 递归删除
            fs.delete(filePath,true);
        } catch (IOException e) {
            MinioExceptionHandler.handleException(e);
        }
    }
    /***
     * Description: 获取文件路径对象
     * @param bucketName : 子路径名称
     * @param fileName : 文件名称
     * @return : org.apache.hadoop.fs.Path
     * @author : lijie
     * @date :2021/9/9 0:11
     * Update By lijie 2021/9/9 0:11
     */
    private Path getFilePath(String bucketName, String fileName) {
        if (StrUtil.isNotBlank(fileName)){
            return new Path(String.join("/", BASE_FILE_PATH, bucketName, getHashFilePath(fileName), fileName));
        }
        return new Path(String.join("/", BASE_FILE_PATH, bucketName));
    }

    /**
     * 获取文件key对应的hash散列结果值
     * @param key
     * @return 0-1024之间的数字字符串
     */
    private String getHashFilePath(String key) {
        key = key == null ? "" : key;
        // 得到哈希正整数
        int hashInt = Math.abs(key.hashCode());
        // 设置哈希字符值，长度为4位
        hashInt = 1000000 + hashInt % HASH_FILE_FOLDER_NUM;
        return ("" + hashInt).substring(3);
    }
}
