Spaces:
Sleeping
Sleeping
| # libraries | |
| from flask import Flask, render_template, request, redirect, url_for, flash, session, send_from_directory | |
| import os | |
| import logging | |
| from utility.utils import extract_text_from_images, process_extracted_text, process_resume_data | |
| from backup.backup import NER_Model | |
| from paddleocr import PaddleOCR | |
| # Configure logging | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| handlers=[ | |
| logging.StreamHandler() # Remove FileHandler and log only to the console | |
| ] | |
| ) | |
| # Flask App | |
| app = Flask(__name__) | |
| app.secret_key = 'your_secret_key' | |
| def basename_filter(path): | |
| return os.path.basename(path) | |
| app.config['UPLOAD_FOLDER'] = 'uploads/' | |
| app.config['RESULT_FOLDER'] = 'results/' | |
| UPLOAD_FOLDER = 'static/uploads/' | |
| RESULT_FOLDER = 'static/results/' | |
| os.makedirs(UPLOAD_FOLDER, exist_ok=True) | |
| os.makedirs(RESULT_FOLDER, exist_ok=True) | |
| if not os.path.exists(app.config['UPLOAD_FOLDER']): | |
| os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True) | |
| if not os.path.exists(app.config['RESULT_FOLDER']): | |
| os.makedirs(app.config['RESULT_FOLDER'], exist_ok=True) | |
| # Set the PaddleOCR home directory to a writable location | |
| os.environ['PADDLEOCR_HOME'] = '/tmp/.paddleocr' | |
| # Check if PaddleOCR home directory is writable | |
| if not os.path.exists('/tmp/.paddleocr'): | |
| os.makedirs('/tmp/.paddleocr', exist_ok=True) | |
| logging.info("Created PaddleOCR home directory.") | |
| else: | |
| logging.info("PaddleOCR home directory exists.") | |
| def index(): | |
| uploaded_files = session.get('uploaded_files', []) | |
| logging.info(f"Accessed index page, uploaded files: {uploaded_files}") | |
| return render_template('index.html', uploaded_files=uploaded_files) | |
| def upload_file(): | |
| if 'files' not in request.files: | |
| flash('No file part') | |
| logging.warning("No file part found in the request") | |
| return redirect(request.url) | |
| files = request.files.getlist('files') | |
| if not files or all(file.filename == '' for file in files): | |
| flash('No selected files') | |
| logging.warning("No files selected for upload") | |
| return redirect(request.url) | |
| uploaded_files = session.get('uploaded_files', []) | |
| for file in files: | |
| if file: | |
| filename = file.filename | |
| file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename) | |
| file.save(file_path) | |
| print(f"file path --->{file_path}") | |
| uploaded_files.append(filename) | |
| logging.info(f"Uploaded file: {filename} at {file_path}") | |
| session['uploaded_files'] = uploaded_files | |
| flash('Files successfully uploaded') | |
| logging.info(f"Files successfully uploaded: {uploaded_files}") | |
| return process_file() | |
| def reset_upload(): | |
| """Reset the uploaded files and processed data in the session and filesystem.""" | |
| uploaded_files = session.get('uploaded_files', []) | |
| for filename in uploaded_files: | |
| file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename) | |
| if os.path.exists(file_path): | |
| os.remove(file_path) | |
| logging.info(f"Removed file: {filename}") | |
| else: | |
| logging.warning(f"File not found for removal: {file_path}") | |
| session.pop('uploaded_files', None) | |
| session.pop('processed_data', None) | |
| session.pop('processed_Img', None) | |
| flash('Reset successful. All files removed.') | |
| logging.info("Session and upload folder reset.") | |
| return redirect(url_for('index')) | |
| def process_file(): | |
| uploaded_files = session.get('uploaded_files', []) | |
| if not uploaded_files: | |
| flash('No files selected for processing') | |
| return redirect(url_for('index')) | |
| file_paths = [os.path.join(app.config['UPLOAD_FOLDER'], filename) for filename in uploaded_files] | |
| logging.info(f"Processing files: {file_paths}") | |
| extracted_text = {} | |
| LLMdata = {} | |
| processed_Img = {} | |
| try: | |
| # Primary: Groq VLM Single Pass | |
| LLMdata, extracted_text, processed_Img = extract_text_from_images(file_paths) | |
| LLMdata['meta'] = "Primary: Groq VLM" | |
| # Check if extracted data is essentially empty | |
| is_empty = all(len(v) == 0 for k, v in LLMdata.items() if k != 'meta') | |
| if is_empty: | |
| logging.info("Groq VLM returned empty data. Attempting backup...") | |
| raise ValueError("Empty data from Groq VLM") | |
| except Exception as e: | |
| logging.exception(f"Primary processing failed or returned empty: {e}") | |
| flash('Primary processing failed, using backup model...') | |
| # If extraction failed but we have some text/images, use them | |
| if not extracted_text: | |
| flash('Critical failure: Could not extract text from images.') | |
| return redirect(url_for('index')) | |
| try: | |
| # Backup: NER Model on extracted text | |
| text_for_ner = "" | |
| for path, text in extracted_text.items(): | |
| text_for_ner += str(text) + " " | |
| LLMdata = NER_Model(text_for_ner) | |
| LLMdata['meta'] = "Backup: Local NER" | |
| logging.info(f"NER model data: {LLMdata}") | |
| except Exception as backup_e: | |
| logging.exception(f"Backup processing failed: {backup_e}") | |
| flash('Processing failed completely.') | |
| return redirect(url_for('index')) | |
| # Common merge and finalization step | |
| try: | |
| cont_data = process_extracted_text(extracted_text) | |
| processed_data = process_resume_data(LLMdata, cont_data, extracted_text) | |
| session['processed_data'] = processed_data | |
| session['processed_Img'] = processed_Img | |
| logging.info(f"Final processed data: {processed_data}") | |
| return redirect(url_for('result')) | |
| except Exception as merge_e: | |
| logging.exception(f"Error during data merging: {merge_e}") | |
| flash('Error finalizing data extraction.') | |
| return redirect(url_for('index')) | |
| def result(): | |
| processed_data = session.get('processed_data', {}) | |
| processed_Img = session.get('processed_Img', {}) | |
| logging.info(f"Displaying results: Data - {processed_data}, Images - {processed_Img}") | |
| return render_template('result.html', data=processed_data, Img=processed_Img) | |
| def uploaded_file(filename): | |
| logging.info(f"Serving file: {filename}") | |
| return send_from_directory(app.config['UPLOAD_FOLDER'], filename) | |
| if __name__ == '__main__': | |
| logging.info("Starting Flask app") | |
| app.run(debug=True) | |