package com.google.refine.extension.gdata;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.json.JSONObject;

import com.google.api.client.http.AbstractInputStreamContent;
import com.google.api.client.http.ByteArrayContent;
import com.google.api.client.http.HttpResponseException;
import com.google.api.services.fusiontables.Fusiontables;

import com.google.refine.exporters.TabularSerializer;

final class FusionTableSerializer implements TabularSerializer {
    private static final int BATCH_SIZE = 20;
    Fusiontables service;
    String tableName;
    List<Exception> exceptions;
    
    String tableId;
    List<String> columnNames;
    StringBuffer sbBatch;
    int rows;
    
    FusionTableSerializer(Fusiontables service, String tableName, List<Exception> exceptions) {
        this.service = service;
        this.tableName = tableName;
        this.exceptions = exceptions;
    }
    
    @Override
    public void startFile(JSONObject options) {
    }

    @Override
    public void endFile() {
        if (sbBatch != null) {
            sendBatch(true);
        }
    }

    @Override
    public void addRow(List<CellData> cells, boolean isHeader) {
        if (isHeader) {
            columnNames = new ArrayList<String>(cells.size());
            for (CellData cellData : cells) {
                columnNames.add(cellData.text);
            }                
            try {
                tableId = FusionTableHandler.createTable(service, tableName, columnNames);
            } catch (Exception e) {
                tableId = null;
                exceptions.add(e);
            }
        } else if (tableId != null) {
            if (sbBatch == null) {
                sbBatch = new StringBuffer();
            }
            formatCsv(cells, sbBatch);            
            rows++;
            if (rows % BATCH_SIZE == 0) {
                if (!sendBatch(false)) {
                    return;
                }
            }
        }
    }
    
    private boolean sendBatch(boolean isLastChunk) {
        try {
            // FIXME: text/csv doesn't work even though that's what the content is
            AbstractInputStreamContent content = ByteArrayContent.fromString("application/octet-stream", sbBatch.toString());

//            AbstractInputStreamContent content = new InputStreamContent("application/octet-stream",
//                    // TODO: we really want to do GZIP compression here 
//                            new ByteArrayInputStream(sbBatch.toString().getBytes("UTF-8")));
            Long count = FusionTableHandler.insertRows(service, tableId, content);
            if (!isLastChunk && count != BATCH_SIZE) {
                // FIXME: this message should say numbers instead of %d but we'd need to know the batch number for this
                exceptions.add(new IOException("Only imported %d of %d rows"));
            }
        } catch (IOException e) {
            exceptions.add(e);
            if (e instanceof HttpResponseException) {
                int code = ((HttpResponseException)e).getStatusCode();
                if (code >= 400 && code < 500) {
                    return false;
                }
                // 500s appear to be retried automatically by li
            }
        } finally {
            sbBatch = null;
        }
        return true;
    }

    private void formatCsv(List<CellData> cells, StringBuffer sb) {
        boolean first = true;
        for (int i = 0; i < cells.size() && i < columnNames.size(); i++) {
            CellData cellData = cells.get(i);
            if (!first) {
                sb.append(',');
            } else {
                first = false;
            }
            sb.append("\"");
            if (cellData != null && cellData.text != null) {
                sb.append(cellData.text.replaceAll("\"", "\"\""));
            }
            sb.append("\"");
        }
        sb.append("\n");
    }
    
    public String getUrl() {
        return tableId == null || exceptions.size() > 0 ? null :
            "https://www.google.com/fusiontables/DataSource?docid=" + tableId;
    }
}

@@ -1099,20 +1099,33 @@
 ndBatch(
-true
+rows %25 BATCH_SIZE
 );%0A     
@@ -1929,21 +1929,26 @@
 ndBatch(
-false
+BATCH_SIZE
 )) %7B%0A   
@@ -2059,44 +2059,103 @@
 tch(
-boolean isLastChunk) %7B%0A        try %7B
+int batchSize) %7B%0A        try %7B%0A            // TODO: we really want to do GZIP compression here 
 %0A   
@@ -2364,538 +2364,156 @@
 ));%0A
-%0A//            AbstractInputStreamContent content = new InputStreamContent(%22application/octet-stream%22,%0A//                    // TODO: we really want to do GZIP compression here %0A//                            new ByteArrayInputStream(sbBatch.toString().getBytes(%22UTF-8%22)));%0A            Long count = FusionTableHandler.insertRows(service, tableId, content);%0A            if (!isLastChunk && count != BATCH_SIZE) %7B%0A                // FIXME: this message should say numbers instead of %25d but we'd need to know the batch number for this
+            Long count = FusionTableHandler.insertRows(service, tableId, content);%0A            if (count != null && count.intValue() != batchSize) %7B
 %0A   
@@ -2557,17 +2557,17 @@
 eption(%22
-O
+o
 nly impo
@@ -2575,16 +2575,42 @@
 ted 
-%25d of %25d
+%22 + count + %22 of %22 + batchSize + %22
  row

Improve error message via @annolangen in PR #853