123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110 |
- import datetime as datetime
- from urllib.parse import urlparse
- import webbrowser
- from emailtool.emailer import Emailer
- from flask import Flask, render_template, request, redirect, url_for, session, send_file
- from gevent.pywsgi import WSGIServer
- import socket
- import logging
- from peewee import Model, CharField, DateTimeField, SqliteDatabase, BooleanField, fn
- import hashlib
- import os
- import csv
- import time
- import threading
- import atexit
- db_name = 'app.db'
- db = SqliteDatabase(db_name)
- class User(Model):
- username = CharField()
- password = CharField()
- date_created = DateTimeField(default=datetime.datetime.now)
- logged_in = BooleanField(default=False)
- class Meta:
- database = db
- class IlsUser(Model):
- username = CharField()
- email = CharField()
- reset_datetime = CharField()
- class Meta:
- database = db
- class Settings(Model):
- name = CharField()
- value = CharField()
- class Meta:
- database = db
- class Schedule(Model):
- interval = CharField()
- class Meta:
- database = db
- class Log(Model):
- date = DateTimeField(default=datetime.datetime.now)
- username = CharField()
- action = CharField()
- class Meta:
- database = db
- class PasswordResetLog(Model):
- date = CharField(default=datetime.datetime.now().strftime('%Y-%m-%d'))
- username = CharField()
- class Meta:
- database = db
- class EmailReminderLog(Model):
- date = CharField(default=datetime.datetime.now().strftime('%Y-%m-%d'))
- username = CharField()
- email = CharField()
- interval = CharField()
- sent = BooleanField(default=False)
- ignore = BooleanField(default=False)
- class Meta:
- database = db
- def reminder_cleanup():
- reminders = EmailReminderLog.select().where(EmailReminderLog.date < datetime.datetime.now().strftime('%Y-%m-%d')).execute()
- for reminder in reminders:
- reminder.delete_instance()
- def scheduler(stop_event):
- while not stop_event.is_set():
- db.connect()
- # Get all scheduled intervals from the DB
- intervals = Schedule.select().execute()
- users = IlsUser.select().execute()
- # Get settings
- domain_name_setting = Settings.get(Settings.name == 'Domain Name')
- domain_name = domain_name_setting.value
- password_reset_interval_setting = Settings.get(Settings.name == 'Password Reset Interval')
- password_reset_interval = password_reset_interval_setting.value
- smtp_host_setting = Settings.get(Settings.name == 'SMTP Host')
- smtp_host = smtp_host_setting.value
- smtp_port_setting = Settings.get(Settings.name == 'SMTP Port')
- smtp_port = smtp_port_setting.value
- smtp_username_setting = Settings.get(Settings.name == 'SMTP Username')
- smtp_username = smtp_username_setting.value
- smtp_password_setting = Settings.get(Settings.name == 'SMTP Password')
- smtp_password = smtp_password_setting.value
- reset_link_setting = Settings.get(Settings.name == 'Password Reset URL')
- reset_link = reset_link_setting.value
- # Loop through each user and calculate the number of days until the users password
- # expires based on the password_reset_interval value.
- # Loop through all users
- # Schedule emails to be sent for each user and interval
- for user in users:
- # get the scheduled intervals
- for interval in intervals:
- # Calculate the date (yyyy-mm-dd) that the password will expire and subtract the interval
- # from that date to get the date that the email should be sent.
- # Example: If the password reset interval is 90 days and the current date is 2019-01-01
- # then the password will expire on 2019-04-01. If the interval is 30 days then the
- # email should be sent on 2019-03-02.
- # Calculate the date that the password will expire
- password_expiration_date = datetime.datetime.strptime(user.reset_datetime, "%Y-%m-%d") + datetime.timedelta(days=int(password_reset_interval))
- # Subtract the interval from the password expiration date to get the date that the email should be sent
- email_date = password_expiration_date - datetime.timedelta(days=int(interval.interval))
- # check if reminder has already been saved to the DB
- reminder = EmailReminderLog.select().where(EmailReminderLog.date == email_date.strftime('%Y-%m-%d'), EmailReminderLog.username == user.username, EmailReminderLog.interval == interval.interval).execute()
- if reminder:
- continue
- else:
- # Store the date in the DB
- EmailReminderLog.create(date=email_date.strftime('%Y-%m-%d'), email=user.email, username=user.username, interval=interval.interval).save()
- # Get all email reminders from the DB
- # if today's date matches the EmailReminderLog date then send the email and mark the reminder as sent
- reminders = EmailReminderLog.select().where(EmailReminderLog.date == datetime.datetime.now().strftime('%Y-%m-%d'), EmailReminderLog.ignore == 0, EmailReminderLog.sent == 0).execute()
- for reminder in reminders:
- # Send the email
- emailer = Emailer(smtp_host, smtp_port, smtp_username, smtp_password)
- if reminder.interval == 1:
- # 1 day reminder email
- emailer.send_email(reminder.email, 'ILS Password Reset Reminder',
- 'The password for ILS account with the username "'
- + reminder.username + '" will expire in ' + reminder.interval +
- ' day. To reset your password, please click the following link: ' + reset_link + '. If the password is not reset today the account will be locked.')
- else:
- # more than a day reminder email
- emailer.send_email(reminder.email, 'ILS Password Reset Reminder',
- 'The password for ILS account with the username "'
- + reminder.username + '" will expire in ' + reminder.interval +
- ' days. To reset your password, please click the following link: ' + reset_link)
- # Mark the reminder as sent
- reminder.sent = True
- reminder.save()
- db.close()
- # Remove all reminders from the DB that are older than today's date
- reminder_cleanup()
- time.sleep(10) # Sleep for 15 minutes
- # Encrypt the password with SHA256
- def encrypt_password(password):
- hash = hashlib.sha256(password.encode('utf-8')).hexdigest()
- return hash
- # Check for DB tables and create if they don't exist
- if db.table_exists('user') is False:
- db.create_tables([User, ])
- User.create(username='admin', password=encrypt_password('admin'),
- date_created=datetime.datetime.now(), logged_in=False).save()
- if db.table_exists('ilsuser') is False:
- db.create_tables([IlsUser, ])
- if db.table_exists('settings') is False:
- db.create_tables([Settings, ])
- Settings.create(name='Debug Mode', value=False).save()
- Settings.create(name='HTTP Port', value=5055).save()
- Settings.create(name='SMTP Host', value="").save()
- Settings.create(name='SMTP Port', value="587").save()
- Settings.create(name='SMTP Username', value="").save()
- Settings.create(name='SMTP Password', value="").save()
- Settings.create(name='Domain Name', value="lynx").save()
- Settings.create(name='Password Reset URL', value="https://terminal.idaho-lynx.org").save()
- Settings.create(name='Password Reset Interval', value="90").save()
- if db.table_exists('log') is False:
- db.create_tables([Log, ])
- if db.table_exists('password_reset_log') is False:
- db.create_tables([PasswordResetLog, ])
- if db.table_exists('schedule') is False:
- db.create_tables([Schedule, ])
- if db.table_exists('email_reminder_log') is False:
- db.create_tables([EmailReminderLog, ])
- db.close()
- settings = Settings.select().execute()
- debug_setting = Settings.get(Settings.name == 'Debug Mode')
- debug = debug_setting.value
- http_port_setting = Settings.get(Settings.name == 'HTTP Port')
- http_port = http_port_setting.value
- domain_name_setting = Settings.get(Settings.name == 'Domain Name')
- domain_name = domain_name_setting.value
- password_reset_url_setting = Settings.get(Settings.name == 'Password Reset URL')
- password_reset_url = password_reset_url_setting.value
- password_reset_interval_setting = Settings.get(Settings.name == 'Password Reset Interval')
- password_reset_interval = password_reset_interval_setting.value
- log = logging.getLogger('werkzeug')
- log.setLevel(logging.INFO)
- if debug.lower() == 'true':
- debug = True
- else:
- debug = False
- # send_email('aday@twinfallspubliclibrary.org', 'TEST', 'This is a test email')
- def shutdown_session(exception=None):
- print('Stopping HTTP Service...')
- http_server.stop()
- # Get the systems hostname
- def get_hostname():
- return socket.gethostname()
- # Get systems IP address
- def get_ip_address():
- return socket.gethostbyname(socket.gethostname())
- # Method to check if a URL is valid using regex
- def is_valid_url(url):
- try:
- result = urlparse(url)
- if all([result.scheme, result.netloc]):
- url = '{uri.scheme}://{uri.netloc}/'.format(uri=result)
- else:
- url = False
- return url
- except:
- return False
- def requires_auth():
- if 'username' in session:
- username = session['username']
- user = User.get(User.username == username)
- if user.logged_in is True:
- return True
- else:
- return False
- else:
- return False
- def admin_password_check():
- admin_user = User.get(User.username == 'admin')
- if admin_user.password == encrypt_password('admin'):
- return True
- db.close()
- # Start the scheduler loop in another thread
- shutdown_scheduler = threading.Event()
- scheduler_thread = threading.Thread(target=scheduler, args=(shutdown_scheduler,))
- scheduler_thread.start()
- # Start the HTTP Server
- app = Flask(__name__)
- #app.secret_key = os.urandom(24)
- app.secret_key = "DEVELOPMENT"
- def format_time_ago(timestamp):
- """Calculate the time passed since a datetime stamp and format it as a human-readable string."""
- now = datetime.datetime.now()
- diff = now - datetime.datetime.strptime(timestamp, "%Y-%m-%d")
- if diff.days > 365:
- years = diff.days // 365
- return f"{years} year{'s' if years > 1 else ''} ago"
- if diff.days > 30:
- months = diff.days // 30
- return f"{months} month{'s' if months > 1 else ''} ago"
- if diff.days > 0:
- return f"{diff.days} day{'s' if diff.days > 1 else ''} ago"
- return ""
- # Create method for time until
- def format_time_until(timestamp):
- """Calculate the time until a datetime stamp and format it as a human-readable string."""
- # Convert timestamp into datetime object
- timestamp = datetime.datetime.strptime(timestamp, '%Y-%m-%d')
- now = datetime.datetime.now()
- diff = timestamp - now
- if diff.days > 365:
- years = diff.days // 365
- return f"in {years} year{'s' if years > 1 else ''}"
- # return months and days
- if diff.days > 30:
- months = diff.days // 30
- days = diff.days % 30
- return f"in {months} month{'s' if months > 1 else ''} and {days} day{'s' if days > 1 else ''}"
- # return days
- if diff.days > 0:
- return f"in {diff.days} day{'s' if diff.days > 1 else ''}"
- # return hours and minutes
- if diff.seconds > 3600:
- hours = diff.seconds // 3600
- minutes = (diff.seconds % 3600) // 60
- return f"in {hours} hour{'s' if hours > 1 else ''} and {minutes} minute{'s' if minutes > 1 else ''}"
- # return minutes
- if diff.seconds > 60:
- minutes = diff.seconds // 60
- return f"in {minutes} minute{'s' if minutes > 1 else ''}"
- if diff.days > 30:
- months = diff.days // 30
- return f"in {months} month{'s' if months > 1 else ''}"
- if diff.days > 0:
- return f"in {diff.days} day{'s' if diff.days > 1 else ''}"
- return "Today"
- app.jinja_env.filters['time_since'] = format_time_ago
- app.jinja_env.filters['time_until'] = format_time_until
- @app.before_request
- def before_request():
- db.connect()
- @app.after_request
- def after_request(response):
- db.close()
- return response
- @app.route('/admin/password/reset', methods=['GET', 'POST'])
- def admin_password_reset():
- message = None
- if request.method == 'POST':
- password = request.form.get('password')
- password_confirm = request.form.get('password_confirm')
- if password != password_confirm:
- message = 'Passwords do not match'
- else:
- try:
- user = User.get(User.username == 'admin')
- user.password = encrypt_password(password)
- user.logged_in = True
- user.save()
- session['username'] = 'admin'
- return redirect(url_for('admin'))
- except Exception as e:
- print(e)
- message = 'Username not found'
- context = {
- 'message': message,
- }
- return render_template('admin_password_reset.html', context=context)
- # Create a route for the home page
- @app.route('/', methods=['GET', 'POST'])
- def index():
- error = None
- reset = False
- reset_url = is_valid_url(password_reset_url)
- reset_url_error = False
- if reset_url is False:
- reset_url_error = True
- if request.method == 'POST':
- username = request.form.get('username')
- # Check for the username in the DB
- try:
- user = IlsUser.filter(IlsUser.username == username).first()
- except Exception as e:
- print(e)
- user = None
- if user:
- # Reset login datetime
- user.reset_datetime = datetime.datetime.now().strftime("%Y-%m-%d")
- PasswordResetLog.create(username=user.username, date_created=datetime.datetime.now()).save()
- user.save()
- # Open the reset URL in a new tab if the URL is valid
- if reset_url is not False:
- webbrowser.open_new_tab(str(reset_url))
- # Set reset to True to pass back to the view to display the correct content back to the user.
- reset = True
- else:
- error = 'Invalid username'
- context = {
- 'domain': domain_name,
- 'error': error,
- 'reset': reset,
- 'reset_url': reset_url,
- 'reset_url_error': reset_url_error,
- }
- return render_template('index.html', context=context)
- # Create a route for admin page
- @app.route('/admin/')
- def admin():
- if admin_password_check():
- return redirect(url_for('admin_password_reset'))
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- # Get a count of all IlsUsers
- try:
- ils_users = IlsUser.select().count()
- except Exception as e:
- ils_users = None
- # Get a count of all EmailReminders that are not ignored or sent
- try:
- total_email_reminders = EmailReminderLog.select().where(EmailReminderLog.sent == False, EmailReminderLog.ignore == False).count()
- except Exception as e:
- total_email_reminders = None
- # Get a list of coming email reminders for the next 7 days
- try:
- email_reminders = PasswordResetLog.select().where(PasswordResetLog.date_created < datetime.datetime.now() + datetime.timedelta(days=7)).execute()
- except Exception as e:
- email_reminders = None
- # Get a list ils users that have passwords expiring in the next 7 days
- try:
- ils_users_expiring = IlsUser.select().where(IlsUser.password_expires < datetime.datetime.now() + datetime.timedelta(days=7)).execute()
- except Exception as e:
- ils_users_expiring = None
- # get a list of email notifications sent in the last 7 days
- try:
- # Get a list of all future email reminders
- email_reminders = EmailReminderLog.select().order_by(-EmailReminderLog.date).limit(15).execute()
- except Exception as e:
- email_reminders = None
- context = {
- 'ils_user_count': ils_users,
- 'ils_users_expiring': ils_users_expiring,
- 'email_reminders': email_reminders,
- 'total_email_reminders': total_email_reminders,
- }
- return render_template('admin.html', context=context)
- @app.route('/admin/users/', methods=['GET', 'POST'])
- def admin_users():
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- message = None
- if request.method == 'POST':
- username = request.form.get('username')
- password = request.form.get('password')
- confirm_password = request.form.get('confirm_password')
- # Check to see if username already exists
- try:
- user = User.filter(User.username == username).first()
- except Exception as e:
- print(e)
- user = None
- if user:
- message = 'Username already exists'
- else:
- if password == confirm_password:
- User.create(username=username, password=encrypt_password(password),
- date_created=datetime.datetime.now(), logged_in=False).save()
- message = 'User created successfully'
- Log.create(username=session['username'], action='Created admin user: %s' % username, ).save()
- else:
- message = 'Passwords do not match'
- # Get all admin users from the DB
- users = User.select().execute()
- context = {
- 'users': users,
- 'message': message,
- }
- return render_template('admin_users.html', context=context)
- @app.route('/admin/users/edit/<int:id>', methods=['GET', 'POST'])
- def admin_users_edit(id):
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- # Get the user from the DB
- user = User.get(User.id == id)
- message = None
- if request.method == 'POST':
- username = request.form.get('username')
- password = request.form.get('password')
- confirm_password = request.form.get('confirm_password')
- # Check to see if username already exists
- all_users = list()
- users = User.select().execute()
- for u in users:
- if u.username != user.username:
- all_users.append(u.username)
- if username in all_users:
- message = 'Username already exists'
- else:
- user.username = username
- if password is not None or password != '':
- if password == confirm_password:
- user.password = encrypt_password(password)
- else:
- message = 'Passwords do not match'
- user.save()
- message = 'User updated successfully'
- Log.create(username=session['username'], action='Updated admin user: %s' % username, ).save()
- context = {
- 'user': user,
- 'message': message,
- }
- return render_template('admin_user_edit.html', context=context)
- @app.route('/admin/users/delete/<int:id>', methods=['GET', 'POST'])
- def admin_users_delete(id):
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- # Get the user from the DB
- user = User.get(User.id == id)
- if user.username != 'admin':
- # Unset session variable if the user is deleting their own account
- if user.username == session['username']:
- session.pop('username', None)
- username = user.username
- user.delete_instance()
- Log.create(username=session['username'], action='Removed admin user: %s' % username, ).save()
- return redirect(url_for('admin_users'))
- @app.route('/admin/users/ils', methods=['GET', 'POST'])
- def admin_ils_users():
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- message = None
- if request.method == 'POST':
- username = request.form.get('username')
- email = request.form.get('email')
- date = request.form.get('date')
- print(date)
- # Check to see if username already exists
- try:
- user = IlsUser.filter(IlsUser.username == username).first()
- except Exception as e:
- print(e)
- user = None
- # Check if date field was submitted if so convert to datetime object
- if date is None or date == '':
- date = datetime.datetime.strftime(datetime.datetime.now(), '%Y-%m-%d')
- if user:
- message = 'Username already exists'
- else:
- IlsUser.create(username=username, email=email, reset_datetime=date).save()
- message = 'ILS User: %s created successfully' % username
- Log.create(username=session['username'], action='Created ILS User: %s' % username, ).save()
- # Get all admin users from the DB
- users = IlsUser.select().execute()
- context = {
- 'users': users,
- 'message': message,
- }
- return render_template('admin_ils_users.html', context=context)
- @app.route('/admin/users/ils/delete/<int:id>', methods=['GET', 'POST'])
- def admin_ils_users_delete(id):
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- # Get the user from the DB
- user = IlsUser.get(IlsUser.id == id)
- username = user.username
- # Remove scheduled email reminders
- try:
- email_reminders = EmailReminderLog.select().where(EmailReminderLog.username == user.username).execute()
- for reminder in email_reminders:
- reminder.delete_instance()
- print('here')
- except Exception as e:
- print('there', e)
- user.delete_instance()
- Log.create(username=session['username'], action='Removed ILS user: %s' % username, ).save()
- return redirect(url_for('admin_ils_users'))
- @app.route('/admin/users/ils/edit/<int:id>', methods=['GET', 'POST'])
- def admin_ils_users_edit(id):
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- # Get the user from the DB
- user = IlsUser.get(IlsUser.id == id)
- message = None
- if request.method == 'POST':
- username = request.form.get('username')
- email = request.form.get('email')
- date = request.form.get('date')
- # Check to see if username already exists
- all_users = list()
- users = IlsUser.select().execute()
- for u in users:
- if u.username != user.username:
- all_users.append(u.username)
- if username in all_users:
- message = 'Username already exists'
- else:
- # Remove all scheduled email reminders. They will automatically be rescheduled with the new information.
- try:
- email_reminders = EmailReminderLog.select().where(EmailReminderLog.username == user.username).execute()
- for reminder in email_reminders:
- reminder.delete_instance()
- except Exception as e:
- print(e)
- user.username = username
- user.email = email
- if date is not None and date != '':
- user.reset_datetime = date
- else:
- user.reset_datetime = datetime.datetime.strftime(datetime.datetime.now(), '%Y-%m-%d')
- user.save()
- message = 'User updated successfully'
- Log.create(username=session['username'], action='Updated ILS user: %s' % username, ).save()
- context = {
- 'user': user,
- 'message': message,
- }
- return render_template('admin_ils_user_edit.html', context=context)
- # create a route for generating a CSV file for download
- @app.route('/admin/users/ils/csv/download', methods=['GET', 'POST'])
- def admin_ils_users_csv_download():
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- # Create a CSV file with the users and don't add a blank line between rows
- with open('ils-users_%s.csv' % datetime.datetime.now().strftime('%Y-%m-%d'), 'w', newline='') as f:
- writer = csv.writer(f)
- writer.writerow(['ILS Username', 'Email', 'Last Password Reset (YYYY-MM-DD)'])
- users = IlsUser.select().execute()
- for user in users:
- date = datetime.datetime.strptime(user.reset_datetime, '%Y-%m-%d').strftime('%Y-%m-%d')
- writer.writerow([user.username, user.email, date])
- Log.create(username=session['username'], action='Downloaded ILS user CSV file.').save()
- # return the CSV file to the user
- return send_file('ils-users_%s.csv' % datetime.datetime.now().strftime('%Y-%m-%d'), as_attachment=True)
- @app.route('/admin/users/ils/csv/import', methods=['GET', 'POST'])
- def admin_ils_users_csv_import():
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- message = None
- if request.method == 'POST':
- csv_file = request.files['csv']
- if csv_file.filename != '':
- csv_file.save(os.path.join('uploads', csv_file.filename))
- with open(os.path.join('uploads', csv_file.filename), 'r') as f:
- reader = csv.reader(f)
- for row in reader:
- username = row[0]
- email = row[1]
- reset_date = row[2]
- try:
- reset_date = datetime.datetime.strptime(reset_date, '%m/%d/%Y').strftime('%Y-%m-%d')
- except Exception as e:
- print(e)
- try:
- reset_date = datetime.datetime.strptime(reset_date, '%Y-%m-%d').strftime('%Y-%m-%d')
- except Exception as e:
- print(e)
- # ignore the header row
- if username == 'ILS Username':
- continue
- # ignore blank rows
- if username == '':
- continue
- # Check if user already exists and if it does update the entry
- try:
- user = IlsUser.filter(IlsUser.username == username).first()
- except Exception as e:
- print(e)
- user = None
- if user:
- user.email = email
- # Check if the reset date is blank and if it is set it to the current date
- if reset_date == '' or reset_date is None:
- user.reset_datetime = datetime.datetime.strftime(datetime.datetime.now(), '%Y-%m-%d')
- else:
- user.reset_datetime = reset_date
- user.save()
- else:
- if reset_date == '' or reset_date is None:
- user.reset_datetime = datetime.datetime.strftime(datetime.datetime.now(), '%Y-%m-%d')
- else:
- reset_datetime = reset_date
- IlsUser.create(username=username, email=email, reset_datetime=reset_datetime).save()
- # Delete the uploaded file
- os.remove(os.path.join('uploads', csv_file.filename))
- return redirect(url_for('admin_ils_users'))
- context = {
- 'message': message,
- }
- return render_template('csv.html', context=context)
- @app.route('/admin/settings', methods=['GET', 'POST'])
- def settings():
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- message = None
- # Process form submission
- if request.method == 'POST':
- # Assign form values to variables
- id = request.form.get('id')
- value = request.form.get('value')
- # Check if the setting exists
- try:
- setting = Settings.get(Settings.id == id)
- except Exception as e:
- print(e)
- setting = None
- if setting:
- # Scrub the values based on the setting name
- if setting.name == 'Debug Mode':
- if value.lower() == 'true':
- value = True
- else:
- value = False
- value = bool(value)
- elif setting.name == 'HTTP Port':
- value = str(value)
- elif setting.name == 'SMTP Host':
- value = str(value)
- elif setting.name == 'SMTP Port':
- value = str(value)
- elif setting.name == 'SMTP Username':
- value = str(value)
- elif setting.name == 'SMTP Password':
- value = str(value)
- elif setting.name == 'Domain Name':
- value = str(value)
- elif setting.name == 'Password Reset URL':
- value = str(value)
- elif setting.name == 'Password Reset Interval':
- if value.isdigit():
- value = int(value)
- else:
- value = setting.value
- # Remove all scheduled reminders
- reminders = EmailReminderLog.select().execute()
- for reminder in reminders:
- reminder.delete_instance()
- # Update the setting
- old_value = setting.value
- setting.value = value
- setting.save()
- Log.create(username=session['username'], action='Changed %s setting from "%s" to "%s"' % (setting.name,
- old_value,
- value)).save()
- message = '%s updated successfully' % setting.name
- # Get settings from DB
- all_settings = Settings.select().execute()
- context = {
- 'settings': all_settings,
- 'message': message,
- }
- return render_template('settings.html', context=context)
- @app.route('/admin/schedule', methods=['GET', 'POST'])
- def schedule():
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- message = None
- # Get all schedules from the DB
- schedules = Schedule.select().order_by(Schedule.interval.cast("INTEGER")).execute()
- # add schedule
- if request.method == 'POST':
- # Assign form values to variables
- interval = request.form.get('interval')
- # Check if interval is a number
- try:
- int(interval)
- except Exception as e:
- print(e)
- message = 'Error creating schedule. Value submitted must be a whole number.'
- return render_template('schedule.html', context={'message': message, 'schedules': schedules})
- # Check if the schedule already exists
- try:
- schedule = Schedule.get(Schedule.interval == interval)
- except Exception as e:
- print(e)
- schedule = None
- if schedule:
- message = 'Schedule already exists'
- else:
- # Create the schedule
- Schedule.create(interval=interval).save()
- Log.create(username=session['username'], action='Created schedule for %s day interval.' % interval).save()
- message = 'Schedule: %s created successfully' % interval
- schedules = Schedule.select().order_by(Schedule.interval.cast("INTEGER")).execute()
- context = {
- 'schedules': schedules,
- 'message': message,
- }
- return render_template('schedule.html', context=context)
- @app.route('/admin/schedule/emails')
- def scheduled_emails():
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- # Get all future emailreminderlogs
- reminders = EmailReminderLog.select().where(EmailReminderLog.date > datetime.datetime.now(), EmailReminderLog.ignore == False).order_by(
- -EmailReminderLog.date).execute()
- context = {
- 'reminders': reminders,
- }
- return render_template('scheduled_emails.html', context=context)
- @app.route('/admin/schedule/emails/clear')
- def reset_scheduled_emails():
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- # Get all future emailreminderlogs
- reminders = EmailReminderLog.select().execute()
- for reminder in reminders:
- reminder.delete_instance()
- return redirect(url_for('scheduled_emails'))
- # remove schedule
- @app.route('/admin/schedule/remove/reminder/<int:id>')
- def reminder_remove(id):
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- # Get EmailReminderLogs with the interval of the schedule being removed
- email_reminder_log = EmailReminderLog.get(EmailReminderLog.id == id)
- Log.create(username=session['username'],
- action='Removed %s day reminder for %s (%s)' % (email_reminder_log.interval, email_reminder_log.username, email_reminder_log.email)).save()
- email_reminder_log.ignore = True
- email_reminder_log.save()
- return redirect(url_for('scheduled_emails'))
- # remove schedule
- @app.route('/admin/schedule/remove/<int:id>', methods=['GET', 'POST'])
- def schedule_remove(id):
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- # Get Schedule by id
- try:
- schedule = Schedule.get(Schedule.id == id)
- except Exception as e:
- schedule = None
- if schedule:
- # Get EmailReminderLogs with the interval of the schedule being removed
- email_reminder_logs = EmailReminderLog.select().where(EmailReminderLog.interval == schedule.interval).execute()
- for email_reminder_log in email_reminder_logs:
- email_reminder_log.delete_instance()
- schedule.delete_instance()
- Log.create(username=session['username'], action='Removed schedule for a %s day reminder.' % schedule.interval).save()
- return redirect(url_for('schedule'))
- @app.route('/admin/system/log')
- def system_log():
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- # Get all logs from the DB
- logs = Log.select().order_by(Log.id.desc()).execute()
- context = {
- 'logs': logs,
- }
- return render_template('system_log.html', context=context)
- @app.route('/admin/system/log/password/resets')
- def password_reset_log():
- # Check to see if user is logged in
- if not requires_auth():
- return redirect(url_for('login'))
- # Get all logs from the DB
- logs = PasswordResetLog.select().order_by(PasswordResetLog.id.desc()).execute()
- context = {
- 'logs': logs,
- }
- return render_template('password_reset_log.html', context=context)
- @app.route('/logout')
- def logout():
- if 'username' in session:
- username = session['username']
- user = User.get(User.username == username)
- user.logged_in = False
- Log.create(username=session['username'], action='Logged out').save()
- user.save()
- session.pop('username', None)
- return redirect(url_for('login'))
- @app.route('/login', methods=['GET', 'POST'])
- def login():
- if request.method == 'POST':
- username = request.form.get('username')
- password = encrypt_password(request.form.get('password'))
- try:
- user = User.filter(User.username == username and User.password == password).first()
- except Exception as e:
- print(e)
- user = None
- session.pop('username', None)
- if user:
- # Login user
- session['username'] = request.form.get('username')
- user.logged_in = True
- Log.create(username=session['username'], action='Logged in').save()
- user.save()
- return redirect(url_for('admin'))
- else:
- error = 'Invalid Credentials. Please try again.'
- context = {
- 'error': error
- }
- return render_template('login.html', context=context)
- context = {
- }
- return render_template('login.html', context=context)
- def clean_up():
- shutdown_scheduler.set()
- http_server.stop()
- if __name__ == "__main__":
- print("------------------------- Start up -----------------------------")
- print("Starting HTTP Service on port %s..." % http_port)
- if debug is True:
- print("Debug mode is enabled.")
- http_server = WSGIServer(('0.0.0.0', int(http_port)), app)
- else:
- http_server = WSGIServer(('0.0.0.0', int(http_port)), app, log=log, error_log=log)
- print("HTTP Service Started.")
- print("--------------------- Application Details ---------------------")
- print("Application started at %s" % datetime.datetime.now())
- print("System IP Address: %s" % get_ip_address())
- print("System Hostname: %s" % get_hostname())
- print("Access the Dashboard using a web browser using any of the following:")
- print("http://%s:%s or http://%s:%s" % (get_hostname(), http_port, get_ip_address(), http_port))
- print("---------------------------------------------------------------")
- print("To stop the application close this window.")
- print("---------------------------------------------------------------")
- http_server.serve_forever()
- atexit.register(clean_up)
|