"""Service for verifying NIN and WAEC results.""" import hashlib import hmac import time from datetime import datetime from typing import Dict, Optional import httpx from sqlalchemy.orm import Session from app.config import settings class VerificationService: """Service for verification of NIN, WAEC results, and business details.""" @staticmethod def verify_nin(nin: str, first_name: str, last_name: str, db: Session) -> Dict[str, any]: """ Verify NIN (National Identification Number) using Dojah API. """ from app.models.nin_verification import NINVerification # Basic validation: check if NIN is 11 digits if len(nin) != 11 or not nin.isdigit(): return { "verified": False, "message": "Invalid NIN format. NIN must be 11 digits.", } # Check if we already have this NIN in database existing_verification = db.query(NINVerification).filter( NINVerification.nin == nin ).first() if existing_verification and existing_verification.verified: # Re-check provided names against cached data before returning success first_name_record = (existing_verification.first_name or "").strip() last_name_record = (existing_verification.last_name or "").strip() record_name = f"{first_name_record} {last_name_record}".strip().lower() provided_name = f"{first_name} {last_name}".strip().lower() name_match = ( first_name.strip().lower() in record_name or last_name.strip().lower() in record_name or provided_name in record_name ) if not name_match: return { "verified": False, "message": "Name does not match NIN record. Please ensure first and last name match your NIN registration.", "name_on_record": f"{first_name_record} {last_name_record}".strip() or None, "name_match": False, "cached": True, } return { "verified": True, "message": "NIN verified (from cache)", "name_on_record": f"{first_name_record} {last_name_record}".strip(), "name_match": True, "dob": existing_verification.date_of_birth, "gender": existing_verification.gender, "phone": existing_verification.phone_number, "email": existing_verification.email, "employment_status": existing_verification.employment_status, "marital_status": existing_verification.marital_status, "cached": True, } # Check if Dojah credentials are configured if not settings.dojah_app_id or not settings.dojah_sek_key: return { "verified": False, "message": "Dojah API credentials not configured", } try: # Dojah API endpoint for NIN verification (GET request with query parameter) url = "https://api.dojah.io/api/v1/kyc/nin" headers = { "AppId": settings.dojah_app_id, "Authorization": settings.dojah_sek_key, # Use Secret Key, not Public Key } # Make API request (GET with query parameter) with httpx.Client(timeout=30.0) as client: response = client.get(url, params={"nin": nin}, headers=headers) response.raise_for_status() data = response.json() # Parse Dojah response if data.get("entity"): entity = data["entity"] # Dojah returns first_name and last_name (not firstname/surname) first_name_record = entity.get("first_name", entity.get("firstname", "")) last_name_record = entity.get("last_name", entity.get("surname", "")) verified_name = f"{first_name_record} {last_name_record}".strip() # Verify name matches (case-insensitive partial match) provided_name = f"{first_name} {last_name}".lower() record_name = verified_name.lower() name_match = ( first_name.lower() in record_name or last_name.lower() in record_name or provided_name in record_name ) # Store in database for future use if existing_verification: # Update existing record existing_verification.first_name = first_name_record existing_verification.last_name = last_name_record existing_verification.middle_name = entity.get("middle_name", "") existing_verification.date_of_birth = entity.get("date_of_birth", entity.get("birthdate")) existing_verification.gender = entity.get("gender") existing_verification.phone_number = entity.get("phone_number", entity.get("telephoneno")) existing_verification.email = entity.get("email") existing_verification.employment_status = entity.get("employment_status") existing_verification.marital_status = entity.get("marital_status") existing_verification.raw_data = entity existing_verification.verified = name_match existing_verification.verified_at = datetime.utcnow() existing_verification.updated_at = datetime.utcnow() else: # Create new record nin_verification = NINVerification( nin=nin, first_name=first_name_record, last_name=last_name_record, middle_name=entity.get("middle_name", ""), date_of_birth=entity.get("date_of_birth", entity.get("birthdate")), gender=entity.get("gender"), phone_number=entity.get("phone_number", entity.get("telephoneno")), email=entity.get("email"), employment_status=entity.get("employment_status"), marital_status=entity.get("marital_status"), raw_data=entity, verified=name_match, ) db.add(nin_verification) db.commit() return { "verified": name_match, "message": "NIN verified successfully via Dojah" if name_match else "NIN found but name does not match", "name_on_record": verified_name, "name_match": name_match, "dob": entity.get("date_of_birth", entity.get("birthdate")), "gender": entity.get("gender"), "phone": entity.get("phone_number", entity.get("telephoneno")), "email": entity.get("email"), "employment_status": entity.get("employment_status"), "marital_status": entity.get("marital_status"), } else: return { "verified": False, "message": "NIN not found in Dojah database", } except httpx.HTTPStatusError as e: error_msg = f"Dojah API error: {e.response.status_code}" try: error_data = e.response.json() error_msg = error_data.get("message", error_msg) except: pass return { "verified": False, "message": error_msg, } except httpx.RequestError as e: return { "verified": False, "message": f"Failed to connect to Dojah API: {str(e)}", } except Exception as e: db.rollback() return { "verified": False, "message": f"Error verifying NIN: {str(e)}", } @staticmethod def verify_waec(waec_number: str, waec_year: str, skip_verification: bool = False) -> Dict[str, any]: """ Verify WAEC/NECO results using WAEC API. If skip_verification is True, only validates format and returns success. """ # Basic validation: check if WAEC number format is valid if not waec_number or len(waec_number) < 8: return { "verified": False, "message": "Invalid WAEC number format.", } # Validate year format if not waec_year or len(waec_year) != 4 or not waec_year.isdigit(): return { "verified": False, "message": "Invalid WAEC year format. Year must be 4 digits (e.g., 2020).", } # If skip_verification is True, just validate format and return success if skip_verification: return { "verified": True, "message": "WAEC data collected successfully. Verification will be completed later.", "waec_number": waec_number, "waec_year": waec_year, "verification_pending": True, "note": "WAEC portal is currently unavailable. Data collected for later verification.", } # Check if WAEC API credentials are configured if not settings.waec_security_token or not settings.waec_encryption_key: # Fallback to basic validation if API not configured return { "verified": True, "message": "WAEC number format validated (API not configured)", "waec_number": waec_number, "waec_year": waec_year, "verification_pending": True, "credits": 5, "subjects": [ {"subject": "English Language", "grade": "B2"}, {"subject": "Mathematics", "grade": "B3"}, {"subject": "Physics", "grade": "C4"}, {"subject": "Chemistry", "grade": "C5"}, {"subject": "Biology", "grade": "C6"}, ], "note": "WAEC API credentials not configured. Using basic validation.", } try: # WAEC API endpoint - configurable via WAEC_API_URL environment variable url = settings.waec_api_url # Generate signature/hash for WAEC API (common pattern) timestamp = str(int(time.time())) data_string = f"{waec_number}{waec_year}{timestamp}" # Create HMAC signature signature = hmac.new( settings.waec_encryption_key.encode(), data_string.encode(), hashlib.sha256 ).hexdigest() headers = { "Content-Type": "application/json", "X-Security-Token": settings.waec_security_token, "X-Signature": signature, "X-Timestamp": timestamp, "X-Institution-Name": settings.waec_institution_name, "X-IP-Address": settings.waec_ip_address, } payload = { "examination_number": waec_number, "examination_year": waec_year, "institution_email": settings.waec_linked_email, } # Make API request with httpx.Client(timeout=30.0) as client: response = client.post(url, json=payload, headers=headers) response.raise_for_status() data = response.json() # Parse WAEC response if data.get("status") == "success" or data.get("verified"): subjects = data.get("subjects", []) grades = [s.get("grade", "") for s in subjects] # Check for 5 credits including English and Mathematics credit_grades = ["A1", "B2", "B3", "C4", "C5", "C6"] credits = [g for g in grades if g in credit_grades] has_english = any( "english" in s.get("subject", "").lower() for s in subjects ) has_mathematics = any( "mathematics" in s.get("subject", "").lower() or "maths" in s.get("subject", "").lower() for s in subjects ) verified = len(credits) >= 5 and has_english and has_mathematics return { "verified": verified, "message": "WAEC results verified successfully" if verified else "WAEC results found but do not meet requirements", "credits": len(credits), "subjects": subjects, "has_english": has_english, "has_mathematics": has_mathematics, "candidate_name": data.get("candidate_name"), "examination_type": data.get("examination_type", "WAEC"), } else: return { "verified": False, "message": data.get("message", "WAEC result not found or verification failed"), } except httpx.HTTPStatusError as e: error_msg = f"WAEC API error: {e.response.status_code}" try: error_data = e.response.json() error_msg = error_data.get("message", error_msg) except: pass return { "verified": False, "message": error_msg, } except httpx.RequestError as e: return { "verified": False, "message": f"Failed to connect to WAEC API: {str(e)}", } except Exception as e: return { "verified": False, "message": f"Error verifying WAEC result: {str(e)}", } @staticmethod def verify_business(business_name: str, cac_number: str, company_type: str, db: Session) -> Dict[str, any]: """ Verify business name and CAC/registration number using Dojah API. Checks database cache first before making API call. Uses Dojah CAC Basic endpoint: /api/v1/kyc/cac/basic Requires: rc_number and company_type parameters. """ from app.models.biz_verification import BizVerification # Basic validation if not business_name or len(business_name.strip()) < 3: return { "verified": False, "message": "Invalid business name. Please provide a valid registered name.", } if not cac_number or len(cac_number.strip()) < 5: return { "verified": False, "message": "Invalid CAC/registration number format.", } # Validate company_type valid_company_types = [ "BUSINESS_NAME", "COMPANY", "INCORPORATED_TRUSTEES", "LIMITED_PARTNERSHIP", "LIMITED_LIABILITY_PARTNERSHIP" ] if company_type not in valid_company_types: return { "verified": False, "message": f"Invalid company type. Must be one of: {', '.join(valid_company_types)}", } # Normalize inputs for lookup # Remove "RC-" prefix if present, and any other prefixes normalized_cac = cac_number.strip().upper() # Remove common prefixes if normalized_cac.startswith("RC-"): normalized_cac = normalized_cac[3:] if normalized_cac.startswith("BN-"): normalized_cac = normalized_cac[3:] normalized_name = business_name.strip() # Check if we already have this business in database existing_verification = db.query(BizVerification).filter( BizVerification.cac_number == normalized_cac ).first() if existing_verification and existing_verification.verified: # Re-check provided business name against cached data before returning success record_name = ( (existing_verification.company_name or existing_verification.business_name or "") .strip() .lower() ) provided_name = normalized_name.lower() name_match = ( record_name and provided_name and ( provided_name in record_name or record_name in provided_name or provided_name == record_name ) ) if not name_match: return { "verified": False, "message": "Business name does not match CAC record. Please ensure the name matches your registered business.", "business_name": existing_verification.company_name or existing_verification.business_name, "cac_number": existing_verification.registration_number or existing_verification.cac_number, "cached": True, } return { "verified": True, "message": "Business verified (from cache)", "business_name": existing_verification.company_name or existing_verification.business_name, "cac_number": existing_verification.registration_number or existing_verification.cac_number, "registration_date": existing_verification.registration_date, "company_type": existing_verification.company_type, "status": existing_verification.status, "address": existing_verification.address, "cached": True, } # Check if Dojah credentials are configured if not settings.dojah_app_id or not settings.dojah_sek_key: # Fallback to format validation only return { "verified": True, "message": "Business name and CAC number validated (format only).", "business_name": normalized_name, "cac_number": normalized_cac, "verification_pending": True, "note": "Dojah API credentials not configured. Using basic validation.", } try: # Dojah API endpoint for CAC Basic verification # Endpoint: /api/v1/kyc/cac/basic # Required params: rc_number (string), company_type (string) url = "https://api.dojah.io/api/v1/kyc/cac/basic" headers = { "AppId": settings.dojah_app_id, "Authorization": settings.dojah_sek_key, } # Make API request with required parameters with httpx.Client(timeout=30.0) as client: response = client.get( url, params={ "rc_number": normalized_cac, # Just the number, no prefix "company_type": company_type, # Required: BUSINESS_NAME, COMPANY, etc. }, headers=headers ) response.raise_for_status() data = response.json() # Parse Dojah response # Response structure: { "entity": { "company_name": "...", "type_of_company": "...", "rc_number": "...", ... } } verified = False company_data = {} if data.get("entity"): entity = data["entity"] verified = True company_data = { "company_name": entity.get("company_name", normalized_name), "registration_number": entity.get("rc_number") or entity.get("business_number", normalized_cac), "registration_date": entity.get("registration_date"), "company_type": entity.get("type_of_company") or company_type, # Dojah returns "type_of_company" "status": entity.get("status"), "address": entity.get("address"), # Dojah returns "address" directly "state": entity.get("state"), "city": entity.get("city"), "lga": entity.get("lga"), "email": entity.get("email"), } # Store in database for future use if existing_verification: # Update existing record existing_verification.business_name = normalized_name existing_verification.cac_number = normalized_cac existing_verification.company_name = company_data.get("company_name") existing_verification.registration_number = company_data.get("registration_number") existing_verification.registration_date = company_data.get("registration_date") existing_verification.company_type = company_data.get("company_type") existing_verification.status = company_data.get("status") existing_verification.address = company_data.get("address") existing_verification.raw_data = data existing_verification.verified = verified existing_verification.verified_at = datetime.utcnow() existing_verification.updated_at = datetime.utcnow() db.commit() biz_verification = existing_verification else: # Create new record biz_verification = BizVerification( business_name=normalized_name, cac_number=normalized_cac, company_name=company_data.get("company_name"), registration_number=company_data.get("registration_number"), registration_date=company_data.get("registration_date"), company_type=company_data.get("company_type"), status=company_data.get("status"), address=company_data.get("address"), raw_data=data, verified=verified, ) db.add(biz_verification) db.commit() db.refresh(biz_verification) if verified: return { "verified": True, "message": "Business verified successfully", "business_name": company_data.get("company_name", normalized_name), "cac_number": company_data.get("registration_number", normalized_cac), "registration_date": company_data.get("registration_date"), "company_type": company_data.get("company_type"), "status": company_data.get("status"), "address": company_data.get("address"), } else: return { "verified": False, "message": "Business verification failed. Please check your CAC number and business name.", "business_name": normalized_name, "cac_number": normalized_cac, } except httpx.HTTPStatusError as e: # Handle API errors error_msg = "Business verification service temporarily unavailable" if e.response.status_code == 404: error_msg = "Business not found. Please verify your CAC number and business name." elif e.response.status_code == 400: error_msg = "Invalid request. Please check your CAC number format." return { "verified": False, "message": error_msg, "business_name": normalized_name, "cac_number": normalized_cac, } except Exception as e: return { "verified": False, "message": f"Error verifying business: {str(e)}", "business_name": normalized_name, "cac_number": normalized_cac, }