#
# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
#


from abc import ABC
from typing import Any, Iterable, List, Mapping, MutableMapping, Optional, Tuple, Union
from urllib.parse import parse_qsl, urlparse

import pendulum
import requests

from airbyte_cdk.models import SyncMode
from airbyte_cdk.sources import AbstractSource
from airbyte_cdk.sources.streams import Stream
from airbyte_cdk.sources.streams.http import HttpStream, HttpSubStream

from .auth import OutbrainAmplifyAuthenticator


DEFAULT_END_DATE = pendulum.now()
DEFAULT_GEO_LOCATION_BREAKDOWN = "region"
DEFAULT_REPORT_GRANULARITY = "daily"
DEFAULT_REPORT_CONVERSION_COUNT_BY_CLICK_DATE = "conversion_time"

# Default limit for Outbrain API, setting it higher will cause an API error
DEFAULT_LIMIT = 50


# Basic full refresh stream
class OutbrainAmplifyStream(HttpStream, ABC):
    url_base = "https://api.outbrain.com/amplify/v0.1/"

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        if response.json():
            # Number of records in the current page as defined in Outbrain API for
            # all endpoints. Example: https://developer.outbrain.com/home-page/amplify-api/documentation/#/reference/campaigns/campaigns-collection-via-marketer/retrieve-multiple-campaigns?console=1
            current_count = response.json().get("count", 0)

            offset = 0
            limit = DEFAULT_LIMIT

            if response.request and response.request.url:
                parsed = urlparse(response.request.url)
                params = dict(parse_qsl(parsed.query))

                offset = int(params.get("offset", 0))
                limit = int(params.get("limit", DEFAULT_LIMIT))

            if current_count < limit:
                return None
            else:
                next_offset = offset + limit

                return {"offset": next_offset, "limit": limit}
        else:
            return None

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None, **kwargs
    ) -> MutableMapping[str, Any]:
        params = super().request_params(next_page_token=next_page_token, stream_state=stream_state, **kwargs)
        if next_page_token:
            params.update(next_page_token)
        return params

    def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapping]:
        yield {response.json()}

    @staticmethod
    def _get_time_interval(
        start_date: Union[pendulum.datetime, str], ending_date: Union[pendulum.datetime, str]
    ) -> Iterable[Tuple[pendulum.datetime, pendulum.datetime]]:
        if isinstance(start_date, str):
            start_date = pendulum.parse(start_date)
        end_date = pendulum.parse(ending_date) if ending_date else DEFAULT_END_DATE
        if end_date < start_date:
            raise ValueError(f"Specified start date: {start_date} is later than the end date: {end_date}")
        return start_date, end_date

    @staticmethod
    def _get_bool_conversion_count_by_click_date(value: str) -> str:
        if value.lower() == "click/view_time":
            return "true"
        elif value.lower() == "conversion_time":
            return "false"
        else:
            raise ValueError(f"Invalid value for conversion count by click date: {value}")


class Marketers(OutbrainAmplifyStream):
    primary_key = "id"

    def __init__(self, authenticator, config, **kwargs):
        super().__init__(**kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    @property
    def use_cache(self) -> bool:
        return True

    @property
    def cache_filename(self):
        return "marketers.yml"

    def parse_response(
        self, response: requests.Response, stream_state: Mapping[str, Any], stream_slice: Mapping[str, Any] = None, **kwargs
    ) -> Iterable[Mapping]:
        if response.json():
            for x in response.json().get("marketers"):
                yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        return "marketers/"


class CampaignsByMarketers(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: Marketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    @property
    def use_cache(self) -> bool:
        return True

    @property
    def cache_filename(self):
        return "campaigns.yml"

    @property
    def name(self) -> str:
        return "campaigns"

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for x in response.json().get("campaigns"):
                yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        return f"marketers/{stream_slice['marketer_id']}/campaigns?limit={DEFAULT_LIMIT}"


# Retrieve Campaign GeoLocations.
# A new endpoint has been added which returns all targeted and excluded locations of a given campaign. It can be called in order to retrieve a campaign's geotargeting.
class CampaignsGeoLocation(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: CampaignsByMarketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"campaign_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for x in response.json().get("geoLocations"):
                x["campaign_id"] = stream_slice["campaign_id"]
                yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        return f"campaigns/{stream_slice['campaign_id']}/locations"


# List PromotedLinks for Campaign.
# Collection of all PromotedLinks for the specified Campaign.
class PromotedLinksForCampaigns(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: CampaignsByMarketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    @property
    def name(self) -> str:
        return "promoted_links"

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"campaign_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for x in response.json().get("promotedLinks"):
                yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        return f"campaigns/{stream_slice['campaign_id']}/promotedLinks?limit={DEFAULT_LIMIT}"


# List PromotedLinksSequences for Campaign.
# Collection of all PromotedLinksSequences for the specified Campaign.
class PromotedLinksSequenceForCampaigns(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: CampaignsByMarketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"campaign_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for x in response.json().get("sequences"):
                yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        return f"campaigns/{stream_slice['campaign_id']}/promotedLinksSequences"


# List Budgets for a Marketer.
# Retrieve a collection of all Budgets for the specified Marketer.
class BudgetsForMarketers(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: Marketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    @property
    def name(self) -> str:
        return "budgets"

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for x in response.json().get("budgets"):
                x["marketer_id"] = stream_slice["marketer_id"]
                yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        return f"marketers/{stream_slice['marketer_id']}/budgets"


# Retrieve campaigns with performance statistics for a Marketer.
# The API in this sub-section allows retrieving marketer campaigns data with performance statistics.
class PerformanceReportCampaignsByMarketers(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: Marketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for x in response.json().get("results"):
                x["marketer_id"] = stream_slice["marketer_id"]
                yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        stream_start, stream_end = self._get_time_interval(self.config.get("start_date"), self.config.get("end_date"))
        stream_conversion_count = self._get_bool_conversion_count_by_click_date(
            self.config.get("conversion_count", DEFAULT_REPORT_CONVERSION_COUNT_BY_CLICK_DATE)
        )
        return (
            f"reports/marketers/{stream_slice['marketer_id']}/campaigns?from="
            + str(stream_start.date())
            + "&to="
            + str(stream_end.date())
            + "&limit=500"
            + "&includeVideoStats=true"
            + "&conversionsByClickDate="
            + str(stream_conversion_count)
        )


# Retrieve periodic performance statistics for a Marketer.
# The API in this sub-section allows retrieving performance statistics by periodic breakdown at different levels: marketer, budget, campaign and promoted link.
class PerformanceReportPeriodicByMarketers(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: Marketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for x in response.json().get("results"):
                x["marketer_id"] = stream_slice["marketer_id"]
                yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        stream_start, stream_end = self._get_time_interval(self.config.get("start_date"), self.config.get("end_date"))
        stream_conversion_count = self._get_bool_conversion_count_by_click_date(
            self.config.get("conversion_count", DEFAULT_REPORT_CONVERSION_COUNT_BY_CLICK_DATE)
        )
        return (
            f"reports/marketers/{stream_slice['marketer_id']}/periodic?from="
            + str(stream_start.date())
            + "&to="
            + str(stream_end.date())
            + "&breakdown="
            + str(self.config.get("report_granularity", DEFAULT_REPORT_GRANULARITY))
            + "&limit=500"
            + "&includeVideoStats=true"
            + "&conversionsByClickDate="
            + str(stream_conversion_count)
        )


# Retrieve performance statistics for all marketer campaigns by periodic breakdown.
# A special endpoint for retrieving periodic data by campaign breakdown. Now supports all breakdowns: daily, weekly, monthly, hourOfDay, dayOfWeek and dayOfWeekByHour.
class PerformanceReportPeriodicByMarketersCampaign(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: Marketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for results in response.json().get("campaignResults"):
                for x in results.get("results"):
                    x["marketer_id"] = stream_slice["marketer_id"]
                    x["campaign_id"] = results.get("campaignId")
                    yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        stream_start, stream_end = self._get_time_interval(self.config.get("start_date"), self.config.get("end_date"))
        stream_conversion_count = self._get_bool_conversion_count_by_click_date(
            self.config.get("conversion_count", DEFAULT_REPORT_CONVERSION_COUNT_BY_CLICK_DATE)
        )
        return (
            f"reports/marketers/{stream_slice['marketer_id']}/campaigns/periodic?from="
            + str(stream_start.date())
            + "&to="
            + str(stream_end.date())
            + "&breakdown="
            + str(self.config.get("report_granularity", DEFAULT_REPORT_GRANULARITY))
            + "&limit=500"
            + "&includeVideoStats=true"
            + "&conversionsByClickDate="
            + str(stream_conversion_count)
        )


# Retrieve periodic performance statistics by promoted link for a campaign. HERE
# A special endpoint for retrieving periodic data by promoted link breakdown for a given campaign.
class PerformanceReportPeriodicContentByPromotedLinksCampaign(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: CampaignsByMarketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    @property
    def name(self) -> str:
        return "performance_promoted_links"

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("marketerId"), "campaign_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for results in response.json().get("promotedLinkResults"):
                for x in results.get("results"):
                    x["marketer_id"] = stream_slice["marketer_id"]
                    x["campaign_id"] = stream_slice["campaign_id"]
                    x["promoted_link_id"] = results.get("promotedLinkId")
                    yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        stream_start, stream_end = self._get_time_interval(self.config.get("start_date"), self.config.get("end_date"))
        stream_conversion_count = self._get_bool_conversion_count_by_click_date(
            self.config.get("conversion_count", DEFAULT_REPORT_CONVERSION_COUNT_BY_CLICK_DATE)
        )
        return (
            f"reports/marketers/{stream_slice['marketer_id']}/campaigns/{stream_slice['campaign_id']}/periodicContent?from="
            + str(stream_start.date())
            + "&to="
            + str(stream_end.date())
            + "&breakdown="
            + str(self.config.get("report_granularity", DEFAULT_REPORT_GRANULARITY))
            + "&limit=500"
            + "&includeVideoStats=true"
            + "&conversionsByClickDate="
            + str(stream_conversion_count)
        )


# Retrieve performance statistics for a Marketer by publisher.
# Marketer performance statistics with breakdown by publisher with publishers that are already blocked at the marketer or campaign level.
class PerformanceReportMarketersByPublisher(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: Marketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for x in response.json().get("results"):
                x["marketer_id"] = stream_slice["marketer_id"]
                yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        stream_start, stream_end = self._get_time_interval(self.config.get("start_date"), self.config.get("end_date"))
        stream_conversion_count = self._get_bool_conversion_count_by_click_date(
            self.config.get("conversion_count", DEFAULT_REPORT_CONVERSION_COUNT_BY_CLICK_DATE)
        )
        return (
            f"reports/marketers/{stream_slice['marketer_id']}/publishers?from="
            + str(stream_start.date())
            + "&to="
            + str(stream_end.date())
            + "&limit=500"
            + "&includeVideoStats=true"
            + "&conversionsByClickDate="
            + str(stream_conversion_count)
        )


# Retrieve performance statistics for all marketer campaigns by publisher
# A special endpoint for retrieving publishers data by campaign breakdown.
class PerformanceReportPublishersByCampaigns(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: Marketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for fetched in response.json().get("campaignResults"):
                for x in fetched.get("results"):
                    x["marketer_id"] = stream_slice["marketer_id"]
                    x["campaign_id"] = fetched.get("campaignId")
                    yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        stream_start, stream_end = self._get_time_interval(self.config.get("start_date"), self.config.get("end_date"))
        stream_conversion_count = self._get_bool_conversion_count_by_click_date(
            self.config.get("conversion_count", DEFAULT_REPORT_CONVERSION_COUNT_BY_CLICK_DATE)
        )
        return (
            f"reports/marketers/{stream_slice['marketer_id']}/campaigns/publishers?from="
            + str(stream_start.date())
            + "&to="
            + str(stream_end.date())
            + "&limit=500"
            + "&includeVideoStats=true"
            + "&conversionsByClickDate="
            + str(stream_conversion_count)
        )


# Retrieve performance statistics for a Marketer by platform.
# The API in this sub-section allows retrieving performance statistics by platform at different levels: marketer, budget, and campaign.
class PerformanceReportMarketersByPlatforms(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: Marketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for x in response.json().get("results"):
                x["marketer_id"] = stream_slice["marketer_id"]
                yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        stream_start, stream_end = self._get_time_interval(self.config.get("start_date"), self.config.get("end_date"))
        stream_conversion_count = self._get_bool_conversion_count_by_click_date(
            self.config.get("conversion_count", DEFAULT_REPORT_CONVERSION_COUNT_BY_CLICK_DATE)
        )
        return (
            f"reports/marketers/{stream_slice['marketer_id']}/platforms?from="
            + str(stream_start.date())
            + "&to="
            + str(stream_end.date())
            + "&limit=500"
            + "&includeVideoStats=true"
            + "&conversionsByClickDate="
            + str(stream_conversion_count)
        )


# Retrieve performance statistics for all marketer campaigns by platform.
# A special endpoint for retrieving platforms data by campaign breakdown.
class PerformanceReportMarketersCampaignsByPlatforms(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: Marketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for fetched in response.json().get("campaignResults"):
                for x in fetched.get("results"):
                    x["marketer_id"] = stream_slice["marketer_id"]
                    x["campaign_id"] = fetched.get("campaignId")
                    yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        stream_start, stream_end = self._get_time_interval(self.config.get("start_date"), self.config.get("end_date"))
        stream_conversion_count = self._get_bool_conversion_count_by_click_date(
            self.config.get("conversion_count", DEFAULT_REPORT_CONVERSION_COUNT_BY_CLICK_DATE)
        )
        return (
            f"reports/marketers/{stream_slice['marketer_id']}/campaigns/platforms?from="
            + str(stream_start.date())
            + "&to="
            + str(stream_end.date())
            + "&limit=500"
            + "&includeVideoStats=true"
            + "&conversionsByClickDate="
            + str(stream_conversion_count)
        )


# Retrieve geo performance statistics for a Marketer.
# The API in this sub-section allows retrieving performance statistics by geographic breakdown at different levels: country, region, and subregion.
class PerformanceReportMarketersByGeoPerformance(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: Marketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for x in response.json().get("results"):
                x["marketer_id"] = stream_slice["marketer_id"]
                yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        stream_start, stream_end = self._get_time_interval(self.config.get("start_date"), self.config.get("end_date"))
        stream_conversion_count = self._get_bool_conversion_count_by_click_date(
            self.config.get("conversion_count", DEFAULT_REPORT_CONVERSION_COUNT_BY_CLICK_DATE)
        )
        return (
            f"reports/marketers/{stream_slice['marketer_id']}/geo?from="
            + str(stream_start.date())
            + "&to="
            + str(stream_end.date())
            + "&breakdown="
            + str(self.config.get("geo_location_breakdown", DEFAULT_GEO_LOCATION_BREAKDOWN))
            + "&limit=500"
            + "&includeVideoStats=true"
            + "&conversionsByClickDate="
            + str(stream_conversion_count)
        )


# Retrieve performance statistics for all marketer campaigns by geo.
# A special endpoint for retrieving geo data by campaign breakdown.
class PerformanceReportMarketersCampaignsByGeo(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: Marketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for fetched in response.json().get("campaignResults"):
                for x in fetched.get("results"):
                    x["marketer_id"] = stream_slice["marketer_id"]
                    x["campaign_id"] = fetched.get("campaignId")
                    yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        stream_start, stream_end = self._get_time_interval(self.config.get("start_date"), self.config.get("end_date"))
        stream_conversion_count = self._get_bool_conversion_count_by_click_date(
            self.config.get("conversion_count", DEFAULT_REPORT_CONVERSION_COUNT_BY_CLICK_DATE)
        )
        return (
            f"reports/marketers/{stream_slice['marketer_id']}/campaigns/geo?from="
            + str(stream_start.date())
            + "&to="
            + str(stream_end.date())
            + "&breakdown="
            + str(self.config.get("geo_location_breakdown", DEFAULT_GEO_LOCATION_BREAKDOWN))
            + "&limit=500"
            + "&includeVideoStats=true"
            + "&conversionsByClickDate="
            + str(stream_conversion_count)
        )


# Retrieve performance statistics for a Marketer by interest.
# The API in this sub-section allows retrieving performance statistics by interest at different levels: marketer and campaign.
class PerformanceReportMarketersByInterest(OutbrainAmplifyStream, HttpSubStream):
    primary_key = None

    def __init__(self, authenticator, config, parent: Marketers, **kwargs):
        super().__init__(parent=parent, **kwargs)
        self.config = config
        self._authenticator = authenticator
        self._session = requests.sessions.Session()

    def request_params(
        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
    ) -> MutableMapping[str, Any]:
        return {}

    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
        return None

    def stream_slices(
        self, sync_mode: SyncMode.full_refresh, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
    ) -> Iterable[Optional[Mapping[str, Any]]]:
        parent_stream_slices = self.parent.stream_slices(
            sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_state=stream_state
        )

        for stream_slice in parent_stream_slices:
            parent_records = self.parent.read_records(
                sync_mode=SyncMode.full_refresh, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state
            )

            for record in parent_records:
                yield {"marketer_id": record.get("id")}

    def parse_response(
        self,
        response: requests.Response,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Iterable[Mapping]:
        if response.json():
            for x in response.json().get("results"):
                x["marketer_id"] = stream_slice["marketer_id"]
                yield x

    def path(
        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
    ) -> str:
        stream_start, stream_end = self._get_time_interval(self.config.get("start_date"), self.config.get("end_date"))
        stream_conversion_count = self._get_bool_conversion_count_by_click_date(
            self.config.get("conversion_count", DEFAULT_REPORT_CONVERSION_COUNT_BY_CLICK_DATE)
        )
        return (
            f"reports/marketers/{stream_slice['marketer_id']}/interests?from="
            + str(stream_start.date())
            + "&to="
            + str(stream_end.date())
            + "&limit=500"
            + "&includeVideoStats=true"
            + "&conversionsByClickDate="
            + str(stream_conversion_count)
        )


class IncrementalOutbrainAmplifyStream(OutbrainAmplifyStream, ABC):
    state_checkpoint_interval = None

    @property
    def cursor_field(self) -> str:
        return []

    def get_updated_state(self, current_stream_state: MutableMapping[str, Any], latest_record: Mapping[str, Any]) -> Mapping[str, Any]:
        return {}


# Source
class SourceOutbrainAmplify(AbstractSource):
    def check_connection(self, logger, config) -> Tuple[bool, any]:
        url_base = OutbrainAmplifyStream.url_base
        auth = OutbrainAmplifyAuthenticator(url_base=url_base, config=config)
        try:
            auth.get_auth_header()
            marketer_stream = Marketers(authenticator=auth, config=config)
            next(marketer_stream.read_records(SyncMode.full_refresh))
            return True, None
        except Exception as e:
            return False, e

    def streams(self, config: Mapping[str, Any]) -> List[Stream]:
        url_base = OutbrainAmplifyStream.url_base
        auth = OutbrainAmplifyAuthenticator(url_base=url_base, config=config)
        # Basic stream marketing
        # 1. All Marketing streams
        stream = [Marketers(authenticator=auth, config=config)]

        # Campaigns Streams.
        # 1. Campaigns by marketers (implemented).
        # 2. Camapings Geo Location (implemented).
        stream.extend(
            [
                CampaignsByMarketers(authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)),
                CampaignsGeoLocation(
                    authenticator=auth,
                    config=config,
                    parent=CampaignsByMarketers(authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)),
                ),
            ]
        )

        # Budget for Marketers stream.
        # 1. Budget stream based on marketers id.
        (stream.extend([BudgetsForMarketers(authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config))]),)

        # Promoted Links stream.
        # 1. Promoted Links stream for campaigns.
        stream.extend(
            [
                PromotedLinksForCampaigns(
                    authenticator=auth,
                    config=config,
                    parent=CampaignsByMarketers(authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)),
                )
            ]
        )

        # Promoted Links Sequences stream.
        # 1. Promoted Links Sequences stream for campaigns.
        stream.extend(
            [
                PromotedLinksSequenceForCampaigns(
                    authenticator=auth,
                    config=config,
                    parent=CampaignsByMarketers(authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)),
                )
            ]
        )

        # Performance Reporting.
        # 1. Streams to retrieve performance statistics.
        stream.extend(
            [
                PerformanceReportCampaignsByMarketers(
                    authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)
                ),
                PerformanceReportPeriodicByMarketers(
                    authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)
                ),
                PerformanceReportPeriodicByMarketersCampaign(
                    authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)
                ),
                PerformanceReportPeriodicContentByPromotedLinksCampaign(
                    authenticator=auth,
                    config=config,
                    parent=CampaignsByMarketers(authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)),
                ),
                PerformanceReportMarketersByPublisher(
                    authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)
                ),
                PerformanceReportPublishersByCampaigns(
                    authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)
                ),
                PerformanceReportMarketersByPlatforms(
                    authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)
                ),
                PerformanceReportMarketersCampaignsByPlatforms(
                    authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)
                ),
                PerformanceReportMarketersByGeoPerformance(
                    authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)
                ),
                PerformanceReportMarketersCampaignsByGeo(
                    authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)
                ),
                PerformanceReportMarketersByInterest(
                    authenticator=auth, config=config, parent=Marketers(authenticator=auth, config=config)
                ),
            ]
        )
        return stream
