package io.delphiplatform.api.v3.bigtable.reader;

import com.google.cloud.bigtable.data.v2.models.Row;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import com.sonymusic.delphi.etl.apps.proto.Playlists.PlaylistTrackCountryStats;
import com.sonymusic.delphi.etl.apps.proto.Playlists.PlaylistTrackStats;

import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import io.delphiplatform.api.util.CollectionUtils;
import io.delphiplatform.api.util.DateUtils;
import io.delphiplatform.api.util.model.ModelUtils;
import io.delphiplatform.api.v3.bigtable.ParseProtoException;
import io.delphiplatform.api.v3.bigtable.entity.PlaylistPosition;
import io.delphiplatform.api.v3.constant.DspConstants;
import lombok.extern.slf4j.Slf4j;

import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.CURRENT;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.DSP;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.ISRC;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.PLAYLIST_ID;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.PLAYLIST_TRACK;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.PREVIOUS;

@Deprecated
@Slf4j
@Service
public class PlaylistPositionReader extends BigtableEntityReader<PlaylistPosition> {

    @Override
    public PlaylistPosition readRow(Row row) {
        PlaylistPosition playlistPosition = new PlaylistPosition();

        playlistPosition.setDate(readDate(row));
        playlistPosition.setDsp(readString(row, DSP));
        playlistPosition.setPlaylistId(readString(row, PLAYLIST_ID));
        playlistPosition.setIsrc(readString(row, ISRC));

        Integer current = readInt(row, pos, CURRENT);
        Integer previous = readInt(row, pos, PREVIOUS);
        playlistPosition.setCurrent(current != null ? Long.valueOf(current) : null);
        playlistPosition.setPrevious(previous != null ? Long.valueOf(previous) : null);

        playlistPosition.setTrend(ModelUtils.calculatePlaylistTrackPositionTrend(
            playlistPosition.getPrevious(), playlistPosition.getCurrent()));

        readBytes(row, PLAYLIST_TRACK).ifPresent(
            bytes -> playlistPosition.setPlaylistTrackCountryStats(parsePlaylistTrackCountryStats(bytes))
        );

        return playlistPosition;
    }

    /**
     * Converts position rows to PlaylistPosition.
     * <p>
     * Special handling for DSP with per-country playlist stats available in protobuf field (e.g. Amazon) - one position row is expanded into many
     * PlaylistPosition's since one row will contain protobuf map with stats for different country variants of this playlist
     */
    @Override
    public Stream<PlaylistPosition> readRows(Stream<Row> rowsStream, EnumSet<EntityReadingParams> entityReadingParams) {
        List<PlaylistPosition> rawPositions = rowsStream.map(this::readRow).collect(Collectors.toList());
        List<PlaylistPosition> expandedPositions = new ArrayList<>();

        rawPositions.forEach(position -> {
            Map<String, PlaylistTrackStats> playlistTrackCountryStatsMap = position.getPlaylistTrackCountryStatsMap();

            if (CollectionUtils.isEmpty(playlistTrackCountryStatsMap)) {
                expandedPositions.add(position);
            } else {
                /* If position's DSP has per-country playlist stats available in protobuf field (e.g. Amazon) -
                   expand position into multiple */

                playlistTrackCountryStatsMap.forEach((playlistCountryCode, countryPlaylistData) -> {
                    PlaylistPosition parentPositionCopy = new PlaylistPosition(position);

                    if (DspConstants.AMAZON.equals(position.getDsp())) {
                        //for Amazon - Chartmetric playlistId is appended with countryCode like 'amazon_123' + '_us'
                        parentPositionCopy.setPlaylistId(
                            ModelUtils.getPlaylistIdWithCountryCode(position.getPlaylistId(), playlistCountryCode)
                        );
                    }

                    //for DSP with per-country playlist stats - set 'current', 'previous', 'trend' fields per-country
                    // from protobuf instead of BT table fields.
                    parentPositionCopy.setCurrent(ModelUtils.extractLong(countryPlaylistData.getCurrentPosition()));
                    parentPositionCopy.setPrevious(ModelUtils.extractLong(countryPlaylistData.getPreviousPosition()));
                    parentPositionCopy.setTrend(ModelUtils.extractLong(countryPlaylistData.getTrend()));
                    //if trend value is empty in protobuf (but current/previous are not) - calculate in place
                    if (parentPositionCopy.getTrend() == null) {
                        parentPositionCopy.setTrend(ModelUtils.calculatePlaylistTrackPositionTrend(
                            parentPositionCopy.getPrevious(), parentPositionCopy.getCurrent()));
                    }

                    parentPositionCopy.setTrackId(ModelUtils.extractString(countryPlaylistData.getTrackId()));
                    parentPositionCopy.setCountryCode(playlistCountryCode);
                    parentPositionCopy.setDaysInPlaylist(ModelUtils.extractLong(countryPlaylistData.getDaysInPlaylist()));
                    parentPositionCopy.setEarliestPositionDate(
                        DateUtils.parseSafe(ModelUtils.extractString(countryPlaylistData.getEarliestPositionDate())));
                    parentPositionCopy.setLatestPositionDate(
                        DateUtils.parseSafe(ModelUtils.extractString(countryPlaylistData.getLatestPositionDate())));
                    parentPositionCopy.setPreviousPositionChange14Days(
                        ModelUtils.extractLong(countryPlaylistData.getPreviousPositionChange14Days()));
                    parentPositionCopy.setLastDateChange14Days(
                        DateUtils.parseSafe(ModelUtils.extractString(countryPlaylistData.getLastDateChange14Days())));
                    parentPositionCopy.setTrendChange14Days(ModelUtils.extractLong(countryPlaylistData.getTrendChange14Days()));

                    expandedPositions.add(parentPositionCopy);
                });
            }
        });

        return expandedPositions.stream();
    }

    private PlaylistTrackCountryStats parsePlaylistTrackCountryStats(ByteString bytes) {
        try {
            return PlaylistTrackCountryStats.parseFrom(bytes);
        } catch (InvalidProtocolBufferException e) {
            throw new ParseProtoException("Cannot parse per-country playlist stats proto.", e);
        }
    }

    @Override
    public Class<PlaylistPosition> getParsedClass() {
        return PlaylistPosition.class;
    }

}