140 lines
4.6 KiB
Python
140 lines
4.6 KiB
Python
from flask import Flask, render_template, request, redirect, flash, session, url_for
|
|
from models import db, Users, Annotations, Samples
|
|
from dotenv import load_dotenv
|
|
from sqlalchemy.orm import aliased
|
|
import sqlalchemy
|
|
import os
|
|
|
|
|
|
import logging
|
|
|
|
load_dotenv()
|
|
|
|
app = Flask(__name__)
|
|
|
|
app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DB_URI')
|
|
app.secret_key = os.getenv('SECRET_KEY')
|
|
|
|
db.init_app(app)
|
|
|
|
|
|
|
|
@app.route('/', methods=['GET'])
|
|
def home():
|
|
session.pop('id_user', None)
|
|
return render_template('home.html')
|
|
|
|
@app.route('/login', methods=['POST'])
|
|
def login():
|
|
if request.method == 'POST':
|
|
email = request.form['email']
|
|
if '@' in email:
|
|
try:
|
|
splitted = email.split('@')
|
|
name, surname = splitted[0].split('.')[:2]
|
|
domain = splitted[1]
|
|
if 'tuke' not in domain:
|
|
raise ValueError
|
|
except ValueError:
|
|
flash('Nie je validný TUKE email')
|
|
return redirect('/')
|
|
try:
|
|
db.session.add(Users(name, surname, email))
|
|
db.session.commit()
|
|
except sqlalchemy.exc.IntegrityError as err:
|
|
db.session.rollback()
|
|
logging.info('Logged existing email')
|
|
user = Users.query.filter_by(email=email).first()
|
|
session['email'] = email
|
|
session['id_user'] = user._id
|
|
return redirect('/anot')
|
|
flash('Nie je validný TUKE email')
|
|
return redirect('/')
|
|
|
|
@app.route('/anot', methods=['GET'])
|
|
def anot():
|
|
if 'id_user' in session:
|
|
try:
|
|
annotated_count = Annotations.query.filter_by(user_id=session['id_user']).count()
|
|
|
|
# query = text(
|
|
# f'''
|
|
# SELECT samples._id, text
|
|
# FROM samples
|
|
# LEFT JOIN annotations
|
|
# ON samples._id = annotations.sample_id
|
|
# WHERE samples._id NOT IN (
|
|
# SELECT sample_id
|
|
# FROM annotations
|
|
# GROUP BY sample_id
|
|
# HAVING COUNT(sample_id) > 5
|
|
# )
|
|
# AND samples._id NOT IN (
|
|
# SELECT samples._id
|
|
# FROM samples
|
|
# LEFT JOIN annotations
|
|
# ON samples._id = annotations.sample_id
|
|
# WHERE annotations.user_id IS {session['id_user']}
|
|
# )
|
|
# ORDER BY samples._id ASC
|
|
# LIMIT 1;
|
|
# '''
|
|
# )
|
|
|
|
annotations_alias = aliased(Annotations)
|
|
|
|
# Construct the query
|
|
query = (
|
|
db.session.query(Samples._id, Samples.text)
|
|
.outerjoin(annotations_alias, Samples._id == annotations_alias.sample_id)
|
|
.filter(
|
|
~Samples._id.in_(
|
|
db.session.query(Annotations.sample_id)
|
|
.group_by(Annotations.sample_id)
|
|
.having(db.func.count(Annotations.sample_id) > 5)
|
|
),
|
|
~Samples._id.in_(
|
|
db.session.query(Samples._id)
|
|
.outerjoin(Annotations, Samples._id == Annotations.sample_id)
|
|
.filter(Annotations.user_id == session['id_user'])
|
|
)
|
|
)
|
|
.order_by(Samples._id.asc())
|
|
.limit(1)
|
|
)
|
|
sample_id, sample_text = query.one_or_none()
|
|
|
|
data = {
|
|
'email': session.get('email'),
|
|
'text': sample_text,
|
|
'sample_id': sample_id,
|
|
'annotated_count': annotated_count
|
|
}
|
|
|
|
except (sqlalchemy.exc.OperationalError) as err:
|
|
print(err)
|
|
logging.info('Annotationss started')
|
|
data = {
|
|
'email': session.get('email'),
|
|
'text': Samples.query.order_by(Samples._id.asc()).first().text,
|
|
'sample_id': Samples.query.order_by(Samples._id.asc()).first()._id,
|
|
'annotated_count': annotated_count
|
|
}
|
|
return render_template('anot.html', **data)
|
|
return redirect('/')
|
|
|
|
@app.route('/process_anot', methods=['POST'])
|
|
def process():
|
|
if request.method == 'POST':
|
|
data = request.get_json()
|
|
print(data)
|
|
db.session.add(Annotations(
|
|
user_id=session['id_user'],
|
|
sample_id=data['sample_id'],
|
|
label=data['value']
|
|
))
|
|
db.session.commit()
|
|
return redirect(url_for('anot'))
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=5050) |