Spaces:
Build error
Build error
| import streamlit as st | |
| from deepface import DeepFace | |
| import easyocr | |
| from PIL import Image | |
| import io | |
| import pandas as pd | |
| import sqlite3 | |
| import requests | |
| import random # for simulate QTE | |
| conn = sqlite3.connect('kyc_logs.db') | |
| c = conn.cursor() | |
| c.execute('''CREATE TABLE IF NOT EXISTS verifications | |
| (id INTEGER PRIMARY KEY, name TEXT, phone TEXT, address TEXT, trs REAL, status TEXT, qte_details TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP)''') | |
| conn.commit() | |
| reader = easyocr.Reader(['en'], gpu=False) | |
| st.set_page_config(page_title="Bright KYC AI", layout="wide") | |
| st.title("🔒 Bright KYC AI – Strict Enforcer with Quantum Trust Echo 🚀") | |
| st.write("Ghana-Focused | No Shortcuts | Founder: Gov") | |
| tab1, tab2 = st.tabs(["Customer Onboarding", "Admin Dashboard"]) | |
| with tab1: | |
| st.header("Strict Verification Flow") | |
| full_name = st.text_input("Full Name") | |
| phone = st.text_input("Phone Number (+233...)") | |
| gps_address = st.text_input("GhanaPost Digital Address (e.g., GA-123-4567)") | |
| col1, col2 = st.columns(2) | |
| with col1: id_file = st.file_uploader("Upload ID (Ghana Card/Passport/Voter’s ID)", type=['jpg', 'jpeg', 'png']) | |
| with col2: selfie_file = st.file_uploader("Upload Live Selfie", type=['jpg', 'jpeg', 'png']) | |
| if st.button("Run Strict Verification") and all([full_name, phone, gps_address, id_file, selfie_file]): | |
| with st.spinner("Enforcing + Quantum Trust Echo... 🔍"): | |
| id_bytes = id_file.read() | |
| selfie_bytes = selfie_file.read() | |
| id_img = Image.open(io.BytesIO(id_bytes)) | |
| selfie_img = Image.open(io.BytesIO(selfie_bytes)) | |
| col1, col2 = st.columns(2) | |
| with col1: st.image(id_img, caption="ID") | |
| with col2: st.image(selfie_img, caption="Selfie") | |
| # Step 2: ID Verification (OCR) - Simple extract for name match | |
| extracted = reader.readtext(id_bytes) | |
| full_text = " ".join([text for _, text, _ in extracted]).lower() | |
| if full_name.lower() not in full_text: | |
| st.warning("Name no match ID text! Flag ⚠️") | |
| # Step 3: Address Confirmation | |
| try: | |
| resp = requests.get(f"https://ghanapostgps.sperixlabs.org/get-address/{gps_address}") | |
| address_valid = resp.status_code == 200 and resp.json().get("status") | |
| address_score = 20 if address_valid else 0 | |
| st.success("Address Valid ✅") if address_valid else st.warning("Invalid Address ⚠️") | |
| except: | |
| address_score = 10 | |
| st.info("Address check skipped") | |
| # Step 4: Biometric Capture | |
| try: | |
| live_result = DeepFace.extract_faces(img_path=selfie_bytes, anti_spoofing=True) | |
| is_live = live_result[0]['is_real'] | |
| live_score = live_result[0]['antispoof_score'] | |
| if not is_live: | |
| st.error("SPOOF DETECTED → BLOCKED 🚫") | |
| status = "BLOCKED" | |
| trs = 0 | |
| else: | |
| verify = DeepFace.verify(img1_path=id_bytes, img2_path=selfie_bytes) | |
| confidence = (1 - verify["distance"]) * 100 | |
| except Exception as e: | |
| confidence = 0 | |
| live_score = 0 | |
| status = "ERROR" | |
| st.error(f"Biometric error: {e}") | |
| # Step 5: Quantum Trust Echo (QTE) Simulation - Your Brilliant Feature! | |
| # Temporal Consistency: Check duplicate in DB (identity exist before?) | |
| c.execute("SELECT COUNT(*) FROM verifications WHERE name=? OR phone=?", (full_name, phone)) | |
| duplicate_count = c.fetchone()[0] | |
| temporal_score = 30 if duplicate_count == 0 else -20 # New good, duplicate bad | |
| # Digital Gravity: How many systems know am? (Simulate with phone check + random) | |
| gravity_score = 25 if phone.startswith("+233") and len(phone) == 13 else -10 | |
| gravity_score += random.randint(-5, 5) # Simulate other systems | |
| # Echo Stability: Consistency across devices/locations (Simulate with address + name) | |
| stability_score = 20 if len(gps_address) > 5 and address_score > 0 else -15 | |
| qte_details = f"Temporal: {temporal_score} | Gravity: {gravity_score} | Stability: {stability_score}" | |
| # Final TRS with QTE + Biometric | |
| biometric_base = confidence * 0.5 if 'confidence' in locals() else 0 | |
| trs = biometric_base + address_score + (live_score * 100 * 0.2) + temporal_score + gravity_score + stability_score | |
| if trs >= 70: | |
| status = "AUTO-APPROVED ✅" | |
| elif trs >= 40: | |
| status = "MANUAL REVIEW ⚠️" | |
| else: | |
| status = "BLOCKED 🚫" | |
| # Step 6: Risk Scoring (Simple, add more later) | |
| risk_flag = duplicate_count > 0 or confidence < 50 | |
| if risk_flag: | |
| status = "FLAGGED FOR REVIEW ⚠️" if "APPROVED" in status else status | |
| trs -= 10 | |
| st.write(f"**Trust Resonance Score (TRS): {trs:.2f}/100**") | |
| st.write(f"QTE Breakdown: {qte_details}") | |
| st.write(f"Final Decision: {status}") | |
| # Save to DB | |
| c.execute("INSERT INTO verifications (name, phone, address, trs, status, qte_details) VALUES (?, ?, ?, ?, ?, ?)", | |
| (full_name, phone, gps_address, trs, status, qte_details)) | |
| conn.commit() | |
| with tab2: | |
| st.header("Admin & Compliance Dashboard") | |
| df = pd.read_sql_query("SELECT * FROM verifications ORDER BY timestamp DESC", conn) | |
| st.dataframe(df, use_container_width=True) | |
| st.subheader("Manual Override") | |
| selected_id = st.number_input("Enter Verification ID to Override", min_value=1, step=1) | |
| new_status = st.selectbox("New Status", ["APPROVED ✅", "MANUAL REVIEW ⚠️", "BLOCKED 🚫"]) | |
| if st.button("Apply Override"): | |
| if selected_id: | |
| c.execute("UPDATE verifications SET status=? WHERE id=?", (new_status, selected_id)) | |
| conn.commit() | |
| st.success(f"ID {selected_id} updated to {new_status}") | |
| st.experimental_rerun() # Refresh table | |
| else: | |
| st.warning("Enter valid ID") | |
| st.write(f"Total Verifications: {len(df)} | Flagged/Blocked: {len(df[df['status'].str.contains('REVIEW|BLOCKED|ERROR')])}") |