Separated ŠOLAR from other portals.

This commit is contained in:
msinkec 2021-10-12 11:11:37 +02:00
parent 7d9d2b175c
commit cead80ed79
16 changed files with 803 additions and 1317 deletions

318
app.py
View File

@ -12,16 +12,10 @@ from flask_script import Manager
from flask_login import LoginManager, login_required, login_user, current_user, logout_user
from portal.model import db, RegisteredUser
import portal.base
import portal.solar
import portal.regular
# TODO: Implement user registration.
# TODO: Integrate Shibboleth login.
# TODO: make logging level configurable
logging.basicConfig(level=logging.DEBUG, format='[APP LOGGER] %(asctime)s %(levelname)s: %(message)s')
@ -44,8 +38,6 @@ CONTRACT_CLIENT_CONTACT = config['CONTRACT_CLIENT_CONTACT']
MAIL_SUBJECT = config['MAIL_SUBJECT']
MAIL_BODY = config['MAIL_BODY']
SQL_CONN_STR = config['SQL_CONN_STR']
DESC_PREVODI = config['DESC_PREVODI']
DESC_GIGAFIDA = config['DESC_GIGAFIDA']
if 'UPLOADS_DIR' in config:
UPLOADS_DIR = Path(config['UPLOADS_DIR'])
@ -81,13 +73,6 @@ if 'PORTALDS4DS1_MAIL_BODY' in os.environ:
MAIL_BODY = os.environ['PORTALDS4DS1_MAIL_BODY']
if 'PORTALDS4DS1_SQL_CONN_STR' in os.environ:
SQL_CONN_STR = os.environ['PORTALDS4DS1_SQL_CONN_STR']
if 'PORTALDS4DS1_DESC_PREVODI' in os.environ:
DESC_PREVODI = os.environ['PORTALDS4DS1_DESC_PREVODI']
if 'PORTALDS4DS1_DESC_GIGAFIDA' in os.environ:
DESC_GIGAFIDA = os.environ['PORTALDS4DS1_DESC_GIGAFIDA']
ENABLED_CORPUSES = ['prevodi', 'gigafida', 'solar']
CORPUSES_LOGIN_REQUIRED = ['solar']
######################
@ -113,20 +98,6 @@ manager.add_command('db', MigrateCommand)
# Set up dropzone.js to serve all the stuff for "file dropping" on the web interface.
dropzone = Dropzone(app)
upload_handler_regular = portal.regular.UploadHandlerRegular(
UPLOADS_DIR=UPLOADS_DIR,
MAIL_HOST=MAIL_HOST,
MAIL_LOGIN=MAIL_LOGIN,
MAIL_PASS=MAIL_PASS,
SMTP_PORT=SMTP_PORT,
IMAP_PORT=IMAP_PORT,
MAIL_SUBJECT=MAIL_SUBJECT,
MAIL_BODY=MAIL_BODY,
CONTRACT_CLIENT_CONTACT=CONTRACT_CLIENT_CONTACT,
MAX_FILES_PER_UPLOAD=MAX_FILES_PER_UPLOAD,
APP_SECRET_KEY=APP_SECRET_KEY
)
upload_handler_solar = portal.solar.UploadHandlerSolar(
UPLOADS_DIR=UPLOADS_DIR,
MAIL_HOST=MAIL_HOST,
@ -155,26 +126,9 @@ def redirect_url(default='/'):
@app.route('/')
def index():
return render_template('index.html')
@app.route('/<corpus_name>')
def index_corpus(corpus_name):
if corpus_name not in ENABLED_CORPUSES:
return 'Korpus "{}" ne obstaja.'.format(corpus_name), 404
description = ""
if corpus_name == 'prevodi':
description = DESC_PREVODI
elif corpus_name == 'gigafida':
description = DESC_GIGAFIDA
elif corpus_name == 'solar':
if current_user.is_authenticated:
return redirect('/solar/oddaja')
return redirect('/solar/login')
return render_template('basic.html',
corpus_name=corpus_name, description=description, max_files=MAX_FILES_PER_UPLOAD)
if current_user.is_authenticated:
return redirect('/oddaja')
return redirect('/login')
@login_manager.user_loader
@ -183,120 +137,116 @@ def load_user(user_id):
return user
@app.route('/solar/login')
@app.route('/login')
def solar_login_get():
return render_template('solar-login.html')
@app.route('/solar/register')
@app.route('/register')
def solar_register_get():
return render_template('solar-register.html')
@app.route('/solar/login', methods=['POST'])
@app.route('/login', methods=['POST'])
def solar_login_post():
email = request.form.get('email')
password = request.form.get('password')
remember = True if request.form.get('remember') else False
user = portal.base.get_user_obj_by_email(email)
user = portal.solar.get_user_obj_by_email(email)
if not user or not check_password_hash(user.pass_hash, password):
flash('Napačni podatki za prijavo. Poskusite ponovno.')
return redirect('/solar/login')
return redirect('/login')
if not user.active:
flash('Vaš uporabniški račun še ni bil aktiviran.')
return redirect('/solar/login')
return redirect('/login')
#portal.base.add_user_session(user.id)
#portal.solar.add_user_session(user.id)
login_user(user, remember=remember)
return redirect('/solar/oddaja')
return redirect('/oddaja')
@app.route('/solar/register', methods=['POST'])
@app.route('/register', methods=['POST'])
def solar_register_post():
name = request.form.get('name')
email = request.form.get('email')
password = request.form.get('password')
institution_name = request.form.get('institution')
institution_role = request.form.get('role')
institution = portal.base.get_institution_obj_by_name(institution_name)
institution = portal.solar.get_institution_obj_by_name(institution_name)
user = RegisteredUser.query.filter_by(email=email).first()
if user:
flash('Uporabniški račun s tem emailom je že registriran.')
return redirect('/solar/register')
return redirect('/register')
if not name:
flash('Prazno polje za ime.')
return redirect('/solar/register')
return redirect('/register')
if len(name) > 100:
flash('Predolgo ime.')
return redirect('/solar/register')
return redirect('/register')
if not email:
flash('Prazno polje za elektronsko pošto.')
return redirect('/solar/register')
return redirect('/register')
if len(email) > 100:
flash('Predolgi email naslov')
return redirect('/solar/register')
elif not re.search(portal.base.REGEX_EMAIL, email):
return redirect('/register')
elif not re.search(portal.solar.REGEX_EMAIL, email):
flash('Email napačnega formata.')
return redirect('/solar/register')
return redirect('/register')
if not password:
flash('Prazno polje za geslo.')
return redirect('/solar/register')
return redirect('/register')
if len(password) > 8:
flash('Geslo mora biti vsaj 8 znakov dolgo.')
return redirect('/solar/register')
return redirect('/register')
if len(password) > 100:
flash('Predolgo geslo.')
return redirect('/solar/register')
return redirect('/register')
if institution_role not in ['coordinator', 'mentor', 'other']:
flash('Neveljavna vloga v instituciji.')
return redirect('/solar/register')
return redirect('/register')
if not institution:
institution_id = portal.base.add_institution(institution_name, "")
portal.base.grant_institution_corpus_access(institution_id, "solar")
institution_id = portal.solar.add_institution(institution_name, "")
portal.solar.grant_institution_corpus_access(institution_id, "solar")
else:
institution_id = institution.id
user_id = portal.base.register_new_user(name, email, password, active=False)
portal.base.add_user_to_institution(user_id, institution_id, institution_role)
user_id = portal.solar.register_new_user(name, email, password, active=False)
portal.solar.add_user_to_institution(user_id, institution_id, institution_role)
portal.base.send_admins_new_user_notification_mail(user_id, upload_handler_solar.config)
portal.solar.send_admins_new_user_notification_mail(user_id, upload_handler_solar.config)
flash('Podatki so bili poslani v potrditev. Ko bo registracija potrjena, boste o tem obveščeni po e-mailu, ki ste ga posredovali zgoraj.')
return redirect('/solar/login')
return redirect('/login')
# TODO: Move solar stuff to seperate file using Flask blueprints.
# TODO: Better routing logic.
@app.route('/solar/logout')
@app.route('/logout')
@login_required
def logout():
logout_user()
return redirect('/solar/login')
return redirect('/login')
@app.route('/solar/<path:text>')
@app.route('/<path:text>')
@login_required
def solar(text):
is_admin = current_user.role == 'admin'
current_user_institution = portal.base.get_user_institution(current_user.id)
current_user_obj = portal.base.get_user_obj(current_user.get_id())
current_user_institution = portal.solar.get_user_institution(current_user.id)
current_user_obj = portal.solar.get_user_obj(current_user.get_id())
institution_contract = None
if current_user_institution:
current_user_institution_coordinator = portal.base.is_institution_coordinator(current_user.id, current_user_institution.id)
institution_contract = portal.base.get_institution_contract(current_user_institution.id)
current_user_institution_coordinator = portal.solar.is_institution_coordinator(current_user.id, current_user_institution.id)
institution_contract = portal.solar.get_institution_contract(current_user_institution.id)
else:
current_user_institution_coordinator = False
@ -313,8 +263,8 @@ def solar(text):
uploader_names = []
institution_names = []
for item in upload_items:
uploader_names.append(portal.base.get_user_obj(item.upload_user).name)
institution = portal.base.get_institution_obj(item.institution)
uploader_names.append(portal.solar.get_user_obj(item.upload_user).name)
institution = portal.solar.get_institution_obj(item.institution)
if not institution:
institution_names.append(None)
else:
@ -334,9 +284,9 @@ def solar(text):
f_hash = filename.split('.')[0]
if contract_type == 'institucije':
actual_filename = portal.base.get_actual_institution_contract_filename(f_hash)
actual_filename = portal.solar.get_actual_institution_contract_filename(f_hash)
else:
actual_filename = portal.base.get_actual_studentparent_contract_filename(f_hash)
actual_filename = portal.solar.get_actual_studentparent_contract_filename(f_hash)
safe_path = safe_join(str(upload_handler_solar.get_uploads_subdir('contracts')), prefix, suffix)
try:
@ -351,11 +301,11 @@ def solar(text):
collaborators = []
cooperation_history = dict()
if current_user_institution:
collaborators = portal.base.get_all_active_institution_users(current_user_institution.id)
collaborators = portal.solar.get_all_active_institution_users(current_user_institution.id)
show_upload_form = True
contract_school = portal.solar.get_institution_contract(current_user_institution.id)
cooperation_history = portal.base.get_institution_cooperation_history(current_user_institution.id)
if portal.base.is_institution_coordinator(current_user_obj.id, current_user_institution.id):
cooperation_history = portal.solar.get_institution_cooperation_history(current_user_institution.id)
if portal.solar.is_institution_coordinator(current_user_obj.id, current_user_institution.id):
contracts_students = portal.solar.get_institution_student_contracts(current_user_institution.id)
enable_upload_school_contract = True
else:
@ -370,23 +320,23 @@ def solar(text):
user_id=current_user.id,
is_admin=is_admin, is_institution_coordinator=current_user_institution_coordinator)
elif text.startswith('admin/') or text == 'admin':
users = portal.base.get_all_users_join_institutions()
inactive_users = portal.base.get_all_users_join_institutions(active=False)
users = portal.solar.get_all_users_join_institutions()
inactive_users = portal.solar.get_all_users_join_institutions(active=False)
solar_institutions = portal.solar.get_all_institutions()
cooperation_history = portal.base.get_cooperation_history()
cooperation_history = portal.solar.get_cooperation_history()
uploads = portal.solar.get_all_upload_history(-1)
if is_admin:
return render_template('solar-admin.html', users=users, user_cooperation_history=cooperation_history,
institutions=solar_institutions, inactive_users=inactive_users, uploads=uploads)
elif text.startswith('manage-institution/') or text == 'manage-institution':
if portal.base.is_institution_coordinator(current_user.id, current_user_institution.id):
solar_users = portal.base.get_all_active_users()
institution_users = portal.base.get_all_active_institution_users(current_user_institution.id)
if portal.solar.is_institution_coordinator(current_user.id, current_user_institution.id):
solar_users = portal.solar.get_all_active_users()
institution_users = portal.solar.get_all_active_institution_users(current_user_institution.id)
return render_template('solar-manage-institution.html', users=solar_users,
institution_users=institution_users)
return '', 404
@app.route('/solar/pogodbe', methods=['POST'])
@app.route('/pogodbe', methods=['POST'])
@login_required
def solar_upload_contract():
msg = upload_handler_solar.handle_contract_upload(request, current_user.get_id())
@ -394,11 +344,11 @@ def solar_upload_contract():
return redirect(redirect_url())
@app.route('/solar/adduser', methods=['POST'])
@app.route('/adduser', methods=['POST'])
@login_required
def solar_add_user():
if not portal.base.is_admin(current_user.id):
if not portal.solar.is_admin(current_user.id):
return '', 404
name = request.form.get('name')
@ -418,7 +368,7 @@ def solar_add_user():
if len(email) > 100:
flash('Predolg email naslov.')
return redirect(redirect_url())
elif not re.search(portal.base.REGEX_EMAIL, email):
elif not re.search(portal.solar.REGEX_EMAIL, email):
flash('Email napačnega formata.')
return redirect(redirect_url())
@ -429,22 +379,22 @@ def solar_add_user():
flash('Predolgo geslo.')
return redirect(redirect_url())
user = portal.base.get_user_obj_by_email(email)
user = portal.solar.get_user_obj_by_email(email)
if user:
#portal.base.undo_remove_user(user.id)
#portal.solar.undo_remove_user(user.id)
flash('Uporabnik s tem emailom je že vnešen v sistem.')
return redirect(redirect_url())
portal.base.register_new_user(name, email, password)
portal.solar.register_new_user(name, email, password)
flash('Uporabnik je bil uspešno dodan.')
return redirect(redirect_url())
@app.route('/solar/activateuser', methods=['POST'])
@app.route('/activateuser', methods=['POST'])
@login_required
def solar_activate_user():
if not portal.base.is_admin(current_user.id):
if not portal.solar.is_admin(current_user.id):
return '', 404
user_id = request.form.get('id')
@ -452,34 +402,34 @@ def solar_activate_user():
flash('Prazno polje za ID uporabnika.')
return redirect(redirect_url())
rowcount = portal.base.activate_user(user_id)
rowcount = portal.solar.activate_user(user_id)
if rowcount == 0:
return '', 404
portal.base.send_user_activation_mail(user_id, upload_handler_solar.config)
portal.solar.send_user_activation_mail(user_id, upload_handler_solar.config)
flash('Uporabnik je bil aktiviran.')
return redirect(redirect_url())
@app.route('/solar/forgotpass')
@app.route('/forgotpass')
def solar_forgotpass():
return render_template('solar-forgotpass.html')
@app.route('/solar/sendresetpass', methods=['POST'])
@app.route('/sendresetpass', methods=['POST'])
def solar_sendresetpass():
email = request.form.get('email')
portal.base.send_resetpass_mail(email, upload_handler_solar.config)
portal.solar.send_resetpass_mail(email, upload_handler_solar.config)
flash('Povezava za ponastavitev gesla je bila poslana na vpisan email naslov.')
return redirect(redirect_url())
@app.route('/solar/resetpass/<token>')
@app.route('/resetpass/<token>')
def solar_resetpass(token):
user = portal.base.verify_reset_token(token, upload_handler_solar.config['APP_SECRET_KEY'])
user = portal.solar.verify_reset_token(token, upload_handler_solar.config['APP_SECRET_KEY'])
if not user:
return '', 404
@ -487,42 +437,42 @@ def solar_resetpass(token):
return render_template('solar-resetpass.html', user=user, token=token)
@app.route('/solar/resetpass/<token>', methods=['POST'])
@app.route('/resetpass/<token>', methods=['POST'])
def solar_resetpass_post(token):
new_password = request.form.get('new_password')
user = portal.base.verify_reset_token(token, upload_handler_solar.config['APP_SECRET_KEY'])
user = portal.solar.verify_reset_token(token, upload_handler_solar.config['APP_SECRET_KEY'])
if not user:
return '', 404
rowcount = portal.base.update_user_password(user.id, new_password)
rowcount = portal.solar.update_user_password(user.id, new_password)
if rowcount == 0:
return '', 404
flash('Ponastavitev gesla je bila uspešna.')
return redirect('/solar/login')
return redirect('/login')
@app.route('/solar/topuploads')
@app.route('/topuploads')
@login_required
def solar_topuploads_srednje():
return jsonify(portal.solar.get_top_uploading_institutions())
@app.route('/solar/deluser', methods=['POST'])
@app.route('/deluser', methods=['POST'])
@login_required
def solar_del_user():
if not portal.base.is_admin(current_user.id):
if not portal.solar.is_admin(current_user.id):
return '', 404
user_id = request.form.get('user_id')
portal.base.remove_user(user_id)
portal.solar.remove_user(user_id)
flash('Uporabnik je bil odstranjen.')
return redirect(redirect_url())
@app.route('/solar/addinstitution', methods=['POST'])
@app.route('/addinstitution', methods=['POST'])
@login_required
def add_institution():
if not portal.base.is_admin(current_user.id):
if not portal.solar.is_admin(current_user.id):
return '', 404
name = request.form.get('name')
@ -539,15 +489,15 @@ def add_institution():
flash('Neveljavna vrednost za regijo.')
return redirect(redirect_url())
institution_id = portal.base.add_institution(name, region)
portal.base.grant_institution_corpus_access(institution_id, "solar") # TODO: throw out
institution_id = portal.solar.add_institution(name, region)
portal.solar.grant_institution_corpus_access(institution_id, "solar") # TODO: throw out
flash('Institucija je bila dodana.')
return redirect(redirect_url())
@app.route('/solar/mergeinstitutions', methods=['POST'])
@app.route('/mergeinstitutions', methods=['POST'])
@login_required
def merge_institutions():
if not portal.base.is_admin(current_user.id):
if not portal.solar.is_admin(current_user.id):
return '', 404
id_from = request.form.get('id-from')
@ -557,8 +507,8 @@ def merge_institutions():
flash('Prazno polje.')
return redirect(redirect_url())
institution_from = portal.base.get_institution_obj(id_from)
institution_to = portal.base.get_institution_obj(id_to)
institution_from = portal.solar.get_institution_obj(id_from)
institution_to = portal.solar.get_institution_obj(id_to)
if not institution_from:
flash('Institucija z ID "{}" ne obstaja.'.format(id_from))
@ -569,18 +519,18 @@ def merge_institutions():
return redirect(redirect_url())
portal.base.transfer_users_institution(institution_from.id, institution_to.id)
portal.base.transfer_uploads_institution(institution_from.id, institution_to.id)
portal.base.transfer_contracts_institution(institution_from.id, institution_to.id)
portal.base.remove_institution(institution_from.id)
portal.solar.transfer_users_institution(institution_from.id, institution_to.id)
portal.solar.transfer_uploads_institution(institution_from.id, institution_to.id)
portal.solar.transfer_contracts_institution(institution_from.id, institution_to.id)
portal.solar.remove_institution(institution_from.id)
flash('Instituciji uspešno združeni')
return redirect(redirect_url())
@app.route('/solar/addcooperationhistoryitem', methods=['POST'])
@app.route('/addcooperationhistoryitem', methods=['POST'])
@login_required
def add_cooperation_history_item():
if not portal.base.is_admin(current_user.id):
if not portal.solar.is_admin(current_user.id):
return '', 404
user_id = request.form.get('user')
@ -589,8 +539,8 @@ def add_cooperation_history_item():
school_year = request.form.get('school-year')
badge_text = request.form.get('badge-text')
user = portal.base.get_user_obj(user_id)
institution = portal.base.get_institution_obj(institution_id)
user = portal.solar.get_user_obj(user_id)
institution = portal.solar.get_institution_obj(institution_id)
if not user:
flash('Uporabnik s tem ID-jem ne obstaja.')
@ -608,15 +558,15 @@ def add_cooperation_history_item():
flash('Šolsko leto mora biti formata "2021/22".')
return redirect(redirect_url())
portal.base.add_cooperation_history_item(user_id, institution_id, role, school_year, badge_text)
portal.solar.add_cooperation_history_item(user_id, institution_id, role, school_year, badge_text)
flash('Vnos dodan.')
return redirect(redirect_url())
@app.route('/solar/updateuploaditem', methods=['POST'])
@app.route('/updateuploaditem', methods=['POST'])
@login_required
def update_upload_item():
if not portal.base.is_admin(current_user.id):
if not portal.solar.is_admin(current_user.id):
return '', 404
err_msg = portal.solar.UploadHandlerSolar.check_form(request.form)
@ -652,22 +602,22 @@ def update_upload_item():
return redirect(redirect_url())
@app.route('/solar/delcooperationhistoryitem', methods=['POST'])
@app.route('/delcooperationhistoryitem', methods=['POST'])
@login_required
def del_cooperation_history_item():
if not portal.base.is_admin(current_user.id):
if not portal.solar.is_admin(current_user.id):
return '', 404
entry_id = request.form.get('entry-id')
portal.base.del_cooperation_history_item(entry_id)
portal.solar.del_cooperation_history_item(entry_id)
flash('Vnos odstranjen.')
return redirect(redirect_url())
@app.route('/solar/changeinstitutiondata', methods=['POST'])
@app.route('/changeinstitutiondata', methods=['POST'])
@login_required
def change_institution_data():
if not portal.base.is_admin(current_user.id):
if not portal.solar.is_admin(current_user.id):
return '', 404
institution_id = request.form.get('id')
@ -685,34 +635,34 @@ def change_institution_data():
flash('Neveljavna vrednost za regijo.')
return redirect(redirect_url())
portal.base.update_institution_data(institution_id, new_name, new_region)
portal.solar.update_institution_data(institution_id, new_name, new_region)
flash('Podatki institucije so bili spremenjeni.')
return redirect(redirect_url())
@app.route('/solar/changeuseremail', methods=['POST'])
@app.route('/changeuseremail', methods=['POST'])
@login_required
def change_user_email():
if not portal.base.is_admin(current_user.id):
if not portal.solar.is_admin(current_user.id):
return '', 404
user_id = request.form.get('user-id')
email = request.form.get('email')
if not re.search(portal.base.REGEX_EMAIL, email):
if not re.search(portal.solar.REGEX_EMAIL, email):
flash('Email napačnega formata.')
return redirect(redirect_url())
portal.base.update_user_email(user_id, email)
portal.solar.update_user_email(user_id, email)
flash('Email spremenjen.')
return redirect(redirect_url())
@app.route('/solar/changeuserrole', methods=['POST'])
@app.route('/changeuserrole', methods=['POST'])
@login_required
def change_user_role():
if not portal.base.is_admin(current_user.id):
if not portal.solar.is_admin(current_user.id):
return '', 404
user_id = request.form.get('user-id')
@ -722,38 +672,35 @@ def change_user_role():
flash('Neveljavna vloga.')
return redirect(redirect_url())
portal.base.update_user_role(user_id, role)
portal.solar.update_user_role(user_id, role)
flash('Vloga spremenjena.')
return redirect(redirect_url())
@app.route('/solar/changeusername', methods=['POST'])
@app.route('/changeusername', methods=['POST'])
@login_required
def change_user_name():
if not portal.base.is_admin(current_user.id):
if not portal.solar.is_admin(current_user.id):
return '', 404
user_id = request.form.get('user-id')
name = request.form.get('name')
portal.base.update_user_name(user_id, name)
portal.solar.update_user_name(user_id, name)
flash('Ime in priimek spremenjena.')
return redirect(redirect_url())
@app.route('/<corpus_name>/addusertoinstitution', methods=['POST'])
@app.route('/addusertoinstitution', methods=['POST'])
@login_required
def add_user_institution_mapping(corpus_name):
if not corpus_name in ENABLED_CORPUSES:
return '', 404
def add_user_institution_mapping():
institution_id = request.form.get('institution_id')
if not institution_id:
institution = portal.base.get_user_institution(current_user.id)
institution = portal.solar.get_user_institution(current_user.id)
if institution:
institution_id = institution.id
if not (portal.base.is_admin(current_user.id) or portal.base.is_institution_coordinator(current_user.id, institution_id)):
if not (portal.solar.is_admin(current_user.id) or portal.solar.is_institution_coordinator(current_user.id, institution_id)):
return '', 404
user_id = request.form['user_id']
@ -761,54 +708,37 @@ def add_user_institution_mapping(corpus_name):
if role not in ['coordinator', 'mentor', 'other']:
return '', 404
if portal.base.get_user_institution(user_id):
if portal.solar.get_user_institution(user_id):
flash('Uporabnik je že dodeljen instituciji.')
return redirect(redirect_url())
portal.base.add_user_to_institution(user_id, institution_id, role)
portal.solar.add_user_to_institution(user_id, institution_id, role)
flash('Uporabnik je bil dodeljen instituciji.')
return redirect(redirect_url())
@app.route('/solar/deluserfrominstitution', methods=['POST'])
@app.route('/deluserfrominstitution', methods=['POST'])
@login_required
def del_user_institution_mapping():
user_id = request.form['user_id']
institution = portal.base.get_user_institution(user_id)
institution = portal.solar.get_user_institution(user_id)
if not institution:
flash('Uporabnik ni član nobene institucije.')
return redirect(redirect_url())
if not portal.base.is_admin(current_user.id) \
and not portal.base.is_institution_coordinator(current_user.id, institution.id):
if not portal.solar.is_admin(current_user.id) \
and not portal.solar.is_institution_coordinator(current_user.id, institution.id):
flash('Nimate ustreznih pravic za odstranitev uporabnika iz institucije.')
return redirect(redirect_url())
portal.base.del_user_from_institution(user_id, institution.id)
portal.solar.del_user_from_institution(user_id, institution.id)
flash('Uporabnik je bil odstranjen iz institucije.')
return redirect(redirect_url())
@app.route('/<corpus_name>/delinstitution', methods=['POST'])
@login_required
def del_institution(corpus_name):
# TODO: check if valid corpus_name
# TODO: check if user is admin
# TODO: delete cascade - institution, user_institution_mapping, corpus_access, institution_contract
return '', 404
@app.route('/<corpus_name>/upload', methods=['POST'])
def handle_upload(corpus_name):
if corpus_name not in ENABLED_CORPUSES:
@app.route('/upload', methods=['POST'])
def handle_upload():
if not current_user.is_authenticated:
return '', 404
if corpus_name == 'solar':
if not current_user.is_authenticated:
return '', 404
#if not portal.base.has_user_corpus_access(current_user.id, corpus_name):
# return '', 404
return upload_handler_solar.handle_upload(request, current_user.get_id())
else:
return upload_handler_regular.handle_upload(request, corpus_name)
return upload_handler_solar.handle_upload(request, current_user.get_id())
if __name__ == '__main__':

