Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import cv2 | |
| import numpy as np | |
| from PIL import Image, ImageChops, ImageEnhance | |
| import hashlib | |
| import datetime | |
| import json | |
| import base64 | |
| from io import BytesIO | |
| import PyPDF2 | |
| import imagehash | |
| from reportlab.lib.pagesizes import letter | |
| from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Image as RLImage | |
| from reportlab.lib.styles import getSampleStyleSheet | |
| from reportlab.lib.units import inch | |
| import xml.etree.ElementTree as ET | |
| import os | |
| # Authentication credentials | |
| AUTH_USERNAME = os.getenv("AUTH_USERNAME", "forensic_admin") | |
| AUTH_PASSWORD = os.getenv("AUTH_PASSWORD", "SecureForensics2025!") | |
| class DocumentForensicsSystem: | |
| def __init__(self): | |
| self.case_id = 0 | |
| self.audit_log = [] | |
| def generate_case_id(self): | |
| self.case_id += 1 | |
| return f"DF-{datetime.datetime.now().strftime('%Y%m%d')}-{self.case_id:04d}" | |
| def compute_image_hash(self, img_array): | |
| return hashlib.sha256(img_array.tobytes()).hexdigest() | |
| def ela_analysis(self, image, quality=90): | |
| """Error Level Analysis for JPEG compression artifacts""" | |
| temp_filename = 'temp_ela.jpg' | |
| image.save(temp_filename, 'JPEG', quality=quality) | |
| temp_image = Image.open(temp_filename) | |
| ela_image = ImageChops.difference(image, temp_image) | |
| extrema = ela_image.getextrema() | |
| max_diff = max([ex[1] for ex[1] in extrema]) | |
| if max_diff == 0: | |
| max_diff = 1 | |
| scale = 255.0 / max_diff | |
| ela_image = ImageEnhance.Brightness(ela_image).enhance(scale) | |
| if os.path.exists(temp_filename): | |
| os.remove(temp_filename) | |
| return np.array(ela_image) | |
| def detect_copy_move(self, image): | |
| """Detect copy-move forgery using SIFT features""" | |
| gray = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2GRAY) | |
| sift = cv2.SIFT_create() | |
| keypoints, descriptors = sift.detectAndCompute(gray, None) | |
| if descriptors is None or len(keypoints) < 10: | |
| return False, 0, "Insufficient features detected" | |
| bf = cv2.BFMatcher(cv2.NORM_L2, crossCheck=False) | |
| matches = bf.knnMatch(descriptors, descriptors, k=2) | |
| good_matches = 0 | |
| for match_pair in matches: | |
| if len(match_pair) == 2: | |
| m, n = match_pair | |
| if m.distance < 0.7 * n.distance and m.queryIdx != m.trainIdx: | |
| pt1 = keypoints[m.queryIdx].pt | |
| pt2 = keypoints[m.trainIdx].pt | |
| distance = np.sqrt((pt1[0]-pt2[0])**2 + (pt1[1]-pt2[1])**2) | |
| if distance > 10: | |
| good_matches += 1 | |
| is_forged = good_matches > 15 | |
| confidence = min(100, (good_matches / 15) * 100) if is_forged else 0 | |
| return is_forged, confidence, f"Found {good_matches} suspicious duplicate regions" | |
| def analyze_signature(self, image): | |
| """Analyze signature for authenticity""" | |
| gray = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2GRAY) | |
| edges = cv2.Canny(gray, 50, 150) | |
| contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) | |
| if len(contours) > 0: | |
| stroke_consistency = np.std([cv2.contourArea(c) for c in contours]) | |
| pressure_variation = np.std(gray) | |
| authenticity_score = 100 - min(100, (stroke_consistency / 10 + pressure_variation / 5)) | |
| return { | |
| 'authentic_probability': max(0, authenticity_score), | |
| 'stroke_count': len(contours), | |
| 'pressure_variation': float(pressure_variation), | |
| 'consistency_score': float(100 - min(100, stroke_consistency / 10)) | |
| } | |
| return {'authentic_probability': 50, 'stroke_count': 0, 'pressure_variation': 0, 'consistency_score': 50} | |
| def analyze_document(self, image): | |
| if image is None: | |
| return None, None, "Please upload a document image", None, None, None | |
| case_id = self.generate_case_id() | |
| timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
| pil_image = Image.fromarray(image) if isinstance(image, np.ndarray) else image | |
| img_array = np.array(pil_image) | |
| doc_hash = self.compute_image_hash(img_array) | |
| # ELA Analysis | |
| ela_result = self.ela_analysis(pil_image) | |
| ela_mean = np.mean(ela_result) | |
| tampering_detected = ela_mean > 15 | |
| # Copy-Move Detection | |
| copy_move_detected, cm_confidence, cm_details = self.detect_copy_move(pil_image) | |
| # Signature Analysis | |
| sig_analysis = self.analyze_signature(pil_image) | |
| # Generate perceptual hash | |
| phash = str(imagehash.phash(pil_image)) | |
| dhash = str(imagehash.dhash(pil_image)) | |
| report_data = { | |
| 'case_id': case_id, | |
| 'timestamp': timestamp, | |
| 'document_hash': doc_hash, | |
| 'perceptual_hash': {'phash': phash, 'dhash': dhash}, | |
| 'ela_analysis': {'mean_error': float(ela_mean), 'tampering_detected': tampering_detected}, | |
| 'copy_move': {'detected': copy_move_detected, 'confidence': cm_confidence, 'details': cm_details}, | |
| 'signature_analysis': sig_analysis | |
| } | |
| report_md = f"""## DOCUMENT FORGERY ANALYSIS REPORT | |
| ### Case ID: {case_id} | |
| **Timestamp**: {timestamp} | |
| **Document Hash (SHA-256)**: {doc_hash[:32]}... | |
| **Perceptual Hashes**: pHash={phash}, dHash={dhash} | |
| ### 1. Error Level Analysis (ELA) | |
| - **Mean Error Level**: {ela_mean:.2f} | |
| - **Tampering Detected**: {'YES - SUSPICIOUS' if tampering_detected else 'NO - AUTHENTIC'} | |
| - **Analysis**: {'High compression inconsistency detected - possible manipulation' if tampering_detected else 'Consistent compression throughout document'} | |
| ### 2. Copy-Move Forgery Detection | |
| - **Status**: {'DETECTED' if copy_move_detected else 'NOT DETECTED'} | |
| - **Confidence**: {cm_confidence:.1f}% | |
| - **Details**: {cm_details} | |
| ### 3. Signature Verification | |
| - **Authenticity Probability**: {sig_analysis['authentic_probability']:.1f}% | |
| - **Stroke Count**: {sig_analysis['stroke_count']} | |
| - **Pressure Variation**: {sig_analysis['pressure_variation']:.2f} | |
| - **Consistency Score**: {sig_analysis['consistency_score']:.1f}% | |
| - **Verdict**: {'AUTHENTIC SIGNATURE' if sig_analysis['authentic_probability'] > 70 else 'SUSPICIOUS - REQUIRES MANUAL REVIEW'} | |
| ### Final Assessment | |
| **Overall Risk Level**: {'HIGH' if tampering_detected or copy_move_detected else 'LOW'} | |
| **Court Admissibility**: {'QUESTIONABLE - Further verification required' if tampering_detected or copy_move_detected else 'ACCEPTABLE - Document appears authentic'} | |
| ### Chain of Custody | |
| Cryptographic hash recorded | |
| Case ID assigned | |
| Audit trail created | |
| Timestamp certified | |
| """ | |
| pdf_bytes = self.export_pdf(report_data, report_md) | |
| json_bytes = self.export_json(report_data) | |
| xml_bytes = self.export_xml(report_data) | |
| return pil_image, Image.fromarray(ela_result), report_md, pdf_bytes, json_bytes, xml_bytes | |
| def export_pdf(self, data, markdown_report): | |
| buffer = BytesIO() | |
| doc = SimpleDocTemplate(buffer, pagesize=letter) | |
| styles = getSampleStyleSheet() | |
| story = [] | |
| story.append(Paragraph("DOCUMENT FORGERY ANALYSIS REPORT", styles['Title'])) | |
| story.append(Spacer(1, 0.2*inch)) | |
| story.append(Paragraph(f"Case ID: {data['case_id']}", styles['Heading2'])) | |
| story.append(Paragraph(f"Timestamp: {data['timestamp']}", styles['Normal'])) | |
| story.append(Spacer(1, 0.3*inch)) | |
| for line in markdown_report.split('\n'): | |
| if line.strip(): | |
| if line.startswith('###'): | |
| story.append(Paragraph(line.replace('#', '').strip(), styles['Heading3'])) | |
| elif line.startswith('**'): | |
| story.append(Paragraph(line.replace('**', '').strip(), styles['Normal'])) | |
| else: | |
| story.append(Paragraph(line.strip(), styles['Normal'])) | |
| story.append(Spacer(1, 0.1*inch)) | |
| doc.build(story) | |
| return buffer.getvalue() | |
| def export_json(self, data): | |
| return json.dumps(data, indent=2).encode('utf-8') | |
| def export_xml(self, data): | |
| root = ET.Element('ForensicReport') | |
| for key, value in data.items(): | |
| child = ET.SubElement(root, key) | |
| child.text = str(value) | |
| tree = ET.ElementTree(root) | |
| buffer = BytesIO() | |
| tree.write(buffer, encoding='utf-8', xml_declaration=True) | |
| return buffer.getvalue() | |
| forensic_sys = DocumentForensicsSystem() | |
| with gr.Blocks(title="Document Forgery Detection Lab", theme=gr.themes.Soft()) as demo: | |
| gr.Markdown("# Document Forgery Detection Lab") | |
| gr.Markdown("### Signature Verification | Copy-Move Detection | ELA Analysis | Export Reports (PDF/JSON/XML)") | |
| with gr.Tabs(): | |
| with gr.Tab(" Document Analysis"): | |
| with gr.Row(): | |
| with gr.Column(): | |
| doc_input = gr.Image(label="Upload Document/Signature", type="numpy") | |
| analyze_btn = gr.Button("Analyze Document", variant="primary") | |
| with gr.Column(): | |
| doc_output = gr.Image(label="Original Document") | |
| ela_output = gr.Image(label="ELA Analysis Result") | |
| report_output = gr.Markdown(label="Forensic Analysis Report") | |
| with gr.Row(): | |
| pdf_download = gr.File(label="Download PDF Report") | |
| json_download = gr.File(label="Download JSON Data") | |
| xml_download = gr.File(label="Download XML Report") | |
| analyze_btn.click( | |
| forensic_sys.analyze_document, | |
| doc_input, | |
| [doc_output, ela_output, report_output, pdf_download, json_download, xml_download] | |
| ) | |
| gr.Markdown("---") | |
| gr.Markdown(" Technologies: ELA | SIFT | OpenCV | SHA-256 | PDF/JSON/XML Export | Authenticated Access") | |
| if __name__ == "__main__": | |
| demo.launch(auth=(AUTH_USERNAME, AUTH_PASSWORD)) |