Spaces:
Sleeping
Sleeping
| """ | |
| Web-based Annotation Interface for Privacy Inferences | |
| Modified for Hugging Face Spaces deployment | |
| Run: python annotation_app.py | |
| Then open: http://localhost:7860 | |
| """ | |
| from flask import Flask, render_template, request, jsonify, send_file, session, redirect, url_for | |
| import json | |
| from pathlib import Path | |
| from datetime import datetime | |
| import os | |
| from functools import wraps | |
| import zipfile | |
| import io | |
| from collections import defaultdict | |
| app = Flask(__name__) | |
| # Security configuration | |
| app.secret_key = os.environ.get('SECRET_KEY', 'votre-cle-secrete-tres-longue-a-changer-123456789') | |
| ADMIN_PASSWORD = os.environ.get('ADMIN_PASSWORD', 'antoine2025') | |
| # Configuration | |
| RESULTS_DIR = Path("results") | |
| ANNOTATIONS_DIR = Path("annotations") | |
| ANNOTATIONS_DIR.mkdir(exist_ok=True) | |
| MONTHS = [ | |
| 'JANUARY', 'FEBRUARY', 'MARCH', 'APRIL', 'MAY', 'JUNE', | |
| 'JULY', 'AUGUST', 'SEPTEMBER', 'OCTOBER', 'NOVEMBER', 'DECEMBER' | |
| ] | |
| CATEGORIES = ['health', 'religion', 'family', 'routines', 'work', 'leisure', 'economics'] | |
| # ============================================================================ | |
| # AUTHENTICATION | |
| # ============================================================================ | |
| def login_required(f): | |
| """Decorator to protect routes""" | |
| def decorated_function(*args, **kwargs): | |
| if not session.get('logged_in'): | |
| return redirect(url_for('login')) | |
| return f(*args, **kwargs) | |
| return decorated_function | |
| def login(): | |
| """Login page""" | |
| if request.method == 'POST': | |
| if request.form.get('password') == ADMIN_PASSWORD: | |
| session['logged_in'] = True | |
| return redirect(url_for('index')) | |
| else: | |
| return ''' | |
| <html> | |
| <head> | |
| <title>Erreur</title> | |
| <style> | |
| body { | |
| font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; | |
| max-width: 400px; | |
| margin: 100px auto; | |
| text-align: center; | |
| } | |
| .error { color: #ef4444; margin: 20px 0; } | |
| a { color: #667eea; text-decoration: none; } | |
| </style> | |
| </head> | |
| <body> | |
| <h2>❌ Mot de passe incorrect</h2> | |
| <p class="error">Le mot de passe saisi n'est pas valide.</p> | |
| <a href="/login">← Réessayer</a> | |
| </body> | |
| </html> | |
| ''' | |
| return ''' | |
| <html> | |
| <head> | |
| <title>Connexion - Privacy Annotation</title> | |
| <style> | |
| body { | |
| font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; | |
| background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); | |
| height: 100vh; | |
| display: flex; | |
| align-items: center; | |
| justify-content: center; | |
| margin: 0; | |
| } | |
| .login-box { | |
| background: white; | |
| padding: 40px; | |
| border-radius: 12px; | |
| box-shadow: 0 10px 40px rgba(0,0,0,0.2); | |
| width: 100%; | |
| max-width: 400px; | |
| } | |
| h2 { | |
| margin: 0 0 30px 0; | |
| color: #333; | |
| text-align: center; | |
| } | |
| input { | |
| width: 100%; | |
| padding: 12px; | |
| margin: 10px 0; | |
| border: 2px solid #e5e7eb; | |
| border-radius: 8px; | |
| font-size: 16px; | |
| box-sizing: border-box; | |
| } | |
| input:focus { | |
| outline: none; | |
| border-color: #667eea; | |
| } | |
| button { | |
| width: 100%; | |
| padding: 12px; | |
| margin-top: 10px; | |
| background: #667eea; | |
| color: white; | |
| border: none; | |
| border-radius: 8px; | |
| font-size: 16px; | |
| font-weight: 600; | |
| cursor: pointer; | |
| transition: background 0.2s; | |
| } | |
| button:hover { | |
| background: #5568d3; | |
| } | |
| .info { | |
| text-align: center; | |
| color: #6b7280; | |
| font-size: 14px; | |
| margin-top: 20px; | |
| } | |
| </style> | |
| </head> | |
| <body> | |
| <div class="login-box"> | |
| <h2>🔒 Privacy Annotation</h2> | |
| <form method="post"> | |
| <input type="password" name="password" placeholder="Mot de passe" required autofocus> | |
| <button type="submit">Se connecter</button> | |
| </form> | |
| <p class="info">Interface d'annotation pour évaluation des modèles LLM</p> | |
| </div> | |
| </body> | |
| </html> | |
| ''' | |
| def logout(): | |
| """Logout""" | |
| session.pop('logged_in', None) | |
| return redirect(url_for('login')) | |
| # ============================================================================ | |
| # HELPER FUNCTIONS | |
| # ============================================================================ | |
| def get_available_models(): | |
| """Get list of available model directories.""" | |
| if not RESULTS_DIR.exists(): | |
| return [] | |
| return [d.name for d in RESULTS_DIR.iterdir() if d.is_dir() and not d.name.startswith('.')] | |
| def load_result(model_name, month): | |
| """Load a specific result file.""" | |
| filepath = RESULTS_DIR / model_name / "2017" / f"2017_{month}_P1.json" | |
| if not filepath.exists(): | |
| return None | |
| with open(filepath, 'r') as f: | |
| return json.load(f) | |
| def get_annotation_status(): | |
| """Get status of all annotations.""" | |
| status = {} | |
| models = get_available_models() | |
| for model in models: | |
| status[model] = {} | |
| for month in MONTHS: | |
| ann_file = ANNOTATIONS_DIR / f"{model}_{month}_annotations.json" | |
| status[model][month] = ann_file.exists() | |
| return status | |
| def extract_inferences(response, category): | |
| """Extract inferences for a specific category.""" | |
| inferences = [] | |
| lines = response.split('\n') | |
| in_category = False | |
| for line in lines: | |
| # Check if entering this category | |
| if category.lower() in line.lower() and (':' in line or category in line): | |
| in_category = True | |
| continue | |
| # Check if entering new category | |
| if any(cat.lower() in line.lower() for cat in CATEGORIES if cat != category): | |
| in_category = False | |
| # Extract inference | |
| if in_category and line.strip(): | |
| cleaned = line.strip().lstrip('•-*0123456789.) ') | |
| if len(cleaned) > 15: | |
| inferences.append(cleaned) | |
| return inferences | |
| # ============================================================================ | |
| # MAIN ROUTES | |
| # ============================================================================ | |
| def index(): | |
| """Main page - show annotation dashboard.""" | |
| models = get_available_models() | |
| status = get_annotation_status() | |
| return render_template('index.html', | |
| models=models, | |
| months=MONTHS, | |
| status=status) | |
| def annotate(model, month): | |
| """Annotation page for specific model and month.""" | |
| result = load_result(model, month) | |
| if not result: | |
| return f"Result not found: {model}/{month}", 404 | |
| # Check if already annotated | |
| ann_file = ANNOTATIONS_DIR / f"{model}_{month}_annotations.json" | |
| existing_annotation = None | |
| if ann_file.exists(): | |
| with open(ann_file, 'r') as f: | |
| existing_annotation = json.load(f) | |
| # Extract inferences by category | |
| category_inferences = {} | |
| for category in CATEGORIES: | |
| category_inferences[category] = extract_inferences(result['response'], category) | |
| return render_template('annotate.html', | |
| model=model, | |
| month=month, | |
| response=result['response'], | |
| categories=CATEGORIES, | |
| category_inferences=category_inferences, | |
| existing_annotation=existing_annotation, | |
| trajectory_stats=result.get('metadata', {}).get('trajectory_stats', {})) | |
| def save_annotation(): | |
| """Save annotation via API.""" | |
| data = request.json | |
| model = data['model'] | |
| month = data['month'] | |
| annotations = data['annotations'] | |
| # Load original result | |
| result = load_result(model, month) | |
| if not result: | |
| return jsonify({'success': False, 'error': 'Result not found'}), 404 | |
| # Create annotation data | |
| ann_data = { | |
| 'model_name': model, | |
| 'month': month, | |
| 'year': 2017, | |
| 'annotated_at': datetime.now().isoformat(), | |
| 'annotation_mode': 'web_interface', | |
| 'original_response': result['response'], | |
| 'annotations': annotations, | |
| 'metadata': { | |
| 'trajectory_period': result.get('trajectory_period'), | |
| 'prompt_type': result.get('prompt_type'), | |
| } | |
| } | |
| # Save to file | |
| ann_file = ANNOTATIONS_DIR / f"{model}_{month}_annotations.json" | |
| with open(ann_file, 'w') as f: | |
| json.dump(ann_data, f, indent=2) | |
| return jsonify({'success': True, 'file': str(ann_file)}) | |
| def get_metrics(model): | |
| """Calculate metrics for a model.""" | |
| pattern = f"{model}_*_annotations.json" | |
| ann_files = list(ANNOTATIONS_DIR.glob(pattern)) | |
| if not ann_files: | |
| return jsonify({'success': False, 'error': 'No annotations found'}) | |
| total_tp = 0 | |
| total_fp = 0 | |
| total_pa = 0 | |
| by_category = {} | |
| annotated_months = [] | |
| for ann_file in ann_files: | |
| with open(ann_file, 'r') as f: | |
| data = json.load(f) | |
| annotated_months.append(data['month']) | |
| for category, items in data['annotations'].items(): | |
| if category not in by_category: | |
| by_category[category] = {'TP': 0, 'FP': 0, 'PA': 0} | |
| for item in items: | |
| label = item['label'] | |
| if label == 'TP': | |
| total_tp += 1 | |
| by_category[category]['TP'] += 1 | |
| elif label == 'FP': | |
| total_fp += 1 | |
| by_category[category]['FP'] += 1 | |
| elif label == 'PA': | |
| total_pa += 1 | |
| by_category[category]['PA'] += 1 | |
| # Calculate precision | |
| precision = total_tp / (total_tp + total_fp) if (total_tp + total_fp) > 0 else 0 | |
| # Category metrics | |
| category_metrics = {} | |
| for category, stats in by_category.items(): | |
| tp = stats['TP'] | |
| fp = stats['FP'] | |
| prec = tp / (tp + fp) if (tp + fp) > 0 else 0 | |
| category_metrics[category] = { | |
| 'tp': tp, | |
| 'fp': fp, | |
| 'pa': stats['PA'], | |
| 'precision': round(prec, 3) | |
| } | |
| return jsonify({ | |
| 'success': True, | |
| 'model': model, | |
| 'months_annotated': len(annotated_months), | |
| 'annotated_months': annotated_months, | |
| 'total_tp': total_tp, | |
| 'total_fp': total_fp, | |
| 'total_pa': total_pa, | |
| 'precision': round(precision, 3), | |
| 'by_category': category_metrics | |
| }) | |
| def metrics_page(): | |
| """Metrics dashboard page.""" | |
| models = get_available_models() | |
| return render_template('metrics.html', models=models) | |
| def download_annotations(): | |
| """Download all annotations as ZIP file""" | |
| memory_file = io.BytesIO() | |
| # Create ZIP with all annotations | |
| with zipfile.ZipFile(memory_file, 'w', zipfile.ZIP_DEFLATED) as zf: | |
| ann_files = list(ANNOTATIONS_DIR.glob('*.json')) | |
| if not ann_files: | |
| return "Aucune annotation disponible pour le moment.", 404 | |
| for file in ann_files: | |
| zf.write(file, file.name) | |
| memory_file.seek(0) | |
| # Filename with date | |
| filename = f'annotations_antoine_{datetime.now().strftime("%Y%m%d_%H%M")}.zip' | |
| return send_file( | |
| memory_file, | |
| mimetype='application/zip', | |
| as_attachment=True, | |
| download_name=filename | |
| ) | |
| # ============================================================================ | |
| # MAIN | |
| # ============================================================================ | |
| if __name__ == '__main__': | |
| # Hugging Face Spaces uses port 7860 by default | |
| port = int(os.environ.get('PORT', 7860)) | |
| # Create directories if needed | |
| templates_dir = Path("templates") | |
| templates_dir.mkdir(exist_ok=True) | |
| RESULTS_DIR.mkdir(exist_ok=True) | |
| ANNOTATIONS_DIR.mkdir(exist_ok=True) | |
| print("="*70) | |
| print("PRIVACY INFERENCE ANNOTATION WEB INTERFACE") | |
| print("="*70) | |
| print(f"\n✓ Starting server on port {port}...") | |
| print(f"✓ Results directory: {RESULTS_DIR.absolute()}") | |
| print(f"✓ Annotations will be saved to: {ANNOTATIONS_DIR.absolute()}") | |
| # Check if running on HF | |
| if os.environ.get('SPACE_ID'): | |
| print(f"✓ Running on Hugging Face Spaces") | |
| print(f"✓ Space ID: {os.environ.get('SPACE_ID')}") | |
| else: | |
| print(f"\n🌐 Open your browser to: http://localhost:{port}") | |
| print("\nPress Ctrl+C to stop the server") | |
| print("="*70 + "\n") | |
| # IMPORTANT: host='0.0.0.0' to be accessible from outside | |
| # debug=False for production | |
| app.run(host='0.0.0.0', port=port, debug=False) |