package com.moex.asts.export;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.micex.client.API.ServerInfo;
import com.micex.client.Binder;
import com.micex.client.Client;
import com.micex.client.Client.Request;
import com.micex.client.ClientException;
import com.micex.client.Filler;
import com.micex.client.Meta;
import com.micex.client.Meta.Message;
import com.micex.client.Parser;
import com.moex.asts.export.Description.ParamInfo;
import com.moex.asts.export.Description.TableInfo;
import com.moex.memory.Column;
import com.moex.memory.Record;
import com.moex.memory.Type;
import com.moex.util.Fixed;
import com.moex.util.HMS;
import com.moex.util.YMD;
import java.io.File;

public class Export implements Runnable, Binder {

    final Logger logger = LoggerFactory.getLogger(getClass());

    private final Settings settings;
    private final Map<String, ClientTable> database;

    private final Client client;
    private final Encoder encoder;

    final File output;

    public Export(Settings settings, String filename) throws FileNotFoundException, UnsupportedEncodingException {
        this.settings = settings;
        this.database = new LinkedHashMap<>();
        this.client = new Client();
        this.output = new File(filename);
        this.encoder = Encoder.create(filename);
    }

    boolean isXML() {
        return encoder instanceof XmlEncoder;
    }

    static ClientTable create(Meta.Message source) {
        final List<Column> columns = new ArrayList<>();
        for (int i = 0; i < source.output().size(); i++) {
            final Meta.Field field = source.output().item(i);
            Type type = ClientTable.mapMetaType(field.type());
            int flags = field.isKey() ? Column.PRIMARY : 0;
            if (Type.STRING == type && flags == 0)
                flags |= Column.AUTOTRIM;
            columns.add(new Column(field.name, type, field.size(), flags));
        }
        return new ClientTable(source.name, columns.toArray(new Column[columns.size()]));
    }

    private ClientTable needTable(Message source) {
        ClientTable target = database.get(source.name);
        if (target == null) {
            target = create(source);
            database.put(source.name, target);
        }
        return target;
    }

    @Override
    public Filler getFiller(Message source) {
        return needTable(source);
    }

    private boolean active = true;

    public void cancel() {
        active = false;
    }

    @Override
    public void run() {
        try {
            try {
                execute();
            } catch (Throwable x) {
                logger.error("Unexpected error", x);
            }
        } finally {
            close();
        }
    }

    private void close() {
        try {
            encoder.close();
        } catch (IOException x) {
            // swallow
        }
        client.close();
        logger.info("ASTS connection closed");
    }

    final static Description.TableInfo SESSION = new TableInfo("SESSION",
            null, null, 0, Description.NO_PARAMS, Description.NO_FIELDS, 0);

    private void exportSessionInfo(Meta.Market market, ServerInfo server) {
        final ClientTable session = new ClientTable("SESSION", new Column[] {
                new Column("MARKET", Type.STRING, 32),
                new Column("SYSTEM", Type.STRING, 1),
                new Column("SSN_ID", Type.INTEGER),
                new Column("USERID", Type.STRING, 12),
                new Column("DATE", Type.DATE, 8),
                new Column("TIME", Type.TIME, 6)
        });
        int systemDate = 0;
        int systemTime = 0;
        Message origin = market.tables().find(Meta.TableType.SysTime);
        ClientTable source = origin == null ? null : database.get(origin.name());
        if (source != null && source.getRecordCount() > 0) {
            Record r = source.getRow(0);
            systemDate = r.getDate(source.getColumnIndex("DATE"));
            systemTime = r.getTime(source.getColumnIndex("TIME"));
        }
        try {
            session.addRow(session.newRow()
                    .setString(0, market.name)
                    .setString(1, Character.toString(server.systemID))
                    .setInt(2, server.sessionID)
                    .setString(3, server.userID)
                    .setDate(4, systemDate)
                    .setTime(5, systemTime)
            );
            encoder.encode(SESSION, session, Collections.<String,Object>emptyMap());
        } finally {
            session.close();
        }
    }

    private void execute() throws Exception {
        logger.info("ASTS connecting...");
        client.start(settings.parameters);

        final ServerInfo server = client.getServerInfo();
        final Meta.Market market = client.getMarketInfo();
        logger.info("ASTS connected: session={}, system={}, user={}, market={}",
                server.sessionID, server.systemID, server.userID, market.name);

        encoder.start();
        Message time = market.tables().find(Meta.TableType.SysTime);
        if (time == null)
            logger.warn("No {}.(TE)SYSTIME found: " +
                    "session date and time are not valid", market.name);
        else {
            client.load(time.name(), null).execute(this);
            exportSessionInfo(market, server);
        }

        for (TableInfo item : settings.description.values()) {
            if (!active)
                throw new InterruptedException("Interrupted");

            Meta.Message source = market.tables().find(item.source);
            if (source == null) {
                if (item.isOptional())
                    continue;
                else
                    throw new Exception(
                            "Market " + market.name + " has no table "
                            + item.source + " (not optional)");
            }

            export(item, source);
        }
    }

