package com.huawei.hwclouds.drs.avro;

import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.specific.SpecificDatumReader;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class AvroObject {

    /*    private Schema schema = null;

        private static final String USER_SCHEMA = "record.avsc";*/
    private Map<String, java.lang.Integer> tableCount = new HashMap<String, java.lang.Integer>();
    private static Map<String, String> typeMap = new HashMap<String, String>() {
        {
            put("-5", "Integer");
            put("2", "Numberic");
            put("1", "Character");
            put("4", "Integer");
            put("12", "Character");
            put("3", "Decimal");
            put("6", "Float");
            put("8", "Double");
            put("-7", "Bit");
            put("91", "Date");
            put("92", "Time");
            put("94", "DateTime");
            put("93", "Timestamp");
            put("2013", "Timestamp");
            put("2014", "Timestamp");
            put("2004", "BinaryObject");
            put("2005", "BinaryObject");
        }
    };

    public void init() throws IOException {
        //init(ClassLoader.getSystemResource(USER_SCHEMA).getFile());
    }

    public void init(String sche) throws IOException {
/*        InputStream inputStream = new FileInputStream(sche);
        Schema.Parser parser = new Schema.Parser();
        schema = parser.parse(inputStream);*/
    }

    public Record getData(byte[] data) {
        try {
            SpecificDatumReader<Record> reader = new SpecificDatumReader<Record>(Record.class);
            Decoder decoder = DecoderFactory.get().binaryDecoder(data, null);
            Record record = reader.read(null, decoder);
            return record;
        } catch (Exception e) {
            System.out.println(e);
            return null;
        }
    }

    public void showRecord(Record rec) {
        System.out.print("version=" + rec.getVersion());    //版本号
        if ("1.4".equals(rec.getVersion().toString())) {
            System.out.print(",sourceDb=" + rec.getSourceDb()); //源库类型,mysql,oracle
        }
        System.out.print(",seqno=" + rec.getSeqno());       //日志流水号
        if ("1.4".equals(rec.getVersion().toString())) {
            System.out.print(",transactionId=" + rec.getTransactionId());   //事务id
        }
        System.out.print(",tableName=" + rec.getTableName());   //数据库和表名
        System.out.print(",operation=" + rec.getOperation());   //操作类型：INSERT,DELETE,UPDATE,DDL
        System.out.print(",updateTimestamp=" + rec.getUpdateTimestamp());   //事件时间
        String time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(rec.getUpdateTimestamp());
        System.out.print("(" + time + ")");
        System.out.print(",columnCount=" + rec.getColumnCount());   //字段个数
        if ("1.4".equals(rec.getVersion().toString())) {
            System.out.print(",rowid=" + rec.getRowid());   //oracle源库的rowid
        }
        System.out.println(",eventid=" + rec.getEventid()); //mysql源库eventid

        int columnCount = (int) (long) rec.getColumnCount();
        List<Field> fields = rec.getFields();   //字段列表，包括字段名称和字段类型
        List<Object> value = rec.getAfterImages();  //记录后镜像
        List<Object> beforeValue = rec.getBeforeImages();   ////记录前镜像

        for (int col = 0; col < columnCount; col++) {
            Field field = fields.get(col);
            System.out.print((col + 1) + " fieldName=" + field.getName());
            int type = field.getDataTypeNumber();
            String typeNmae = typeMap.get(String.valueOf(type));
            String aValue = getValue(type, value.get(col));
            if (beforeValue != null && beforeValue.size() > 0) {
                String bValue = getValue(type, beforeValue.get(col));
                if (!aValue.equals(bValue)) {
                    aValue = aValue  + ",beforeValue=" + bValue;
                }
            }
            System.out.println(",type=" + type + "(" + typeNmae + "),value=" + aValue);
        }
    }

    private static String getValue(int type, Object value) {
        if (value == null) {
            return "";
        }

        if (value instanceof Integer) {
            Integer integer = (Integer) value;
            return integer.getValue().toString();
        }

        if (value instanceof Character) {
            Character character = (Character) value;
            try {
                return new String(character.getValue().array(), "utf8");
            } catch (UnsupportedEncodingException ignored) {
            }
        }

        if (value instanceof Decimal) {
            Decimal decimal = (Decimal) value;
            return decimal.getValue().toString();
        }

        if (value instanceof Float) {
            Float aFloat = (Float) value;
            return String.valueOf(aFloat.getValue());
        }

        if (value instanceof Bit) {
            Bit bit = (Bit) value;
            return bit.getValue().toString();
        }

        if (value instanceof Date) {
            Date date = (Date) value;
            if (type == 91) {
                java.sql.Date date1 = java.sql.Date.valueOf(date.getValue().toString());
                return date1.toString();
            }
            if (type == 94) {
                java.sql.Timestamp timestamp = java.sql.Timestamp.valueOf(date.getValue().toString());
                return timestamp.toString();
            }
            //type = 93
            java.sql.Time time = java.sql.Time.valueOf(date.getValue().toString());
            return time.toString();
        }

        if (value instanceof Timestamp) {
            Timestamp timestamp = (Timestamp) value;
            java.util.Date date = new java.util.Date(timestamp.getTimestamp() * 1000 + timestamp.getMillis());
            return date.toString();
        }

        if (value instanceof TimestampWithTimezone) {
            String val = String.valueOf(((TimestampWithTimezone) value).getValue());
            String timeZone = String.valueOf(((TimestampWithTimezone) value).getTimezone());
            return val + " " + timeZone;
        }

        if (value instanceof BinaryObject) {
            BinaryObject binaryObject = (BinaryObject) value;
            try {
                return new String(binaryObject.getValue().array(), "utf8");
            } catch (UnsupportedEncodingException ignored) {
            }
        }

        return "";
    }

    public void updateTableCount(String tableName) {
        updateTableCount(tableName,1);
    }

    public void updateTableCount(String tableName,int num) {
        java.lang.Integer count = tableCount.get(tableName);
        if (count == null)  {
            count = 0;
        }
        tableCount.put(tableName,num + count);
    }

    public int showTableCount()    {
        int sum = 0;
        for(String table : tableCount.keySet()) {
            int count = tableCount.get(table);
            System.out.println(table + " = " + count);
            sum = sum + count;
        }
        return sum;
    }
}
