"""Flask application for the ESConv Annotation Management System.""" import os import secrets from datetime import datetime, timezone from functools import wraps import bcrypt from flask import ( Flask, jsonify, redirect, render_template, request, session, url_for ) from models import db, User, CsvFile, Assignment, Review, AnnotationHistory from csv_utils import parse_csv_file, count_samples, scan_csv_folder # ── Config ────────────────────────────────────────────────────────────────── BASE_DIR = os.path.abspath(os.path.dirname(__file__)) CSV_FOLDER = os.path.abspath( os.environ.get('CSV_DATA_DIR', os.path.join(BASE_DIR, 'csv_data')) ) DATABASE_PATH = os.path.abspath( os.environ.get('DATABASE_PATH', os.path.join(BASE_DIR, 'annotation.db')) ) app = Flask(__name__) app.secret_key = os.environ.get('SECRET_KEY') or secrets.token_hex(32) app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{DATABASE_PATH.replace(os.sep, "/")}' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.config['CSV_DATA_DIR'] = CSV_FOLDER app.config['SESSION_COOKIE_HTTPONLY'] = True app.config['SESSION_COOKIE_SAMESITE'] = os.environ.get( 'SESSION_COOKIE_SAMESITE', 'Lax' ) app.config['SESSION_COOKIE_SECURE'] = ( os.environ.get('SESSION_COOKIE_SECURE', 'false').lower() == 'true' ) db.init_app(app) # ── Helpers ───────────────────────────────────────────────────────────────── def hash_password(pw): return bcrypt.hashpw(pw.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') def check_password(pw, hashed): return bcrypt.checkpw(pw.encode('utf-8'), hashed.encode('utf-8')) def login_required(f): @wraps(f) def wrapper(*args, **kwargs): if 'user_id' not in session: if request.path.startswith('/api/'): return jsonify({'error': 'Not authenticated'}), 401 return redirect(url_for('login_page')) return f(*args, **kwargs) return wrapper def admin_required(f): @wraps(f) def wrapper(*args, **kwargs): if 'user_id' not in session: if request.path.startswith('/api/'): return jsonify({'error': 'Not authenticated'}), 401 return redirect(url_for('login_page')) if session.get('role') != 'admin': if request.path.startswith('/api/'): return jsonify({'error': 'Admin access required'}), 403 return redirect(url_for('login_page')) return f(*args, **kwargs) return wrapper def sync_csv_files(): """Scan the CSV folder and sync with the database.""" os.makedirs(CSV_FOLDER, exist_ok=True) disk_files = scan_csv_folder(CSV_FOLDER) disk_names = {f['filename'] for f in disk_files} # Add new files for finfo in disk_files: existing = CsvFile.query.filter_by(filename=finfo['filename']).first() total = count_samples(finfo['filepath']) if existing: # Keep persisted paths portable when the app moves between hosts. existing.filepath = finfo['filepath'] existing.total_samples = total else: new_file = CsvFile( filename=finfo['filename'], filepath=finfo['filepath'], total_samples=total, ) db.session.add(new_file) # Remove DB entries for files no longer on disk all_db_files = CsvFile.query.all() for dbf in all_db_files: if dbf.filename not in disk_names: db.session.delete(dbf) db.session.commit() # ── Auth Routes ───────────────────────────────────────────────────────────── @app.route('/login', methods=['GET']) def login_page(): if 'user_id' in session: return redirect(url_for('dashboard')) return render_template('login.html') @app.route('/api/login', methods=['POST']) def api_login(): data = request.get_json() username = data.get('username', '').strip() password = data.get('password', '') user = User.query.filter_by(username=username).first() if not user or not check_password(password, user.password_hash): return jsonify({'error': 'Invalid username or password'}), 401 session['user_id'] = user.id session['username'] = user.username session['role'] = user.role return jsonify({'ok': True, 'role': user.role, 'username': user.username}) @app.route('/api/logout', methods=['POST']) def api_logout(): session.clear() return jsonify({'ok': True}) @app.route('/api/me') @login_required def api_me(): return jsonify({ 'user_id': session['user_id'], 'username': session['username'], 'role': session['role'], }) # ── Page Routes ───────────────────────────────────────────────────────────── @app.route('/') @login_required def dashboard(): if session.get('role') == 'admin': return redirect(url_for('admin_page')) return redirect(url_for('annotator_page')) @app.route('/admin') @admin_required def admin_page(): return render_template('admin.html') @app.route('/annotator') @login_required def annotator_page(): return render_template('annotator.html') @app.route('/review/') @login_required def review_page(assignment_id): assignment = Assignment.query.get_or_404(assignment_id) # Only the assigned user or admin can access if session.get('role') != 'admin' and assignment.user_id != session['user_id']: return redirect(url_for('dashboard')) return render_template('review.html') # ── Admin API: Users ──────────────────────────────────────────────────────── @app.route('/api/users', methods=['GET']) @admin_required def api_get_users(): users = User.query.order_by(User.created_at).all() return jsonify([u.to_dict() for u in users]) @app.route('/api/users', methods=['POST']) @admin_required def api_create_user(): data = request.get_json() username = data.get('username', '').strip() password = data.get('password', '') role = data.get('role', 'annotator') if not username or not password: return jsonify({'error': 'Username and password required'}), 400 if len(password) < 4: return jsonify({'error': 'Password must be at least 4 characters'}), 400 if role not in ('admin', 'annotator'): return jsonify({'error': 'Invalid role'}), 400 if User.query.filter_by(username=username).first(): return jsonify({'error': 'Username already exists'}), 400 user = User(username=username, password_hash=hash_password(password), role=role) db.session.add(user) db.session.commit() return jsonify(user.to_dict()), 201 @app.route('/api/users/', methods=['DELETE']) @admin_required def api_delete_user(user_id): user = User.query.get_or_404(user_id) if user.id == session['user_id']: return jsonify({'error': 'Cannot delete yourself'}), 400 db.session.delete(user) db.session.commit() return jsonify({'ok': True}) # ── Admin API: Files ──────────────────────────────────────────────────────── @app.route('/api/files', methods=['GET']) @admin_required def api_get_files(): sync_csv_files() files = CsvFile.query.order_by(CsvFile.filename).all() result = [] for f in files: d = f.to_dict() # Count total reviews across all assignments for this file total_reviews = ( db.session.query(Review) .join(Assignment) .filter(Assignment.file_id == f.id) .count() ) total_assignees = Assignment.query.filter_by(file_id=f.id).count() d['total_reviews'] = total_reviews d['total_assignees'] = total_assignees result.append(d) return jsonify(result) # @app.route('/api/files//data', methods=['GET']) # @login_required # def api_get_file_data(file_id): # csv_file = CsvFile.query.get_or_404(file_id) # rows = parse_csv_file(csv_file.filepath) # return jsonify({'filename': csv_file.filename, 'samples': rows}) @app.route('/api/files//data', methods=['GET']) @login_required def api_get_file_data(file_id): csv_file = CsvFile.query.get_or_404(file_id) # Extract JUST the filename from whatever is stored in the DB filename = os.path.basename(csv_file.filepath) # Construct the correct path for the CURRENT operating system safe_filepath = os.path.join(CSV_FOLDER, filename) # Optional: Verify it exists before trying to parse if not os.path.exists(safe_filepath): return jsonify({'error': f'File not found on server at {safe_filepath}'}), 404 rows = parse_csv_file(safe_filepath) return jsonify({'filename': csv_file.filename, 'samples': rows}) # ── Admin API: Assignments ────────────────────────────────────────────────── @app.route('/api/assignments', methods=['GET']) @admin_required def api_get_assignments(): assignments = Assignment.query.order_by(Assignment.assigned_at.desc()).all() return jsonify([a.to_dict() for a in assignments]) @app.route('/api/assignments', methods=['POST']) @admin_required def api_create_assignment(): data = request.get_json() user_id = data.get('user_id') file_id = data.get('file_id') if not user_id or not file_id: return jsonify({'error': 'user_id and file_id required'}), 400 user = User.query.get(user_id) csv_file = CsvFile.query.get(file_id) if not user: return jsonify({'error': 'User not found'}), 404 if not csv_file: return jsonify({'error': 'File not found'}), 404 existing = Assignment.query.filter_by(user_id=user_id, file_id=file_id).first() if existing: return jsonify({'error': 'Assignment already exists'}), 400 assignment = Assignment(user_id=user_id, file_id=file_id) db.session.add(assignment) db.session.commit() return jsonify(assignment.to_dict()), 201 @app.route('/api/assignments/', methods=['DELETE']) @admin_required def api_delete_assignment(assignment_id): assignment = Assignment.query.get_or_404(assignment_id) db.session.delete(assignment) db.session.commit() return jsonify({'ok': True}) # ── Annotator API ─────────────────────────────────────────────────────────── @app.route('/api/my-assignments', methods=['GET']) @login_required def api_my_assignments(): assignments = Assignment.query.filter_by(user_id=session['user_id']).all() return jsonify([a.to_dict() for a in assignments]) @app.route('/api/assignments//reviews', methods=['GET']) @login_required def api_get_reviews(assignment_id): assignment = Assignment.query.get_or_404(assignment_id) if session.get('role') != 'admin' and assignment.user_id != session['user_id']: return jsonify({'error': 'Access denied'}), 403 reviews = Review.query.filter_by(assignment_id=assignment_id).all() return jsonify({r.sample_index: r.to_dict() for r in reviews}) @app.route('/api/reviews', methods=['POST']) @login_required def api_submit_review(): data = request.get_json() assignment_id = data.get('assignment_id') sample_index = data.get('sample_index') status = data.get('status') # 'accept' or 'reject' comment = data.get('comment', '') if not assignment_id or sample_index is None or status not in ('accept', 'reject'): return jsonify({'error': 'assignment_id, sample_index, and valid status required'}), 400 assignment = Assignment.query.get(assignment_id) if not assignment: return jsonify({'error': 'Assignment not found'}), 404 if session.get('role') != 'admin' and assignment.user_id != session['user_id']: return jsonify({'error': 'Access denied'}), 403 review = Review.query.filter_by( assignment_id=assignment_id, sample_index=sample_index ).first() if review: review.status = status review.comment = comment review.reviewed_at = datetime.now(timezone.utc) else: review = Review( assignment_id=assignment_id, sample_index=sample_index, status=status, comment=comment, ) db.session.add(review) db.session.commit() return jsonify(review.to_dict()) @app.route('/api/reviews/annotations', methods=['PATCH']) @login_required def api_save_annotations(): """Save annotator-edited annotations for a specific sample.""" import json as _json data = request.get_json() assignment_id = data.get('assignment_id') sample_index = data.get('sample_index') annotations = data.get('annotations') # dict: {emotion, distortions, empathy} if not assignment_id or sample_index is None or annotations is None: return jsonify({'error': 'assignment_id, sample_index, and annotations required'}), 400 assignment = Assignment.query.get(assignment_id) if not assignment: return jsonify({'error': 'Assignment not found'}), 404 if session.get('role') != 'admin' and assignment.user_id != session['user_id']: return jsonify({'error': 'Access denied'}), 403 # Upsert review — annotations can be saved without status yet (use 'pending' placeholder) review = Review.query.filter_by( assignment_id=assignment_id, sample_index=sample_index ).first() annotations_json = _json.dumps( annotations, ensure_ascii=False, sort_keys=True, separators=(',', ':') ) previous_annotations = None if review and review.annotations_edit: try: previous_annotations = _json.loads(review.annotations_edit) except (TypeError, _json.JSONDecodeError): pass if review: review.annotations_edit = annotations_json review.reviewed_at = datetime.now(timezone.utc) else: # Create a minimal review record to hold the annotation edits review = Review( assignment_id=assignment_id, sample_index=sample_index, status='reject', # default, annotator must explicitly submit verdict comment='', annotations_edit=annotations_json, ) db.session.add(review) # Store only meaningful changes; repeated saves of the same data do not # create duplicate history entries. if previous_annotations != annotations: db.session.add(AnnotationHistory( assignment_id=assignment_id, sample_index=sample_index, annotations_json=annotations_json, edited_by=session.get('username', 'unknown'), )) db.session.commit() return jsonify(review.to_dict()) @app.route( '/api/assignments//samples//annotation-history', methods=['GET'], ) @login_required def api_get_annotation_history(assignment_id, sample_index): """Return newest-first annotation snapshots for one assigned sample.""" assignment = Assignment.query.get_or_404(assignment_id) if session.get('role') != 'admin' and assignment.user_id != session['user_id']: return jsonify({'error': 'Access denied'}), 403 entries = ( AnnotationHistory.query .filter_by(assignment_id=assignment_id, sample_index=sample_index) .order_by(AnnotationHistory.created_at.desc(), AnnotationHistory.id.desc()) .all() ) return jsonify([entry.to_dict() for entry in entries]) # ── Export API ────────────────────────────────────────────────────────────── @app.route('/api/assignments//export', methods=['GET']) @login_required def api_export(assignment_id): """Export reviewed data as JSON or CSV.""" import json as _json import csv import io fmt = request.args.get('format', 'json') # 'json' or 'csv' assignment = Assignment.query.get_or_404(assignment_id) if session.get('role') != 'admin' and assignment.user_id != session['user_id']: return jsonify({'error': 'Access denied'}), 403 csv_file = assignment.csv_file safe_filepath = os.path.join(CSV_FOLDER, os.path.basename(csv_file.filepath)) if not os.path.exists(safe_filepath): return jsonify({'error': 'Source CSV file not found'}), 404 rows = parse_csv_file(safe_filepath) reviews = {r.sample_index: r for r in Review.query.filter_by(assignment_id=assignment_id).all()} export_rows = [] for i, row in enumerate(rows): review = reviews.get(i) # Get effective annotations (edited or original) if review and review.annotations_edit: anno = _json.loads(review.annotations_edit) else: anno = { 'emotion': _json.loads(row.get('Emotion', 'null')) if row.get('Emotion') else None, 'distortions': _json.loads(row.get('Distortions', 'null')) if row.get('Distortions') else None, 'empathy': _json.loads(row.get('Empathy_Level', 'null')) if row.get('Empathy_Level') else None, } export_rows.append({ 'sample_index': i, 'conversation_cut': row.get('conversation_cut', ''), 'emotion_type': row.get('emotion_type', ''), 'problem_type': row.get('problem_type', ''), 'Emotion': anno.get('emotion'), 'Distortions': anno.get('distortions'), 'Empathy_Level': anno.get('empathy'), 'review_status': review.status if review else 'pending', 'review_comment': review.comment if review else '', 'is_edited': bool(review and review.annotations_edit), }) if fmt == 'csv': output = io.StringIO() writer = csv.writer(output) writer.writerow([ 'sample_index', 'conversation_cut', 'emotion_type', 'problem_type', 'Emotion', 'Distortions', 'Empathy_Level', 'review_status', 'review_comment', 'is_edited' ]) for r in export_rows: writer.writerow([ r['sample_index'], r['conversation_cut'], r['emotion_type'], r['problem_type'], _json.dumps(r['Emotion'], ensure_ascii=False) if r['Emotion'] else '', _json.dumps(r['Distortions'], ensure_ascii=False) if r['Distortions'] else '', _json.dumps(r['Empathy_Level'], ensure_ascii=False) if r['Empathy_Level'] else '', r['review_status'], r['review_comment'], r['is_edited'], ]) resp = app.response_class( output.getvalue(), mimetype='text/csv', headers={'Content-Disposition': f'attachment; filename=export_{csv_file.filename}'} ) return resp else: return jsonify(export_rows) # ── Init DB ───────────────────────────────────────────────────────────────── def init_db(): """Create tables and seed admin user.""" db.create_all() # ``create_all`` does not retrofit indexes onto tables created by an older # app version, so ensure the history lookup index also exists on upgrades. for index in AnnotationHistory.__table__.indexes: index.create(bind=db.engine, checkfirst=True) if not User.query.filter_by(username='admin').first(): admin = User( username='admin', password_hash=hash_password('admin123'), role='admin', ) db.session.add(admin) db.session.commit() print(' ✓ Created default admin user (admin / admin123)') # Initial CSV scan os.makedirs(CSV_FOLDER, exist_ok=True) sync_csv_files() file_count = CsvFile.query.count() print(f' ✓ Found {file_count} CSV file(s) in {CSV_FOLDER}') if __name__ == '__main__': with app.app_context(): init_db() print(f'\n 🚀 Server running at http://localhost:5000\n') app.run(debug=True, host='0.0.0.0', port=5000)