404

[ Avaa Bypassed ]




Upload:

Command:

botdev@18.191.158.217: ~ $
import json
import time
import os

from abc import abstractmethod

from ...constants import PROTON_XDG_CACHE_HOME_NOTIFICATION_ICONS
from ...logger import logger
from ...enums import NotificationEnum
from ..utils import SubclassesMixin


class NotificationData:

    def __init__(self):
        self.__data = None

    def json_dumps(self):
        return json.dumps(self.__data)

    def json_loads(self, data):
        self.__data = json.loads(data)

    def update_notifications_data(self, data):
        assert "Code" in data
        assert "Notifications" in data

        if data["Code"] != 1000:
            raise ValueError("Invalid data with code != 1000")

        data["NotificationsUpdateTimestamp"] = time.time()
        self.__data = data

    def get_notification(self, notification_type):
        try:
            _data = self.__data.get("Notifications", None)[0]
        except (IndexError, AttributeError) as e:
            logger.exception(e)
            _data = {}

        return BaseNotificationType.factory(
            _data,
            notification_type
        )

    def get_all_notifications(self):
        notifications = self.__data.get("Notifications", None)
        if len(notifications) < 1:
            logger.info("Nofitications are empty: {}".format(notifications))
            _data = {}
        else:
            _data = notifications[0]

        return BaseNotificationType.factory(_data)

    @property
    def notifications_timestamp(self):
        try:
            return self.__data.get("NotificationsUpdateTimestamp", 0.0)
        except AttributeError:
            return 0.0


class BaseNotificationType(SubclassesMixin):

    def __init__(self, data):
        self.__data = data

    @classmethod
    def factory(cls, data, attribute=None):
        if not data or len(data) == 0:
            return cls._get_subclasses_dict("notification_type")[
                NotificationEnum.EMPTY.value
            ](data)

        if not attribute:
            return [_intance(data) for _intance in cls._get_all_subclasses()]
        else:
            subclasses_dict = cls._get_subclasses_dict("notification_type")
            return subclasses_dict[attribute](data)

    @property
    def start_time(self):
        return self.__data.get("StartTime", 0)

    @property
    def end_time(self):
        return self.__data.get("EndTime", 0)

    @property
    @abstractmethod
    def can_be_displayed():
        raise NotImplementedError("Should be implemented")

    @property
    def type_of_notification(self):
        return self.__data.get("Type", None)

    @property
    def url(self):
        return self.offer.get("URL", None)

    @property
    def icon(self):
        return self.offer.get("Icon", None)

    @property
    def label(self):
        return self.offer.get("Label", None)

    @property
    def incentive(self):
        return self.panel.get("Incentive", None)

    @property
    def incentive_price(self):
        return self.panel.get("IncentivePrice", None)

    @property
    def pill(self):
        return self.panel.get("Pill", None)

    @property
    def picture_url(self):
        return self.panel.get("PictureURL", None)

    @property
    def title(self):
        return self.panel.get("Title", None)

    @property
    def features(self):
        return self.panel.get("Features", [])

    @property
    def features_footer(self):
        return self.panel.get("FeaturesFooter", None)

    @property
    def button_text(self):
        return self.button.get("Text", None)

    @property
    def button_url(self):
        return self.button.get("URL", None)

    @property
    def page_footer(self):
        return self.panel.get("PageFooter", None)

    @property
    def offer(self):
        return self.__data.get("Offer", {})

    @property
    def panel(self):
        return self.offer.get("Panel", {})

    @property
    def button(self):
        return self.panel.get("Button", {})


class EmptyNotificationObject(BaseNotificationType):
    notification_type = NotificationEnum.EMPTY.value

    """This class is used only when there is no data available,
    so that instead of creating multiple crashes and multiple error handling
    paths due to lack of data, it just returns an object with empty data, thus
    lowering the risk of incurring into exceptions and thus managing them.
    """
    def __init__(self, data):
        self.icon_paths = set()
        super().__init__(data)

    @property
    def can_be_displayed(self):
        return False