View File

@ -0,0 +1,41 @@
"""Removed non-solar stuff.
Revision ID: 44dae32b13af
Revises: 84168f439c55
Create Date: 2021-10-05 13:22:23.539298
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision = '44dae32b13af'
down_revision = '84168f439c55'
branch_labels = None
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('upload_regular')
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('upload_regular',
sa.Column('id', sa.INTEGER(), autoincrement=True, nullable=False),
sa.Column('upload_hash', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('timestamp', postgresql.TIMESTAMP(), autoincrement=False, nullable=False),
sa.Column('name', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('org', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('address', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('zipcode', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('email', sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column('file_contract', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('upload_file_hashes', postgresql.ARRAY(sa.VARCHAR()), autoincrement=False, nullable=True),
sa.Column('corpus_name', sa.TEXT(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='upload_regular_pkey')
)
# ### end Alembic commands ###

View File

@ -1,604 +0,0 @@
import os
import hashlib
import time
import ssl
import traceback
import re
import logging
from pathlib import Path
from datetime import datetime
import imaplib
from smtplib import SMTP_SSL
import email
from email import encoders
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication
import pdfkit
from jinja2 import Environment, FileSystemLoader
import jwt
from werkzeug.security import generate_password_hash
from . model import *
#REGEX_EMAIL = re.compile('^[a-z0-9]+[\._]?[a-z0-9]+[@]\w+[.]\w{2,3}$')
REGEX_EMAIL = re.compile('^(?:[a-z0-9!#$%&\'*+/=?^_`{|}~-]+(?:\.[a-z0-9!#$%&\'*+/=?^_`{|}~-]+)*|\"(?:[\x01-\x08\x0b\x0c\x0e-\x1f\x21\x23-\x5b\x5d-\x7f]|\\[\x01-\x09\x0b\x0c\x0e-\x7f])*\")@(?:(?:[a-z0-9](?:[a-z0-9-]*[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]*[a-z0-9])?|\[(?:(?:(2(5[0-5]|[0-4][0-9])|1[0-9][0-9]|[1-9]?[0-9]))\.){3}(?:(2(5[0-5]|[0-4][0-9])|1[0-9][0-9]|[1-9]?[0-9])|[a-z0-9-]*[a-z0-9]:(?:[\x01-\x08\x0b\x0c\x0e-\x1f\x21-\x5a\x53-\x7f]|\\[\x01-\x09\x0b\x0c\x0e-\x7f])+)\])$')
MAX_FNAME_LEN = 100
class ContractCreator:
def __init__(self, base_path, template_path):
self.base = base_path
template_loader = FileSystemLoader(searchpath="./")
template_env = Environment(loader=template_loader)
self.template = template_env.get_template(template_path)
self.pdfkit_options = {
'page-size': 'A4',
'margin-top': '0.75in',
'margin-right': '0.75in',
'margin-bottom': '0.75in',
'margin-left': '0.75in',
'encoding': "UTF-8",
'custom-header' : [
('Accept-Encoding', 'gzip')
]
}
def fill_template(self, **kwargs):
return self.template.render(**kwargs)
def create_pdf(self, f_name, fields_dict):
sub_dir = self.base / Path(f_name[:2])
if not sub_dir.exists():
sub_dir.mkdir()
out_f = sub_dir / Path(f_name[2:])
html_str = self.fill_template(**fields_dict)
pdfkit.from_string(html_str, out_f, options=self.pdfkit_options)
class UploadHandler:
ENABLED_FILETYPES = ['txt', 'csv', 'pdf', 'doc', 'docx', 'xls', 'xlsx', 'ppt', 'pptx', 'xml', 'mxliff', 'tmx', 'jpg', 'jpeg', 'png']
def __init__(self, **kwargs):
self.config = kwargs
def set_contract_creator(self, contract_creator):
assert isinstance(contract_creator, ContractCreator)
self._contract_creator = contract_creator
def get_uploads_subdir(self, dir_name):
subdir = Path(self.config['UPLOADS_DIR']) / dir_name
if not subdir.exists():
subdir.mkdir(parents=True)
return subdir
@staticmethod
def extract_upload_metadata(corpus_name, request):
upload_metadata = dict()
file_hashes = UploadHandler.create_file_hashes(request.files)
file_names = file_hashes.keys()
form_data = request.form.copy()
upload_timestamp = int(time.time())
upload_id = UploadHandler.create_upload_id(corpus_name, form_data, upload_timestamp, file_hashes)
# Strip form fieds.
for key, val in form_data.items():
form_data[key] = val.strip()
upload_metadata['corpus_name'] = corpus_name
upload_metadata['form_data'] = form_data
upload_metadata['upload_id'] = upload_id
upload_metadata['timestamp'] = upload_timestamp
upload_metadata['file_hashes_dict'] = file_hashes
upload_metadata['file_names'] = file_names
upload_metadata['contract_file'] = upload_id + '.pdf'
return upload_metadata
@staticmethod
def create_upload_id(corpus_name, form_data, upload_timestamp, file_hashes):
# Order is important while hashing, hence the sorting.
val_buff = [str(upload_timestamp)]
for key in sorted(form_data):
val_buff.append(form_data[key])
# This hash serves as an unique identifier for the whole upload.
metahash = hashlib.md5((''.join(val_buff)).encode())
# Include file hashes to avoid metafile name collisions if they have the same form values,
# but different data files. Sort hashes first so upload order doesn't matter.
sorted_f_hashes = list(file_hashes.values())
sorted_f_hashes.sort()
metahash.update(''.join(sorted_f_hashes).encode())
metahash = metahash.hexdigest()
return metahash
@staticmethod
def create_file_hashes(files):
res = dict()
for key, f in files.items():
if key.startswith('file'):
h = hashlib.md5(f.filename.encode())
h.update(f.stream.read())
res[f.filename] = h.hexdigest()
f.seek(0)
return res
@staticmethod
def store_model(model_obj):
try:
db.session.add(model_obj)
db.session.commit()
except Exception:
traceback.print_exc()
def store_datafiles(self, files, upload_metadata):
base = self.get_uploads_subdir('files')
file_hashes = upload_metadata['file_hashes_dict']
for key, f in files.items():
if key.startswith('file'):
f_hash = file_hashes[f.filename]
# First byte used for indexing, similarly like git does for example.
sub_dir = base / f_hash[:2]
if not sub_dir.exists():
sub_dir.mkdir()
path = sub_dir / f_hash[2:]
if not path.exists():
path.mkdir()
f.save(path / f.filename)
def send_confirm_mail(self, upload_metadata, attach_contract_file=False):
upload_id = upload_metadata['upload_id']
message = MIMEMultipart()
message['From'] = self.config['MAIL_LOGIN']
message['To'] = upload_metadata['form_data']['email']
message['Subject'] = self.config['MAIL_SUBJECT'].format(upload_id=upload_id)
body = self.config['MAIL_BODY'].format(upload_id=upload_id)
message.attach(MIMEText(body, "plain"))
if attach_contract_file:
contracts_dir = self.contract_creator.base
f_name = upload_metadata['contract_file']
sub_dir = contracts_dir / Path(f_name[:2])
contract_file = sub_dir / Path(f_name[2:])
with open(contract_file, "rb") as f:
part = MIMEApplication(
f.read(),
Name = f_name
)
part['Content-Disposition'] = 'attachment; filename="%s"' % f_name
message.attach(part)
text = message.as_string()
# Create a secure SSL context
context = ssl.create_default_context()
# TODO: Implement timeout.
try:
with SMTP_SSL(self.config['MAIL_HOST'], self.config['SMTP_PORT'], context=context) as server:
server.login(self.config['MAIL_LOGIN'], self.config['MAIL_PASS'])
server.sendmail(message['From'], message['To'], text)
# Save copy of sent mail in Sent mailbox
#imap = imaplib.IMAP4_SSL(self.config['MAIL_HOST'], self.config['IMAP_PORT'])
#imap.login(self.config['MAIL_LOGIN'], self.config['MAIL_PASS'])
#imap.append('Sent', '\\Seen', imaplib.Time2Internaldate(time.time()), text.encode('utf8'))
#imap.logout()
except Exception:
traceback.print_exc()
def check_suffixes(self, files):
for key, f in files.items():
if key.startswith('file'):
suffix = f.filename.split('.')[-1]
if self.ENABLED_FILETYPES and suffix.lower() not in self.ENABLED_FILETYPES:
return 'Datoteka "{}" ni pravilnega formata.'.format(f.filename)
return None
@staticmethod
def check_fname_lengths(files):
for key, f in files.items():
if key.startswith('file'):
if len(f.filename) > MAX_FNAME_LEN:
return 'Ime datoteke presega dolžino {} znakov.'.format(MAX_FNAME_LEN)
return None
def check_upload_request(self, request):
files = request.files
max_files = self.config['MAX_FILES_PER_UPLOAD']
if len(files) > max_files:
return 'Naložite lahko do {} datotek hkrati.'.format(max_files), 400
elif len(files) < 1:
return 'Priložena ni bila nobena datoteka.', 400
err = self.check_suffixes(files)
if err:
return err, 400
err = UploadHandler.check_fname_lengths(files)
if err:
return err, 400
return None
def get_user_institution(user_id):
mapping = UserInstitutionMapping.query.filter_by(user=user_id).first()
if mapping:
return Institution.query.filter_by(id=mapping.institution).first()
return None
def get_institution_contract(institution_id):
return InstitutionContract.query.filter_by(institution=institution_id).first()
def get_institution_cooperation_history(institution_id):
#return CooperationToken.query.join(UserCooperationTokenMapping,
# UserCooperationTokenMapping.cooperation_token == CooperationToken.id).filter(UserCooperationTokenMapping.user == user_id).all()
#
res = dict()
uch_rows = UserCooperationHistory.query.filter_by(institution=institution_id).order_by(UserCooperationHistory.school_year.desc()).all()
for row in uch_rows:
if row.user not in res:
res[row.user] = {
'coordinator': [],
'mentor': [],
'other': []
}
res[row.user][row.role].append((row.school_year, row.badge_text))
return res
def get_cooperation_history():
return UserCooperationHistory.query.all()
def add_cooperation_history_item(user_id, institution_id, role, school_year, badge_text):
model_obj = UserCooperationHistory(
user=user_id,
institution=institution_id,
role=role,
school_year=school_year,
badge_text=badge_text
)
db.session.add(model_obj)
db.session.commit()
return model_obj.id
def del_cooperation_history_item(entry_id):
db.session.query(UserCooperationHistory).filter_by(id=entry_id).delete()
db.session.commit()
def has_user_corpus_access(user_id, corpus_name):
user = RegisteredUser.query.filter_by(id=user_id).first()
# TODO: check if user even is active?
# Admins always have access to everything.
if user.role == 'admin':
return True
# Check if user belongs to an institution, that has access to this corpus.
institution = get_user_institution(user_id)
has_access = False
row = CorpusAccess.query.filter_by(institution=institution.id, corpus=corpus_name).first()
if row:
has_access = True
return has_access
def is_admin(user_id):
user = RegisteredUser.query.filter_by(id=user_id).first()
if user.role == 'admin':
return True
return False
def get_user_obj(user_id):
return RegisteredUser.query.filter_by(id=user_id).first()
def get_user_obj_by_email(email):
return RegisteredUser.query.filter_by(email=email).first()
def get_institution_obj(institution_id):
return Institution.query.filter_by(id=institution_id).first()
def get_institution_obj_by_name(institution_name):
return Institution.query.filter_by(name=institution_name).first()
def register_new_user(name, email, password, active=True, admin=False):
model_obj = RegisteredUser(
name=name,
email=email,
role='admin' if admin else 'user',
pass_hash=generate_password_hash(password),
active=active,
registered=datetime.now()
)
db.session.add(model_obj)
db.session.commit()
return model_obj.id
def add_institution(name, region):
model_obj = Institution(
name=name,
region=region
)
db.session.add(model_obj)
db.session.commit()
return model_obj.id
def grant_institution_corpus_access(institution_id, corpus_name):
model_obj = CorpusAccess(
institution=institution_id,
corpus=corpus_name
)
db.session.add(model_obj)
db.session.commit()
return model_obj.id
def add_user_to_institution(user_id, institution_id, role):
model_obj = UserInstitutionMapping(
user=user_id,
institution=institution_id,
role=role
)
db.session.add(model_obj)
db.session.commit()
return model_obj.id
def activate_user(user_id):
rowcount = db.session.query(RegisteredUser).filter_by(id=user_id).update({'active': True})
db.session.commit()
return rowcount
def update_user_password(user_id, new_password):
phash = generate_password_hash(new_password)
rowcount = db.session.query(RegisteredUser).filter_by(id=user_id).update({'pass_hash': phash})
db.session.commit()
return rowcount
def update_user_role(user_id, role):
rowcount = db.session.query(RegisteredUser).filter_by(id=user_id).update({'role': role})
db.session.commit()
return rowcount
def update_user_email(user_id, new_email):
rowcount = db.session.query(RegisteredUser).filter_by(id=user_id).update({'email': new_email})
db.session.commit()
return rowcount
def update_user_name(user_id, new_name):
rowcount = db.session.query(RegisteredUser).filter_by(id=user_id).update({'name': new_name})
db.session.commit()
return rowcount
def update_institution_data(institution_id, new_name, new_region):
rowcount = db.session.query(Institution).filter_by(id=institution_id).update({'name': new_name, 'region': new_region})
db.session.commit()
return rowcount
def remove_user(user_id):
db.session.query(UserCooperationHistory).filter(UserCooperationHistory.user == user_id).delete()
db.session.query(UserInstitutionMapping).filter(UserInstitutionMapping.user == user_id).delete()
db.session.query(RegisteredUser).filter(RegisteredUser.id == user_id).delete()
db.session.commit()
#db.session.query(RegisteredUser).filter(RegisteredUser.id == user_id).update({'is_removed': True})
#db.session.commit()
#def undo_remove_user(user_id):
# db.session.query(RegisteredUser).filter(RegisteredUser.id == user_id).update({'is_removed': False})
# db.session.commit()
def remove_institution(institution_id):
db.session.query(CorpusAccess).filter(CorpusAccess.institution == institution_id).delete()
db.session.query(Institution).filter(Institution.id == institution_id).delete()
db.session.commit()
#db.session.query(Institution).filter(Institution.id == institution_id).update({'is_removed': True})
#db.session.commit()
def del_user_from_institution(user_id, institution_id):
db.session.query(UserInstitutionMapping).filter(UserInstitutionMapping.institution == institution_id).filter(UserInstitutionMapping.user == user_id).delete()
db.session.commit()
def get_all_active_users():
return RegisteredUser.query.filter_by(active=True).order_by(RegisteredUser.id).all()
def get_all_inactive_users():
return RegisteredUser.query.filter_by(active=False).order_by(RegisteredUser.id).all()
def get_all_users_join_institutions(active=True):
#return RegisteredUser.query.filter_by(active=True).order_by(RegisteredUser.id).all()
return db.session.query(RegisteredUser, UserInstitutionMapping).outerjoin(UserInstitutionMapping,
RegisteredUser.id == UserInstitutionMapping.user).filter(RegisteredUser.active == active).order_by(RegisteredUser.id).all()
def get_all_active_institution_users(institution_id):
return RegisteredUser.query.filter_by(active=True).join(UserInstitutionMapping,
RegisteredUser.id == UserInstitutionMapping.user).filter(UserInstitutionMapping.institution == institution_id).all()
def is_institution_coordinator(user_id, institution_id):
user_inst_mapping = UserInstitutionMapping.query.filter_by(user=user_id).filter_by(institution=institution_id).first()
if not user_inst_mapping:
return False
if user_inst_mapping.role != 'coordinator':
return False
return True
def is_institution_member(user_id, institution_id):
user_inst_mapping = UserInstitutionMapping.query.filter_by(user=user_id).filter_by(institution=institution_id).first()
if not user_inst_mapping:
return False
return True
def get_actual_institution_contract_filename(f_hash):
contract = InstitutionContract.query.filter_by(file_contract=f_hash).first()
if contract:
return contract.original_filename
return None
def get_actual_studentparent_contract_filename(f_hash):
contract = ContractsSolar.query.filter_by(file_contract=f_hash).first()
if contract:
return contract.original_filename
return None
def get_password_reset_token(email, key, expires=600):
return jwt.encode({'reset_password': email,
'exp': int(time.time()) + expires},
key=key, algorithm='HS256')
def transfer_users_institution(institution_id_from, institution_id_to):
rowcount = db.session.query(UserInstitutionMapping).filter_by(institution=institution_id_from).update(
{'institution': institution_id_to})
db.session.commit()
return rowcount
def transfer_uploads_institution(institution_id_from, institution_id_to):
rowcount = db.session.query(UploadSolar).filter_by(institution=institution_id_from).update(
{'institution': institution_id_to})
db.session.commit()
return rowcount
def transfer_contracts_institution(institution_id_from, institution_id_to):
rowcount = db.session.query(ContractsSolar).filter_by(institution=institution_id_from).update(
{'institution': institution_id_to})
db.session.commit()
return rowcount
def verify_reset_token(token, key):
try:
message = jwt.decode(token,
key=key, algorithms=["HS256"])
email = message['reset_password']
exp = message['exp']
if int(time.time()) >= exp:
# Token timed out
return None
except Exception as e:
logging.error(e)
return
return RegisteredUser.query.filter_by(email=email).first()
def send_resetpass_mail(email, config):
jwt_token = get_password_reset_token(email, config['APP_SECRET_KEY'])
body = '''
Zahtevali ste ponastavitev gesla vašega uporabniškega računa.
Geslo lahko ponastavite na naslednji povezavi: https://zbiranje.slovenscina.eu/solar/resetpass/{}'''.format(jwt_token)
message = MIMEMultipart()
message['From'] = config['MAIL_LOGIN']
message['To'] = email
message['Subject'] = 'Ponastavitev gesla'
message.attach(MIMEText(body, "plain"))
text = message.as_string()
# Create a secure SSL context
context = ssl.create_default_context()
try:
with SMTP_SSL(config['MAIL_HOST'], config['SMTP_PORT'], context=context) as server:
server.login(config['MAIL_LOGIN'], config['MAIL_PASS'])
server.sendmail(config['MAIL_LOGIN'], email, text)
except Exception:
traceback.print_exc()
def send_admins_new_user_notification_mail(user_id, config):
user = RegisteredUser.query.filter_by(id=user_id).first()
body = '''
Nov uporabnik "{}" je ustvaril uporabniški račun na portalu za oddajanje besedil Šolar in čaka na odobritev.
'''.format(user.name)
admins = RegisteredUser.query.filter_by(role="admin").all()
# Create a secure SSL context
context = ssl.create_default_context()
for admin in admins:
message = MIMEMultipart()
message['From'] = config['MAIL_LOGIN']