Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_schedule_interval_in_seconds(self, seconds_ago=0):
"""Return the job interval in seconds or None if there is no interval
:params seconds_ago: return an interval the job had in the past
"""
schedule = self.get_schedule()
if schedule is None:
return None
if CronSlices.is_valid(schedule):
try:
job_tz = pytz.timezone(self.get_schedule_time_zone())
except (pytz.exceptions.UnknownTimeZoneError, AttributeError):
job_tz = pytz.utc
c = croniter(schedule, datetime.datetime.now(job_tz) - datetime.timedelta(seconds=seconds_ago))
return c.get_next() - c.get_prev()
else:
try:
_, _, interval = self.get_schedule().split('/')
return int(float(isodate.parse_duration(interval).total_seconds()))
except (isodate.ISO8601Error, ValueError):
return None
elif args.mode == "schedule":
DEFAULT_CRON = "0 * * * *" # every hour
if CONFIG_BACKUP_ENABLED is None or CONFIG_BACKUP_ENABLED.lower() == "false" or CONFIG_BACKUP_ENABLED.lower() == "off":
log.info("Configuration Backup is not activated.")
sys.exit()
if not os.path.exists(CONFIG_BACKUP_FOLDER):
os.makedirs(CONFIG_BACKUP_FOLDER)
from crontab import CronTab, CronSlices
cron_schedule = DEFAULT_CRON
# env variable can also be a cron scheadule
if CronSlices.is_valid(CONFIG_BACKUP_ENABLED):
cron_schedule = CONFIG_BACKUP_ENABLED
# Cron does not provide enviornment variables, source them manually
environment_file = os.path.join(RESOURCE_FOLDER, "environment.sh")
with open(environment_file, 'w') as fp:
for env in os.environ:
if env != "LS_COLORS":
fp.write("export " + env + "=\"" + os.environ[env] + "\"\n")
os.chmod(environment_file, 0o777)
script_file_path = os.path.realpath(__file__)
command = ". " + environment_file + "; " + sys.executable + " '" + script_file_path + "' backup> /proc/1/fd/1 2>/proc/1/fd/2"
cron = CronTab(user=True)
# remove all other backup tasks
'cd',
quoted(out_dir),
'&&',
quoted(ARCHIVEBOX_BINARY),
*(['add', f'"{import_path}"'] if import_path else ['update']),
'2>&1',
'>',
quoted(os.path.join(LOGS_DIR, 'archivebox.log')),
]
new_job = cron.new(command=' '.join(cmd), comment=CRON_COMMENT)
if every in ('minute', 'hour', 'day', 'week', 'month', 'year'):
set_every = getattr(new_job.every(), every)
set_every()
elif CronSlices.is_valid(every):
new_job.setall(every)
else:
stderr('{red}[X] Got invalid timeperiod for cron task.{reset}'.format(**ANSI))
stderr(' It must be one of minute/hour/day/week/month')
stderr(' or a quoted cron-format schedule like:')
stderr(' archivebox init --every=day https://example.com/some/rss/feed.xml')
stderr(' archivebox init --every="0/5 * * * *" https://example.com/some/rss/feed.xml')
raise SystemExit(1)
cron = dedupe_cron_jobs(cron)
cron.write()
total_runs = sum(j.frequency_per_year() for j in cron)
existing_jobs = list(cron.find_comment(CRON_COMMENT))
print()
def check_schedule(self):
msgs = []
schedule = self.get_schedule()
if schedule is not None:
if not CronSlices.is_valid(schedule):
try:
repeat, start_time, interval = schedule.split('/') # the parts have separate validators
except ValueError:
return (
False, (
'The specified schedule "%s" is neither a valid cron schedule nor a valid'
' ISO 8601 schedule' % schedule
),
)
# an empty start time is not valid ISO8601 but Chronos accepts it: '' == current time
if start_time == '':
msgs.append('The specified schedule "%s" does not contain a start time' % schedule)
else:
# Check if start time contains time zone information
try:
"""
Add a single event in the crontab.
Will add a line like:
python /path/to/kalliope.py start --brain-file /path/to/brain.yml --run-synapse ""
E.g:
30 7 * * * python /home/me/kalliope/kalliope.py start --brain-file /home/me/brain.yml --run-synapse "Say-hello"
:param period_string: crontab period
:type period_string: str
:param event_id:
:type event_id: str
:return:
"""
my_user_cron = CronTab(user=True)
job = my_user_cron.new(command=self.base_command+" "+str("\"" + event_id + "\""), comment=CRONTAB_COMMENT)
if CronSlices.is_valid(period_string):
job.setall(period_string)
job.enable()
else:
raise InvalidCrontabPeriod("The crontab period %s is not valid" % period_string)
# write the file
my_user_cron.write()
Utils.print_info("Synapse \"%s\" added to the crontab" % event_id)
DEFAULT_CRON = "0 * * * *" # every hour
from crontab import CronTab, CronSlices
cron_schedule = DEFAULT_CRON
script_file_path = os.path.realpath(__file__)
command = sys.executable + " '" + script_file_path + "' check> /proc/1/fd/1 2>/proc/1/fd/2"
cron = CronTab(user=True)
# remove all other tasks
cron.remove_all(command=command)
job = cron.new(command=command)
if CronSlices.is_valid(cron_schedule):
log.info("Scheduling cron check xfdesktop task with with cron: " + cron_schedule)
job.setall(cron_schedule)
job.enable()
cron.write()
else:
log.info("Failed to schedule check xfdesktop. Cron is not valid.")
log.info("Running cron jobs:")
for job in cron:
log.info(job)
def _is_valid_crontab(crontab):
if CronSlices.is_valid(crontab):
return crontab
raise argparse.ArgumentTypeError(
'Invalid crontab.',
)
with open(environment_file, 'w') as fp:
for env in os.environ:
if env != "LS_COLORS":
fp.write("export " + env + "=\"" + os.environ[env] + "\"\n")
os.chmod(environment_file, 0o777)
script_file_path = os.path.realpath(__file__)
command = ". " + environment_file + "; " + sys.executable + " '" + script_file_path + "' backup> /proc/1/fd/1 2>/proc/1/fd/2"
cron = CronTab(user=True)
# remove all other backup tasks
cron.remove_all(command=command)
job = cron.new(command=command)
if CronSlices.is_valid(cron_schedule):
log.info("Scheduling cron config backup task with with cron: " + cron_schedule)
job.setall(cron_schedule)
job.enable()
cron.write()
else:
log.info("Failed to schedule config backup. Cron is not valid.")
log.info("Running cron jobs:")
for job in cron:
log.info(job)