diff --git a/analytix/__init__.py b/analytix/__init__.py index 0d6488f..361d2b8 100644 --- a/analytix/__init__.py +++ b/analytix/__init__.py @@ -1,9 +1,10 @@ __productname__ = "analytix" -__version__ = "1.2.0" +__version__ = "2.0.0.dev4" __description__ = "A simple yet powerful API wrapper to make getting analytical information from the YouTube Analytics API easier than ever." __url__ = "https://github.com/parafoxia/analytix" __docs__ = "https://analytix.readthedocs.io/en/latest/" __author__ = "Ethan Henderson" __license__ = "BSD-3-Clause" +__bugtracker__ = "https://github.com/parafoxia/analytix/issues" -from .errors import * +from .youtube.analytics.api import YouTubeAnalytics diff --git a/analytix/errors.py b/analytix/errors.py index 8c16f8f..38bbbca 100644 --- a/analytix/errors.py +++ b/analytix/errors.py @@ -4,25 +4,21 @@ class AnalytixError(Exception): pass -class NoAuthorisedService(AnalytixError): - """Exception thrown when an operatoion that requires authorisation is attempted while no service is no authorised.""" - +class InvalidScopes(AnalytixError): pass -class ServiceAlreadyExists(AnalytixError): - """Exception thrown when an attempt to create a service is made while one already exists.""" - +class InvalidRequest(AnalytixError): pass -class IncompleteRequest(AnalytixError): - """Exception throws when not enough information has been passed to a request.""" - +class Deprecated(AnalytixError): pass -class InvalidRequest(AnalytixError): - """Exception throws when invalid information has been passed to a request.""" +class MissingOptionalComponents(AnalytixError): + pass + +class HTTPError(AnalytixError): pass diff --git a/analytix/iso.py b/analytix/iso.py new file mode 100644 index 0000000..0c64e6d --- /dev/null +++ b/analytix/iso.py @@ -0,0 +1,482 @@ +COUNTRIES = ( + "AW", + "AF", + "AO", + "AI", + "AX", + "AL", + "AD", + "AE", + "AR", + "AM", + "AS", + "AQ", + "TF", + "AG", + "AU", + "AT", + "AZ", + "BI", + "BE", + "BJ", + "BQ", + "BF", + "BD", + "BG", + "BH", + "BS", + "BA", + "BL", + "BY", + "BZ", + "BM", + "BO", + "BR", + "BB", + "BN", + "BT", + "BV", + "BW", + "CF", + "CA", + "CC", + "CH", + "CL", + "CN", + "CI", + "CM", + "CD", + "CG", + "CK", + "CO", + "KM", + "CV", + "CR", + "CU", + "CW", + "CX", + "KY", + "CY", + "CZ", + "DE", + "DJ", + "DM", + "DK", + "DO", + "DZ", + "EC", + "EG", + "ER", + "EH", + "ES", + "EE", + "ET", + "FI", + "FJ", + "FK", + "FR", + "FO", + "FM", + "GA", + "GB", + "GE", + "GG", + "GH", + "GI", + "GN", + "GP", + "GM", + "GW", + "GQ", + "GR", + "GD", + "GL", + "GT", + "GF", + "GU", + "GY", + "HK", + "HM", + "HN", + "HR", + "HT", + "HU", + "ID", + "IM", + "IN", + "IO", + "IE", + "IR", + "IQ", + "IS", + "IL", + "IT", + "JM", + "JE", + "JO", + "JP", + "KZ", + "KE", + "KG", + "KH", + "KI", + "KN", + "KR", + "KW", + "LA", + "LB", + "LR", + "LY", + "LC", + "LI", + "LK", + "LS", + "LT", + "LU", + "LV", + "MO", + "MF", + "MA", + "MC", + "MD", + "MG", + "MV", + "MX", + "MH", + "MK", + "ML", + "MT", + "MM", + "ME", + "MN", + "MP", + "MZ", + "MR", + "MS", + "MQ", + "MU", + "MW", + "MY", + "YT", + "NA", + "NC", + "NE", + "NF", + "NG", + "NI", + "NU", + "NL", + "NO", + "NP", + "NR", + "NZ", + "OM", + "PK", + "PA", + "PN", + "PE", + "PH", + "PW", + "PG", + "PL", + "PR", + "KP", + "PT", + "PY", + "PS", + "PF", + "QA", + "RE", + "RO", + "RU", + "RW", + "SA", + "SD", + "SN", + "SG", + "GS", + "SH", + "SJ", + "SB", + "SL", + "SV", + "SM", + "SO", + "PM", + "RS", + "SS", + "ST", + "SR", + "SK", + "SI", + "SE", + "SZ", + "SX", + "SC", + "SY", + "TC", + "TD", + "TG", + "TH", + "TJ", + "TK", + "TM", + "TL", + "TO", + "TT", + "TN", + "TR", + "TV", + "TW", + "TZ", + "UG", + "UA", + "UM", + "UY", + "US", + "UZ", + "VA", + "VC", + "VE", + "VG", + "VI", + "VN", + "VU", + "WF", + "WS", + "YE", + "ZA", + "ZM", + "ZW", +) +SUBDIVISIONS = ( + "US-OH", + "US-IN", + "US-OK", + "US-KS", + "US-OR", + "US-KY", + "US-PA", + "US-LA", + "US-AK", + "US-PR", + "US-MA", + "US-AL", + "US-RI", + "US-MD", + "US-AR", + "US-SC", + "US-ME", + "US-AS", + "US-SD", + "US-MI", + "US-AZ", + "US-TN", + "US-MN", + "US-CA", + "US-TX", + "US-MO", + "US-CO", + "US-UM", + "US-MP", + "US-CT", + "US-UT", + "US-MS", + "US-DC", + "US-VA", + "US-MT", + "US-DE", + "US-VI", + "US-NC", + "US-FL", + "US-VT", + "US-ND", + "US-GA", + "US-WA", + "US-NE", + "US-GU", + "US-WI", + "US-NH", + "US-HI", + "US-WV", + "US-NJ", + "US-IA", + "US-WY", + "US-NM", + "US-ID", + "US-NV", + "US-IL", + "US-NY", +) +CURRENCIES = ( + "AED", + "AFN", + "ALL", + "AMD", + "ANG", + "AOA", + "ARS", + "AUD", + "AWG", + "AZN", + "BAM", + "BBD", + "BDT", + "BGN", + "BHD", + "BIF", + "BMD", + "BND", + "BOB", + "BRL", + "BSD", + "BTN", + "BWP", + "BYN", + "BZD", + "CAD", + "CDF", + "CHF", + "CLP", + "CNY", + "COP", + "CRC", + "CUC", + "CUP", + "CVE", + "CZK", + "DJF", + "DKK", + "DOP", + "DZD", + "EGP", + "ERN", + "ETB", + "EUR", + "FJD", + "FKP", + "GBP", + "GEL", + "GHS", + "GIP", + "GMD", + "GNF", + "GTQ", + "GYD", + "HKD", + "HNL", + "HRK", + "HTG", + "HUF", + "IDR", + "ILS", + "INR", + "IQD", + "IRR", + "ISK", + "JMD", + "JOD", + "JPY", + "KES", + "KGS", + "KHR", + "KMF", + "KPW", + "KRW", + "KWD", + "KYD", + "KZT", + "LAK", + "LBP", + "LKR", + "LRD", + "LSL", + "LYD", + "MAD", + "MDL", + "MGA", + "MKD", + "MMK", + "MNT", + "MOP", + "MRO", + "MUR", + "MVR", + "MWK", + "MXN", + "MYR", + "MZN", + "NAD", + "NGN", + "NIO", + "NOK", + "NPR", + "NZD", + "OMR", + "PAB", + "PEN", + "PGK", + "PHP", + "PKR", + "PLN", + "PYG", + "QAR", + "RON", + "RSD", + "RUB", + "RWF", + "SAR", + "SBD", + "SCR", + "SDG", + "SEK", + "SGD", + "SHP", + "SLL", + "SOS", + "SRD", + "SSP", + "STD", + "SVC", + "SYP", + "SZL", + "THB", + "TJS", + "TMT", + "TND", + "TOP", + "TRY", + "TTD", + "TWD", + "TZS", + "UAH", + "UGX", + "USD", + "UYU", + "UZS", + "VEF", + "VND", + "VUV", + "WST", + "XAF", + "XAG", + "XAU", + "XBA", + "XBB", + "XBC", + "XBD", + "XCD", + "XDR", + "XOF", + "XPD", + "XPF", + "XPT", + "XSU", + "XTS", + "XUA", + "XXX", + "YER", + "ZAR", + "ZMW", + "ZWL", +) diff --git a/analytix/packages.py b/analytix/packages.py new file mode 100644 index 0000000..c539f3d --- /dev/null +++ b/analytix/packages.py @@ -0,0 +1,34 @@ +__all__ = ( + "PANDAS_AVAILABLE", + "NUMPY_AVAILABLE", + "requires_pandas", + "requires_numpy", +) + +from pkg_resources import working_set + +from analytix.errors import MissingOptionalComponents + +_packages = [p.key for p in working_set] +PANDAS_AVAILABLE = "pandas" in _packages +NUMPY_AVAILABLE = "numpy" in _packages + + +def requires_pandas(func): + def wrapper(*args, **kwargs): + if not PANDAS_AVAILABLE: + raise MissingOptionalComponents("pandas is not installed") + + return func(*args, **kwargs) + + return wrapper + + +def requires_numpy(func): + def wrapper(*args, **kwargs): + if not NUMPY_AVAILABLE: + raise MissingOptionalComponents("numpy is not installed") + + return func(*args, **kwargs) + + return wrapper diff --git a/analytix/secrets.py b/analytix/secrets.py new file mode 100644 index 0000000..77ff557 --- /dev/null +++ b/analytix/secrets.py @@ -0,0 +1,9 @@ +import os +from pathlib import Path + +if os.name == "nt": + TOKEN_STORE = Path("%USERPROFILE%/.analytix") +else: + TOKEN_STORE = Path(f"/home/{os.environ['USER']}/.analytix") + +YT_ANALYTICS_TOKEN = "youtube-analytics-token.json" diff --git a/analytix/youtube/__init__.py b/analytix/youtube/__init__.py index 3ebed60..e69de29 100644 --- a/analytix/youtube/__init__.py +++ b/analytix/youtube/__init__.py @@ -1,9 +0,0 @@ -YOUTUBE_ANALYTICS_API_SERVICE_NAME = "youtubeAnalytics" -YOUTUBE_ANALYTICS_API_VERSION = "v2" -YOUTUBE_ANALYTICS_SCOPES = ( - "https://www.googleapis.com/auth/yt-analytics.readonly", - "https://www.googleapis.com/auth/yt-analytics-monetary.readonly", -) - -from .analytics import YouTubeAnalytics, YouTubeAnalyticsReport -from .service import YouTubeService diff --git a/analytix/youtube/analytics.py b/analytix/youtube/analytics.py deleted file mode 100644 index e457d5c..0000000 --- a/analytix/youtube/analytics.py +++ /dev/null @@ -1,196 +0,0 @@ -import datetime as dt -import json - -import pandas as pd - -from analytix import InvalidRequest, NoAuthorisedService -from analytix.youtube import features, reports - - -class YouTubeAnalyticsReport: - """A class that provides an interface for working with data retrieved from the YouTube Analytics API. - - Args: - data (dict): The response data from the YouTube Analytics API. - type_ (ReportType): The type of report generated. - - Attributes: - data (dict): The response data from the YouTube Analytics API. - ncolumns (int): The number of columns in the report. - nrows (int): The number of rows in the report. - type (ReportType): The type of report generated. - """ - - __slots__ = ("data", "ncolumns", "nrows", "type") - - def __init__(self, data, type_): - self.data = data - self.ncolumns = len(data["columnHeaders"]) - self.nrows = len(data["rows"]) - self.type = type_ - - def to_dataframe(self): - """Returns the report as a Pandas DataFrame. - - Returns: - pandas.DataFrame: The DataFrame object. - """ - df = pd.DataFrame() - df = df.append(self.data["rows"]) - df.columns = [c["name"] for c in self.data["columnHeaders"]] - return df - - def to_json(self, file, indent=4): - """Exports the report to a JSON file in the same format as it was provided by the YouTube Analytics API. - - Args: - file (str | os.PathLike): The path in which the report should be saved. - indent (int): The number of spaces to use for indentation. Defaults to 4. - """ - if not file.endswith(".json"): - file += ".json" - with open(file, "w", encoding="utf-8") as f: - json.dump(self.data, f, indent=indent, ensure_ascii=False) - - def to_csv(self, file): - """Exports the report to a CSV file. - - Args: - file (str | os.PathLike): The path in which the report should be saved. - """ - if not file.endswith(".csv"): - file += ".csv" - self.to_dataframe().to_csv(file) - - def __str__(self): - return self.type - - -class YouTubeAnalytics: - """A class to retrieve data from the YouTube Analytics API. - - Args: - service (YouTubeService): The YouTube service to perform the operation on. - - Attributes: - service (YouTubeService): The YouTube service to perform the operation on. - """ - - __slots__ = ("service",) - - def __init__(self, service): - self.service = service - - def get_report_type(self, metrics=(), dimensions=(), filters={}, *, verify=True): - """Gets the report type that best matches the metrics, dimensions, and filters given. If :code:`verify` is False, this will return a :code:`Generic` report type. - - Args: - metrics (tuple[str, ...] | list[str]): The metrics (or columns) to retrieve. Defaults to an empty tuple. - dimensions (tuple[str, ...] | list[str]): The dimensions in which data is split. Defaults to an empty tuple. - filters (dict[str, str]): The filters to be applied when retrieving data. Defaults to an empty dictionary. - verify (bool): Whether to attempt to determine the report type dynamically. Defaults to True. - - Returns: - ReportType: The selected report type. - """ - if not verify: - return reports.Generic() - return reports.determine(metrics, dimensions, filters)() - - def get_verified_report_type(self, metrics=(), dimensions=(), filters={}): - """Like :code:`get_report_type`, but only returns a ReportType object if verification succeeds. - - Args: - metrics (tuple[str, ...] | list[str]): The metrics (or columns) to retrieve. Defaults to an empty tuple. - dimensions (tuple[str, ...] | list[str]): The dimensions in which data is split. Defaults to an empty tuple. - filters (dict[str, str]): The filters to be applied when retrieving data. Defaults to an empty dictionary. - - Returns: - ReportType | InvalidRequest: The selected report type or the error that caused verification to fail. - """ - r = self.get_report_type(metrics, dimensions, filters) - try: - r.verify(metrics, dimensions, filters, 5, ("views",)) - return r - except InvalidRequest as exc: - return exc - - def retrieve(self, metrics="all", verify=True, **kwargs): - """Executes an API request to pull down analytical information. - - .. warning:: - - The :code:`start_date` and :code:`end_date` parameters do not account for delayed analytics such as revenue. - - .. note:: - - This will retrieve video reports by default. To retrieve playlist reports, include '"isCurated": "1"' in your filters. - - Args: - metrics (tuple[str, ...] | list[str]): The metrics (or columns) to retrieve. Defaults to "all". - verify (bool): Whether to verify the requests before passing them to the API. Defaults to True. - start_date (datetime.date): The earliest date to fetch data for. Defaults to 28 days before the current date. - end_date (datetime.date): The latest date to fetch data for. Defaults to the current date. - currency (str): The currency to use for monetary analytics. This should be passed in ISO 4217 format. Defaults to USD. - dimensions (tuple[str, ...] | list[str]): The dimensions in which data is split. Defaults to an empty tuple. - filters (dict[str, str]): The filters to be applied when retrieving data. Defaults to an empty dictionary. - include_historical_data (bool): Whether data before the point in which the specified channel(s) were linked to the content owner should be retrieved. Defaults to False. - max_results (int | None): The maximum number of rows to fetch. Defaults to None. - sort_by (tuple[str, ...] | list[str]): The dimensions in which to sort the data. Defaults to an empty tuple. - start_index (int): The index of the first row to retrieve. Note that this is one-indexed. Defaults to 1. - - Returns: - YouTubeAnalyticsReport: The report object containing the data fetched from the YouTube Analytics API. - - Raises: - NoAuthorisedService: The given service has not been authorised. - InvalidRequest: The request was invalid. - """ - if not self.service.active: - raise NoAuthorisedService("the YouTube service has not been authorised") - - start_date = kwargs.get("start_date", dt.date.today() - dt.timedelta(days=28)) - end_date = kwargs.get("end_date", dt.date.today()) - currency = kwargs.get("currency", "USD") - dimensions = kwargs.get("dimensions", ()) - filters = kwargs.get("filters", {}) - include_historical_data = kwargs.get("include_historical_data", False) - max_results = kwargs.get("max_results", None) - sort_by = kwargs.get("sort_by", ()) - start_index = kwargs.get("start_index", 1) - - if not isinstance(dimensions, (tuple, list, set)): - raise InvalidRequest(f"expected tuple, list, or set of dimensions, got {type(dimensions).__name__}") - - if not isinstance(filters, dict): - raise InvalidRequest(f"expected dict of filters, got {type(filters).__name__}") - - r = self.get_report_type_for(metrics, dimensions, filters, verify=verify) - r.verify(metrics, dimensions, filters, max_results, sort_by) - - if metrics == "all": - if not verify: - raise InvalidRequest("you must manually specify a list of metrics when not verifying") - metrics = r.metrics - - if filters: - filters = tuple(f"{k}=={v}" for k, v in filters.items()) - - return YouTubeAnalyticsReport( - self.service.active.reports() - .query( - ids="channel==MINE", - metrics=",".join(metrics), - startDate=start_date, - endDate=end_date, - currency=currency, - dimensions=",".join(dimensions), - filters=";".join(filters), - includeHistoricalChannelData=include_historical_data, - maxResults=max_results, - sort=",".join(sort_by), - startIndex=start_index, - ) - .execute(), - r, - ) diff --git a/analytix/youtube/analytics/__init__.py b/analytix/youtube/analytics/__init__.py new file mode 100644 index 0000000..b1db611 --- /dev/null +++ b/analytix/youtube/analytics/__init__.py @@ -0,0 +1,6 @@ +YOUTUBE_ANALYTICS_API_SERVICE_NAME = "youtubeAnalytics" +YOUTUBE_ANALYTICS_API_VERSION = "v2" +YOUTUBE_ANALYTICS_SCOPES = ( + "https://www.googleapis.com/auth/yt-analytics.readonly", + "https://www.googleapis.com/auth/yt-analytics-monetary.readonly", +) diff --git a/analytix/youtube/analytics/api.py b/analytix/youtube/analytics/api.py new file mode 100644 index 0000000..5cc36a7 --- /dev/null +++ b/analytix/youtube/analytics/api.py @@ -0,0 +1,308 @@ +import csv +import datetime as dt +import json +import logging +import os +import time +import typing as t + +import requests +from requests_oauthlib import OAuth2Session + +from analytix.errors import * +from analytix.iso import CURRENCIES +from analytix.packages import * +from analytix.secrets import TOKEN_STORE, YT_ANALYTICS_TOKEN +from analytix.youtube.analytics import * +from analytix.youtube.analytics import verify + +if PANDAS_AVAILABLE: + import pandas as pd +if NUMPY_AVAILABLE: + import numpy as np + + +class YouTubeAnalytics: + __slots__ = ("_session", "secrets", "project_id", "_token") + + def __init__(self, session, secrets, **kwargs): + self._session = session + self.secrets = secrets + self.project_id = secrets["project_id"] + self._token = self._get_token() + + def __str__(self): + return self.project_id + + def __repr__(self): + return f"" + + @property + def authorised(self): + return bool(self._token) + + authorized = authorised + + @classmethod + def from_file(cls, path, *, scopes="all", **kwargs): + if not os.path.isfile(path): + raise FileNotFoundError( + "you must provide a valid path to a secrets file" + ) + + with open(path, mode="r", encoding="utf-8") as f: + logging.debug("Secrets file loaded") + secrets = json.load(f)["installed"] + + scopes = cls._resolve_scopes(scopes) + session = OAuth2Session( + secrets["client_id"], + redirect_uri=secrets["redirect_uris"][0], + scope=scopes, + **kwargs, + ) + return cls(session, secrets) + + @classmethod + def from_dict(cls, secrets, *, scopes="all", **kwargs): + scopes = cls._resolve_scopes(scopes) + session = OAuth2Session( + secrets["installed"]["client_id"], + redirect_uri=secrets["redirect_uris"][0], + scope=scopes, + **kwargs, + ) + return cls(session, secrets["installed"]) + + @staticmethod + def _resolve_scopes(scopes): + if scopes == "all": + logging.debug(f"Scopes set to {YOUTUBE_ANALYTICS_SCOPES}") + return YOUTUBE_ANALYTICS_SCOPES + + if not isinstance(scopes, (tuple, list, set)): + raise InvalidScopes( + f"expected tuple, list, or set of scopes, got {type(scopes).__name__}" + ) + + for i, scope in enumerate(scopes[:]): + if not scope.startswith("https://www.googleapis.com/auth/"): + scopes[i] = "https://www.googleapis.com/auth/" + scope + + diff = set(scopes) - set(YOUTUBE_ANALYTICS_SCOPES) + if diff: + raise InvalidScopes( + "one or more scopes you provided are invalid: " + + ", ".join(diff) + ) + + logging.debug(f"Scopes set to {scopes}") + return scopes + + @staticmethod + def _get_token(): + if not os.path.isfile(TOKEN_STORE / YT_ANALYTICS_TOKEN): + logging.info("No token found. You will need to authorise") + return "" + + with open( + TOKEN_STORE / YT_ANALYTICS_TOKEN, mode="r", encoding="utf-8" + ) as f: + data = json.load(f) + if time.time() > data["expires"]: + logging.info( + "Token found, but it has expired. You will need to authorise" + ) + return "" + + logging.info( + "Valid token found! analytix will use this, so you don't need to authorise" + ) + return data["token"] + + def authorise(self, store_token=True, force=False, **kwargs): + if self._token and not force: + logging.warning("Client is already authorised! Skipping...") + return + + url, _ = self._session.authorization_url( + self.secrets["auth_uri"], **kwargs + ) + code = input(f"You need to authorise the session: {url}\nCODE > ") + token = self._session.fetch_token( + self.secrets["token_uri"], + code=code, + client_secret=self.secrets["client_secret"], + ) + self._token = token["access_token"] + logging.info("Token retrieved") + + if not store_token: + logging.info("Not storing token, as instructed") + return + + os.makedirs(TOKEN_STORE, exist_ok=True) + with open( + TOKEN_STORE / YT_ANALYTICS_TOKEN, mode="w", encoding="utf-8" + ) as f: + json.dump( + {"token": self._token, "expires": token["expires_at"]}, + f, + ensure_ascii=False, + ) + logging.info(f"Key stored in {TOKEN_STORE / YT_ANALYTICS_TOKEN}") + + authorize = authorise + + def retrieve( + self, start_date, end_date=dt.date.today(), metrics="all", **kwargs + ): + currency = kwargs.pop("currency", "USD") + dimensions = kwargs.pop("dimensions", ()) + filters = kwargs.pop("filters", {}) + include_historical_data = kwargs.pop("include_historical_data", False) + max_results = kwargs.pop("max_results", 0) + sort_by = kwargs.pop("sort_by", ()) + start_index = kwargs.pop("start_index", 1) + + logging.debug("Verifying options...") + if "7DayTotals" in dimensions or "30DayTotals" in dimensions: + raise Deprecated( + "the '7DayTotals' and '30DayTotals' dimensions were deprecated, and can no longer be used" + ) + if not isinstance(start_date, dt.date): + raise InvalidRequest( + f"expected start date as date object, got {type(start_date).__name__}" + ) + if not isinstance(end_date, dt.date): + raise InvalidRequest( + f"expected end date as date object, got {type(end_date).__name__}" + ) + if end_date <= start_date: + raise InvalidRequest( + f"the start date should be earlier than the end date" + ) + if currency not in CURRENCIES: + raise InvalidRequest( + f"expected existing currency as ISO 4217 alpha-3 code, got {currency}" + ) + if not isinstance(dimensions, (tuple, list, set)): + raise InvalidRequest( + f"expected tuple, list, or set of dimensions, got {type(dimensions).__name__}" + ) + if not isinstance(filters, dict): + raise InvalidRequest( + f"expected dict of filters, got {type(filters).__name__}" + ) + if not isinstance(include_historical_data, bool): + raise InvalidRequest( + f"expected bool for 'include_historical_data', got {type(include_historical_data).__name__}" + ) + if not isinstance(max_results, int): + raise InvalidRequest( + f"expected int for 'max_results', got {type(max_results).__name__}" + ) + if max_results < 0: + raise InvalidRequest( + f"the maximum number of results should be no less than 0 (0 for unlimited results)" + ) + if not isinstance(sort_by, (tuple, list, set)): + raise InvalidRequest( + f"expected tuple, list, or set of sorting columns, got {type(sort_by).__name__}" + ) + if not isinstance(start_index, int): + raise InvalidRequest( + f"expected int for 'start_index', got {type(start_index).__name__}" + ) + if start_index < 1: + raise InvalidRequest(f"the start index should be no less than 1") + + logging.debug("Determining report type...") + rtype = verify.rtypes.determine(dimensions, metrics, filters)() + logging.info(f"Report type determined as: {rtype}") + + if metrics == "all": + metrics = tuple(rtype.metrics) + elif not isinstance(metrics, (tuple, list, set)): + raise InvalidRequest( + f"expected tuple, list, or set of metrics, got {type(metrics).__name__}" + ) + logging.debug("Using these metrics: " + ", ".join(metrics)) + + logging.debug("Verifying report...") + rtype.verify(dimensions, metrics, filters, max_results, sort_by) + logging.debug("Verification complete") + + url = ( + f"https://youtubeanalytics.googleapis.com/{YOUTUBE_ANALYTICS_API_VERSION}/reports" + "?ids=channel==MINE" + f"&metrics={','.join(metrics)}" + f"&startDate={start_date.strftime('%Y-%m-%d')}" + f"&endDate={end_date.strftime('%Y-%m-%d')}" + f"¤cy={currency}" + f"&dimensions={','.join(dimensions)}" + f"&filters={';'.join(f'{k}=={v}' for k, v in filters.items())}" + f"&includeHistorialChannelData={f'{include_historical_data}'.lower()}" + f"&maxResults={max_results}" + f"&sort={','.join(sort_by)}" + f"&startIndex={start_index}" + ) + logging.debug(f"URL: {url}") + + if not self._token: + self.authorise() + + with requests.get( + url, headers={"Authorization": f"Bearer {self._token}"} + ) as r: + data = r.json() + + if next(iter(data)) == "error": + error = data["error"] + raise HTTPError(f"{error['code']}: {error['message']}") + + logging.info("Creating report...") + return YouTubeAnalyticsReport( + f"{rtype}", [c["name"] for c in data["columnHeaders"]], data["rows"] + ) + + +class YouTubeAnalyticsReport: + __slots__ = ("type", "columns", "rows", "_ncolumns", "_nrows") + + def __init__(self, type, columns, rows): + self.type = type + self.columns = columns + if NUMPY_AVAILABLE: + self.rows = np.array(rows) + else: + self.rows = rows + self._ncolumns = len(self.columns) + self._nrows = len(self.rows) + + def __repr__(self): + return f"" + + @property + def shape(self): + return (self._nrows, self._ncolumns) + + @requires_pandas + def to_dataframe(self): + df = pd.DataFrame(self.rows) + df.columns = self.columns + if "day" in df.columns: + df["day"] = pd.to_datetime(df["day"], format="%Y-%m-%d") + if "month" in df.columns: + df["month"] = pd.to_datetime(df["month"], format="%Y-%m") + return df + + def to_csv(self, path, *, delimiter=","): + if not path.endswith(".csv"): + path += ".csv" + + with open(path, mode="w", encoding="utf-8") as f: + writer = csv.writer(f, delimiter=delimiter) + writer.writerow(self.columns) + for r in self.rows: + writer.writerow(r) diff --git a/analytix/youtube/analytics/verify/__init__.py b/analytix/youtube/analytics/verify/__init__.py new file mode 100644 index 0000000..fe88705 --- /dev/null +++ b/analytix/youtube/analytics/verify/__init__.py @@ -0,0 +1,2 @@ +from . import features, rtypes +from .constants import * diff --git a/analytix/youtube/analytics/verify/constants.py b/analytix/youtube/analytics/verify/constants.py new file mode 100644 index 0000000..6f8625b --- /dev/null +++ b/analytix/youtube/analytics/verify/constants.py @@ -0,0 +1,436 @@ +from analytix.iso import COUNTRIES, SUBDIVISIONS + + +YOUTUBE_ANALYTICS_CORE_DIMENSIONS = ( + "ageGroup", + "channel", + "country", + "day", + "gender", + "month", + "sharingService", + "uploaderType", + "video", +) + +YOUTUBE_ANALYTICS_CONTENT_OWNER_DIMENSIONS = ( + "claimedStatus", + "uploaderType", +) + +YOUTUBE_ANALYTICS_ALL_DIMENSIONS = ( + "video", + "playlist", + "channel", + "country", + "province", + "day", + "month", + "insightPlaybackLocationType", + "insightPlaybackLocationDetail", + "liveOrOnDemand", + "subscribedStatus", + "youtubeProduct", + "insightTrafficSourceType", + "insightTrafficSourceDetail", + "deviceType", + "operatingSystem", + "ageGroup", + "gender", + "sharingService", + "elapsedVideoTimeRatio", + "audienceType", + "adType", + "claimedStatus", + "uploaderType", +) + +YOUTUBE_ANALYTICS_VALID_FILTER_OPTIONS = { + "video": (), + "playlist": (), + "channel": (), + "group": (), + "country": COUNTRIES, + "province": SUBDIVISIONS, + "continent": ( + "002", + "019", + "142", + "150", + "009", + ), + "subContinent": ( + "014", + "017", + "015", + "018", + "011", + "029", + "013", + "021", + "005", + "143", + "030", + "034", + "035", + "145", + "151", + "154", + "039", + "155", + "053", + "054", + "057", + "061", + ), + "day": (), + "month": (), + "insightPlaybackLocationType": ( + "BROWSE", + "CHANNEL", + "EMBEDDED", + "EXTERNAL_APP", + "MOBILE", + "SEARCH", + "WATCH", + "YT_OTHER", + ), + "insightPlaybackLocationDetail": (), + "liveOrOnDemand": ( + "LIVE", + "ON_DEMAND", + ), + "subscribedStatus": ( + "SUBSCRIBED", + "UNSUBSCRIBED", + ), + "youtubeProduct": ( + "CORE", + "GAMING", + "KIDS", + "UNKNOWN", + ), + "insightTrafficSourceType": ( + "ADVERTISING", + "ANNOTATION", + "CAMPAIGN_CARD", + "END_SCREEN", + "EXT_URL", + "NO_LINK_EMBEDDED", + "NO_LINK_OTHER", + "NOTIFICATION", + "PLAYLIST", + "PROMOTED", + "RELATED_VIDEO", + "SHORTS", + "SUBSCRIBER", + "YT_CHANNEL", + "YT_OTHER_PAGE", + "YT_PLAYLIST_PAGE", + "YT_SEARCH", + ), + "insightTrafficSourceDetail": ( + "ADVERTISING", + "CAMPAIGN_CARD", + "END_SCREEN", + "EXT_URL", + "NOTIFICATION", + "RELATED_VIDEO", + "SUBSCRIBER", + "YT_CHANNEL", + "YT_OTHER_PAGE", + "YT_SEARCH", + ), + "deviceType": ( + "DESKTOP", + "GAME_CONSOLE", + "MOBILE", + "TABLET", + "TV", + "UNKNOWN_PLATFORM", + ), + "operatingSystem": ( + "ANDROID", + "BADA", + "BLACKBERRY", + "CHROMECAST", + "DOCOMO", + "FIREFOX", + "HIPTOP", + "IOS", + "KAIOS", + "LINUX", + "MACINTOSH", + "MEEGO", + "NINTENDO_3DS", + "OTHER", + "PLAYSTATION", + "PLAYSTATION_VITA", + "REALMEDIA", + "SMART_TV", + "SYMBIAN", + "TIZEN", + "WEBOS", + "WII", + "WINDOWS", + "WINDOWS_MOBILE", + "XBOX", + ), + "ageGroup": ( + "age13-17", + "age18-24", + "age25-34", + "age35-44", + "age45-54", + "age55-64", + "age65-", + ), + "gender": ( + "female", + "male", + ), + "sharingService": ( + "AMEBA", + "ANDROID_EMAIL", + "ANDROID_MESSENGER", + "ANDROID_MMS", + "BBM", + "BLOGGER", + "COPY_PASTE", + "CYWORLD", + "DIGG", + "DROPBOX", + "EMBED", + "MAIL", + "FACEBOOK", + "FACEBOOK_MESSENGER", + "FACEBOOK_PAGES", + "FOTKA", + "GMAIL", + "GOO", + "GOOGLEPLUS", + "GO_SMS", + "GROUPME", + "HANGOUTS", + "HI5", + "HTC_MMS", + "INBOX", + "IOS_SYSTEM_ACTIVITY_DIALOG", + "KAKAO_STORY", + "KAKAO", + "KIK", + "LGE_EMAIL", + "LINE", + "LINKEDIN", + "LIVEJOURNAL", + "MENEAME", + "MIXI", + "MOTOROLA_MESSAGING", + "MYSPACE", + "NAVER", + "NEARBY_SHARE", + "NUJIJ", + "ODNOKLASSNIKI", + "OTHER", + "PINTEREST", + "RAKUTEN", + "REDDIT", + "SKYPE", + "SKYBLOG", + "SONY_CONVERSATIONS", + "STUMBLEUPON", + "TELEGRAM", + "TEXT_MESSAGE", + "TUENTI", + "TUMBLR", + "TWITTER", + "UNKNOWN", + "VERIZON_MMS", + "VIBER", + "VKONTATKE", + "WECHAT", + "WEIBO", + "WHATS_APP", + "WYKOP", + "YAHOO", + "YOUTUBE_GAMING", + "YOUTUBE_KIDS", + "YOUTUBE_MUSIC", + "YOUTUBE_TV", + ), + "elapsedVideoTimeRatio": tuple(f"{n/100}" for n in range(1, 101)), + "audienceType": ( + "ORGANIC", + "AD_INSTREAM", + "AD_INDISPLAY", + ), + "adType": ( + "auctionBumperInstream", + "auctionDisplay", + "auctionInstream", + "auctionTrueviewInslate", + "auctionTrueviewInstream", + "auctionUnknown", + "reservedBumperInstream", + "reservedClickToPlay", + "reservedDisplay", + "reservedInstream", + "reservedInstreamSelect", + "reservedMasthead", + "reservedUnknown", + "unknown", + ), + "claimedStatus": ("claimed",), + "uploaderType": ( + "self", + "thirdParty", + ), +} + +YOUTUBE_ANALYTICS_ALL_FILTERS = tuple( + YOUTUBE_ANALYTICS_VALID_FILTER_OPTIONS.keys() +) + +YOUTUBE_ANALYTICS_CORE_METRICS = ( + "annotationClickThroughRate", + "annotationCloseRate", + "averageViewDuration", + "comments", + "dislikes", + "estimatedMinutesWatched", + "estimatedRevenue", + "likes", + "shares", + "subscribersGained", + "subscribersLost", + "viewerPercentage", + "views", +) + +YOUTUBE_ANALYTICS_ALL_METRICS = ( + "views", + "redViews", + "comments", + "likes", + "dislikes", + "videosAddedToPlaylists", + "videosRemovedFromPlaylists", + "shares", + "estimatedMinutesWatched", + "estimatedRedMinutesWatched", + "averageViewDuration", + "averageViewPercentage", + "annotationClickThroughRate", + "annotationCloseRate", + "annotationImpressions", + "annotationClickableImpressions", + "annotationClosableImpressions", + "annotationClicks", + "annotationCloses", + "cardClickRate", + "cardTeaserClickRate", + "cardImpressions", + "cardTeaserImpressions", + "cardClicks", + "cardTeaserClicks", + "subscribersGained", + "subscribersLost", + "estimatedRevenue", + "estimatedAdRevenue", + "grossRevenue", + "estimatedRedPartnerRevenue", + "monetizedPlaybacks", + "playbackBasedCpm", + "adImpressions", + "cpm", +) + +YOUTUBE_ANALYTICS_ALL_PROVINCE_METRICS = ( + "views", + "redViews", + "estimatedMinutesWatched", + "estimatedRedMinutesWatched", + "averageViewDuration", + "averageViewPercentage", + "annotationClickThroughRate", + "annotationCloseRate", + "annotationImpressions", + "annotationClickableImpressions", + "annotationClosableImpressions", + "annotationClicks", + "annotationCloses", + "cardClickRate", + "cardTeaserClickRate", + "cardImpressions", + "cardTeaserImpressions", + "cardClicks", + "cardTeaserClicks", +) + +YOUTUBE_ANALYTICS_SUBSCRIPTION_METRICS = ( + "views", + "redViews", + "likes", + "dislikes", + "videosAddedToPlaylists", + "videosRemovedFromPlaylists", + "shares", + "estimatedMinutesWatched", + "estimatedRedMinutesWatched", + "averageViewDuration", + "averageViewPercentage", + "annotationClickThroughRate", + "annotationCloseRate", + "annotationImpressions", + "annotationClickableImpressions", + "annotationClosableImpressions", + "annotationClicks", + "annotationCloses", + "cardClickRate", + "cardTeaserClickRate", + "cardImpressions", + "cardTeaserImpressions", + "cardClicks", + "cardTeaserClicks", +) + +YOUTUBE_ANALYTICS_LIVE_PLAYBACK_DETAIL_METRICS = ( + "views", + "redViews", + "estimatedMinutesWatched", + "estimatedRedMinutesWatched", + "averageViewDuration", +) + +YOUTUBE_ANALYTICS_VIEW_PERCENTAGE_PLAYBACK_DETAIL_METRICS = ( + "views", + "redViews", + "estimatedMinutesWatched", + "estimatedRedMinutesWatched", + "averageViewDuration", + "averageViewPercentage", +) + +YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS = ( + "views", + "estimatedMinutesWatched", +) + +YOUTUBE_ANALYTICS_ALL_PLAYLIST_METRICS = ( + "views", + "redViews", + "estimatedMinutesWatched", + "estimatedRedMinutesWatched", + "averageViewDuration", + "playlistStarts", + "viewsPerPlaylistStart", + "averageTimeInPlaylist", +) + +YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_PLAYLIST_METRICS = ( + "views", + "estimatedMinutesWatched", + "playlistStarts", + "viewsPerPlaylistStart", + "averageTimeInPlaylist", +) diff --git a/analytix/youtube/analytics/verify/features.py b/analytix/youtube/analytics/verify/features.py new file mode 100644 index 0000000..238df70 --- /dev/null +++ b/analytix/youtube/analytics/verify/features.py @@ -0,0 +1,185 @@ +__all__ = ( + "Dimensions", + "Filters", + "Metrics", + "Required", + "ExactlyOne", + "OneOrMore", + "Optional", + "ZeroOrOne", + "ZeroOrMore", +) + +import abc + +from analytix.errors import InvalidRequest + +from .constants import * + + +class FeatureType(metaclass=abc.ABCMeta): + __slots__ = ("sets",) + + def __init__(self, *sets): + self.sets = sets + self.every = [] + for s in sets: + self.every.extend(s.values) + + def __iter__(self): + return iter(self.every) + + def verify(self, against): + raise NotImplementedError( + "you should not attempt to verify this ABC directly" + ) + + +class FeatureSet(metaclass=abc.ABCMeta): + __slots__ = ("values",) + + def __init__(self, *values): + self.values = set(values) + + def __iter__(self): + return iter(self.values) + + def verify(self, against, ftype): + raise NotImplementedError( + "you should not attempt to verify this ABC directly" + ) + + +class Dimensions(FeatureType): + def verify(self, against): + against = set(against) + + diff = against - set(YOUTUBE_ANALYTICS_ALL_DIMENSIONS) + if diff: + raise InvalidRequest( + f"one or more dimensions you provided are invalid ({', '.join(diff)})" + ) + + diff = against - set(self.every) + if diff: + raise InvalidRequest( + "one or more dimensions you provided are not supported by the " + f"selected report type ({', '.join(diff)})" + ) + + for s in self.sets: + s.verify(against, "dimension") + + +class Filters(FeatureType): + def __init__(self, *sets): + self.sets = sets + self.every = [] + for s in sets: + self.every.extend([v.split("==")[0] for v in s.values]) + + def verify(self, against): + keys = set(against.keys()) + + diff = keys - set(YOUTUBE_ANALYTICS_ALL_FILTERS) + if diff: + raise InvalidRequest( + f"one or more filters you provided are invalid ({', '.join(diff)})" + ) + + diff = keys - set(self.every) + if diff: + raise InvalidRequest( + "one or more filters you provided are not supported by the " + f"selected report type ({', '.join(diff)})" + ) + + for s in self.sets: + s.verify(against, "filter") + + for k, v in against.items(): + if ( + against[k] + and v not in YOUTUBE_ANALYTICS_VALID_FILTER_OPTIONS[k] + ): + raise InvalidRequest( + f"'{v}' is not a valid value for filter '{k}'" + ) + + +class Metrics: + def __init__(self, *values): + self.values = values + + def __iter__(self): + return iter(self.values) + + def verify(self, against): + against = set(against) + + diff = against - set(YOUTUBE_ANALYTICS_ALL_METRICS) + if diff: + raise InvalidRequest( + f"one or more metrics you provided are invalid ({', '.join(diff)})" + ) + + diff = against - set(self.values) + if diff: + raise InvalidRequest( + "one or more metrics you provided are not supported by the " + f"selected report type ({', '.join(diff)})" + ) + + +class Required(FeatureSet): + def verify(self, against, ftype): + values = [] + for v in self.values: + k = v.split("==") + if len(k) == 2: + if k[1] != against[k[0]]: + raise InvalidRequest( + f"filter '{k[0]}' must be set to '{k[1]}' for the selected report type" + ) + values.append(k[0]) + + if len(set(values) & set(against)) != len(self.values): + raise InvalidRequest( + f"expected all {ftype}s from '{', '.join(values)}', got {len(against)}" + ) + + +class ExactlyOne(FeatureSet): + def verify(self, against, ftype): + if len(self.values & set(against)) != 1: + raise InvalidRequest( + f"expected 1 {ftype} from '{', '.join(self.values)}', got {len(against)}" + ) + + +class OneOrMore(FeatureSet): + def verify(self, against, ftype): + if len(self.values & set(against)) == 0: + raise InvalidRequest( + f"expected at least 1 {ftype} from '{', '.join(self.values)}', got 0" + ) + + +class Optional(FeatureSet): + def verify(self, against, ftype): + # This doesn't need any verification. + pass + + +class ZeroOrOne(FeatureSet): + def verify(self, against, ftype): + if len(self.values & set(against)) > 1: + raise InvalidRequest( + f"expected 0 or 1 {ftype}s from '{', '.join(self.values)}', got {len(against)}" + ) + + +class ZeroOrMore(FeatureSet): + def verify(self, against, ftype): + # This doesn't need any verification. + pass diff --git a/analytix/youtube/analytics/verify/rtypes.py b/analytix/youtube/analytics/verify/rtypes.py new file mode 100644 index 0000000..835df01 --- /dev/null +++ b/analytix/youtube/analytics/verify/rtypes.py @@ -0,0 +1,688 @@ +from analytix.errors import InvalidRequest + +from .features import * +from .constants import * + +_D = Dimensions +_F = Filters +_M = Metrics + + +class ReportType: + _friendly_name = "Generic" + + def __init__(self): + self.dimensions = _D() + self.metrics = _M() + self.filters = _F() + + def __str__(self): + return self._friendly_name + + def verify(self, dim, met, fil, *args): + self.dimensions.verify(dim) + self.metrics.verify(met) + self.filters.verify(fil) + + +class DetailedReportType(ReportType): + def __init__(self): + super().__init__() + self.max_results = 0 + + def verify(self, dim, met, fil, max, srt): + super().verify(dim, met, fil) + + if not max or max > self.max_results: + raise InvalidRequest( + "the 'max_results' parameter must be provided and no larger " + f"than {self.max_results} for the selected report type" + ) + + if not srt: + raise InvalidRequest( + f"you must provide at least 1 sort parameter for the selected report type" + ) + + if any(s.strip("-") not in met for s in srt): + raise InvalidRequest( + f"the sort parameter must be one or more valid metrics" + ) + + if any(not s[0] == "-" for s in srt): + raise InvalidRequest( + ( + "you can only sort in descending order for this report type. " + "You can do this by prefixing the sort metrics with '-'" + ) + ) + + +class BasicUserActivity(ReportType): + _friendly_name = "Basic user activity" + + def __init__(self): + self.dimensions = _D() + self.metrics = _M(*YOUTUBE_ANALYTICS_ALL_METRICS) + self.filters = _F( + ZeroOrOne("country", "continent", "subContinent"), + ZeroOrOne("video", "group"), + ) + + +class BasicUserActivityUS(ReportType): + _friendly_name = "Basic user activity (US)" + + def __init__(self): + self.dimensions = _D() + self.metrics = _M(*YOUTUBE_ANALYTICS_ALL_PROVINCE_METRICS) + self.filters = _F( + Required("province"), + ZeroOrOne("video", "group"), + ) + + +class TimeBasedActivity(ReportType): + _friendly_name = "Time-based activity" + + def __init__(self): + self.dimensions = _D(ExactlyOne("day", "month")) + self.metrics = _M(*YOUTUBE_ANALYTICS_ALL_METRICS) + self.filters = _F( + ZeroOrOne("country", "continent", "subContinent"), + ZeroOrOne("video", "group"), + ) + + +class TimeBasedActivityUS(ReportType): + _friendly_name = "Time-based activity (US)" + + def __init__(self): + self.dimensions = _D(ExactlyOne("day", "month")) + self.metrics = _M(*YOUTUBE_ANALYTICS_ALL_PROVINCE_METRICS) + self.filters = _F( + Required("province"), + ZeroOrOne("video", "group"), + ) + + +class GeographyBasedActivity(ReportType): + _friendly_name = "Geography-based activity" + + def __init__(self): + self.dimensions = _D(Required("country")) + self.metrics = _M(*YOUTUBE_ANALYTICS_ALL_METRICS) + self.filters = _F( + ZeroOrOne("continent", "subContinent"), + ZeroOrOne("video", "group"), + ) + + +class GeographyBasedActivityUS(ReportType): + _friendly_name = "Geography-based activity (US)" + + def __init__(self): + self.dimensions = _D(Required("province")) + self.metrics = _M(*YOUTUBE_ANALYTICS_ALL_PROVINCE_METRICS) + self.filters = _F( + Required("country==US"), + ZeroOrOne("video", "group"), + ) + + +class PlaybackDetailsSubscribedStatus(ReportType): + _friendly_name = "User activity by subscribed status" + + def __init__(self): + self.dimensions = _D( + Optional("subscribedStatus"), + ZeroOrOne("day", "month"), + ) + self.metrics = _M(*YOUTUBE_ANALYTICS_SUBSCRIPTION_METRICS) + self.filters = _F( + ZeroOrOne("country", "continent", "subContinent"), + ZeroOrOne("video", "group"), + Optional("subscribedStatus"), + ) + + +class PlaybackDetailsSubscribedStatusUS(ReportType): + _friendly_name = "User activity by subscribed status (US)" + + def __init__(self): + self.dimensions = _D( + Optional("subscribedStatus"), + ZeroOrOne("day", "month"), + ) + self.metrics = _M( + "views", + "redViews", + "estimatedMinutesWatched", + "estimatedRedMinutesWatched", + "averageViewDuration", + "averageViewPercentage", + "annotationClickThroughRate", + "annotationCloseRate", + "annotationImpressions", + "annotationClickableImpressions", + "annotationClosableImpressions", + "annotationClicks", + "annotationCloses", + "cardClickRate", + "cardTeaserClickRate", + "cardImpressions", + "cardTeaserImpressions", + "cardClicks", + "cardTeaserClicks", + ) + self.filters = _F( + ZeroOrOne("video", "group"), + ZeroOrMore("province", "subscribedStatus"), + ) + + +class PlaybackDetailsLiveTimeBased(ReportType): + _friendly_name = "Time-based playback details (live)" + + def __init__(self): + self.dimensions = _D( + ZeroOrMore("liveOrOnDemand", "subscribedStatus", "youtubeProduct"), + ZeroOrOne("day", "month"), + ) + self.metrics = _M(*YOUTUBE_ANALYTICS_LIVE_PLAYBACK_DETAIL_METRICS) + self.filters = _F( + ZeroOrOne("country", "province", "continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus", "youtubeProduct"), + ) + + +class PlaybackDetailsViewPercentageTimeBased(ReportType): + _friendly_name = "Time-based playback details (view percentage)" + + def __init__(self): + self.dimensions = _D( + Required("country"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus", "youtubeProduct"), + ) + self.metrics = _M(*YOUTUBE_ANALYTICS_LIVE_PLAYBACK_DETAIL_METRICS) + self.filters = _F( + ZeroOrOne("continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus", "youtubeProduct"), + ) + + +class PlaybackDetailsLiveGeographyBased(ReportType): + _friendly_name = "Geography-based playback details (live)" + + def __init__(self): + self.dimensions = _D( + Required("country"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus", "youtubeProduct"), + ) + self.metrics = _M(*YOUTUBE_ANALYTICS_LIVE_PLAYBACK_DETAIL_METRICS) + self.filters = _F( + ZeroOrOne("continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus", "youtubeProduct"), + ) + + +class PlaybackDetailsViewPercentageGeographyBased(ReportType): + _friendly_name = "Geography-based playback details (view percentage)" + + def __init__(self): + self.dimensions = _D( + Required("country"), + ZeroOrMore("subscribedStatus", "youtubeProduct"), + ) + self.metrics = _M( + *YOUTUBE_ANALYTICS_VIEW_PERCENTAGE_PLAYBACK_DETAIL_METRICS + ) + self.filters = _F( + ZeroOrOne("continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore("subscribedStatus", "youtubeProduct"), + ) + + +class PlaybackDetailsLiveGeographyBasedUS(ReportType): + _friendly_name = "Geography-based playback details (live, US)" + + def __init__(self): + self.dimensions = _D( + Required("province"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus", "youtubeProduct"), + ) + self.metrics = _M(*YOUTUBE_ANALYTICS_LIVE_PLAYBACK_DETAIL_METRICS) + self.filters = _F( + Required("country==US"), + ZeroOrOne("video", "group"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus", "youtubeProduct"), + ) + + +class PlaybackDetailsViewPercentageGeographyBasedUS(ReportType): + _friendly_name = "Geography-based playback details (view percentage, US)" + + def __init__(self): + self.dimensions = _D( + Required("province"), + ZeroOrMore("subscribedStatus", "youtubeProduct"), + ) + self.metrics = _M( + *YOUTUBE_ANALYTICS_VIEW_PERCENTAGE_PLAYBACK_DETAIL_METRICS + ) + self.filters = _F( + Required("country==US"), + ZeroOrOne("video", "group"), + ZeroOrMore("subscribedStatus", "youtubeProduct"), + ) + + +class PlaybackLocation(ReportType): + _friendly_name = "Playback locations" + + def __init__(self): + self.dimensions = _D( + Required("insightPlaybackLocationType"), + ZeroOrMore("day", "liveOrOnDemand", "subscribedStatus"), + ) + self.metrics = _M(*YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS) + self.filters = _F( + ZeroOrOne("country", "province", "continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus"), + ) + + +class PlaybackLocationDetail(DetailedReportType): + _friendly_name = "Playback locations (detailed)" + + def __init__(self): + self.dimensions = _D(Required("insightPlaybackLocationDetail")) + self.metrics = _M(*YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS) + self.filters = _F( + Required("insightPlaybackLocationType==EMBEDDED"), + ZeroOrOne("country", "province", "continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus"), + ) + self.max_results = 25 + + +class TrafficSource(ReportType): + _friendly_name = "Traffic sources" + + def __init__(self): + self.dimensions = _D( + Required("insightTrafficSourceType"), + ZeroOrMore("day", "liveOrOnDemand", "subscribedStatus"), + ) + self.metrics = _M(*YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS) + self.filters = _F( + ZeroOrOne("country", "province", "continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus"), + ) + + +class TrafficSourceDetail(DetailedReportType): + _friendly_name = "Traffic sources (detailed)" + + def __init__(self): + self.dimensions = _D(Required("insightTrafficSourceDetail")) + self.metrics = _M(*YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS) + self.filters = _F( + Required("insightTrafficSourceType"), + ZeroOrOne("country", "province", "continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus"), + ) + self.max_results = 25 + + # TODO: Need a custom verifier here + + +class DeviceType(ReportType): + _friendly_name = "Device types" + + def __init__(self): + self.dimensions = _D( + Required("deviceType"), + ZeroOrMore( + "day", "liveOrOnDemand", "subscribedStatus", "youtubeProduct" + ), + ) + self.metrics = _M(*YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS) + self.filters = _F( + ZeroOrOne("country", "province", "continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore( + "operatingSystem", + "liveOrOnDemand", + "subscribedStatus", + "youtubeProduct", + ), + ) + + +class OperatingSystem(ReportType): + _friendly_name = "Operating systems" + + def __init__(self): + self.dimensions = _D( + Required("operatingSystem"), + ZeroOrMore( + "day", "liveOrOnDemand", "subscribedStatus", "youtubeProduct" + ), + ) + self.metrics = _M(*YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS) + self.filters = _F( + ZeroOrOne("country", "province", "continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore( + "deviceType", + "liveOrOnDemand", + "subscribedStatus", + "youtubeProduct", + ), + ) + + +class DeviceTypeAndOperatingSystem(ReportType): + _friendly_name = "Device types and operating systems" + + def __init__(self): + self.dimensions = _D( + Required("deviceType", "operatingSystem"), + ZeroOrMore( + "day", "liveOrOnDemand", "subscribedStatus", "youtubeProduct" + ), + ) + self.metrics = _M(*YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS) + self.filters = _F( + ZeroOrOne("country", "province", "continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus", "youtubeProduct"), + ) + + +class ViewerDemographics(ReportType): + _friendly_name = "Viewer demographics" + + def __init__(self): + self.dimensions = _D( + OneOrMore("ageGroup", "gender"), + ZeroOrMore("liveOrOnDemand", "subscribedStatus"), + ) + self.metrics = _M("viewerPercentage") + self.filters = _F( + ZeroOrOne("country", "province", "continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore( + "deviceType", + "liveOrOnDemand", + "subscribedStatus", + "youtubeProduct", + ), + ) + + +class EngagementAndContentSharing(ReportType): + _friendly_name = "Engagement and content sharing" + + def __init__(self): + self.dimensions = _D( + Required("sharingService"), + Optional("subscribedStatus"), + ) + self.metrics = _M("shares") + self.filters = _F( + ZeroOrOne("country", "continent", "subContinent"), + ZeroOrOne("video", "group"), + ZeroOrMore("subscribedStatus"), + ) + + +class AudienceRetention(ReportType): + _friendly_name = "Audience retention" + + def __init__(self): + self.dimensions = _D(Required("elapsedVideoTimeRatio")) + self.metrics = _M("audienceWatchRatio", "relativeRetentionPerformance") + self.filters = _F( + Required("video"), + ZeroOrMore("audienceType", "subscribedStatus", "youtubeProduct"), + ) + + # TODO: Custom verify + + +class TopVideosRegional(DetailedReportType): + _friendly_name = "Top videos by region" + + def __init__(self): + self.dimensions = _D(Required("video")) + self.metrics = _M(*YOUTUBE_ANALYTICS_ALL_METRICS) + self.filters = _F(ZeroOrOne("country", "continent", "subContinent")) + self.max_results = 200 + + +class TopVideosUS(DetailedReportType): + _friendly_name = "Top videos by state" + + def __init__(self): + self.dimensions = _D(Required("video")) + self.metrics = _M(*YOUTUBE_ANALYTICS_ALL_PROVINCE_METRICS) + self.filters = _F( + Required("province"), + Optional("subscribedStatus"), + ) + self.max_results = 200 + + +class TopVideosSubscribed(DetailedReportType): + _friendly_name = "Top videos by subscription status" + + def __init__(self): + self.dimensions = _D(Required("video")) + self.metrics = _M(*YOUTUBE_ANALYTICS_SUBSCRIPTION_METRICS) + self.filters = _F( + Optional("subscribedStatus"), + ZeroOrOne("country", "continent", "subContinent"), + ) + self.max_results = 200 + + +class TopVideosYouTubeProduct(DetailedReportType): + _friendly_name = "Top videos by YouTube product" + + def __init__(self): + self.dimensions = _D(Required("video")) + self.metrics = _M( + *YOUTUBE_ANALYTICS_VIEW_PERCENTAGE_PLAYBACK_DETAIL_METRICS + ) + self.filters = _F( + ZeroOrOne("country", "province", "continent", "subContinent"), + ZeroOrMore("subscribedStatus", "youtubeProduct"), + ) + self.max_results = 200 + + +class TopVideosPlaybackDetail(DetailedReportType): + _friendly_name = "Top videos by playback detail" + + def __init__(self): + self.dimensions = _D(Required("video")) + self.metrics = _M(*YOUTUBE_ANALYTICS_LIVE_PLAYBACK_DETAIL_METRICS) + self.filters = _F( + ZeroOrOne("country", "province", "continent", "subContinent"), + ZeroOrMore("subscribedStatus", "youtubeProduct"), + ) + self.max_results = 200 + + +def determine(dimensions, metrics, filters): + curated = filters.get("isCurated", "0") == "1" + + if "sharingService" in dimensions: + return EngagementAndContentSharing + + if "elapsedVideoTimeRatio" in dimensions: + return AudienceRetention + + if "playlist" in dimensions: + return TopPlaylists + + if "insightPlaybackLocationType" in dimensions: + if curated: + return PlaybackLocationPlaylist + return PlaybackLocation + + if "insightPlaybackLocationDetail" in dimensions: + if curated: + return PlaybackLocationDetailPlaylist + return PlaybackLocationDetail + + if "insightTrafficSourceType" in dimensions: + if curated: + return TrafficSourcePlaylist + return TrafficSource + + if "insightTrafficSourceDetail" in dimensions: + if curated: + return TrafficSourceDetailPlaylist + return TrafficSourceDetail + + if "ageGroup" in dimensions or "gender" in dimensions: + if curated: + return ViewerDemographicsPlaylist + return ViewerDemographics + + if "deviceType" in dimensions: + if "operatingSystem" in dimensions: + if curated: + return DeviceTypeAndOperatingSystemPlaylist + return DeviceTypeAndOperatingSystem + if curated: + return DeviceTypePlaylist + return DeviceType + + if "operatingSystem" in dimensions: + if curated: + return OperatingSystemPlaylist + return OperatingSystem + + if "video" in dimensions: + if "province" in filters: + return TopVideosUS + if "subscribedStatus" not in filters: + return TopVideosRegional + if "province" not in filters and "youtubeProduct" not in filters: + return TopVideosSubscribed + if "averageViewPercentage" in metrics: + return TopVideosYouTubeProduct + return TopVideosPlaybackDetail + + if "country" in dimensions: + if "liveOrOnDemand" in dimensions or "liveOrOnDemand" in filters: + return PlaybackDetailsLiveGeographyBased + if curated: + return GeographyBasedActivityPlaylist + if ( + "subscribedStatus" in dimensions + or "subscribedStatus" in filters + or "youtubeProduct" in dimensions + or "youtubeProduct" in filters + ): + return PlaybackDetailsViewPercentageGeographyBased + return GeographyBasedActivity + + if "province" in dimensions: + if "liveOrOnDemand" in dimensions or "liveOrOnDemand" in filters: + return PlaybackDetailsLiveGeographyBasedUS + if curated: + return GeographyBasedActivityPlaylistUS + if ( + "subscribedStatus" in dimensions + or "subscribedStatus" in filters + or "youtubeProduct" in dimensions + or "youtubeProduct" in filters + ): + return PlaybackDetailsViewPercentageGeographyBasedUS + return GeographyBasedActivityUS + + if "youtubeProduct" in dimensions or "youtubeProduct" in filters: + if "liveOrOnDemand" in dimensions or "liveOrOnDemand" in filters: + return PlaybackDetailsLiveTimeBased + return PlaybackDetailsViewPercentageTimeBased + + if "liveOrOnDemand" in dimensions or "liveOrOnDemand" in filters: + return PlaybackDetailsLiveTimeBased + + if "subscribedStatus" in dimensions: + if "province" in filters: + return PlaybackDetailsSubscribedStatusUS + return PlaybackDetailsSubscribedStatus + + if "day" in dimensions or "month" in dimensions: + if curated: + return TimeBasedActivityPlaylist + if "province" in filters: + return TimeBasedActivityUS + return TimeBasedActivity + + if curated: + return BasicUserActivityPlaylist + if "province" in filters: + return BasicUserActivityUS + return BasicUserActivity + + +_ALL_REPORTS = { + "BASIC_USER_ACTIVITY": BasicUserActivity, + "BASIC_USER_ACTIVITY_US": BasicUserActivityUS, + "TIME_BASED_ACTIVITY": TimeBasedActivity, + "TIME_BASED_ACTIVITY_US": TimeBasedActivityUS, + "GEOGRAPHY_BASED_ACTIVITY": object(), + "GEOGRAPHY_BASED_ACTIVITY_US": object(), + "PLAYBACK_DETAILS_SUBSCRIBED_STATUS": object(), + "PLAYBACK_DETAILS_SUBSCRIBED_STATUS_US": object(), + "PLAYBACK_DETAILS_LIVE_TIME_BASED": object(), + "PLAYBACK_DETAILS_VIEW_PERCENTAGE_TIME_BASED": object(), + "PLAYBACK_DETAILS_LIVE_GEOGRAPHY_BASED": object(), + "PLAYBACK_DETAILS_VIEW_PERCENTAGE_GEOGRAPHY_BASED": object(), + "PLAYBACK_DETAILS_LIVE_GEOGRAPHY_BASED_US": object(), + "PLAYBACK_DETAILS_VIEW_PERCENTAGE_GEOGRAPHY_BASED_US": object(), + "PLAYBACK_LOCATION": object(), + "PLAYBACK_LOCATION_DETAIL": object(), + "TRAFFIC_SOURCE": object(), + "TRAFFIC_SOURCE_DETAIL": object(), + "DEVICE_TYPE": object(), + "OPERATING_SYSTEM": object(), + "DEVICE_TYPE_AND_OPERATING_SYSTEM": object(), + "VIEWER_DEMOGRAPHICS": object(), + "ENGAGEMENT_AND_CONTENT_SHARING": object(), + "AUDIENCE_RETENTION": object(), + "TOP_VIDEOS_REGIONAL": object(), + "TOP_VIDEOS_US": object(), + "TOP_VIDEOS_SUBSCRIBED": object(), + "TOP_VIDEOS_YOUTUBE_PRODUCT": object(), + "TOP_VIDEOS_PLAYBACK_DETAIL": object(), + "BASIC_USER_ACTIVITY_PLAYLIST": object(), + "TIME_BASED_ACTIVITY_PLAYLIST": object(), + "GEOGRAPHY_BASED_ACTIVITY_PLAYLIST": object(), + "GEOGRAPHY_BASED_ACTIVITY_US_PLAYLIST": object(), + "PLAYBACK_LOCATION_PLAYLIST": object(), + "PLAYBACK_LOCATION_DETAIL_PLAYLIST": object(), + "TRAFFIC_SOURCE_PLAYLIST": object(), + "TRAFFIC_SOURCE_DETAIL_PLAYLIST": object(), + "DEVICE_TYPE_PLAYLIST": object(), + "OPERATING_SYSTEM_PLAYLIST": object(), + "DEVICE_TYPE_AND_OPERATING_SYSTEM_PLAYLIST": object(), + "VIEWER_DEMOGRAPHICS_PLAYLIST": object(), + "TOP_PLAYLISTS": object(), + "AD_PERFORMANCE": object(), +} diff --git a/analytix/youtube/features.py b/analytix/youtube/features.py deleted file mode 100644 index 0ceee8b..0000000 --- a/analytix/youtube/features.py +++ /dev/null @@ -1,130 +0,0 @@ -from enum import Enum, auto - -YOUTUBE_ANALYTICS_ALL_METRICS = ( - "views", - "redViews", - "comments", - "likes", - "dislikes", - "videosAddedToPlaylists", - "videosRemovedFromPlaylists", - "shares", - "estimatedMinutesWatched", - "estimatedRedMinutesWatched", - "averageViewDuration", - "averageViewPercentage", - "annotationClickThroughRate", - "annotationCloseRate", - "annotationImpressions", - "annotationClickableImpressions", - "annotationClosableImpressions", - "annotationClicks", - "annotationCloses", - "cardClickRate", - "cardTeaserClickRate", - "cardImpressions", - "cardTeaserImpressions", - "cardClicks", - "cardTeaserClicks", - "subscribersGained", - "subscribersLost", - "estimatedRevenue", - "estimatedAdRevenue", - "grossRevenue", - "estimatedRedPartnerRevenue", - "monetizedPlaybacks", - "playbackBasedCpm", - "adImpressions", - "cpm", -) -YOUTUBE_ANALYTICS_ALL_PROVINCE_METRICS = ( - "views", - "redViews", - "estimatedMinutesWatched", - "estimatedRedMinutesWatched", - "averageViewDuration", - "averageViewPercentage", - "annotationClickThroughRate", - "annotationCloseRate", - "annotationImpressions", - "annotationClickableImpressions", - "annotationClosableImpressions", - "annotationClicks", - "annotationCloses", - "cardClickRate", - "cardTeaserClickRate", - "cardImpressions", - "cardTeaserImpressions", - "cardClicks", - "cardTeaserClicks", -) -YOUTUBE_ANALYTICS_SUBSCRIPTION_METRICS = ( - "views", - "redViews", - "likes", - "dislikes", - "videosAddedToPlaylists", - "videosRemovedFromPlaylists", - "shares", - "estimatedMinutesWatched", - "estimatedRedMinutesWatched", - "averageViewDuration", - "averageViewPercentage", - "annotationClickThroughRate", - "annotationCloseRate", - "annotationImpressions", - "annotationClickableImpressions", - "annotationClosableImpressions", - "annotationClicks", - "annotationCloses", - "cardClickRate", - "cardTeaserClickRate", - "cardImpressions", - "cardTeaserImpressions", - "cardClicks", - "cardTeaserClicks", -) -YOUTUBE_ANALYTICS_LIVE_PLAYBACK_DETAIL_METRICS = ( - "views", - "redViews", - "estimatedMinutesWatched", - "estimatedRedMinutesWatched", - "averageViewDuration", -) -YOUTUBE_ANALYTICS_VIEW_PERCENTAGE_PLAYBACK_DETAIL_METRICS = ( - "views", - "redViews", - "estimatedMinutesWatched", - "estimatedRedMinutesWatched", - "averageViewDuration", - "averageViewPercentage", -) -YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS = ( - "views", - "estimatedMinutesWatched", -) -YOUTUBE_ANALYTICS_ALL_PLAYLIST_METRICS = ( - "views", - "redViews", - "estimatedMinutesWatched", - "estimatedRedMinutesWatched", - "averageViewDuration", - "playlistStarts", - "viewsPerPlaylistStart", - "averageTimeInPlaylist", -) -YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_PLAYLIST_METRICS = ( - "views", - "estimatedMinutesWatched", - "playlistStarts", - "viewsPerPlaylistStart", - "averageTimeInPlaylist", -) - - -class FeatureAmount(Enum): - ANY = auto() - NON_ZERO = auto() - ZERO_OR_ONE = auto() - EXACTLY_ONE = auto() - REQUIRED = auto() diff --git a/analytix/youtube/reports.py b/analytix/youtube/reports.py deleted file mode 100644 index 8f36fe0..0000000 --- a/analytix/youtube/reports.py +++ /dev/null @@ -1,1061 +0,0 @@ -from analytix import InvalidRequest -from analytix.youtube import features -from analytix.youtube.features import FeatureAmount - - -class ReportType: - __slots__ = ("dimensions", "metrics", "filters") - - def __init__(self): - raise NotImplementedError("you should not use this class directly, nor call its __init__ method using super()") - - @property - def all_dimensions(self): - e = [] - for x in self.dimensions: - e.extend(x[1]) - return e - - @property - def all_filters(self): - e = [] - for x in self.filters: - e.extend(x[1]) - return e - - def verify(self, metrics, dimensions, filters, *args, **kwargs): - dimensions = set(dimensions) - filters = set(filters) - - if metrics != "all": - diff = set(metrics) - set(self.metrics) - if diff: - raise InvalidRequest("unexpected metric(s): " + ", ".join(diff)) - - diff = dimensions - set(self.all_dimensions) - if diff: - raise InvalidRequest("unexpected dimension(s): " + ", ".join(diff)) - - diff = filters - set(self.all_filters) - if diff: - raise InvalidRequest("unexpected filter(s): " + ", ".join(diff)) - - for amount, values in self.dimensions: - similarities = len(dimensions & values) - if amount == FeatureAmount.REQUIRED and similarities != len(values): - raise InvalidRequest(f"expected all dimensions from '{', '.join(values)}'") - elif amount == FeatureAmount.ZERO_OR_ONE and similarities > 1: - raise InvalidRequest(f"expected 0 or 1 dimensions from '{', '.join(values)}', got {len(dimensions)}") - elif amount == FeatureAmount.EXACTLY_ONE and similarities != 1: - raise InvalidRequest(f"expected 1 dimension from '{', '.join(values)}', got {len(dimensions)}") - elif amount == FeatureAmount.NON_ZERO and similarities == 0: - raise InvalidRequest( - f"expected at least 1 dimension from '{', '.join(values)}', got {len(dimensions)}" - ) - - for amount, values in self.filters: - similarities = len(filters & values) - if amount == FeatureAmount.REQUIRED and similarities != len(values): - raise InvalidRequest(f"expected all filters from '{', '.join(values)}'") - elif amount == FeatureAmount.ZERO_OR_ONE and similarities > 1: - raise InvalidRequest(f"expected 0 or 1 filters from '{', '.join(values)}', got {len(filters)}") - elif amount == FeatureAmount.EXACTLY_ONE and similarities != 1: - raise InvalidRequest(f"expected 1 filter from '{', '.join(values)}', got {len(filters)}") - elif amount == FeatureAmount.NON_ZERO and similarities == 0: - raise InvalidRequest(f"expected at least 1 filter from '{', '.join(values)}', got {len(filters)}") - - -class Generic(ReportType): - def __init__(self): - self.dimensions = [] - self.metrics = [] - self.filters = [] - - def verify(self, *args): - pass - - def __str__(self): - return "Generic" - - -class BasicUserActivity(ReportType): - def __init__(self): - self.dimensions = [] - self.metrics = features.YOUTUBE_ANALYTICS_ALL_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - ] - - def __str__(self): - return "Basic user activity" - - -class BasicUserActivityUS(ReportType): - def __init__(self): - self.dimensions = [] - self.metrics = features.YOUTUBE_ANALYTICS_ALL_PROVINCE_METRICS - self.filters = [(FeatureAmount.REQUIRED, {"province"}), (FeatureAmount.ZERO_OR_ONE, {"video", "group"})] - - def __str__(self): - return "Basic user activity (US)" - - -class TimeBasedActivity(ReportType): - def __init__(self): - self.dimensions = [(FeatureAmount.EXACTLY_ONE, {"day", "month"})] - self.metrics = features.YOUTUBE_ANALYTICS_ALL_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - ] - - def __str__(self): - return "Time-based activity" - - -class TimeBasedActivityUS(ReportType): - def __init__(self): - self.dimensions = [(FeatureAmount.EXACTLY_ONE, {"day", "month"})] - self.metrics = features.YOUTUBE_ANALYTICS_ALL_PROVINCE_METRICS - self.filters = [(FeatureAmount.REQUIRED, {"province"}), (FeatureAmount.ZERO_OR_ONE, {"video", "group"})] - - def __str__(self): - return "Time-based activity (US)" - - -class GeographyBasedActivity(ReportType): - def __init__(self): - self.dimensions = [(FeatureAmount.REQUIRED, {"country"})] - self.metrics = features.YOUTUBE_ANALYTICS_ALL_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - ] - - def __str__(self): - return "Geography-based activity" - - -class GeographyBasedActivityUS(ReportType): - def __init__(self): - self.dimensions = [(FeatureAmount.REQUIRED, {"province"})] - self.metrics = features.YOUTUBE_ANALYTICS_ALL_PROVINCE_METRICS - self.filters = [ - (FeatureAmount.REQUIRED, {"country"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - ] - - def verify(self, metrics, dimensions, filters): - super().verify(metrics, dimensions, filters) - - if filters["country"] != "US": - raise InvalidRequest("the 'country' filter must be set to 'US'") - - def __str__(self): - return "Geography-based activity (US)" - - -class PlaybackDetailsSubscribedStatus(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.ZERO_OR_ONE, {"subscribedStatus"}), - (FeatureAmount.ZERO_OR_ONE, {"day", "month"}), - ] - self.metrics = features.YOUTUBE_ANALYTICS_SUBSCRIPTION_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ZERO_OR_ONE, {"subscribedStatus"}), - ] - - def __str__(self): - return "User activity by subscribed status" - - -class PlaybackDetailsSubscribedStatusUS(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.ZERO_OR_ONE, {"subscribedStatus"}), - (FeatureAmount.ZERO_OR_ONE, {"day", "month"}), - ] - self.metrics = ( - "views", - "redViews", - "estimatedMinutesWatched", - "estimatedRedMinutesWatched", - "averageViewDuration", - "averageViewPercentage", - "annotationClickThroughRate", - "annotationCloseRate", - "annotationImpressions", - "annotationClickableImpressions", - "annotationClosableImpressions", - "annotationClicks", - "annotationCloses", - "cardClickRate", - "cardTeaserClickRate", - "cardImpressions", - "cardTeaserImpressions", - "cardClicks", - "cardTeaserClicks", - ) - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"province", "subscribedStatus"}), - ] - - def __str__(self): - return "User activity by subscribed status (US)" - - -class PlaybackDetailsLiveTimeBased(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus", "youtubeProduct"}), - (FeatureAmount.ZERO_OR_ONE, {"day", "month"}), - ] - self.metrics = features.YOUTUBE_ANALYTICS_LIVE_PLAYBACK_DETAIL_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus", "youtubeProduct"}), - ] - - def __str__(self): - return "Time-based playback details (live)" - - -class PlaybackDetailsViewPercentageTimeBased(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - (FeatureAmount.ZERO_OR_ONE, {"day", "month"}), - ] - self.metrics = features.YOUTUBE_ANALYTICS_VIEW_PERCENTAGE_PLAYBACK_DETAIL_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ] - - def __str__(self): - return "Time-based playback details (view percentage)" - - -class PlaybackDetailsLiveGeographyBased(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.REQUIRED, {"country"}), - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus", "youtubeProduct"}), - ] - self.metrics = features.YOUTUBE_ANALYTICS_LIVE_PLAYBACK_DETAIL_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus", "youtubeProduct"}), - ] - - def __str__(self): - return "Geography-based playback details (live)" - - -class PlaybackDetailsViewPercentageGeographyBased(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.REQUIRED, {"country"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ] - self.metrics = features.YOUTUBE_ANALYTICS_VIEW_PERCENTAGE_PLAYBACK_DETAIL_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ] - - def __str__(self): - return "Geography-based playback details (view percentage)" - - -class PlaybackDetailsLiveGeographyBasedUS(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.REQUIRED, {"province"}), - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus", "youtubeProduct"}), - ] - self.metrics = features.YOUTUBE_ANALYTICS_LIVE_PLAYBACK_DETAIL_METRICS - self.filters = [ - (FeatureAmount.REQUIRED, {"country"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus", "youtubeProduct"}), - ] - - def verify(self, metrics, dimensions, filters): - super().verify(metrics, dimensions, filters) - - if filters["country"] != "US": - raise InvalidRequest("the 'country' filter must be set to 'US'") - - def __str__(self): - return "Geography-based playback details (live, US)" - - -class PlaybackDetailsViewPercentageGeographyBasedUS(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.REQUIRED, {"province"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ] - self.metrics = features.YOUTUBE_ANALYTICS_VIEW_PERCENTAGE_PLAYBACK_DETAIL_METRICS - self.filters = [ - (FeatureAmount.REQUIRED, {"country"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ] - - def verify(self, metrics, dimensions, filters): - super().verify(metrics, dimensions, filters) - - if filters["country"] != "US": - raise InvalidRequest("the 'country' filter must be set to 'US'") - - def __str__(self): - return "Geography-based playback details (view percentage, US)" - - -class PlaybackLocation(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.REQUIRED, {"insightPlaybackLocationType"}), - (FeatureAmount.ANY, {"day", "liveOrOnDemand", "subscribedStatus"}), - ] - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus"}), - ] - - def __str__(self): - return "Playback locations" - - -class PlaybackLocationDetail(ReportType): - def __init__(self): - self.dimensions = [(FeatureAmount.REQUIRED, {"insightPlaybackLocationDetail"})] - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS - self.filters = [ - (FeatureAmount.REQUIRED, {"insightPlaybackLocationType"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus"}), - ] - - def verify(self, metrics, dimensions, filters, max_results, sort_by): - super().verify(metrics, dimensions, filters) - - if filters["insightPlaybackLocationType"] != "EMBEDDED": - raise InvalidRequest("the 'insightPlaybackLocationType' filter must be set to 'EMBEDDED'") - - if not max_results or max_results >= 25: - raise InvalidRequest("the 'max_results' parameter must not be set above 25") - - if not sort_by: - raise InvalidRequest("you must provide at least 1 sort parameter") - - def __str__(self): - return "Playback locations (detailed)" - - -class TrafficSource(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.REQUIRED, {"insightTrafficSourceType"}), - (FeatureAmount.ANY, {"day", "liveOrOnDemand", "subscribedStatus"}), - ] - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus"}), - ] - - def __str__(self): - return "Traffic sources" - - -class TrafficSourceDetail(ReportType): - def __init__(self): - self.dimensions = [(FeatureAmount.REQUIRED, {"insightTrafficSourceDetail"})] - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS - self.filters = [ - (FeatureAmount.REQUIRED, {"insightTrafficSourceType"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus"}), - ] - - def verify(self, metrics, dimensions, filters, max_results, sort_by): - super().verify(metrics, dimensions, filters) - - if not max_results or max_results >= 25: - raise InvalidRequest("the 'max_results' parameter must not be set above 25") - - if not sort_by: - raise InvalidRequest("you must provide at least 1 sort parameter") - - def __str__(self): - return "Traffic sources (detailed)" - - -class DeviceType(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.REQUIRED, {"deviceType"}), - (FeatureAmount.ANY, {"day", "liveOrOnDemand", "subscribedStatus", "youtubeProduct"}), - ] - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"operatingSystem", "liveOrOnDemand", "subscribedStatus", "youtubeProduct"}), - ] - - def __str__(self): - return "Device types" - - -class OperatingSystem(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.REQUIRED, {"operatingSystem"}), - (FeatureAmount.ANY, {"day", "liveOrOnDemand", "subscribedStatus", "youtubeProduct"}), - ] - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"deviceType", "liveOrOnDemand", "subscribedStatus", "youtubeProduct"}), - ] - - def __str__(self): - return "Operating systems" - - -class DeviceTypeAndOperatingSystem(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.REQUIRED, {"deviceType", "operatingSystem"}), - (FeatureAmount.ANY, {"day", "liveOrOnDemand", "subscribedStatus", "youtubeProduct"}), - ] - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus", "youtubeProduct"}), - ] - - def __str__(self): - return "Device types and operating systems" - - -class ViewerDemographics(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.NON_ZERO, {"ageGroup", "gender"}), - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus"}), - ] - self.metrics = ("viewerPercentage",) - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ANY, {"liveOrOnDemand", "subscribedStatus"}), - ] - - def __str__(self): - return "Viewer demographics" - - -class EngagementAndContentSharing(ReportType): - def __init__(self): - self.dimensions = [ - (FeatureAmount.REQUIRED, {"sharingService"}), - (FeatureAmount.ZERO_OR_ONE, {"subscribedStatus"}), - ] - self.metrics = ("shares",) - self.filters = [(FeatureAmount.ZERO_OR_ONE, {"country", "continent", "subContinent"}), Detail] - - def __str__(self): - return "Engagement and content sharing" - - -class AudienceRetention(ReportType): - def __init__(self): - self.dimensions = [(FeatureAmount.REQUIRED, {"elaspedVideoTimeRatio"})] - self.metrics = ("audienceWatchRatio", "relativeRetentionPerformance") - self.filters = [ - (FeatureAmount.REQUIRED, {"video"}), - (FeatureAmount.ANY, {"audienceType", "subscribedStatus", "youtubeProduct"}), - ] - - def verify(self, metrics, dimensions, filters): - super().verify(metrics, dimensions, filters) - - if len(filters["video"].split(",")) > 1: - raise InvalidRequest("you can only specify 1 video ID") - - def __str__(self): - return "Audience retention" - - -class TopVideosRegional(ReportType): - def __init__(self): - self.dimensions = [(FeatureAmount.REQUIRED, {"video"})] - self.metrics = features.YOUTUBE_ANALYTICS_ALL_METRICS - self.filters = [(FeatureAmount.ZERO_OR_ONE, {"country", "continent", "subContinent"})] - - def verify(self, metrics, dimensions, filters, max_results, sort_by): - super().verify(metrics, dimensions, filters) - - if not max_results or max_results >= 200: - raise InvalidRequest("the 'max_results' parameter must not be set above 200") - - if not sort_by: - raise InvalidRequest("you must provide at least 1 sort parameter") - - def __str__(self): - return "Top videos by region" - - -class TopVideosUS(ReportType): - def __init__(self): - self.dimensions = [(FeatureAmount.REQUIRED, {"video"})] - self.metrics = features.YOUTUBE_ANALYTICS_ALL_PROVINCE_METRICS - self.filters = [(FeatureAmount.REQUIRED, {"province"}), (FeatureAmount.ZERO_OR_ONE, {"subscribedStatus"})] - - def verify(self, metrics, dimensions, filters, max_results, sort_by): - super().verify(metrics, dimensions, filters) - - if not max_results or max_results >= 200: - raise InvalidRequest("the 'max_results' parameter must not be set above 200") - - if not sort_by: - raise InvalidRequest("you must provide at least 1 sort parameter") - - def __str__(self): - return "Top videos by state" - - -class TopVideosSubscribed(ReportType): - def __init__(self): - self.dimensions = [(FeatureAmount.REQUIRED, {"video"})] - self.metrics = features.YOUTUBE_ANALYTICS_SUBSCRIPTION_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"subscribedStatus"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "continent", "subContinent"}), - ] - - def verify(self, metrics, dimensions, filters, max_results, sort_by): - super().verify(metrics, dimensions, filters) - - if not max_results or max_results >= 200: - raise InvalidRequest("the 'max_results' parameter must not be set above 200") - - if not sort_by: - raise InvalidRequest("you must provide at least 1 sort parameter") - - def __str__(self): - return "Top videos by subscription status" - - -class TopVideosYouTubeProduct(ReportType): - def __init__(self): - self.dimensions = [(FeatureAmount.REQUIRED, {"video"})] - self.metrics = features.YOUTUBE_ANALYTICS_VIEW_PERCENTAGE_PLAYBACK_DETAIL_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ] - - def verify(self, metrics, dimensions, filters, max_results, sort_by): - super().verify(metrics, dimensions, filters) - - if not max_results or max_results >= 200: - raise InvalidRequest("the 'max_results' parameter must not be set above 200") - - if not sort_by: - raise InvalidRequest("you must provide at least 1 sort parameter") - - def __str__(self): - return "Top videos by YouTube product" - - -class TopVideosPlaybackDetail(ReportType): - def __init__(self): - self.dimensions = [(FeatureAmount.REQUIRED, {"video"})] - self.metrics = features.YOUTUBE_ANALYTICS_LIVE_PLAYBACK_DETAIL_METRICS - self.filters = [ - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ] - - def verify(self, metrics, dimensions, filters, max_results, sort_by): - super().verify(metrics, dimensions, filters) - - if not max_results or max_results >= 200: - raise InvalidRequest("the 'max_results' parameter must not be set above 200") - - if not sort_by: - raise InvalidRequest("you must provide at least 1 sort parameter") - - def __str__(self): - return "Top videos by playback detail" - - -class BasicUserActivityPlaylist(ReportType): - def __init__(self): - self.dimensions = [] - self.metrics = features.YOUTUBE_ANALYTICS_ALL_PLAYLIST_METRICS - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"playlist", "group"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ) - - def verify(self, metrics, dimensions, filters, *args): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - def __str__(self): - return "Basic user activity for playlists" - - -class TimeBasedActivityPlaylist(ReportType): - def __init__(self): - self.dimensions = ( - (FeatureAmount.EXACTLY_ONE, {"day", "month"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ) - self.metrics = features.YOUTUBE_ANALYTICS_ALL_PLAYLIST_METRICS - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"playlist", "group"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ) - - def verify(self, metrics, dimensions, filters, *args): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - def __str__(self): - return "Time-based activity for playlists" - - -class GeographyBasedActivityPlaylist(ReportType): - def __init__(self): - self.dimensions = ( - (FeatureAmount.REQUIRED, {"country"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ) - self.metrics = features.YOUTUBE_ANALYTICS_ALL_PLAYLIST_METRICS - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"playlist", "group"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ) - - def verify(self, metrics, dimensions, filters, *args): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - def __str__(self): - return "Geography-based activity for playlists" - - -class GeographyBasedActivityUSPlaylist(ReportType): - def __init__(self): - self.dimensions = ( - (FeatureAmount.REQUIRED, {"province"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ) - self.metrics = features.YOUTUBE_ANALYTICS_ALL_PLAYLIST_METRICS - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated", "country"}), - (FeatureAmount.ZERO_OR_ONE, {"playlist", "group"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ) - - def verify(self, metrics, dimensions, filters, *args): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - if filters["country"] != "US": - raise InvalidRequest("the 'country' filter must be set to 'US'") - - def __str__(self): - return "Geography-based activity for playlists (US)" - - -class PlaybackLocationPlaylist(ReportType): - def __init__(self): - self.dimensions = ( - (FeatureAmount.REQUIRED, {"insightPlaybackLocationType"}), - (FeatureAmount.ANY, {"day", "subscribedStatus"}), - ) - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_PLAYLIST_METRICS - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"playlist", "group"}), - (FeatureAmount.ANY, {"subscribedStatus"}), - ) - - def verify(self, metrics, dimensions, filters, *args): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - def __str__(self): - return "Playback locations for playlists" - - -class PlaybackLocationDetailPlaylist(ReportType): - def __init__(self): - self.dimensions = (FeatureAmount.REQUIRED, {"insightPlaybackLocationDetail"}) - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_PLAYLIST_METRICS - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated", "insightPlaybackLocationType"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"playlist", "group"}), - (FeatureAmount.ANY, {"subscribedStatus"}), - ) - - def verify(self, metrics, dimensions, filters, max_results, sort_by): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - if filters["insightPlaybackLocationType"] != "EMBEDDED": - raise InvalidRequest("the 'insightPlaybackLocationType' filter must be set to 'EMBEDDED'") - - if not max_results or max_results >= 25: - raise InvalidRequest("the 'max_results' parameter must not be set above 25") - - if not sort_by: - raise InvalidRequest("you must provide at least 1 sort parameter") - - def __str__(self): - return "Playback locations for playlists (detailed)" - - -class TrafficSourcePlaylist(ReportType): - def __init__(self): - self.dimensions = ( - (FeatureAmount.REQUIRED, {"insightTrafficSourceType"}), - (FeatureAmount.ANY, {"day", "subscribedStatus"}), - ) - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_PLAYLIST_METRICS - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"playlist", "group"}), - (FeatureAmount.ANY, {"subscribedStatus"}), - ) - - def verify(self, metrics, dimensions, filters, *args): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - def __str__(self): - return "Traffic sources for playlists" - - -class TrafficSourceDetailPlaylist(ReportType): - def __init__(self): - self.dimensions = (FeatureAmount.REQUIRED, {"insightTrafficSourceDetail"}) - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_PLAYLIST_METRICS - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated", "insightTrafficSourceType"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"playlist", "group"}), - (FeatureAmount.ANY, {"subscribedStatus"}), - ) - - def verify(self, metrics, dimensions, filters, max_results, sort_by): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - if not max_results or max_results >= 25: - raise InvalidRequest("the 'max_results' parameter must not be set above 25") - - if not sort_by: - raise InvalidRequest("you must provide at least 1 sort parameter") - - def __str__(self): - return "Traffic sources for playlists (detailed)" - - -class DeviceTypePlaylist(ReportType): - def __init__(self): - self.dimensions = ( - (FeatureAmount.REQUIRED, {"deviceType"}), - (FeatureAmount.ANY, {"day", "subscribedStatus", "youtubeProduct"}), - ) - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_PLAYLIST_METRICS - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"playlist", "group"}), - (FeatureAmount.ANY, {"operatingSystem", "subscribedStatus", "youtubeProduct"}), - ) - - def verify(self, metrics, dimensions, filters, *args): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - def __str__(self): - return "Device types for playlists" - - -class OperatingSystemPlaylist(ReportType): - def __init__(self): - self.dimensions = ( - (FeatureAmount.REQUIRED, {"operatingSystem"}), - (FeatureAmount.ANY, {"day", "subscribedStatus", "youtubeProduct"}), - ) - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_PLAYLIST_METRICS - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"playlist", "group"}), - (FeatureAmount.ANY, {"deviceType", "subscribedStatus", "youtubeProduct"}), - ) - - def verify(self, metrics, dimensions, filters, *args): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - def __str__(self): - return "Operating systems for playlists" - - -class DeviceTypeAndOperatingSystemPlaylist(ReportType): - def __init__(self): - self.dimensions = ( - (FeatureAmount.REQUIRED, {"deviceType", "operatingSystem"}), - (FeatureAmount.ANY, {"day", "subscribedStatus", "youtubeProduct"}), - ) - self.metrics = features.YOUTUBE_ANALYTICS_LOCATION_AND_TRAFFIC_PLAYLIST_METRICS - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"playlist", "group"}), - (FeatureAmount.ANY, {"subscribedStatus", "youtubeProduct"}), - ) - - def verify(self, metrics, dimensions, filters, *args): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - def __str__(self): - return "Device types and operating systems for playlists" - - -class ViewerDemographicsPlaylist(ReportType): - def __init__(self): - self.dimensions = ((FeatureAmount.NON_ZERO, {"ageGroup", "gender"}), (FeatureAmount.ANY, {"subscribedStatus"})) - self.metrics = ("viewerPercentage",) - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ZERO_OR_ONE, {"playlist", "group"}), - (FeatureAmount.ANY, {"subscribedStatus"}), - ) - - def verify(self, metrics, dimensions, filters, *args): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - def __str__(self): - return "Viewer demographics for playlists" - - -class TopPlaylists(ReportType): - def __init__(self): - self.dimensions = (FeatureAmount.REQUIRED, {"playlist"}) - self.metrics = features.YOUTUBE_ANALYTICS_ALL_PLAYLIST_METRICS - self.filters = ( - (FeatureAmount.REQUIRED, {"isCurated"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "province", "continent", "subContinent"}), - (FeatureAmount.ANY, {"playlist", "subscribedStatus", "youtubeProduct"}), - ) - - def verify(self, metrics, dimensions, filters, max_results, sort_by): - super().verify(metrics, dimensions, filters) - - if filters["isCurated"] != "1": - raise InvalidRequest("the 'isCurated' filter must be set to '1'") - - if not max_results or max_results >= 200: - raise InvalidRequest("the 'max_results' parameter must not be set above 200") - - if not sort_by: - raise InvalidRequest("you must provide at least 1 sort parameter") - - def __str__(self): - return "Top playlists" - - -class AdPerformance(ReportType): - def __init__(self): - self.dimensions = ((FeatureAmount.REQUIRED, {"adType"}), (FeatureAmount.OPTIONAL, {"day"})) - self.metrics = ("grossRevenue", "adImpressions", "cpm") - self.filters = ( - (FeatureAmount.ZERO_OR_ONE, {"video", "group"}), - (FeatureAmount.ZERO_OR_ONE, {"country", "continent", "subContinent"}), - ) - - def __str__(self): - return "Ad performance" - - -def determine(metrics, dimensions, filters): - curated = filters.get("isCurated", "0") == "1" - - if "sharingService" in dimensions: - return EngagementAndContentSharing - - if "elapsedVideoTimeRatio" in dimensions: - return AudienceRetention - - if "playlist" in dimensions: - return TopPlaylists - - if "insightPlaybackLocationType" in dimensions: - if curated: - return PlaybackLocationPlaylist - return PlaybackLocation - - if "insightPlaybackLocationDetail" in dimensions: - if curated: - return PlaybackLocationDetailPlaylist - return PlaybackLocationDetail - - if "insightTrafficSourceType" in dimensions: - if curated: - return TrafficSourcePlaylist - return TrafficSource - - if "insightTrafficSourceDetail" in dimensions: - if curated: - return TrafficSourceDetailPlaylist - return TrafficSourceDetail - - if "ageGroup" in dimensions or "gender" in dimensions: - if curated: - return ViewerDemographicsPlaylist - return ViewerDemographics - - if "deviceType" in dimensions: - if "operatingSystem" in dimensions: - if curated: - return DeviceTypeAndOperatingSystemPlaylist - return DeviceTypeAndOperatingSystem - if curated: - return DeviceTypePlaylist - return DeviceType - - if "operatingSystem" in dimensions: - if curated: - return OperatingSystemPlaylist - return OperatingSystem - - if "video" in dimensions: - if "province" in filters: - return TopVideosUS - if "subscribedStatus" not in filters: - return TopVideosRegional - if "province" not in filters and "youtubeProduct" not in filters: - return TopVideosSubscribed - if "averageViewPercentage" in metrics: - return TopVideosYouTubeProduct - return TopVideosPlaybackDetail - - if "country" in dimensions: - if "liveOrOnDemand" in dimensions or "liveOrOnDemand" in filters: - return PlaybackDetailsLiveGeographyBased - if curated: - return GeographyBasedActivityPlaylist - if ( - "subscribedStatus" in dimensions - or "subscribedStatus" in filters - or "youtubeProduct" in dimensions - or "youtubeProduct" in filters - ): - return PlaybackDetailsViewPercentageGeographyBased - return GeographyBasedActivity - - if "province" in dimensions: - if "liveOrOnDemand" in dimensions or "liveOrOnDemand" in filters: - return PlaybackDetailsLiveGeographyBasedUS - if curated: - return GeographyBasedActivityPlaylistUS - if ( - "subscribedStatus" in dimensions - or "subscribedStatus" in filters - or "youtubeProduct" in dimensions - or "youtubeProduct" in filters - ): - return PlaybackDetailsViewPercentageGeographyBasedUS - return GeographyBasedActivityUS - - if "youtubeProduct" in dimensions or "youtubeProduct" in filters: - if "liveOrOnDemand" in dimensions or "liveOrOnDemand" in filters: - return PlaybackDetailsLiveTimeBased - return PlaybackDetailsViewPercentageTimeBased - - if "liveOrOnDemand" in dimensions or "liveOrOnDemand" in filters: - return PlaybackDetailsLiveTimeBased - - if "subscribedStatus" in dimensions: - if "province" in filters: - return PlaybackDetailsSubscribedStatusUS - return PlaybackDetailsSubscribedStatus - - if "day" in dimensions or "month" in dimensions: - if curated: - return TimeBasedActivityPlaylist - if "province" in filters: - return TimeBasedActivityUS - return TimeBasedActivity - - if curated: - return BasicUserActivityPlaylist - if "province" in filters: - return BasicUserActivityUS - return BasicUserActivity diff --git a/analytix/youtube/service.py b/analytix/youtube/service.py deleted file mode 100644 index 715d242..0000000 --- a/analytix/youtube/service.py +++ /dev/null @@ -1,88 +0,0 @@ -import json -import os -import typing as t - -from google_auth_oauthlib.flow import InstalledAppFlow -from googleapiclient import discovery - -from analytix import IncompleteRequest, NoAuthorisedService, ServiceAlreadyExists -from analytix.youtube import ( - YOUTUBE_ANALYTICS_API_SERVICE_NAME, - YOUTUBE_ANALYTICS_API_VERSION, - YOUTUBE_ANALYTICS_SCOPES, -) - - -class YouTubeService: - """A YouTube service container to help with authorisation. - - Args: - secrets (dict | os.PathLike | str]): The filepath to a secrets file or a dictionary of credentials used to authorise a YouTube service. - - Parameters: - active (Resource | None): The currently authorised service. If no service is authorised, this None. - """ - - __slots__ = ("active", "_secrets") - - def __init__(self, secrets): - self.active = None - - if not isinstance(secrets, (os.PathLike, str)): - self._secrets = secrets - return - - with open(secrets, "r", encoding="utf-8") as f: - self._secrets = json.load(f) - - def authorise(self, use_console=False): - """Authorises the YouTube service. - - Args: - use_console (bool): Whether to use the console authorisation method. Defaults to False. - - Raises: - ServiceAlreadyExists: A service already exists, and has been authorised. - """ - if self.active: - raise ServiceAlreadyExists("an authorised service already exists") - - flow = InstalledAppFlow.from_client_config(self._secrets, YOUTUBE_ANALYTICS_SCOPES) - - try: - if use_console: - credentials = flow.run_console( - authorization_prompt_message="You need to authorise your service. Head to the below address, and enter the code.\n{url}", - authorization_code_message="CODE > ", - ) - else: - credentials = flow.run_local_server( - open_browser=True, - authorization_prompt_message="A browser window should have opened. Use this to authenticate your service.", - success_message="All done -- you're ready to start pulling reports! You can close this window now.", - ) - except OSError: - print("WARNING: Using console authorisation as server authorisation failed.") - credentials = flow.run_console( - authorization_prompt_message="You need to authorise your service. Head to the below address, and enter the code.\n{url}", - authorization_code_message="CODE > ", - ) - self.active = discovery.build( - YOUTUBE_ANALYTICS_API_SERVICE_NAME, YOUTUBE_ANALYTICS_API_VERSION, credentials=credentials - ) - - def authorize(self, use_console=False): - """An alias to :code:`authorise`.""" - self.authorise(use_console) - - def close(self): - """Closes a YouTube service - - Raises: - NoAuthorisedService: No authorised service currently exists. - """ - if not self.active: - raise NoAuthorisedService("no authorised service currently exists") - - self.active.close() - self.active = None