Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
'.mkv,.avi,.divx,.xvid,.mov,.wmv,.mp4,.mpg .mpeg,.vob,,; ;',
('.mkv,.avi,.divx,.xvid,.mov ', ' .wmv,.mp4;.mpg,.mpeg,'),
'.vob,.iso', ['.vob', ['.vob', '.mkv', StrangeObject(), ], ],
StrangeObject())
assert results == sorted([
'.divx', '.iso', '.mkv', '.mov', '.mpg', '.avi', '.mpeg', '.vob',
'.xvid', '.wmv', '.mp4',
])
# Garbage in is removed
assert utils.parse_list(object(), 42, None) == []
# Now a list with extras we want to add as strings
# empty entries are removed
results = utils.parse_list([
'.divx', '.iso', '.mkv', '.mov', '', ' ', '.avi', '.mpeg', '.vob',
'.xvid', '.mp4'], '.mov,.wmv,.mp4,.mpg')
assert results == sorted([
'.divx', '.wmv', '.iso', '.mkv', '.mov', '.mpg', '.avi', '.vob',
'.xvid', '.mpeg', '.mp4',
])
if not (config or urls) else config), asset=asset)
# Load our inventory up
for url in urls:
a.add(url)
if len(a) == 0:
logger.error(
'You must specify at least one server URL or populated '
'configuration file.')
print_help_msg(main)
sys.exit(1)
# each --tag entry comprises of a comma separated 'and' list
# we or each of of the --tag and sets specified.
tags = None if not tag else [parse_list(t) for t in tag]
if not dry_run:
if body is None:
logger.trace('No --body (-b) specified; reading from stdin')
# if no body was specified, then read from STDIN
body = click.get_text_stream('stdin').read()
# now print it out
result = a.notify(
body=body, title=title, notify_type=notification_type, tag=tags,
attach=attach)
else:
# Number of rows to assume in the terminal. In future, maybe this can
# be detected and made dynamic. The actual row count is 80, but 5
# characters are already reserved for the counter on the left
rows = 75
if not self.port:
if self.secure:
self.port = self.default_secure_port
else:
self.port = self.default_port
# Email SMTP Server Timeout
try:
self.timeout = int(timeout)
except (ValueError, TypeError):
self.timeout = self.connect_timeout
# Acquire targets
self.targets = parse_list(targets)
# Acquire Carbon Copies
self.cc = set()
# Acquire Blind Carbon Copies
self.bcc = set()
# Now we want to construct the To and From email
# addresses from the URL provided
self.from_name = from_name
self.from_addr = from_addr
if not self.from_addr:
# detect our email address
self.from_addr = '{}@{}'.format(
re.split(r'[\s@]+', self.user)[0],
def __init__(self, accesstoken, targets=None, **kwargs):
"""
Initialize PushBullet Object
"""
super(NotifyPushBullet, self).__init__(**kwargs)
# Access Token (associated with project)
self.accesstoken = validate_regex(accesstoken)
if not self.accesstoken:
msg = 'An invalid PushBullet Access Token ' \
'({}) was specified.'.format(accesstoken)
self.logger.warning(msg)
raise TypeError(msg)
self.targets = parse_list(targets)
if len(self.targets) == 0:
self.targets = (PUSHBULLET_SEND_TO_ALL, )
return
raise TypeError(msg)
else:
self.access_token = validate_regex(
access_token, *self.template_tokens['access_token']['regex'])
if not self.access_token:
msg = 'An invalid Slack OAuth Access Token ' \
'({}) was specified.'.format(access_token)
self.logger.warning(msg)
raise TypeError(msg)
if not self.user:
self.logger.warning(
'No user was specified; using "%s".' % self.app_id)
# Build list of channels
self.channels = parse_list(targets)
if len(self.channels) == 0:
# No problem; the webhook is smart enough to just notify the
# channel it was created for; adding 'None' is just used as
# a flag lower to not set the channels
self.channels.append(
None if self.mode is SlackMode.WEBHOOK
else self.default_notification_channel)
# Formatting requirements are defined here:
# https://api.slack.com/docs/message-formatting
self._re_formatting_map = {
# New lines must become the string version
r'\r\*\n': '\\n',
# Escape other special characters
r'&': '&',
r'<': '<',
# Further check our phone # for it's digit count
result = ''.join(re.findall(r'\d+', result.group('phone')))
if len(result) < 11 or len(result) > 14:
msg = 'The MessageBird source # specified ({}) is invalid.'\
.format(source)
self.logger.warning(msg)
raise TypeError(msg)
# Store our source
self.source = result
# Parse our targets
self.targets = list()
targets = parse_list(targets)
if not targets:
# No sources specified, use our own phone no
self.targets.append(self.source)
return
# otherwise, store all of our target numbers
for target in targets:
# Validate targets and drop bad ones:
result = IS_PHONE_NO.match(target)
if result:
# Further check our phone # for it's digit count
result = ''.join(re.findall(r'\d+', result.group('phone')))
if len(result) < 11 or len(result) > 14:
self.logger.warning(
'Dropped invalid phone # '
'({}) specified.'.format(target),
# Set Cache Flag
self.cache = cache
if self.mode not in TWITTER_MESSAGE_MODES:
msg = 'The Twitter message mode specified ({}) is invalid.' \
.format(mode)
self.logger.warning(msg)
raise TypeError(msg)
# Track any errors
has_error = False
# Identify our targets
self.targets = []
for target in parse_list(targets):
match = IS_USER.match(target)
if match and match.group('user'):
self.targets.append(match.group('user'))
continue
has_error = True
self.logger.warning(
'Dropped invalid user ({}) specified.'.format(target),
)
if has_error and not self.targets:
# We have specified that we want to notify one or more individual
# and we failed to load any of them. Since it's also valid to
# notify no one at all (which means we notify ourselves), it's
# important we don't switch from the users original intentions
msg = 'No Twitter targets to notify.'
msg = 'The Account (From) Phone # specified ' \
'({}) is invalid.'.format(source)
self.logger.warning(msg)
raise TypeError(msg)
# else... it as a short code so we're okay
else:
# We're dealing with a phone number; so we need to just
# place a plus symbol at the end of it
self.source = '+{}'.format(self.source)
# Parse our targets
self.targets = list()
for target in parse_list(targets):
# Validate targets and drop bad ones:
result = IS_PHONE_NO.match(target)
if result:
# Further check our phone # for it's digit count
# if it's less than 10, then we can assume it's
# a poorly specified phone no and spit a warning
result = ''.join(re.findall(r'\d+', result.group('phone')))
if len(result) < 11 or len(result) > 14:
self.logger.warning(
'Dropped invalid phone # '
'({}) specified.'.format(target),
)
continue
# store valid phone number
self.targets.append('+{}'.format(result))
self.mode = RocketChatAuthMode.BASIC
if self.mode == RocketChatAuthMode.BASIC \
and not (self.user and self.password):
# Username & Password is required for Rocket Chat to work
msg = 'No Rocket.Chat user/pass combo was specified.'
self.logger.warning(msg)
raise TypeError(msg)
elif self.mode == RocketChatAuthMode.WEBHOOK and not self.webhook:
msg = 'No Rocket.Chat Incoming Webhook was specified.'
self.logger.warning(msg)
raise TypeError(msg)
# Validate recipients and drop bad ones:
for recipient in parse_list(targets):
result = IS_CHANNEL.match(recipient)
if result:
# store valid device
self.channels.append(result.group('name'))
continue
result = IS_ROOM_ID.match(recipient)
if result:
# store valid room
self.rooms.append(result.group('name'))
continue
result = IS_USER.match(recipient)
if result:
# store valid room
self.users.append(result.group('name'))
if not self.token:
msg = 'An invalid Pushover Access Token ' \
'({}) was specified.'.format(token)
self.logger.warning(msg)
raise TypeError(msg)
# User Key (associated with project)
self.user_key = validate_regex(
user_key, *self.template_tokens['user_key']['regex'])
if not self.user_key:
msg = 'An invalid Pushover User Key ' \
'({}) was specified.'.format(user_key)
self.logger.warning(msg)
raise TypeError(msg)
self.targets = parse_list(targets)
if len(self.targets) == 0:
self.targets = (PUSHOVER_SEND_TO_ALL, )
# Setup our sound
self.sound = NotifyPushover.default_pushover_sound \
if not isinstance(sound, six.string_types) else sound.lower()
if self.sound and self.sound not in PUSHOVER_SOUNDS:
msg = 'The sound specified ({}) is invalid.'.format(sound)
self.logger.warning(msg)
raise TypeError(msg)
# The Priority of the message
if priority not in PUSHOVER_PRIORITIES:
self.priority = self.template_args['priority']['default']
else: