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.Youtube.YoutubeCountryStats;

import org.springframework.stereotype.Service;

import io.delphiplatform.api.v3.bigtable.ParseProtoException;
import io.delphiplatform.api.v3.bigtable.entity.YouTubeVideoStats;

import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.CONTENT_TYPE;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.DEMOGRAPHICS;
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.PRODUCT_FAMILY_ID;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.PROJECT_NUMBER;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.STREAMS;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.VIDEO_ID;

@Service
public class YouTubeVideoStatsReader extends BigtableEntityReader<YouTubeVideoStats> {

    @Override
    public YouTubeVideoStats readRow(Row row) {
        YouTubeVideoStats youtubeVideoStats = new YouTubeVideoStats();
        youtubeVideoStats.setDate(readDate(row));
        String videoIdWithPadding = readString(row, VIDEO_ID);
        youtubeVideoStats.setVideoId(videoIdWithPadding);
        youtubeVideoStats.setDsp(readString(row, DSP));
        String isrc = readString(row, ISRC);
        youtubeVideoStats.setIsrc(isrc);
        readBytes(row, STREAMS).ifPresent(bytes -> youtubeVideoStats.setStreams(parseStats(bytes)));
        readBytes(row, DEMOGRAPHICS).ifPresent(bytes -> youtubeVideoStats.setDemographics(parseStats(bytes)));
        String prodFamId = readString(row, PRODUCT_FAMILY_ID);
        youtubeVideoStats.setProductFamilyId(prodFamId);
        String projectNumber = readString(row, PROJECT_NUMBER);
        youtubeVideoStats.setProjectNumber(projectNumber);
        String contentType = readString(row, CONTENT_TYPE);
        youtubeVideoStats.setContentType(contentType);
        return youtubeVideoStats;
    }

    private YoutubeCountryStats parseStats(ByteString bytes) {
        try {
            return YoutubeCountryStats.parseFrom(bytes);
        } catch (InvalidProtocolBufferException e) {
            throw new ParseProtoException("Cannot parse youtube proto stats.", e);
        }
    }

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

}