package com.andy.common;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;

import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.Session;
import javax.mail.Transport;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;
import java.io.*;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * 数据操作常用方法
 * Created by zhangqing on 16-8-31.
 */
public class DataHelper {
    /**
     * 判定参数不为空
     *
     * @param parameters 判断参数列表
     * @param conf       Configuration
     * @return 参数都不为空时，返回true；否则返回false
     */
    public static boolean judgeParameter(String[] parameters, Configuration conf) {
        if (parameters == null) {
            JobLogger.logger.error("String[] parameters is null...you should not call the judgeParameter api...");
            return false;
        }

        for (String parameter : parameters) {
            if (StringUtils.isBlank(conf.get(parameter))) {
                String log = "Jvm args: -D" + parameter + " must be specified!";
                JobLogger.logger.error(log);
                String commands = "The parameters are necessary: " + Arrays.toString(parameters);
                JobLogger.logger.error(commands);
                return false;
            }
        }
        return true;
    }

    // 按照UTF-8编码，将String转化成Bytes
    public static byte[] string2Bytes(String str) {
        try {
            return str.getBytes("UTF-8");
        } catch (UnsupportedEncodingException e) {
            return null;
        }
    }

    // 按照UTF-8编码，将Bytes转化成String
    public static String bytes2String(byte[] bytes) {
        try {
            return new String(bytes, "UTF-8");
        } catch (UnsupportedEncodingException e) {
            return null;
        }
    }

