404

[ Avaa Bypassed ]




Upload:

Command:

botdev@3.21.166.125: ~ $
import datetime
import os
import re
from datetime import tzinfo
from ...constants import (NETWORK_MANAGER_LOGFILE, PROTON_XDG_CACHE_HOME_LOGS,
                          PROTONVPN_RECONNECT_LOGFILE)
from ..subprocess_wrapper import subprocess
from ..utils import Singleton

ZERO = datetime.timedelta(0)
HOUR = datetime.timedelta(hours=1)


class BugReport(metaclass=Singleton):
    DELTA_TIME_IN_DAYS = 3
    COMPILED_LOG_EPOCH_RE = re.compile(r"(\[\d+\.\d+\])")
    IS_USER_UNIT = False

    def generate_logs(self):
        """Generate all logs."""
        self.generate_network_manager_log()
        self.generate_protonvpn_reconnector_log()

    def generate_network_manager_log(self):
        """Generate NetworkManager log file for bug report.

        The log file is created with the help of python-systemd
        package which can easily read journalctl content.
        """
        self._remove_network_manager_log_if_exists()
        self.IS_USER_UNIT = False
        self.__generate_log("NetworkManager.service", NETWORK_MANAGER_LOGFILE)

    def generate_protonvpn_reconnector_log(self):
        """Generate Proton VPN Reconnect log file for bug report.

        The log file is created with the help of python-systemd
        package which can easily read journalctl content.
        """
        self._remove_protonvpn_reconnect_log_if_exists()
        self.IS_USER_UNIT = True
        self.__generate_log("protonvpn_reconnect.service", PROTONVPN_RECONNECT_LOGFILE)

    def _remove_network_manager_log_if_exists(self):
        self.__remove_log_if_exists(NETWORK_MANAGER_LOGFILE)

    def _remove_protonvpn_reconnect_log_if_exists(self):
        self.__remove_log_if_exists(PROTONVPN_RECONNECT_LOGFILE)

    def __generate_log(self, systemd_unit, filepath):
        """Generate log file.

        Args:
            systemd_unit (string): systemd .service name
            filepath (string): filepath to log file
        """
        from systemd import journal

        _journal = journal.Reader()

        if self.IS_USER_UNIT:
            _journal.add_match(_SYSTEMD_USER_UNIT=systemd_unit)
        else:
            _journal.add_match(_SYSTEMD_UNIT=systemd_unit)

        _journal.log_level(journal.LOG_DEBUG)

        self.__add_log_to_file(_journal, filepath)

        _journal.close()

    def __remove_log_if_exists(self, filepath):
        """Remove log file if it exists.

        Args:
            filepath (string): filepath to log file
        """
        if os.path.isfile(filepath):
            os.remove(filepath)

    def __add_log_to_file(self, journal, filepath):
        """Add log entry to file, line by line.

        The log fil will contain information from the last 3 days.

        Args:
            journal (systemd.journal.Reader): journal reader object
            filepath (string): filepath to log file
        """
        start_date = datetime.datetime.today() - datetime.timedelta(
            days=self.DELTA_TIME_IN_DAYS
        )
        with open(filepath, "a") as f:
            for entry in journal:

                # Skip entry if it's older then start date
                try:
                    if entry["_SOURCE_REALTIME_TIMESTAMP"] < start_date:
                        continue

                    edited_entry = self.__convert_time_to_utc(entry, "_SOURCE_REALTIME_TIMESTAMP")
                except KeyError:
                    if entry["__REALTIME_TIMESTAMP"] < start_date:
                        continue

                    edited_entry = self.__convert_time_to_utc(entry, "__REALTIME_TIMESTAMP")

                f.write(self.__format_entry(edited_entry))

    def __convert_time_to_utc(self, entry, key):
        dt = entry[key]
        entry[key] = dt.astimezone(UTC())

        return entry

    def __format_entry(self, entry):
        """Format log entry.

        It will also remove the time in epoch and replace it by human redeable time.

        Args:
            entry (dict): entry containing journalctl data
        """
        try:
            _date = str(entry["_SOURCE_REALTIME_TIMESTAMP"])
            _msg = self.COMPILED_LOG_EPOCH_RE.sub("", entry["MESSAGE"])
        except KeyError:
            _date = str(entry["__REALTIME_TIMESTAMP"])
            _msg = entry["MESSAGE"]

        _entry = _date + " " + _msg + "\n"

        return _entry

    def open_folder_with_logs(self):
        subprocess.run(["xdg-open", PROTON_XDG_CACHE_HOME_LOGS])


# Code snippets below were fetched from https://github.com/newvem/pytz
# Since the project is not available on all distro repos (that we support),
# only relevant parts of the code were imported.
# Some modifications had to be made to accomodate interoperability.


class BaseTzInfo(tzinfo):
    _utcoffset = None
    _tzname = None
    zone = None

    def __str__(self):
        return self.zone


class UTC(BaseTzInfo):
    """UTC

    Optimized UTC implementation. It unpickles using the single module global
    instance defined beneath this class declaration.
    """
    zone = "UTC"

    _utcoffset = ZERO
    _dst = ZERO
    _tzname = zone

    def fromutc(self, dt):
        if dt.tzinfo is None:
            return self.localize(dt)
        return super(UTC, self).fromutc(dt)

    def utcoffset(self, dt):
        return ZERO

    def tzname(self, dt):
        return "UTC"

    def dst(self, dt):
        return ZERO

    def __reduce__(self):
        return _UTC, ()

    def localize(self, dt, is_dst=False):
        '''Convert naive time to local time'''
        if dt.tzinfo is not None:
            raise ValueError('Not naive datetime (tzinfo is already set)')
        return dt.replace(tzinfo=self)

    def normalize(self, dt, is_dst=False):
        '''Correct the timezone information on the given datetime'''
        if dt.tzinfo is self:
            return dt
        if dt.tzinfo is None:
            raise ValueError('Naive time - no tzinfo set')
        return dt.astimezone(self)

    def __repr__(self):
        return "<UTC>"

    def __str__(self):
        return "UTC"


def _UTC():
    """Factory function for utc unpickling.

    Makes sure that unpickling a utc instance always returns the same
    module global.

    These examples belong in the UTC class above, but it is obscured; or in
    the README.rst, but we are not depending on Python 2.4 so integrating
    the README.rst examples with the unit tests is not trivial.

    >>> import datetime, pickle
    >>> dt = datetime.datetime(2005, 3, 1, 14, 13, 21, tzinfo=utc)
    >>> naive = dt.replace(tzinfo=None)
    >>> p = pickle.dumps(dt, 1)
    >>> naive_p = pickle.dumps(naive, 1)
    >>> len(p) - len(naive_p)
    17
    >>> new = pickle.loads(p)
    >>> new == dt
    True
    >>> new is dt
    False
    >>> new.tzinfo is dt.tzinfo
    True
    >>> utc is UTC is timezone('UTC')
    True
    >>> utc is timezone('GMT')
    False
    """
    return UTC


_UTC.__safe_for_unpickling__ = True

Filemanager

Name Type Size Permission Actions
__pycache__ Folder 0755
__init__.py File 52 B 0644
bug.py File 6.76 KB 0644