from flask import Flask, request, jsonify, render_template, send_file import os from ultralytics import YOLO from easyocr import Reader import zipfile import pandas as pd import cv2 from fuzzywuzzy import fuzz, process import re import numpy as np import json import sqlite3 app = Flask(__name__) app.config['UPLOAD_FOLDER'] = 'uploads/' app.config['ALLOWED_EXTENSIONS'] = {'zip', 'xlsx'} app.config['RESULTS_FOLDER'] = 'results/' app.config['DATABASE'] = 'results_ack.db' classifier = YOLO("./models/classification.pt") detector = YOLO("./models/detection.pt") reader = Reader(['en']) def process_image(image_path): if classifier.predict(image_path)[0].probs.numpy().top1 == 0: fields = detector(image_path) image = cv2.imread(image_path) extracted_data = {} for field in fields[0].boxes.data.tolist(): x1, y1, x2, y2, confidence, class_id = map(int, field[:6]) field_class = detector.names[class_id] cropped_roi = image[y1:y2, x1:x2] gray_roi = cv2.cvtColor(cropped_roi, cv2.COLOR_BGR2GRAY) text = reader.readtext(gray_roi, detail=0) extracted_data[field_class] = ' '.join(text) return extracted_data return None # Helper Functions def normalize_text(text): if not text: return "text empty" text = re.sub(r"[^\w\s]", "", text) return " ".join(text.split()).lower() def calculate_match_score(input_value, extracted_value): if pd.isna(input_value) or pd.isna(extracted_value): return 0 return fuzz.ratio(str(input_value), str(extracted_value)) def name_match(input_name, extracted_name): if extracted_name is None: return False input_name = normalize_text(input_name) extracted_name = normalize_text(extracted_name) if input_name == extracted_name: return True input_parts = input_name.split() extracted_parts = extracted_name.split() if sorted(input_parts) == sorted(extracted_parts): return True if len(input_parts) == 2 and len(extracted_parts) == 3: if input_parts[0] == extracted_parts[0] and input_parts[1] == extracted_parts[2]: return True if len(input_parts) == 3 and len(extracted_parts) == 2: if extracted_parts[0] == input_parts[0] and extracted_parts[1] == input_parts[2]: return True for part in input_parts: if part not in extracted_parts: return False return True def address_match(input_address, extracted_address): print(input_address, extracted_address) if input_address is None or extracted_address is None: return False, 0.0, {} # Handle input_address if it's a Series if isinstance(input_address, pd.Series): input_address = input_address.to_dict() print(extracted_address) extracted_address = normalize_text(extracted_address) final_score = 0 print(extracted_address) weights = { "State": 0.2, "Landmark": 0.2, "Premise Building Name": 0.2, "City":0.2, "Street Road Name":0.1, "Floor Number": 0.05, "House Flat Number": 0.05 } tokens = extracted_address.split(" ") # Component matching logic for field, weight in weights.items(): input_value = input_address.get(field, "") match_score = fuzz.token_set_ratio(normalize_text(input_value), extracted_address) if input_value else 0 input_address[field + " Match Score"] = match_score final_score += match_score * weight pincode_score = process.extractOne(input_address.get("PINCODE"), tokens)[1] input_address['PINCODE Match Score'] = pincode_score pincode_matched = True if input_address['PINCODE Match Score'] == 100 else False return final_score >= 70 and pincode_matched, final_score, input_address def compare_data(input_data, json_data): excel_data = input_data.copy() for idx, row in excel_data.iterrows(): serial_no = row.get("SrNo") uid = row.get("UID") extracted = json_data.get(serial_no) if extracted: extracted_uid = extracted.get("uid", "").replace(" ", "") extracted_name = extracted.get("name", "") extracted_address = extracted.get("address", "") row['Extracted UID'] = extracted_uid row['Extracted Name'] = extracted_name row['Extracted Address'] = extracted_address # UID Match uid_match = uid == extracted_uid uid_score = 100 if uid_match else 0 row['UID Match Score'] = uid_score # Name Match name_match_result = name_match(row.get("Name"), extracted_name) name_score = calculate_match_score(row.get("Name"), extracted_name) row['Name Match Score'] = name_score row['Name Match Percentage'] = name_score # Address Match address_match_result, address_score, partial_scores = address_match(row, extracted_address) if partial_scores: row['House Flat Number Match Score'] = partial_scores['House Flat Number Match Score'] row['Street Road Name Match Score'] = partial_scores['Street Road Name Match Score'] row['City Match Score'] = partial_scores['City Match Score'] row['Floor Number Match Score'] = partial_scores['Floor Number Match Score'] row['Premise Building Name Match Score'] = partial_scores['Premise Building Name Match Score'] row['Landmark Match Score'] = partial_scores['Landmark Match Score'] row['State Match Score'] = partial_scores['State Match Score'] row['Final Address Match'] = address_match_result row['Final Address Match Score'] = address_score row['PINCODE Match Score'] = partial_scores['PINCODE Match Score'] # Final Match overall_match = uid_match and name_match_result and address_match_result row['Overall Match'] = overall_match if overall_match: row['Final Remarks'] = "All matched" elif not uid_match: row['Final Remarks'] = "UID mismatch" elif not name_match_result: row['Final Remarks'] = "Name mismatch" elif not address_match_result: row['Final Remarks'] = "Address mismatch" else: if extracted_address is None: row['Final Remarks'] = "Address missing in aadhar" elif extracted_name is None: row['Final Remarks'] = "Name missing in aadhar" else: row['Final Remarks'] = "Non Aadhar" row["Document Type"] = "Aadhaar" if overall_match else "Non-Aadhaar" else: row.replace(float('nan'), 0) row['Final Remarks'] = "Non Aadhar" row['Document Type'] = "Non Aadhar" excel_data.loc[idx] = row return excel_data def create_visualizations(comparison_results): visualization_data = {} # TODO: Implement your visualization logic here # Example: Frequency of 'Final Remarks' (for a bar chart) visualization_data['final_remarks_frequency'] = { 'labels': comparison_results['Final Remarks'].value_counts().index.tolist(), 'values': comparison_results['Final Remarks'].value_counts().values.tolist() } # 2. Document Type Proportion (Pie Chart) visualization_data['document_type_proportion'] = { 'labels': comparison_results['Document Type'].value_counts().index.tolist(), 'values': comparison_results['Document Type'].value_counts().values.tolist() } # 3. Accepted vs. Rejected Proportion (Pie Chart) visualization_data['accepted_rejected_proportion'] = { 'labels': comparison_results['Accepted/Rejected'].value_counts().index.tolist(), 'values': comparison_results['Accepted/Rejected'].value_counts().values.tolist() } # 4. UID Match Score Distribution (Histogram) visualization_data['uid_match_score_distribution'] = { 'values': comparison_results['UID Match Score'].tolist() } # 5. Name Match Score Distribution (Histogram) visualization_data['name_match_score_distribution'] = { 'values': comparison_results['Name Match Score'].tolist() } # 6. Final Address Match Score Distribution (Histogram) visualization_data['address_match_score_distribution'] = { 'values': comparison_results['Final Address Match Score'].tolist() } return visualization_data def create_database_and_table(): """Creates the database and table if they don't exist.""" conn = sqlite3.connect(app.config['DATABASE']) cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS results ( id INTEGER PRIMARY KEY AUTOINCREMENT, SrNo TEXT, DocumentType TEXT, AcceptedRejected TEXT, FinalRemarks TEXT ) ''') conn.commit() conn.close() def save_results_to_database(results): """Saves the provided results to the SQLite database.""" conn = sqlite3.connect(app.config['DATABASE']) cursor = conn.cursor() for result in results: cursor.execute(''' INSERT INTO results (SrNo, DocumentType, AcceptedRejected, FinalRemarks) VALUES (?, ?, ?, ?) ''', ( result.get('SrNo', ''), result.get('Document Type', ''), result.get('Accepted/Rejected', ''), result.get('Final Remarks', '') )) conn.commit() conn.close() @app.route('/download', methods=['GET']) def download_results(): file_path = os.path.join(app.config['RESULTS_FOLDER'], 'results.xlsx') return send_file(file_path, as_attachment=True) @app.route('/') def home(): return render_template('index.html') @app.route('/services', methods=['GET']) def services(): return render_template('services.html') @app.route('/about', methods=['GET']) def about(): return render_template('about.html') @app.route('/contact', methods=['GET']) def contact(): return render_template('contact.html') @app.route('/upload', methods=['POST']) def upload_files(): if 'zipfile' in request.files and 'excelfile' in request.files: zip_file = request.files['zipfile'] excel_file = request.files['excelfile'] # Save files zip_path = os.path.join(app.config['UPLOAD_FOLDER'], zip_file.filename) excel_path = os.path.join(app.config['UPLOAD_FOLDER'], excel_file.filename) zip_file.save(zip_path) excel_file.save(excel_path) # Unzip and process images with zipfile.ZipFile(zip_path, 'r') as zip_ref: zip_ref.extractall(app.config['UPLOAD_FOLDER']) image_paths = [os.path.join(app.config['UPLOAD_FOLDER'], f) for f in os.listdir(app.config['UPLOAD_FOLDER']) if f.endswith(('.jpg', '.png'))] processed_results = {} for image_path in image_paths: file_name = os.path.basename(image_path) key = file_name.split('.')[0][:3] if key not in processed_results: # Check if key already exists extracted_data = process_image(image_path) if extracted_data: processed_results[key] = extracted_data # Read Excel and compare data df = pd.read_excel(excel_path) df = df.astype('str') comparison_results = compare_data(df, processed_results) comparison_results['Accepted/Rejected'] = np.where(comparison_results['Final Remarks'] == 'All matched', 'Accepted', 'Rejected') # Save results to a new Excel file results_df = pd.DataFrame(comparison_results) os.makedirs(app.config['RESULTS_FOLDER'], exist_ok=True) results_file_path = os.path.join(app.config['RESULTS_FOLDER'], 'results.xlsx') results_df.to_excel(results_file_path, index=False) visualization_data = create_visualizations(comparison_results) create_database_and_table() # Ensure database and table exist save_results_to_database(comparison_results[ ['SrNo', 'Document Type', 'Accepted/Rejected', 'Final Remarks'] ].to_dict(orient='records')) return jsonify({"message": "Files processed successfully!", "results": comparison_results[ ['SrNo', 'Document Type', 'Accepted/Rejected', 'Final Remarks'] ].to_dict(orient='records'), "visualization_data": visualization_data}) return jsonify({"error": "Both files are required."}), 400 if __name__ == '__main__': app.run(debug=True, host="0.0.0.0", port=int(os.environ.get("PORT", 7680)))