    /**
     * 以Delimited为分隔符，将文本数据转存为对象
     *
     * @param obj       对象
     * @param content   内容
     * @param delimited 分隔符
     */
    public static void setObjectByContent(Object obj, String content, String delimited) {
        if (obj == null) {
            return;
        }

        Class<?> cls = obj.getClass();
        Field[] fields = cls.getFields();
        List<Field> list = new ArrayList<>();
        for (Field field : fields) {
            // Thrift生成的Java类会有一个metaDataMap成员变量
            if (StringUtils.equalsIgnoreCase(field.getName(), "metaDataMap")) {
                continue;
            }
            list.add(field);
        }

        String[] elems = content.split(delimited, -1);

        // Case1：Thrift升级, 但任务Jar包没有更新
        // Case2：任务包用的最新Jar包，但处理的是历史数据
        if (elems.length != list.size()) {
            JobLogger.logger.error("content.split() != T.getFields() : " + elems.length + "\t" + list.size());
        }
        int minLength = Math.min(elems.length, list.size());

        for (int i = 0; i < minLength; ++i) {
            elems[i] = elems[i].trim();
            if (StringUtils.isBlank(elems[i])) {
                continue;
            }

            boolean flag = true;
            String[] exceptionContents = new String[]{"NULL", "\\N", "None"};
            for (String fuck : exceptionContents) {
                if (StringUtils.equalsIgnoreCase(elems[i], fuck)) {
                    flag = false;
                    break;
                }
            }

            if (!flag) {
                continue;
            }

            Field field = list.get(i);
            String setString = "set" + StringUtils.capitalize(field.getName());
            try {
                Object param = elems[i];
                if (field.getType() == short.class || field.getType() == Short.class)
                    param = Short.valueOf(elems[i]);
                else if (field.getType() == byte.class || field.getType() == Byte.class)
                    param = Byte.valueOf(elems[i]);
                else if (field.getType() == int.class || field.getType() == Integer.class)
                    param = Integer.valueOf(elems[i]);
                else if (field.getType() == long.class || field.getType() == Long.class)
                    param = Long.valueOf(elems[i]);
                else if (field.getType() == float.class || field.getType() == Float.class)
                    param = Float.valueOf(elems[i]);
                else if (field.getType() == double.class || field.getType() == Double.class)
                    param = Double.valueOf(elems[i]);
                else if (field.getType() == boolean.class || field.getType() == Boolean.class) {
                    if (StringUtils.equalsIgnoreCase(elems[i], "false")) {
                        param = false;
                    } else if (StringUtils.equalsIgnoreCase(elems[i], "true")) {
                        param = true;
                    } else {
                        // 非法boolean值，跳过该字段
                        JobLogger.logger.error("the field type is boolean, but the value is " + elems[i]);
                        continue;
                    }
                } else if (field.getType() != String.class) {
                    JobLogger.logger.error("the type is not support : " + field.getType());
                    continue;
                }

                Method setMethod = cls.getMethod(setString, field.getType());
                setMethod.invoke(obj, param);
            } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | NumberFormatException e) {
                e.printStackTrace();
            }
        }
    }

    public static Object getObjectByContent(Class<?> cls, String content, String delimited) throws IllegalAccessException, InstantiationException {
        Object obj = cls.newInstance();
        setObjectByContent(obj, content, delimited);
        return obj;
    }

    /**
     * 获取HDFS写文件的句柄
     *
     * @param conf     HDFS Configuration
     * @param filePath 待写入的HDFS文件
     * @throws IOException
     */
    public static BufferedWriter getFileWriter(Configuration conf, String filePath) throws IOException {
        FileSystem fileSystem = FileSystem.get(conf);
        Path path = new Path(filePath);
        if (fileSystem.exists(path)) {
            fileSystem.delete(path, true);
        }
        fileSystem.createNewFile(path);

        FSDataOutputStream fsDataOutputStream = fileSystem.create(path);
        return new BufferedWriter(new OutputStreamWriter(fsDataOutputStream, "UTF-8"));
    }

    /**
     * HDFS 文件移动, 支持文件夹操作
     *
     * e.g.如果源文件不存在,抛异常
     * e.g.如果B存在   xxx/A  xxx/B  -->  XXX/B/A
     * e.g.如果B不存在 xxx/A  xxx/B  -->  XXX/B
     */
    public static boolean moveFile(Configuration conf, String source, String target) throws IOException {
        FileSystem fs = FileSystem.get(conf);
        return FileUtil.copy(fs, new Path(source), fs, new Path(target), true, true, conf);
    }

    /**
     * moveFile改善版
     * 如果目标文件存在，则覆盖存储
     */
    public static boolean moveFileByCoverage(Configuration conf, String source, String target) throws IOException {
        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(new Path(source))) {
            if (fs.exists(new Path(target))) {
                fs.delete(new Path(target), true);
            }
            return FileUtil.copy(fs, new Path(source), fs, new Path(target), true, true, conf);
        } else {
            JobLogger.logger.error("源数据不存在......" + source);
            return false;
        }
    }

    /**
     * HDFS 文件复制, 支持文件夹操作
     *
     * e.g.如果源文件不存在,抛异常
     * e.g.如果B存在   xxx/A  xxx/B  -->  复制文件A到文件B下, 即XXX/B/A
     * e.g.如果B不存在 xxx/A  xxx/B  -->  复制文件A到文件B, 即XXX/B
     */
    public static boolean copyFile(Configuration conf, String source, String target) throws IOException {
        FileSystem fs = FileSystem.get(conf);
        return FileUtil.copy(fs, new Path(source), fs, new Path(target), false, true, conf);
    }

    /**
     * copyFile改善版
     * 如果目标文件存在，则覆盖存储
     */
    public static boolean copyFileByCoverage(Configuration conf, String source, String target) throws IOException {
        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(new Path(source))) {
            if (fs.exists(new Path(target))) {
                fs.delete(new Path(target), true);
            }
            return FileUtil.copy(fs, new Path(source), fs, new Path(target), false, true, conf);
        } else {
            JobLogger.logger.error("源数据不存在......" + source);
            return false;
        }
    }

    /**
     * 获取字符串的md5值
     *
     * @param content 原内容
     * @return 十六进制，长度为32位的String
     */
    public static String getMD5(final String content) {
        char hexDigits[] = {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F'};
        try {
            byte[] btInput = content.getBytes("UTf-8");
            // 获得MD5摘要算法的 MessageDigest 对象
            MessageDigest mdInst = MessageDigest.getInstance("MD5");
            // 使用指定的字节更新摘要
            mdInst.update(btInput);
            // 获得密文
            byte[] md = mdInst.digest();
            // 把密文转换成十六进制的字符串形式
            int j = md.length;
            char str[] = new char[j * 2];
            int k = 0;
            for (byte elem : md) {
                // 取字节的高4位
                str[k++] = hexDigits[elem >>> 4 & 0xf];
                // 取字节的低4位
                str[k++] = hexDigits[elem & 0xf];
            }
            return new String(str);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * 等待数据点是否生成
     *
     * @param fs        HDFS句柄
     * @param path      数据点路径
     * @param delayTime pending时间(h)
     */
    public static boolean waitForDataPoint(FileSystem fs, String path, int delayTime) throws IOException, InterruptedException {
        long curTime = System.currentTimeMillis() / 1000;
        // 每15分钟检测一次数据源，最多等待delayTime小时
        while (!(fs.exists(new Path(path)) && fs.exists(new Path(path + File.separator + "_SUCCESS")))
                && (curTime + delayTime * 60 * 60 > System.currentTimeMillis() / 1000)) {
            TimeUnit.MINUTES.sleep(15);
        }

        return fs.exists(new Path(path));
    }

    /**
     * 批量等待数据点是否生成
     */
    public static boolean waitForDataPoint(FileSystem fs, List<String> pathList, int delayTime) throws IOException, InterruptedException {
        long curTime = System.currentTimeMillis() / 1000;
        // 每15分钟检测一次数据源，最多等待delayTime小时
        while (curTime + delayTime * 60 * 60 > System.currentTimeMillis() / 1000) {
            boolean flag = true;
            for (String path : pathList) {
                if (!fs.exists(new Path(path)) || !fs.exists(new Path(path + File.separator + "_SUCCESS"))) {
                    flag = false;
                }
            }
            if (flag) {
                break;
            }
            TimeUnit.MINUTES.sleep(15);
        }

        boolean flag = true;
        for (String path : pathList) {
            if (!fs.exists(new Path(path))) {
                flag = false;
            }
        }
        return flag;
    }

    private static boolean isChinese(char c) {
        Character.UnicodeBlock ub = Character.UnicodeBlock.of(c);
        if (ub == Character.UnicodeBlock.CJK_UNIFIED_IDEOGRAPHS
                || ub == Character.UnicodeBlock.CJK_COMPATIBILITY_IDEOGRAPHS
                || ub == Character.UnicodeBlock.CJK_UNIFIED_IDEOGRAPHS_EXTENSION_A
                || ub == Character.UnicodeBlock.GENERAL_PUNCTUATION
                || ub == Character.UnicodeBlock.CJK_SYMBOLS_AND_PUNCTUATION
                || ub == Character.UnicodeBlock.HALFWIDTH_AND_FULLWIDTH_FORMS) {
            return true;
        }
        return false;
    }

    public static boolean isChinese(String strName) {
        char[] ch = strName.toCharArray();
        for (int i = 0; i < ch.length; i++) {
            char c = ch[i];
            if (isChinese(c)) {
                return true;
            }
        }
        return false;
    }

    /**
     * 发送邮件
     *
     * @param to      邮件收件方
     * @param from    邮件发送方
     * @param subject 邮件标题
     * @param info    邮件内容
     */
    public static void send(String to, String from, String subject, String info) {
        Properties properties = System.getProperties();
        // Setup mail server
        properties.setProperty("mail.smtp.host", "mail.srv");
        // Get the default Session object.
        Session session = Session.getDefaultInstance(properties);
        try {
            // Create a default MimeMessage object.
            MimeMessage message = new MimeMessage(session);
            // Set From: header field of the header.
            message.setFrom(new InternetAddress(from));
            // Set To: header field of the header.
            String[] tos = to.split(";");
            // Set To: header field of the header.
            for (String toitem : tos) {
                if (!toitem.trim().isEmpty()) {
                    message.addRecipient(Message.RecipientType.TO, new InternetAddress(toitem));
                }
            }
            // Set Subject: header field
            message.setSubject(subject);
            // Now set the actual message
            message.setContent(info, "text/html;charset=gbk");
            // Send message
            Transport.send(message);
        } catch (MessagingException e) {
            JobLogger.logger.error(e.getMessage());
        }
    }
}