    static boolean isReference(String name) {
        return name != null && name.length() > 2 &&
                name.charAt(0) == '{' &&
                name.charAt(name.length() - 1) == '}';
    }

    static String nameWithParams(String name, Map<String, Object> params) {
        StringBuilder b = new StringBuilder(name);
        if (params.size() > 0) {
            b.append('(');
            for (Object value : params.values()) {
                if (value == null)
                    b.append("null");
                else
                    b.append(value.toString());
                b.append(',');
            }
            b.deleteCharAt(b.length() - 1);
            b.append(')');
        }
        return b.toString();
    }

    /**
     * Размер пакета, при превышении которого предполагается
     * наличие доступной к немедленному скачиванию следующей
     * порции данных; байты.
     */
    static final int SMALL_PACKAGE_THRESHOLD = 16 * 1024;

    private void export(TableInfo specs, Message source, Map<String, Object> params) throws ClientException {
        ClientTable target = needTable(source);
        target.clear();
        if (source.isUpdateable()) {
            logger.info("Loading {}...", nameWithParams(source.name, params));
            Parser parser = client.open(source.name, params, false);
            Request request = null;
            if (parser instanceof Request)
                request = (Request) parser;
            int limit = Math.max(1, specs.cycles);
            int bytes = parser.length();
            int count = parser.execute(this);
            int total = bytes;
            int cycle = 0;
            while (++cycle <= limit && bytes >= SMALL_PACKAGE_THRESHOLD) {
                parser = client.refresh();
                bytes = parser.length();
                total += bytes;
                count += parser.execute(this);
                if (cycle % 100 == 0) {
                    logger.info("Loading {}: bytes={}, rows={} so far...",
                            source.name, total, count);
                }
            }
            if (request != null)
                request.close();
            logger.info("Handled {}: bytes={}, rows={}/{}, cycles={}",
                    source.name, total, count, target.getRecordCount(), cycle);
            encoder.encode(specs, target, params);
        } else {
            logger.info("Loading {}...", nameWithParams(source.name, params));
            final Parser parser = client.load(source.name, params);
            final int bytes = parser.length();
            final int count = parser.execute(this);
            logger.info("Handled {}: bytes={}, rows={}/{}",
                    source.name, bytes, count, target.getRecordCount());
            encoder.encode(specs, target, params);
        }
    }

    private String resolveLink(String name, ClientTable master, Record record) {
        int column = master.getColumnIndex(name);
        if (column >= 0)
            return record.getString(column);
        if ("USERID".equalsIgnoreCase(name))
            return client.getUserID();
        if ("FITMID".equalsIgnoreCase(name))
            return client.getFirmID();
        return null;
    }

    private void export(TableInfo specs, Message source) throws ClientException {
        ClientTable master = null;
        if (specs.isDetail())
            master = database.get(specs.master);

        final Map<String, Object> params = new LinkedHashMap<>();
        for (ParamInfo param : specs.params) {
            Meta.Field actual = source.params().find(param.source);
            if (actual != null) {
                switch (actual.type()) {
                    case Integer:
                        params.put(param.source, Long.parseLong(param.value));
                        break;
                    case Fixed:
                    case Float:
                    case FloatPoint:
                        params.put(param.source, Fixed.parse(param.value));
                        break;
                    case Date:
                        params.put(param.source, YMD.parse(param.value).intValue());
                        break;
                    case Time:
                        params.put(param.source, HMS.parse(param.value).intValue());
                        break;
                    default:
                        params.put(param.source, param.value);
                }
            }
        }

        if (master == null)
            export(specs, source, params);
        else {
            for (int i = 0; i < master.getRecordCount(); i++) {
                Record record = master.getRow(i);
                final Map<String, Object> clone = new LinkedHashMap<>();
                for (Entry<String, Object> param : params.entrySet()) {
                    Object data = param.getValue();
                    String text = data == null ? null : data.toString();
                    if (isReference(text))
                        clone.put(param.getKey(), resolveLink(
                                text.substring(1, text.length()-1), master, record));
                    else
                        clone.put(param.getKey(), data);
                }
                export(specs, source, clone);
            }
        }
    }

}
