Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
formatter = _logging.Formatter('%(asctime)s [%(levelname)s] %(message)s')
for handler in _LOG.handlers: # type: _logging.Handler
handler.setFormatter(formatter)
if not getattr(args, 'func', None):
parser.error('too few arguments')
try:
if not args.config:
args.config = None
feeds = _feeds.Feeds(datafile=args.data, configfiles=args.config)
if args.func != _command.new:
lock = args.func not in [_command.list, _command.opmlexport]
feeds.load(lock=lock)
args.func(feeds=feeds, args=args)
except _error.RSS2EmailError as e:
e.log()
if _logging.ERROR - 10 * args.verbose < _logging.DEBUG:
raise # don't mask the traceback
_sys.exit(1)
'mail server settings are configured properly')
if hasattr(self.__cause__, 'reason'):
_LOG.error('reason: {}'.format(self.__cause__.reason))
class IMAPAuthenticationError (IMAPConnectionError):
def __init__(self, server, port, username):
message = (
'could not authenticate with mail server {}:{} as user {}'.format(
server, port, username))
super(IMAPAuthenticationError, self).__init__(
server=server, port=port, message=message)
self.username = username
class SendmailError (RSS2EmailError):
def __init__(self, status=None, stdout=None, stderr=None):
if status:
message = 'sendmail exited with code {}'.format(status)
else:
message = ''
super(SendmailError, self).__init__(message=message)
self.status = status
self.stdout = stdout
self.stderr = stderr
def log(self):
super(SendmailError, self).log()
_LOG.warning((
'Error attempting to send email via sendmail. You may need '
'to configure rss2email to use an SMTP server. Please refer '
'to the rss2email documentation or website ({}) for complete '
_LOG.warning(_pprint.pformat(self.parsed))
_LOG.warning('rss2email {}'.format(__version__))
_LOG.warning('feedparser {}'.format(_feedparser.__version__))
_LOG.warning('html2text {}'.format(_html2text.__version__))
_LOG.warning('Python {}'.format(_sys.version))
_LOG.warning('=== END HERE ===')
class HTTPError (FeedError):
def __init__(self, status, feed, **kwargs):
message = 'HTTP status {} fetching feed {}'.format(status, feed)
super(HTTPError, self).__init__(feed=feed, message=message)
self.status = status
class FeedsError (RSS2EmailError):
def __init__(self, feeds=None, message=None, **kwargs):
if message is None:
message = 'error with feeds'
super(FeedsError, self).__init__(message=message, **kwargs)
self.feeds = feeds
class DataFileError (FeedsError):
def __init__(self, feeds, message=None):
if message is None:
message = 'problem with the feed data file {}'.format(
feeds.datafile)
super(DataFileError, self).__init__(feeds=feeds, message=message)
class NoDataFile (DataFileError):
class TimeoutError (RSS2EmailError):
def __init__(self, time_limited_function, message=None):
if message is None:
if time_limited_function.error is not None:
message = (
'error while running time limited function: {}'.format(
time_limited_function.error[1]))
else:
message = '{} second timeout exceeded'.format(
time_limited_function.timeout)
super(TimeoutError, self).__init__(message=message)
self.time_limited_function = time_limited_function
class NoValidEncodingError (RSS2EmailError, ValueError):
def __init__(self, string, encodings):
message = 'no valid encoding for {} in {}'.format(string, encodings)
super(NoValidEncodingError, self).__init__(message=message)
self.string = string
self.encodings = encodings
class SMTPConnectionError (RSS2EmailError, ValueError):
def __init__(self, server, message=None):
if message is None:
message = 'could not connect to mail server {}'.format(server)
super(SMTPConnectionError, self).__init__(message=message)
self.server = server
def log(self):
super(SMTPConnectionError, self).log()
message = ''
super(SendmailError, self).__init__(message=message)
self.status = status
self.stdout = stdout
self.stderr = stderr
def log(self):
super(SendmailError, self).log()
_LOG.warning((
'Error attempting to send email via sendmail. You may need '
'to configure rss2email to use an SMTP server. Please refer '
'to the rss2email documentation or website ({}) for complete '
'documentation.').format(__url__))
class FeedError (RSS2EmailError):
def __init__(self, feed, message=None, **kwargs):
if message is None:
message = 'error with feed {}'.format(feed.name)
super(FeedError, self).__init__(message=message, **kwargs)
self.feed = feed
class InvalidFeedConfig (FeedError):
def __init__(self, setting, feed, message=None, **kwargs):
if not message:
message = "invalid feed configuration {}".format(
{setting: getattr(feed, setting)})
super(InvalidFeedConfig, self).__init__(
feed=feed, message=message, **kwargs)
self.setting = setting
last_server = "example.com"
for index in args.index:
feed = feeds.index(index)
# to debug feeds that timeout, run "r2e -VV run"
_LOG.info('refreshing feed {}'.format(feed))
if feed.active:
current_server = _urllib.parse.urlparse(feed.url).netloc
try:
if last_server == current_server:
_LOG.info('fetching from server {current_server} again, sleeping for {interval}s'.format(
current_server = current_server,
interval = interval
))
_time.sleep(interval)
feed.run(send=args.send)
except _error.RSS2EmailError as e:
e.log()
last_server = current_server
finally:
feeds.save()
def __init__(self, message):
super(RSS2EmailError, self).__init__(message)
'mail server settings are configured properly')
if hasattr(self.__cause__, 'reason'):
_LOG.error('reason: {}'.format(self.__cause__.reason))
class SMTPAuthenticationError (SMTPConnectionError):
def __init__(self, server, username):
message = (
'could not authenticate with mail server {} as user {}'.format(
server, username))
super(SMTPAuthenticationError, self).__init__(
server=server, message=message)
self.username = username
class IMAPConnectionError (ValueError, RSS2EmailError):
def __init__(self, server, port, message=None):
if message is None:
message = 'could not connect to mail server {}:{}'.format(
server, port)
super(IMAPConnectionError, self).__init__(message=message)
self.server = server
self.port = port
def log(self):
super(IMAPConnectionError, self).log()
_LOG.warning(
'check your config file to confirm that imap-server and other '
'mail server settings are configured properly')
if hasattr(self.__cause__, 'reason'):
_LOG.error('reason: {}'.format(self.__cause__.reason))
import feedparser as _feedparser
import html2text as _html2text
class RSS2EmailError (Exception):
def __init__(self, message):
super(RSS2EmailError, self).__init__(message)
def log(self):
_LOG.error(str(self))
if self.__cause__ is not None:
_LOG.error('cause: {}'.format(self.__cause__))
class TimeoutError (RSS2EmailError):
def __init__(self, time_limited_function, message=None):
if message is None:
if time_limited_function.error is not None:
message = (
'error while running time limited function: {}'.format(
time_limited_function.error[1]))
else:
message = '{} second timeout exceeded'.format(
time_limited_function.timeout)
super(TimeoutError, self).__init__(message=message)
self.time_limited_function = time_limited_function
class NoValidEncodingError (RSS2EmailError, ValueError):
def __init__(self, string, encodings):
message = 'no valid encoding for {} in {}'.format(string, encodings)
super(NoToEmailAddress, self).log()
_LOG.warning(
"please run 'r2e email emailaddress' or "
"'r2e add name url emailaddress'.")
class FeedIndexError (FeedsError, IndexError):
def __init__(self, index, message=None, **kwargs):
if message is None:
message = 'feed {!r} not found'.format(index)
super(FeedIndexError, self).__init__(
message=message, **kwargs)
self.index = index
class OPMLReadError (RSS2EmailError):
def __init__(self, **kwargs):
message = 'error reading OPML'
super(OPMLReadError, self).__init__(message=message, **kwargs)
class ConfigAlreadyExistsError (FeedsError):
def __init__(self, feeds=None):
message = 'configuration file already exists'
super().__init__(feeds=feeds, message=message)