class GenericNotification(BaseNotificationType):
    notification_type = NotificationEnum.GENERIC.value

    def __init__(self, data):
        self.icon_paths = set()
        super().__init__(data)
        if self.can_be_displayed:
            self.__cache_icons()

    @property
    def incentive(self):
        _incentive = self.panel.get("Incentive", None)
        if not _incentive:
            return None

        return _incentive.replace(" ", "\u00a0")

    @property
    def incentive_price(self):
        _incentive_price = self.panel.get("IncentivePrice", None)
        if not _incentive_price:
            return None

        return _incentive_price.replace("/", "/\u2060")

    @property
    def incentive_template_index_start(self):
        return int(self.incentive.find("%IncentivePrice%"))

    @property
    def features(self):
        features = self.panel.get("Features", None)
        if not features:
            return []

        _f = []
        for feature_dict in features:
            feature_dict.get("IconURL", None)
            _f.append(
                (
                    feature_dict.get("Text", None),
                    feature_dict.get("IconURL", None)
                )
            )

        return _f

    @property
    def can_be_displayed(self):
        now = time.time()
        if (
            0 in [self.start_time, self.end_time]
        ) or (
            now < self.start_time or now > self.end_time
        ):
            return False

        return True

    def __cache_icons(self):
        import concurrent.futures
        import re

        icon_tuple_collection = set()
        pattern = re.compile(r"[\/]{1}([a-zA-Z0-9-]+\.(png|jpeg|jpg))")
        self.__recursive_search_for_icons(self.offer, icon_tuple_collection, pattern)

        if not os.path.isdir(PROTON_XDG_CACHE_HOME_NOTIFICATION_ICONS):
            os.makedirs(PROTON_XDG_CACHE_HOME_NOTIFICATION_ICONS)

        if all(list(map(self.__check_if_icons_exist, icon_tuple_collection))):
            self.icon_paths = {
                os.path.join(PROTON_XDG_CACHE_HOME_NOTIFICATION_ICONS, icon_tuple[0])
                for icon_tuple in icon_tuple_collection
            }
            return

        with concurrent.futures.ThreadPoolExecutor() as executor:
            executor.map(self.__download_and_store_icon, icon_tuple_collection)

    def __check_if_icons_exist(self, data):
        icon_name, url = data
        return self.__check_if_specific_icon_exists(icon_name)

    def __download_and_store_icon(self, data):
        icon_name, url = data

        if not self.__check_if_specific_icon_exists(icon_name):
            content = self.__download_icon(url)
            self.__store_icon(content, icon_name)

    def __download_icon(self, url):
        import requests
        try:
            r = requests.get(url, timeout=3)
        except requests.exceptions.BaseHTTPError as e:
            logger.exception(e)
            return

        return r.content

    def __store_icon(self, content, icon_name):
        from io import BytesIO
        from PIL import Image

        path = os.path.join(
            PROTON_XDG_CACHE_HOME_NOTIFICATION_ICONS,
            icon_name
        )

        i = Image.open(BytesIO(content))
        i.save(path)
        self.icon_paths.add(path)

    def __check_if_specific_icon_exists(self, icon_name):
        if not os.path.isfile(os.path.join(
            PROTON_XDG_CACHE_HOME_NOTIFICATION_ICONS,
            icon_name
        )):
            return False

        return True

    def __recursive_search_for_icons(self, data, icon_collection, pattern):
        if isinstance(data, dict):
            for k, v in data.items():
                if isinstance(v, str):
                    pattern_result = pattern.search(v)
                    if pattern_result:
                        icon_collection.add((pattern_result.group(1), v))
                elif isinstance(v, (dict, list)):
                    self.__recursive_search_for_icons(v, icon_collection, pattern)
        elif isinstance(data, list):
            for v in data:
                if isinstance(v, str):
                    pattern_result = pattern.search(v)
                    if pattern_result:
                        icon_collection.add((pattern_result.group(1), v))
                elif isinstance(v, (list, dict)):
                    self.__recursive_search_for_icons(v, icon_collection, pattern)

Filemanager

Name Type Size Permission Actions
__pycache__ Folder 0755
__init__.py File 76 B 0644
notifications.py File 8.77 KB 0644