package com.ihooyah.flume.sink;

import com.alibaba.fastjson.JSONObject;
import com.aliyun.oss.ClientException;
import com.aliyun.oss.OSS;
import com.aliyun.oss.OSSClientBuilder;
import com.aliyun.oss.OSSException;
import com.aliyun.oss.model.PutObjectRequest;
import com.aliyun.oss.model.PutObjectResult;
import com.ihooyah.flume.sink.util.HttpRequestUtil;
import com.ihooyah.flume.sink.util.aliyunOssUtil;
import org.apache.flume.*;
import org.apache.flume.conf.BatchSizeSupported;
import org.apache.flume.conf.Configurable;
import org.apache.flume.instrumentation.SinkCounter;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 云南项目读取文件到指定目录
 * 文件名请重命名   将存储路径放置于文件名中按照逗号隔开
 */
public class YunnanFileSink extends AbstractSink implements Configurable, BatchSizeSupported {

    private static final Logger logger = LoggerFactory.getLogger(LocalUploadSink.class);

    private int batchSize;

    private static final int DEFAULT_BATCH_SIZE = 1000;

    private SinkCounter sinkCounter;

    private String fileReadName;


    private HttpRequestUtil httpRequestUtil=new HttpRequestUtil();

    private CounterGroup counterGroup = new CounterGroup();

    public Sink.Status process() throws EventDeliveryException {
        Map<String, List<Event>> eventMap = new HashMap<>();
        Sink.Status status = Sink.Status.READY;
        Channel channel = getChannel();
        Transaction transaction = channel.getTransaction();
        transaction.begin();
        try {
            for (int i = 0; i < this.batchSize; i++) {
                Event event = channel.take();
                if (event == null)
                    break;
                this.fileReadName = ((String) event.getHeaders().get("fileName")).toString();
                if (!this.fileReadName.endsWith(".json")) {
                    this.counterGroup.incrementAndGet("event.file");
                    List<Event> eventList = eventMap.get(this.fileReadName);
                    if (eventList == null)
                        eventList = new ArrayList<>();
                    eventList.add(event);
                    eventMap.put(this.fileReadName, eventList);
                }
            }
            for (Map.Entry<String, List<Event>> data : eventMap.entrySet()) {
                try {
                    dealData(data.getKey(), data.getValue());
                } catch (Exception e) {
                    System.out.printf("aliyun::%s%n", e.getMessage());
                    throw new RuntimeException(e);
                }
            }
            transaction.commit();
        } catch (Exception e) {
            transaction.rollback();
            e.printStackTrace();
        } finally {
            transaction.close();
        }
        return status;
    }

    public void configure(Context context) {
        this.batchSize = context.getInteger("batchSize", Integer.valueOf(1000)).intValue();
    }

    public void dealData(String fileName, List<Event> eventList) throws IOException {
        logger.info("没有切割之前的fileName的值"+fileName);
        if (eventList.size() == 0)
            return;
        String[] split = fileName.split(",");
        for(int i=0;i<split.length;i++){
            String a=split[i];
            if("filerecord".equals(a)){
                fileName=fileName.replace("filerecord","file,record");
                logger.info("filerecord 切割后的值"+fileName);
            }
            if("filedefault".equals(a)){
                fileName=fileName.replace("filedefault","file,default");
                logger.info("filedefault 切割后的值"+fileName);
            }
        }
        fileName = fileName.replaceAll(",,", ",").replaceAll(",", "\\/");
        String uploadDir = "/" + fileName;
        String dir = uploadDir.substring(0, uploadDir.lastIndexOf("/"));
        logger.info("uploadDir:" + uploadDir + " dir:" + dir);
        File isDir = new File(dir);
        if (!isDir.exists())
            isDir.mkdirs();
        logger.info("deal data numbers: {}", Integer.valueOf(eventList.size()));
        File localFile = new File(uploadDir);
        FileOutputStream fos = new FileOutputStream(localFile);
        for (int i = 0; i < eventList.size(); i++)
            fos.write(((Event) eventList.get(i)).getBody());

        String url="http://86.1.41.66:8765/api/common/common/uploadFileToAliYun";
        JSONObject param = new JSONObject();
        param.put("filePath", uploadDir);
        String res= HttpRequestUtil.doPostJson(url, param.toJSONString(), null);
        JSONObject resJsonObect = JSONObject.parseObject(res);
        String code = resJsonObect.getString("code");
        String msg=resJsonObect.getString("msg");
        logger.info("调用三方接口的code"+code);
        logger.info("调用三方接口的msg"+msg);
        System.out.println(code);
        System.out.println(msg);
        fos.flush();
        fos.close();
    }

    public long getBatchSize() {
        return this.batchSize;
    }

}
