RT-Client-Tools/scripts/rt-auto-remind

157 lines
4.7 KiB
Text
Raw Normal View History

2018-05-15 10:43:22 +00:00
#!/usr/bin/env python3
import argparse
import datetime
import hashlib
import json
import pathlib
import os
import sqlite3
import subprocess
import sys
import tempfile
DATETIME_FMT = '%Y-%m-%d %H:%M:%S'
try:
_data_home = os.environ['XDG_DATA_HOME']
except KeyError:
_data_home = pathlib.Path(os.path.expanduser('~'), '.local', 'share')
SHARE_DIR = pathlib.Path(_data_home, 'rt-auto-remind')
SHARE_DIR.mkdir(parents=True, exist_ok=True)
class LastRunStorage:
DB_FILENAME = 'RunData.db'
V1_CREATE_SQL = '''
CREATE TABLE IF NOT EXISTS LastRunV1(
key text PRIMARY KEY,
max_date text
);
'''
V1_SELECT = 'SELECT max_date FROM LastRunV1 WHERE key = ?'
V1_INSERT = 'INSERT OR REPLACE INTO LastRunV1 VALUES(?, ?);'
def __init__(self, args):
self.db = sqlite3.connect(str(SHARE_DIR / self.DB_FILENAME))
self.db.executescript(self.V1_CREATE_SQL)
if args.key is None:
self.key = ';'.join(
'{}={!r}'.format(key, str(getattr(args, key)))
for key in ['body_file', 'date_field', 'max_days_diff', 'search']
)
else:
self.key = args.key
def load(self):
cursor = self.db.execute(self.V1_SELECT, (self.key,))
result = cursor.fetchone()
if result is None:
return None
else:
return datetime.datetime.strptime(result[0], DATETIME_FMT)
def save(self, end_date):
end_date_s = end_date.strftime(DATETIME_FMT)
with self.db:
self.db.execute(self.V1_INSERT, (self.key, end_date_s))
def close(self):
self.db.close()
def parse_arguments(arglist):
parser = argparse.ArgumentParser()
action = parser.add_mutually_exclusive_group()
action.add_argument(
'--correspond',
dest='action', action='store_const', const='correspond',
default='correspond',
help="Send correspondence on found tickets (default)",
)
action.add_argument(
'--comment',
dest='action', action='store_const', const='comment',
help="Comment on found tickets",
)
parser.add_argument(
'--key', '-k',
help="Use this string to load and save run times in the database"
" (default is auto-generated)",
)
parser.add_argument(
'--dry-run', '-n',
action='store_true',
help="Show the rt-bulk-send command that would be run; don't run it",
)
parser.add_argument(
'date_field',
help="RT date field to constrain in the search",
)
parser.add_argument(
'min_days_diff',
type=int,
help="The earliest end of the date constraint, in days from today",
)
parser.add_argument(
'max_days_diff',
type=int,
help="The latest end of the date constraint, in days from today",
)
parser.add_argument(
'search',
help="TicketSQL search, like you would pass to `rt search`",
)
parser.add_argument(
'body_file',
type=pathlib.Path,
help="Path to file that has the content of your correspondence/comment",
)
parser.add_argument(
'rt_args', metavar='rt arguments',
nargs=argparse.REMAINDER,
help="Additional arguments to pass to `rt correspond/comment`",
)
args = parser.parse_args(arglist)
args.min_delta = datetime.timedelta(days=args.min_days_diff)
args.max_delta = datetime.timedelta(days=args.max_days_diff)
return args
def main(arglist=None, stdout=sys.stdout, stderr=sys.stderr):
args = parse_arguments(arglist)
last_run_db = LastRunStorage(args)
last_run_range_end = last_run_db.load()
start_datetime = datetime.datetime.utcnow()
date_range_start = start_datetime + args.min_delta
if (last_run_range_end is not None) and (last_run_range_end > date_range_start):
date_range_start = last_run_range_end
date_range_end = start_datetime + args.max_delta
search = '({search}) AND {field} > "{start}" AND {field} <= "{end}"'.format(
search=args.search,
field=args.date_field,
start=date_range_start.strftime(DATETIME_FMT),
end=date_range_end.strftime(DATETIME_FMT),
)
send_cmd = [
'rt-bulk-send', '--{}'.format(args.action),
2018-05-15 10:43:22 +00:00
search, str(args.body_file), *args.rt_args,
]
if args.dry_run:
import pprint
pprint.pprint(send_cmd)
returncode = 0
else:
try:
subprocess.run(send_cmd, check=True)
except subprocess.CalledProcessError as error:
returncode = error.returncode
else:
last_run_db.save(date_range_end)
returncode = 0
last_run_db.close()
return returncode
if __name__ == '__main__':
exit(main())