File: C:/Users/fred/anaconda3/Lib/site-packages/anaconda_navigator/utils/attribution/pool.py
# -*- coding: utf-8 -*-
"""Storage for detected and fetched widgets."""
from __future__ import annotations
__all__ = ['PartnerPool', 'POOL', 'UPDATER']
import json
import re
import typing
from urllib import parse
import yaml
from qtpy import QtCore
from anaconda_navigator import config
from anaconda_navigator.config import preferences, feature_flags
from anaconda_navigator.utils import extra_collections
from anaconda_navigator.utils.logs.loggers import logger
from anaconda_navigator.utils import singletons
from anaconda_navigator.utils import workers
from . import core
from . import resources
from . import simple_advertisement
if typing.TYPE_CHECKING:
from qtpy import QtWidgets
class PartnerConfiguration(typing.TypedDict, total=False):
"""Dictionary of an additional configuration, that might be used by advertisements and the application."""
utm_campaign: str
class PartnerPool(typing.Mapping[core.PartnerWidgetPlacement, 'QtWidgets.QWidget']):
"""Collection of widgets for partner attribution."""
__slots__ = ('__content', '__settings')
def __init__(self) -> None:
"""Initialize new :class:`~PartnerPool` instance."""
self.__settings: core.PartnerSettings = core.PartnerSettings()
self.__content: typing.Final[typing.Dict[core.PartnerWidgetPlacement, core.PartnerWidget]] = {}
@property
def settings(self) -> core.PartnerSettings: # noqa: D401
"""Common configuration that might be used by widgets and different application components."""
return self.__settings
def _set_settings(self, value: typing.Union[core.PartnerSettings, 'core.SettingsDict']) -> None:
"""Update value of the `configuration`."""
if not isinstance(value, core.PartnerSettings):
value = core.PartnerSettings(value)
self.__settings = value
def register(self, widget: core.PartnerWidget) -> None:
"""Register new widget in the pool."""
self.__content[widget.placement] = widget
def __getitem__(self, key: core.PartnerWidgetPlacement) -> typing.Optional[QtWidgets.QWidget]:
"""Retrieve widget that should be placed in `key` location."""
value: typing.Optional[core.PartnerWidget] = self.__content.get(key, None)
if value is None:
return None
return value.widget()
def __iter__(self) -> typing.Iterator[core.PartnerWidgetPlacement]:
"""Iterate through available location options."""
return iter(core.PartnerWidgetPlacement)
def __len__(self) -> int:
"""Total number of """
return len(core.PartnerWidgetPlacement)
POOL: typing.Final[PartnerPool] = PartnerPool()
class Updater(QtCore.QObject):
"""Core for updating advertisement information."""
sig_updated = QtCore.Signal()
def __init__(self) -> None:
"""Initialize new :class:`~Updater` instance."""
super().__init__()
self.__partner_configuration: typing.Optional['PartnerConfiguration'] = None
@property
def partner_configuration(self) -> 'PartnerConfiguration': # noqa: D401
"""Configuration for partners."""
path: str
if self.__partner_configuration is None:
for path in preferences.AD_CONFIGURATION_PATHS:
try:
stream: typing.TextIO
with open(path, 'rt', encoding='utf8') as stream:
result: 'PartnerConfiguration' = yaml.safe_load(stream)
if not isinstance(result, typing.Mapping):
logger.exception('broken partner configuration file at %s', path) # type: ignore
continue
self.__partner_configuration = result
break
except FileNotFoundError:
continue
except (OSError, yaml.YAMLError):
logger.exception('broken partner configuration file at %s', path)
continue
else:
self.__partner_configuration = {}
# normalize utm_campaign value
self.__partner_configuration.setdefault('utm_campaign', '')
if not isinstance(self.__partner_configuration['utm_campaign'], str):
logger.warning('utm_campaign in partner configuration must be a string') # type: ignore
self.__partner_configuration['utm_campaign'] = ''
if len(self.__partner_configuration['utm_campaign']) > 64:
logger.warning('utm_campaign value is too long')
self.__partner_configuration['utm_campaign'] = ''
return self.__partner_configuration
@property
def sources(self) -> extra_collections.OrderedSet[str]: # noqa: D401
"""List of URLs to fetch ads from."""
advertisement_url: str = config.CONF.get('main', 'advertisement_url')
# prepare parameters for URL formatting
source_values: typing.Mapping[str, str] = {
'partner_identity': parse.quote(self.partner_configuration['utm_campaign']),
}
source_placeholders: typing.Mapping[str, str] = {key: '✫' for key in source_values}
source_values = {key: value for key, value in source_values.items() if value}
# format and collect URLs from preferences
source: str
result: extra_collections.OrderedSet[str] = extra_collections.OrderedSet()
for source in preferences.AD_SOURCES:
pending: bool = False
try:
result.add(source.format(**source_values))
except KeyError:
pending = True
if re.fullmatch(re.escape(source.format(**source_placeholders)).replace('✫', '.+'), advertisement_url):
if pending:
result.add(advertisement_url)
advertisement_url = ''
# add custom URL from user preferences
# it would be automatically discarded in case it is already added in one of previous steps
result.add(advertisement_url)
# make sure empty URL is not in the queue
result.discard('')
return result
@workers.Task
def update(self) -> None:
"""Fetch and apply advertisement configuration to the :data:`~POOL`."""
configuration: core.ConfigurationDict = self.__fetch_configuration()
# make configuration available to everything
POOL._set_settings(configuration['settings']) # pylint: disable=protected-access
# create widgets according to the configuration
for widget in configuration['widgets']:
instance: core.PartnerWidget
# NOTE: to support additional built-in advertisement types:
#
# if widget.get('type', 'simple') == 'simple':
instance = simple_advertisement.CompositeAdvertisementWidget(
definition=typing.cast('simple_advertisement.CompositeAdvertisementDict', widget),
settings=POOL.settings,
)
POOL.register(instance)
# NOTE: NAV-753 should be here
#
# Parse other locations for widgets, initialize them with `POOL.settings` and then `POOL.register` them.
self.sig_updated.emit()
def __fetch_configuration(self) -> core.ConfigurationDict:
"""Collect configuration for advertisements."""
result: core.ConfigurationDict = {
'widgets': [],
'settings': {
'url_parameters': {},
},
}
# process advertisement URLs
for advertisement_url in self.sources:
try:
advertisement_content: typing.Optional[bytes] = resources.load_resource(
advertisement_url, extra_headers=feature_flags.prepare_headers()
)
if not advertisement_content:
continue
# Fetch configuration
definition: 'simple_advertisement.CompositeAdvertisementDict' = {
'placement': core.PartnerWidgetPlacement.BOTTOM_LEFT_CORNER,
'advertisements': json.loads(advertisement_content),
}
if not definition['advertisements']:
continue
# Store configuration
result['widgets'].append(definition)
# Extract utm parameters from the redirect_url
item: simple_advertisement.SimpleAdvertisementDict
for item in definition['advertisements']:
if 'redirect_url' not in item:
continue
result['settings']['url_parameters'].update(
(key, value)
for key, value in parse.parse_qsl(parse.urlparse(item['redirect_url']).query)
if key.startswith('utm_')
)
break # NOTE: remove if we need to combine data from all URLs and not just the first valid one
# Remember the working API url
config.CONF.set('main', 'advertisement_url', advertisement_url)
break
except BaseException: # pylint: disable=broad-except
logger.exception('failed to initialize advertisement')
# apply `utm_campaign` from `partner_configuration`
if self.partner_configuration['utm_campaign']:
result['settings']['url_parameters']['utm_campaign'] = self.partner_configuration['utm_campaign']
# apply default values for required parameters
result['settings']['url_parameters'].setdefault('utm_source', 'anaconda_navigator')
return result
UPDATER: typing.Final[singletons.Singleton[Updater]] = singletons.SingleInstanceOf(Updater)