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

import com.google.protobuf.GeneratedMessageV3;
import com.sonymusic.delphi.etl.apps.proto.Amazon.AmazonAggregate.SelectionSourceType;
import com.sonymusic.delphi.etl.apps.proto.Amazon.AmazonStreamStats;
import com.sonymusic.delphi.etl.apps.proto.Apple.AppleStreamStats;
import com.sonymusic.delphi.etl.apps.proto.Spotify.SpotifyStreamStats;

import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import fr.xebia.extras.selma.Selma;
import io.delphiplatform.api.util.CollectionUtils;
import io.delphiplatform.api.v3.bigtable.entity.AmazonTrackStream;
import io.delphiplatform.api.v3.bigtable.entity.TrackStream;
import io.delphiplatform.api.v3.model.IncludeStreams;
import io.delphiplatform.api.v3.model.StreamModel;
import io.delphiplatform.api.v3.model.mapper.AgeBandsMapper;
import io.delphiplatform.api.v3.model.mapper.GendersMapper;
import io.delphiplatform.api.v3.model.mapper.StreamsInfoMapper;
import io.delphiplatform.api.v3.model.mapper.streamsinfo.SpotifyStreamsInfoMapper;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class StreamsConverter {

    private final AgeBandsMapper ageBandsMapper;
    private final GendersMapper gendersMapper;
    private final StreamsInfoMapper streamsInfoMapper;
    private final SpotifyStreamsInfoMapper spotifyStreamsInfoMapper;

    public StreamsConverter() {
        gendersMapper = Selma.builder(GendersMapper.class).build();
        ageBandsMapper = Selma.builder(AgeBandsMapper.class).build();
        streamsInfoMapper = Selma.builder(StreamsInfoMapper.class).build();
        spotifyStreamsInfoMapper = Selma.builder(SpotifyStreamsInfoMapper.class).build();
    }

    public Stream<StreamModel> convertForCountryCode(Set<String> countryCodes, Stream<TrackStream<?, ?>> streams, Set<IncludeStreams> includes) {
        return streams.flatMap(trackStream -> {
            Map<String, ?> streamStatsToConvert = countryCodes == null || countryCodes.isEmpty()
                ? trackStream.getCountryStats()
                : getStatsForCountry(countryCodes, trackStream.getCountryStats());

            Stream<StreamModel> streamModelStream = convertMapOfStats(trackStream, streamStatsToConvert, includes);

            if (IncludeStreams.isDemographicsIncluded(includes)) {
                Map<String, ?> demographicStatsToConvert = countryCodes == null || countryCodes.isEmpty()
                    ? trackStream.getCountryDemographics()
                    : getStatsForCountry(countryCodes, trackStream.getCountryDemographics());

                Stream<StreamModel> demographicsStream = convertMapOfStats(trackStream, demographicStatsToConvert, includes);
                return Stream.concat(streamModelStream, demographicsStream);
            } else {
                return streamModelStream;
            }
        });
    }

    /**
     * Serialize to generic StreamModel
     */
    private Stream<StreamModel> convertMapOfStats(TrackStream<?, ?> trackStream, Map<String, ?> streamStatsToConvert,
        Set<IncludeStreams> includes) {
        return streamStatsToConvert.entrySet().stream()
            .map(entry -> {
                String statCountry = entry.getKey();
                Object stats = entry.getValue();
                return createStreamModel(trackStream, statCountry, stats, includes);
            });
    }

    private <S extends GeneratedMessageV3> Map<String, S> getStatsForCountry(
        Set<String> countryCodes,
        Map<String, S> countryStats) {
        return countryCodes.stream()
            .filter(countryStats::containsKey)
            .collect(Collectors.toMap(code -> code, countryStats::get));
    }

    private StreamModel createStreamModel(
        TrackStream<?, ?> trackStream,
        String statCountry,
        Object streamStatsForCountry,
        Set<IncludeStreams> includes
    ) {
        boolean isDemographicsIncluded = CollectionUtils.contains(includes, IncludeStreams.DEMOGRAPHICS);
        boolean isAllIncluded = CollectionUtils.contains(includes, IncludeStreams.ALL);
        boolean isSavesIncluded = CollectionUtils.contains(includes, IncludeStreams.SAVES);
        boolean isSkipsIncluded = CollectionUtils.contains(includes, IncludeStreams.SKIPS);
        boolean isDspStreamsInfoIncluded = CollectionUtils.contains(includes, IncludeStreams.DSP_STREAMS_INFO);

        StreamModel stream = new StreamModel();
        stream.setDate(trackStream.getDate());
        stream.setMainDsp(trackStream.getDsp());
        if (trackStream instanceof AmazonTrackStream) {
            stream.setSubDsp(((AmazonTrackStream) trackStream).getFullDsp());
        } else {
            stream.setSubDsp(trackStream.getDsp());
        }
        stream.setCountryCode(statCountry);
        stream.setArtistId(trackStream.getArtistId());
        stream.setTrackId(trackStream.getTrackId());
        stream.setPlaylistId(trackStream.getPlaylistId());
        stream.setIsrc(trackStream.getIsrc());

        stream.projectNumber(trackStream.getProjectNumber());
        stream.productFamilyId(trackStream.getProductFamilyId());

        if (streamStatsForCountry instanceof AmazonStreamStats) {
            AmazonStreamStats amazonStreamStats = (AmazonStreamStats) streamStatsForCountry;
            stream.streams(amazonStreamStats.getStreams());
            stream.setHasStationStreams(hasStationStreams(amazonStreamStats.getSelectionSourceType()));

            if (isAllIncluded || isDspStreamsInfoIncluded) {
                stream.amazonStreamsInfo(streamsInfoMapper.toStreamsInfo(amazonStreamStats));
            }
        } else if (streamStatsForCountry instanceof SpotifyStreamStats) {
            SpotifyStreamStats spotifyStreamStats = (SpotifyStreamStats) streamStatsForCountry;
            stream.streams(spotifyStreamStats.getStreams());
            if (isAllIncluded) {
                stream.spotifyStreamsInfo(streamsInfoMapper.toStreamsInfo(spotifyStreamStats));

                stream.spotifyAgeBands(ageBandsMapper.toStreamsAgeBands(spotifyStreamStats.getAgeBand()));
                stream.genders(gendersMapper.toGender(spotifyStreamStats.getGender()));
            } else {
                if (isDspStreamsInfoIncluded) {
                    stream.spotifyStreamsInfo(streamsInfoMapper.toStreamsInfo(spotifyStreamStats));
                }
                if (isDemographicsIncluded) {
                    stream.spotifyAgeBands(ageBandsMapper.toStreamsAgeBands(spotifyStreamStats.getAgeBand()));
                    stream.genders(gendersMapper.toGender(spotifyStreamStats.getGender()));
                }
                if (isSavesIncluded) {
                    stream.initSpotifyStreamsInfoIfNone()
                        .saves(spotifyStreamsInfoMapper.toDto(spotifyStreamStats.getSaves()));
                }
                if (isSkipsIncluded) {
                    stream.initSpotifyStreamsInfoIfNone().skips(spotifyStreamStats.getSkips());
                }
            }
        } else if (streamStatsForCountry instanceof AppleStreamStats) {
            AppleStreamStats appleStreamStats = (AppleStreamStats) streamStatsForCountry;
            stream.streams(appleStreamStats.getStreams());

            if (isAllIncluded) {
                stream.appleStreamsInfo(streamsInfoMapper.toStreamsInfo(appleStreamStats));

                stream.appleAgeBands(ageBandsMapper.toStreamsAgeBands(appleStreamStats.getAppleAgeBands()));
                stream.genders(gendersMapper.toGender(appleStreamStats.getGenders()));
            } else {
                if (isDspStreamsInfoIncluded) {
                    stream.appleStreamsInfo(streamsInfoMapper.toStreamsInfo(appleStreamStats));
                }
                if (isDemographicsIncluded) {
                    stream.appleAgeBands(ageBandsMapper.toStreamsAgeBands(appleStreamStats.getAppleAgeBands()));
                    stream.genders(gendersMapper.toGender(appleStreamStats.getGenders()));
                }
                if (isSkipsIncluded) {
                    stream.initAppleStreamsInfoIfNone().skips(appleStreamStats.getSkips());
                }
            }
        } else {
            throw new IllegalArgumentException(
                "Could not process stats because of unknown type " + streamStatsForCountry);
        }

        return stream;
    }

    public boolean hasStationStreams(SelectionSourceType selectionSourceType) {
        return (
            selectionSourceType.getAmfStation() +
                selectionSourceType.getAmfStationSeed() +
                selectionSourceType.getPrimeStation() +
                selectionSourceType.getPrimeStationSeed() +
                selectionSourceType.getSimilarityStation() +
                selectionSourceType.getStation() +
                selectionSourceType.getUnlStationSeed() +
                selectionSourceType.getUnlimitedStation() +
                selectionSourceType.getUnlimitedStationSe()
        ) > 0;
    }
}