from flask import Flask, request, render_template, jsonify, session, redirect, url_for, flash import os import uuid import json import time from datetime import datetime, timedelta from werkzeug.utils import secure_filename from improved_document_processor import ImprovedDocumentProcessor import openai from summarization_routes import summarization_bp from dotenv import load_dotenv load_dotenv() # Add this with your other imports from document_summarization_processor import DocumentSummarizationProcessor # Import the new citation generator from citation_generator import process_analysis_with_citations # Import database models and functions from database import db, init_db, shutdown_session from models import User, Company, AnalysisJob, AnalysisResult, ActivityLog, Session as UserSession, Case, CaseDocument # Import blueprints from auth_routes import auth_bp, login_required, admin_required, company_admin_required, log_activity from admin_routes import admin_bp from company_admin_routes import company_admin_bp from case_routes import case_bp app = Flask(__name__) app.config['TEMPLATES_AUTO_RELOAD'] = True app.secret_key = os.environ.get('FLASK_SECRET_KEY', 'dev-secret-key') app.config['UPLOAD_FOLDER'] = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'uploads') app.config['MAX_CONTENT_LENGTH'] = 50 * 1024 * 1024 # 50MB max upload app.config['OPENAI_API_KEY'] = os.environ.get('OPENAI_API_KEY', '') app.config['ALLOWED_EXTENSIONS'] = {'pdf', 'txt', 'jpg', 'jpeg', 'png', 'tif', 'tiff', 'bmp'} # Database configuration app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DATABASE_URL', 'mysql+pymysql://lawbot_user:AK$82Z5FOn38i&!@localhost/lawbot') app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False # Ensure upload directory exists os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True) # Initialize database db.init_app(app) # Initialize Flask-Mail from email_config import init_mail init_mail(app) # Register blueprints app.register_blueprint(auth_bp) app.register_blueprint(admin_bp) app.register_blueprint(company_admin_bp) app.register_blueprint(case_bp) app.register_blueprint(summarization_bp) # Close database session when application shuts down @app.teardown_appcontext def shutdown_db_session(exception=None): shutdown_session() def allowed_file(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in app.config['ALLOWED_EXTENSIONS'] @app.route('/') def index(): # Redirect root to the dashboard if logged in, otherwise to login if 'user_id' in session: return redirect(url_for('dashboard')) return redirect(url_for('auth.login')) @app.route('/user-dashboard') @login_required def user_dashboard(): return redirect(url_for('dashboard')) @app.route('/dashboard') @login_required def dashboard(): """ Display the dashboard with analysis statistics and recent activities """ try: # Get current user user_id = session.get('user_id') current_user = User.query.get(user_id) today = datetime.utcnow().date() yesterday = today - timedelta(days=1) # Determine which analyses to show based on user role if current_user.is_admin(): # Admin sees all analyses analyses_query = AnalysisJob.query elif current_user.is_company_admin(): # Company admin sees company analyses analyses_query = AnalysisJob.query.filter_by(company_id=current_user.company_id) else: # Regular user sees their own analyses analyses_query = AnalysisJob.query.filter_by(user_id=current_user.id) # Calculate statistics stats = { "total_analyses": analyses_query.count(), "completed": analyses_query.filter_by(status="completed").count(), "in_progress": analyses_query.filter_by(status="processing").count(), "failed": analyses_query.filter_by(status="failed").count(), "today": analyses_query.filter( AnalysisJob.created_at >= datetime.utcnow().replace( hour=0, minute=0, second=0, microsecond=0 ) ).count() } # Get recent analyses recent_analyses = analyses_query.order_by(AnalysisJob.created_at.desc()).limit(10).all() # Format recent analyses for display formatted_analyses = [] for job in recent_analyses: # Format date for display job_date = job.created_at today = datetime.utcnow().date() yesterday = today - timedelta(days=1) if job_date.date() == today: date_str = f"Today, {job_date.strftime('%I:%M %p')}" elif job_date.date() == yesterday: date_str = f"Yesterday, {job_date.strftime('%I:%M %p')}" else: date_str = job_date.strftime('%b %d, %Y') formatted_analyses.append({ "job_id": job.job_uuid, "filename": job.filename, "perspective": job.perspective, "status": job.status, "date": date_str, "user": job.user.username if job.user else "Unknown", "case": job.case.case_name if job.case else None }) # Get recent activity if current_user.is_admin(): activities = ActivityLog.query.order_by(ActivityLog.created_at.desc()).limit(5).all() elif current_user.is_company_admin(): activities = ActivityLog.query.filter_by(company_id=current_user.company_id).order_by(ActivityLog.created_at.desc()).limit(5).all() else: activities = ActivityLog.query.filter_by(user_id=current_user.id).order_by(ActivityLog.created_at.desc()).limit(5).all() # Format activities for display formatted_activities = [] for activity in activities: activity_date = activity.created_at if activity_date.date() == today: date_str = f"Today, {activity_date.strftime('%I:%M %p')}" elif activity_date.date() == yesterday: date_str = f"Yesterday, {activity_date.strftime('%I:%M %p')}" else: date_str = activity_date.strftime('%b %d, %Y') formatted_activities.append({ "type": activity.activity_type, "message": activity.description, "time": date_str, "user": activity.user.username if activity.user else "Unknown" }) return render_template('dashboard.html', stats=stats, recent_analyses=formatted_analyses, activities=formatted_activities, user=current_user) except Exception as e: import traceback error_details = traceback.format_exc() print(f"Error displaying dashboard: {str(e)}\n{error_details}") return render_template('error.html', error="Error displaying dashboard", message=f"An error occurred: {str(e)}") @app.route('/upload_form') @login_required def upload_form(): # Get case information if provided case_id = request.args.get('case_id') document_id = request.args.get('document_id') case = None document = None if case_id: case = Case.query.get(case_id) user_id = session.get('user_id') user = User.query.get(user_id) # Check if user can access this case if case and not user.can_view_case(case): flash('You do not have permission to access this case', 'danger') return redirect(url_for('dashboard')) if document_id: document = CaseDocument.query.get(document_id) # Render the upload form with case context return render_template('index.html', case=case, document=document) @app.route('/upload', methods=['POST']) @login_required def upload_file(): if 'file' not in request.files: return jsonify({"error": "No file part"}), 400 file = request.files['file'] if file.filename == '': return jsonify({"error": "No selected file"}), 400 perspective = request.form.get('perspective', 'neutral') if perspective not in ['prosecutor', 'defense', 'neutral']: return jsonify({"error": "Invalid perspective"}), 400 if file and allowed_file(file.filename): # Get current user user_id = session.get('user_id') user = User.query.get(user_id) # Get case ID if provided case_id = request.form.get('case_id') # Generate a unique ID for this job job_uuid = str(uuid.uuid4()) # Save the file filename = secure_filename(file.filename) file_path = os.path.join(app.config['UPLOAD_FOLDER'], f"{job_uuid}_{filename}") file.save(file_path) # Create database record for this job job = AnalysisJob( job_uuid=job_uuid, user_id=user_id, company_id=user.company_id, case_id=case_id, filename=f"{job_uuid}_{filename}", original_filename=filename, file_path=file_path, perspective=perspective, status="processing" ) db.session.add(job) db.session.commit() # If this is from a case document, link them document_id = request.form.get('document_id') if case_id and document_id: document = CaseDocument.query.get(document_id) if document and document.case_id == int(case_id): document.analysis_job_id = job_uuid db.session.add(document) db.session.commit() # Log the activity log_activity( user_id, 'analysis_started', f"Started analysis of {filename} with {perspective} perspective" + (f" for case {job.case.case_name}" if job.case else ""), 'analysis_job', job.id ) # Log case activity if applicable if case_id: from case_routes import log_case_activity log_case_activity( case_id, user_id, 'analysis_started', f'{user.username} started {perspective} analysis of {filename}', 'analysis', job.id ) # Start processing in a background thread import threading threading.Thread(target=process_document_job, args=(job_uuid, file_path, perspective, user_id)).start() return jsonify({ "job_id": job_uuid, "message": "Document uploaded and processing started" }) return jsonify({"error": "File type not allowed. Supported formats: PDF, TXT, JPG, JPEG, PNG, TIFF, BMP"}), 400 def process_document_job(job_uuid, file_path, perspective, user_id): """Process document in background thread""" with app.app_context(): try: # Get the job from the database job = AnalysisJob.query.filter_by(job_uuid=job_uuid).first() if not job: print(f"Error: Job with UUID {job_uuid} not found") return processor = ImprovedDocumentProcessor(api_key=app.config['OPENAI_API_KEY']) # Process document result = processor.process_document(file_path, perspective) # Check for errors if "final_analysis" not in result or not result["final_analysis"]: job.status = "failed" job.error_message = "Analysis failed to generate results. Please try a smaller document." db.session.commit() return # Generate citations for the analysis try: print(f"Generating citations for analysis with perspective: {perspective}") processed_text, citations = process_analysis_with_citations( result["final_analysis"], perspective, app.config['OPENAI_API_KEY'] ) # Update the analysis text with citations result["final_analysis"] = processed_text result["citations"] = citations print(f"Generated {len(citations)} citations") except Exception as citation_error: print(f"Error generating citations: {str(citation_error)}") result["citations"] = [] # Create analysis result in database analysis_result = AnalysisResult( job_id=job.id, final_analysis=result["final_analysis"], citations=result["citations"] if "citations" in result else [], section_analyses=result["section_analyses"] if "section_analyses" in result else [] ) # Update job status job.status = "completed" job.completed_at = datetime.utcnow() db.session.add(analysis_result) db.session.commit() # Log the activity log_activity( user_id, 'analysis_completed', f"Completed analysis of {job.original_filename} with {perspective} perspective" + (f" for case {job.case.case_name}" if job.case else ""), 'analysis_job', job.id ) # Log case activity if applicable if job.case_id: from case_routes import log_case_activity log_case_activity( job.case_id, user_id, 'analysis_completed', f'Completed {perspective} analysis of {job.original_filename}', 'analysis', job.id ) # Clean up file after 1 hour (optional) import threading threading.Timer(3600, lambda: os.remove(file_path) if os.path.exists(file_path) else None).start() except Exception as e: import traceback error_details = traceback.format_exc() print(f"Error processing document: {str(e)}\n{error_details}") # Update job status job = AnalysisJob.query.filter_by(job_uuid=job_uuid).first() if job: job.status = "failed" job.error_message = str(e) db.session.commit() # Log the activity log_activity( user_id, 'analysis_failed', f"Failed analysis of {job.original_filename}: {str(e)}", 'analysis_job', job.id ) @app.route('/job_status/', methods=['GET']) @login_required def job_status(job_id): """Check status of a processing job""" job = AnalysisJob.query.filter_by(job_uuid=job_id).first() if not job: return jsonify({"error": "Job not found"}), 404 # Check if user has permission to view this job user_id = session.get('user_id') user = User.query.get(user_id) # Check case access if job is associated with a case if job.case and not user.can_view_case(job.case): return jsonify({"error": "You do not have permission to view this job"}), 403 # Otherwise check standard permissions if not user.is_admin() and not user.is_company_admin() and job.user_id != user_id: return jsonify({"error": "You do not have permission to view this job"}), 403 # Calculate progress (simplified - in a real app, you'd track this more accurately) progress = 0 if job.status == "completed": progress = 100 elif job.status == "processing": # Simulate progress based on time elapsed (up to 90%) elapsed = (datetime.utcnow() - job.created_at).total_seconds() progress = min(90, int(elapsed / 2)) # Rough estimate return jsonify({ "job_id": job.job_uuid, "status": job.status, "progress": progress, "filename": job.original_filename, "perspective": job.perspective, "completed": job.status == "completed" }) @app.route('/results/', methods=['GET']) @login_required def get_results(job_id): """Get results of a completed job""" job = AnalysisJob.query.filter_by(job_uuid=job_id).first() if not job: return jsonify({"error": "Job not found"}), 404 # Check if user has permission to view this job user_id = session.get('user_id') user = User.query.get(user_id) # Check case access if job is associated with a case if job.case and not user.can_view_case(job.case): return jsonify({"error": "You do not have permission to view this job"}), 403 if not user.is_admin() and not user.is_company_admin() and job.user_id != user_id: return jsonify({"error": "You do not have permission to view this job"}), 403 if job.status != "completed": return jsonify({ "error": "Job not completed", "status": job.status }), 400 # Get the results results = job.results if not results: return jsonify({"error": "Results not found"}), 404 return jsonify({ "job_id": job.job_uuid, "status": "completed", "filename": job.original_filename, "perspective": job.perspective, "final_analysis": results.final_analysis, "section_analyses": results.section_analyses_json, "num_sections": len(results.section_analyses_json) if results.section_analyses_json else 0, "citations": results.citations_json, "case": { "id": job.case.id, "name": job.case.case_name, "number": job.case.case_number } if job.case else None }) @app.route('/summarize_form') @login_required def summarize_form(): """Redirect to the new summarization form""" return redirect(url_for('summarization.summarization_form')) @app.route('/analysis/', methods=['GET']) @login_required def show_analysis(job_id): """Display analysis results""" job = AnalysisJob.query.filter_by(job_uuid=job_id).first() if not job: return render_template('error.html', error="Analysis not found") # Check if user has permission to view this job user_id = session.get('user_id') user = User.query.get(user_id) # Check case access if job is associated with a case if job.case and not user.can_view_case(job.case): return render_template('error.html', error="You do not have permission to view this analysis") if not user.is_admin() and not user.is_company_admin() and job.user_id != user_id: return render_template('error.html', error="You do not have permission to view this analysis") if job.status != "completed": return render_template('processing.html', job_id=job_id, job=job) # Get the results results = job.results if not results: return render_template('error.html', error="Analysis results not found") # Check if arguments have been generated has_arguments = results.arguments is not None # Format job data for template formatted_job = { "job_id": job.job_uuid, "filename": job.original_filename, "perspective": job.perspective, "status": job.status, "case": job.case } return render_template('analysis.html', job_id=job_id, job=formatted_job, final_analysis=results.final_analysis, section_analyses=results.section_analyses_json, has_arguments=has_arguments, citations=results.citations_json) @app.route('/summarize', methods=['POST']) @login_required def summarize_document(): """Handle document summarization""" if 'file' not in request.files: return jsonify({"error": "No file part"}), 400 file = request.files['file'] if file.filename == '': return jsonify({"error": "No selected file"}), 400 summary_type = request.form.get('summary_type', 'standard') length = request.form.get('length', 'medium') if summary_type not in ['standard', 'executive', 'bullet_points', 'key_facts']: return jsonify({"error": "Invalid summary type"}), 400 if length not in ['short', 'medium', 'long']: return jsonify({"error": "Invalid length"}), 400 if file and allowed_file(file.filename): try: # Get current user user_id = session.get('user_id') user = User.query.get(user_id) # Save the file temporarily filename = secure_filename(file.filename) temp_path = os.path.join(app.config['UPLOAD_FOLDER'], f"temp_{uuid.uuid4()}_{filename}") file.save(temp_path) # Extract text from the document from improved_document_processor import ImprovedDocumentProcessor processor = ImprovedDocumentProcessor(api_key=app.config['OPENAI_API_KEY']) text = processor.extract_text_from_file(temp_path) if not text: os.remove(temp_path) return jsonify({"error": "Failed to extract text from document"}), 400 # Create summarization processor and generate summary from document_summarization_processor import DocumentSummarizationProcessor summarizer = DocumentSummarizationProcessor( api_key=app.config['OPENAI_API_KEY'] ) result = summarizer.create_summary(text, summary_type, length) # Clean up temp file os.remove(temp_path) if result['success']: # Log the activity log_activity( user_id, 'document_summarized', f"Summarized document {filename} ({summary_type}, {length})", 'summary' ) return jsonify({ "success": True, "filename": filename, "summary": result['summary'], "summary_type": summary_type, "length": length, "word_count": result['word_count'], "compression_ratio": result['compression_ratio'] }) else: return jsonify({ "success": False, "error": result['error'] }), 500 except Exception as e: # Clean up temp file if it exists if 'temp_path' in locals() and os.path.exists(temp_path): os.remove(temp_path) import traceback error_details = traceback.format_exc() print(f"Error summarizing document: {str(e)}\n{error_details}") return jsonify({ "success": False, "error": str(e) }), 500 return jsonify({"error": "File type not allowed"}), 400 @app.route('/reanalyze//', methods=['GET']) @login_required def reanalyze_document(job_id, perspective): """Re-analyze an existing document with a different perspective""" original_job = AnalysisJob.query.filter_by(job_uuid=job_id).first() if not original_job: return jsonify({"error": "Analysis not found"}), 404 # Check if user has permission to view this job user_id = session.get('user_id') user = User.query.get(user_id) # Check case access if job is associated with a case if original_job.case and not user.can_view_case(original_job.case): return jsonify({"error": "You do not have permission to reanalyze this document"}), 403 if not user.is_admin() and not user.is_company_admin() and original_job.user_id != user_id: return jsonify({"error": "You do not have permission to reanalyze this document"}), 403 if perspective not in ['prosecutor', 'defense', 'neutral']: return jsonify({"error": "Invalid perspective"}), 400 # Generate a new job ID new_job_uuid = str(uuid.uuid4()) # Create a new job record new_job = AnalysisJob( job_uuid=new_job_uuid, user_id=user_id, company_id=user.company_id, case_id=original_job.case_id, # Preserve case association filename=original_job.filename, original_filename=original_job.original_filename, file_path=original_job.file_path, perspective=perspective, status="processing" ) db.session.add(new_job) db.session.commit() # Log the activity log_activity( user_id, 'reanalysis_started', f"Started reanalysis of {original_job.original_filename} with {perspective} perspective" + (f" for case {new_job.case.case_name}" if new_job.case else ""), 'analysis_job', new_job.id ) # Log case activity if applicable if new_job.case_id: from case_routes import log_case_activity log_case_activity( new_job.case_id, user_id, 'reanalysis_started', f'Started {perspective} reanalysis of {original_job.original_filename}', 'analysis', new_job.id ) # Start processing in a background thread import threading threading.Thread(target=process_document_job, args=(new_job_uuid, original_job.file_path, perspective, user_id)).start() return jsonify({ "job_id": new_job_uuid, "message": "Document re-analysis started" }) # FIXED: Single copy endpoint that works with your model structure @app.route('/api/copy-analysis/', methods=['GET']) @login_required def copy_analysis(job_id): """ API endpoint to get analysis text for copying to clipboard """ try: print(f"Copy request for job ID: {job_id}") # Find the analysis job using job_uuid (not id) job = AnalysisJob.query.filter_by(job_uuid=job_id).first() if not job: print(f"Job not found: {job_id}") return jsonify({ "success": False, "error": "Analysis not found" }), 404 print(f"Job found: {job.original_filename}, Status: {job.status}") # Check if user has permission to view this job user_id = session.get('user_id') user = User.query.get(user_id) # Check case access if job is associated with a case if job.case and not user.can_view_case(job.case): print(f"User cannot view case") return jsonify({ "success": False, "error": "You do not have permission to view this analysis" }), 403 if not user.is_admin() and not user.is_company_admin() and job.user_id != user_id: print(f"User does not have permission") return jsonify({ "success": False, "error": "You do not have permission to view this analysis" }), 403 if job.status != "completed": print(f"Job not completed: {job.status}") return jsonify({ "success": False, "error": "Analysis not completed", "status": job.status }), 400 # Get the results from the AnalysisResult model results = job.results if not results: print(f"No results found for job: {job_id}") return jsonify({ "success": False, "error": "Results not found" }), 404 # Format the analysis as plain text perspective = job.perspective.capitalize() text_analysis = f"{perspective} Perspective Analysis\n" text_analysis += "=" * 50 + "\n\n" # Add document information text_analysis += f"Document: {job.original_filename}\n" text_analysis += f"Analysis Date: {job.created_at.strftime('%B %d, %Y at %I:%M %p')}\n" text_analysis += f"Perspective: {perspective}\n" # Add case information if available if job.case: text_analysis += f"Case: {job.case.case_name} (#{job.case.case_number})\n" if job.case.client_name: text_analysis += f"Client: {job.case.client_name}\n" text_analysis += "\n" + "-" * 50 + "\n" text_analysis += "LEGAL ANALYSIS\n" text_analysis += "-" * 50 + "\n\n" # Add the final analysis if results.final_analysis: text_analysis += results.final_analysis + "\n\n" # Add section analyses if available if results.section_analyses_json: text_analysis += "-" * 50 + "\n" text_analysis += "DETAILED SECTION ANALYSIS\n" text_analysis += "-" * 50 + "\n\n" for i, section in enumerate(results.section_analyses_json): if isinstance(section, dict) and 'title' in section and 'content' in section: text_analysis += f"Section {i+1}: {section['title']}\n" text_analysis += "-" * 30 + "\n" text_analysis += f"{section['content']}\n\n" else: text_analysis += f"Section {i+1}\n" text_analysis += "-" * 30 + "\n" text_analysis += f"{section}\n\n" # Add arguments if available if results.arguments: text_analysis += "-" * 50 + "\n" text_analysis += "LEGAL ARGUMENTS\n" text_analysis += "-" * 50 + "\n\n" text_analysis += results.arguments + "\n\n" # Add citations if available if results.citations_json: text_analysis += "-" * 50 + "\n" text_analysis += "CITATIONS AND REFERENCES\n" text_analysis += "-" * 50 + "\n\n" for i, citation in enumerate(results.citations_json, 1): text_analysis += f"[{i}] {citation.get('text', '')}\n" if "url" in citation and citation["url"]: text_analysis += f" URL: {citation['url']}\n" text_analysis += "\n" # Add footer text_analysis += "\n" + "=" * 50 + "\n" text_analysis += f"Generated by LawBot on {datetime.utcnow().strftime('%B %d, %Y at %I:%M %p UTC')}\n" text_analysis += "This analysis is for informational purposes only and does not constitute legal advice.\n" print(f"Analysis text generated successfully. Length: {len(text_analysis)} characters") return jsonify({ 'success': True, 'job_id': job_id, 'perspective': perspective, 'analysis': text_analysis, 'filename': job.original_filename, 'word_count': len(text_analysis.split()), 'char_count': len(text_analysis) }) except Exception as e: import traceback error_details = traceback.format_exc() print(f"Error getting analysis for copy: {str(e)}\n{error_details}") return jsonify({ 'success': False, 'error': f"Internal server error: {str(e)}" }), 500 # Debug endpoint @app.route('/api/debug-copy/', methods=['GET']) @login_required def debug_copy_analysis(job_id): """Debug endpoint to see what data is available""" try: job = AnalysisJob.query.filter_by(job_uuid=job_id).first() if not job: return jsonify({"error": "Job not found", "job_id": job_id}), 404 user_id = session.get('user_id') user = User.query.get(user_id) debug_info = { "job_found": True, "job_id": job.id, "job_uuid": job.job_uuid, "status": job.status, "filename": job.original_filename, "perspective": job.perspective, "created_at": job.created_at.isoformat(), "user_id": job.user_id, "current_user_id": user_id, "user_owns_job": job.user_id == user_id, "has_results": job.results is not None, } if job.results: debug_info.update({ "has_final_analysis": job.results.final_analysis is not None, "final_analysis_length": len(job.results.final_analysis) if job.results.final_analysis else 0, "has_citations": job.results.citations_json is not None, "citation_count": len(job.results.citations_json) if job.results.citations_json else 0, "has_section_analyses": job.results.section_analyses_json is not None, "section_count": len(job.results.section_analyses_json) if job.results.section_analyses_json else 0, "has_arguments": job.results.arguments is not None, }) if job.case: debug_info["case_info"] = { "case_id": job.case.id, "case_name": job.case.case_name, "case_number": job.case.case_number } return jsonify(debug_info) except Exception as e: import traceback return jsonify({ "error": str(e), "traceback": traceback.format_exc() }), 500 @app.route('/compare//', methods=['GET']) @login_required def compare_analyses(job_id_1, job_id_2): """Display two analyses side by side for comparison""" # Get both jobs job1 = AnalysisJob.query.filter_by(job_uuid=job_id_1).first() job2 = AnalysisJob.query.filter_by(job_uuid=job_id_2).first() if not job1 or not job2: return render_template('error.html', error="One or both analyses not found", message="The requested analyses could not be found. They may have expired or been removed.") # Check if user has permission to view these jobs user_id = session.get('user_id') user = User.query.get(user_id) # Check case access if jobs are associated with cases if job1.case and not user.can_view_case(job1.case): return render_template('error.html', error="You do not have permission to view one of these analyses") if job2.case and not user.can_view_case(job2.case): return render_template('error.html', error="You do not have permission to view one of these analyses") if not user.is_admin() and not user.is_company_admin(): if job1.user_id != user_id or job2.user_id != user_id: return render_template('error.html', error="You do not have permission to view one or both of these analyses") # Check if both jobs are completed if job1.status != "completed" or job2.status != "completed": # If one is still processing, show that status if job1.status == "processing" or job2.status == "processing": processing_job = job1 if job1.status == "processing" else job2 return render_template('processing.html', job_id=processing_job.job_uuid, job=processing_job, message="One or both analyses are still being processed. Please wait.") # If one failed, show error return render_template('error.html', error="Analysis comparison failed", message="One or both analyses did not complete successfully.") # Get results for both jobs results1 = job1.results results2 = job2.results if not results1 or not results2: return render_template('error.html', error="Analysis results not found") # Format job data for template formatted_job1 = { "job_id": job1.job_uuid, "filename": job1.original_filename, "perspective": job1.perspective, "status": job1.status, "case": job1.case } formatted_job2 = { "job_id": job2.job_uuid, "filename": job2.original_filename, "perspective": job2.perspective, "status": job2.status, "case": job2.case } # Combine unique citations from both analyses all_citations = {} for citation in results1.citations_json: citation_text = citation.get('text', '') if citation_text: all_citations[citation_text] = citation for citation in results2.citations_json: citation_text = citation.get('text', '') if citation_text and citation_text not in all_citations: all_citations[citation_text] = citation combined_citations = list(all_citations.values()) # Log the activity log_activity( user_id, 'analysis_comparison', f"Compared analyses: {job1.original_filename} ({job1.perspective}) and {job2.original_filename} ({job2.perspective})", 'analysis_job', job1.id ) # Render the comparison template with both analyses return render_template('compare.html', job1=formatted_job1, job2=formatted_job2, final_analysis1=results1.final_analysis, final_analysis2=results2.final_analysis, citations=combined_citations) @app.route('/generate_arguments/', methods=['GET']) @login_required def generate_arguments(job_id): """Generate legal arguments based on an existing analysis""" job = AnalysisJob.query.filter_by(job_uuid=job_id).first() if not job: return jsonify({"error": "Analysis not found"}), 404 # Check if user has permission to view this job user_id = session.get('user_id') user = User.query.get(user_id) # Check case access if job is associated with a case if job.case and not user.can_view_case(job.case): return jsonify({"error": "You do not have permission to generate arguments for this analysis"}), 403 if not user.is_admin() and not user.is_company_admin() and job.user_id != user_id: return jsonify({"error": "You do not have permission to generate arguments for this analysis"}), 403 if job.status != "completed": return jsonify({"error": "Analysis not yet completed"}), 400 # Get the results results = job.results if not results: return jsonify({"error": "Results not found"}), 404 # Get the final analysis text analysis_text = results.final_analysis perspective = job.perspective # Set up the prompt based on perspective if perspective == "prosecutor": prompt = f"""Based on the following legal analysis from a prosecutor's perspective, generate the strongest arguments for both prosecution and defense. For each side (prosecution and defense), provide: 1. Three key arguments with concise statements 2. Supporting evidence or reasoning from the analysis for each argument 3. Potential counterarguments or weaknesses Analysis: {analysis_text} Format your response with clear headings for Prosecution Arguments and Defense Arguments. """ elif perspective == "defense": prompt = f"""Based on the following legal analysis from a defense attorney's perspective, generate the strongest arguments for both prosecution and defense. For each side (prosecution and defense), provide: 1. Three key arguments with concise statements 2. Supporting evidence or reasoning from the analysis for each argument 3. Potential counterarguments or weaknesses Analysis: {analysis_text} Format your response with clear headings for Prosecution Arguments and Defense Arguments. """ else: # neutral prompt = f"""Based on the following neutral legal analysis, generate the strongest arguments for both prosecution and defense. For each side (prosecution and defense), provide: 1. Three key arguments with concise statements 2. Supporting evidence or reasoning from the analysis for each argument 3. Potential counterarguments or weaknesses Analysis: {analysis_text} Format your response with clear headings for Prosecution Arguments and Defense Arguments. """ try: # Call OpenAI API client = openai.OpenAI(api_key=app.config['OPENAI_API_KEY']) response = client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": "You are a legal expert skilled at developing arguments for both sides of a case."}, {"role": "user", "content": prompt} ], max_tokens=2000, temperature=0.5 ) arguments = response.choices[0].message.content # Store the arguments with the results results.arguments = arguments db.session.commit() # Log the activity log_activity( user_id, 'arguments_generated', f"Generated arguments for analysis: {job.original_filename}", 'analysis_job', job.id ) return jsonify({ "success": True, "job_id": job_id, "arguments": arguments }) except Exception as e: import traceback error_details = traceback.format_exc() print(f"Error generating arguments: {str(e)}\n{error_details}") return jsonify({ "success": False, "error": str(e) }), 500 @app.route('/arguments/', methods=['GET']) @login_required def view_arguments(job_id): """View generated arguments for an analysis""" job = AnalysisJob.query.filter_by(job_uuid=job_id).first() if not job: return render_template('error.html', error="Analysis not found") # Check if user has permission to view this job user_id = session.get('user_id') user = User.query.get(user_id) # Check case access if job is associated with a case if job.case and not user.can_view_case(job.case): return render_template('error.html', error="You do not have permission to view this analysis") if not user.is_admin() and not user.is_company_admin() and job.user_id != user_id: return render_template('error.html', error="You do not have permission to view this analysis") if job.status != "completed": return render_template('error.html', error="Analysis not yet completed") # Get the results results = job.results if not results or not results.arguments: return render_template('error.html', error="No arguments found", message="Arguments have not been generated for this analysis yet.") # Format job data for template formatted_job = { "job_id": job.job_uuid, "filename": job.original_filename, "perspective": job.perspective, "status": job.status, "case": job.case } return render_template('arguments.html', job_id=job_id, job=formatted_job, arguments=results.arguments) @app.route('/profile', methods=['GET', 'POST']) @login_required def user_profile(): """User profile page""" user_id = session.get('user_id') user = User.query.get(user_id) if request.method == 'POST': # Update user profile first_name = request.form.get('first_name') last_name = request.form.get('last_name') # Update password if provided password = request.form.get('password') if password: confirm_password = request.form.get('confirm_password') if password != confirm_password: flash('Passwords do not match', 'danger') return render_template('profile.html', user=user) user.set_password(password) # Update user user.first_name = first_name user.last_name = last_name db.session.commit() # Log the activity log_activity( user_id, 'profile_update', f"User updated their profile" ) flash('Profile updated successfully', 'success') return redirect(url_for('user_profile')) return render_template('profile.html', user=user) @app.route('/maintenance/cleanup', methods=['POST']) @admin_required def cleanup_old_jobs(): """Admin route to remove old analysis jobs to free up memory""" try: # Get age threshold in days (default to 7 days) days = int(request.form.get('days', 7)) threshold = datetime.utcnow() - timedelta(days=days) # Find old jobs old_jobs = AnalysisJob.query.filter(AnalysisJob.created_at < threshold).all() removed_count = 0 for job in old_jobs: # Clean up file if it exists if job.file_path and os.path.exists(job.file_path): try: os.remove(job.file_path) except Exception as e: print(f"Error removing file {job.file_path}: {e}") # Delete associated results if job.results: db.session.delete(job.results) # Delete job db.session.delete(job) removed_count += 1 db.session.commit() # Log the activity user_id = session.get('user_id') log_activity( user_id, 'cleanup', f"Administrator cleaned up {removed_count} old analysis jobs" ) return jsonify({ "success": True, "removed_count": removed_count, "remaining_count": AnalysisJob.query.count() }) except Exception as e: return jsonify({ "error": f"Cleanup failed: {str(e)}" }), 500 # Initialize database tables when app starts with app.app_context(): db.create_all() # Check if there are any users, if not create an admin if User.query.count() == 0: # Create an initial admin user admin = User( username="admin", email="admin@example.com", first_name="Admin", last_name="User", role="admin" ) admin.set_password("admin123") # Set a default password db.session.add(admin) db.session.commit() print("Created initial admin user: admin@example.com / admin123") print("IMPORTANT: Please change this password immediately after logging in!") if __name__ == '__main__': app.run(host='100.67.138.94', port=5000, ssl_context=('cert.pem', 'key.pem'))