""" compliance.py - Compliance & Audit Pack for DocSentry Three capabilities: 1. KYC validators - IFSC, PAN, Aadhaar format + checksum verification 2. PII redaction - mask Aadhaar/PAN/account/IFSC in extracted text + new PDF 3. Compliance report - RBI Master Direction style audit PDF Imported by app.py Tab 4. None of this needs a network call; the validators use RBI-published format rules. """ import io import re from pathlib import Path from datetime import datetime import fitz # PyMuPDF for PDF rendering from reportlab.lib.pagesizes import A4 from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib.units import cm from reportlab.lib import colors from reportlab.platypus import (SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle) # ============================================================ # 1. KYC validators # ============================================================ # RBI publishes the IFSC structure: 4-char bank code + '0' + 6-char branch code. # Known major bank prefixes (subset of RBI's master list - ~150 banks). RBI_BANK_CODES = { "SBIN": "State Bank of India", "HDFC": "HDFC Bank", "ICIC": "ICICI Bank", "UTIB": "Axis Bank", "PUNB": "Punjab National Bank", "BARB": "Bank of Baroda", "CNRB": "Canara Bank", "UBIN": "Union Bank of India", "IOBA": "Indian Overseas Bank", "IBKL": "IDBI Bank", "KKBK": "Kotak Mahindra Bank", "YESB": "Yes Bank", "RATN": "RBL Bank", "INDB": "IndusInd Bank", "IDIB": "Indian Bank", "MAHB": "Bank of Maharashtra", "UCBA": "UCO Bank", "BKID": "Bank of India", "ANDB": "Andhra Bank", "ALLA": "Allahabad Bank", "CBIN": "Central Bank of India", "ORBC": "Oriental Bank of Commerce", "SYNB": "Syndicate Bank", "VIJB": "Vijaya Bank", "PSIB": "Punjab & Sind Bank", "FDRL": "Federal Bank", "KARB": "Karnataka Bank", "SIBL": "South Indian Bank", "TMBL": "Tamilnad Mercantile Bank", "CIUB": "City Union Bank", "BDBL": "Bandhan Bank", "AUBL": "AU Small Finance Bank", "IDFB": "IDFC First Bank", "ESFB": "Equitas Small Finance Bank", "DCBL": "DCB Bank", "PROG": "Progressive National Bank", } IFSC_RE = re.compile(r"^[A-Z]{4}0[A-Z0-9]{6}$") def validate_ifsc(code): """Returns dict: ok, bank_name, flags.""" code = (code or "").upper().strip() flags = [] if not code: return {"ok": False, "code": "", "flags": ["empty"]} if not IFSC_RE.match(code): flags.append("invalid format (need AAAA0NNNNNN)") return {"ok": False, "code": code, "flags": flags} bank = code[:4] if bank not in RBI_BANK_CODES: flags.append(f"bank code '{bank}' not in known RBI master list") bank_name = "unknown" else: bank_name = RBI_BANK_CODES[bank] return {"ok": not flags, "code": code, "bank_code": bank, "bank_name": bank_name, "branch_code": code[5:], "flags": flags or ["valid"]} # --- PAN: AAAAA9999A; 4th char is entity type, 5th is surname initial --- PAN_RE = re.compile(r"^[A-Z]{5}[0-9]{4}[A-Z]$") PAN_ENTITY_TYPES = { "P": "Individual", "F": "Firm", "C": "Company", "A": "AOP", "T": "Trust", "B": "BOI", "L": "LLP", "J": "Artificial Juridical Person", "G": "Government", "H": "HUF", } def validate_pan(code): code = (code or "").upper().strip() flags = [] if not code: return {"ok": False, "code": "", "flags": ["empty"]} if not PAN_RE.match(code): flags.append("invalid format (need AAAAA9999A)") return {"ok": False, "code": code, "flags": flags} entity = PAN_ENTITY_TYPES.get(code[3], "unknown entity type") if entity == "unknown entity type": flags.append(f"unrecognised 4th character '{code[3]}'") return {"ok": not flags, "code": code, "entity_type": entity, "flags": flags or ["valid"]} # --- Aadhaar: 12 digits, last digit is Verhoeff check --- # Verhoeff algorithm tables (UIDAI standard) _VERHOEFF_D = [ [0,1,2,3,4,5,6,7,8,9], [1,2,3,4,0,6,7,8,9,5], [2,3,4,0,1,7,8,9,5,6], [3,4,0,1,2,8,9,5,6,7], [4,0,1,2,3,9,5,6,7,8], [5,9,8,7,6,0,4,3,2,1], [6,5,9,8,7,1,0,4,3,2], [7,6,5,9,8,2,1,0,4,3], [8,7,6,5,9,3,2,1,0,4], [9,8,7,6,5,4,3,2,1,0], ] _VERHOEFF_P = [ [0,1,2,3,4,5,6,7,8,9], [1,5,7,6,2,8,3,0,9,4], [5,8,0,3,7,9,6,1,4,2], [8,9,1,6,0,4,3,5,2,7], [9,4,5,3,1,2,6,8,7,0], [4,2,8,6,5,7,3,9,0,1], [2,7,9,3,8,0,6,4,1,5], [7,0,4,6,9,1,3,2,5,8], ] def _verhoeff_check(digits): c = 0 for i, d in enumerate(reversed(digits)): c = _VERHOEFF_D[c][_VERHOEFF_P[i % 8][d]] return c == 0 AADHAAR_RE = re.compile(r"^\d{12}$") def validate_aadhaar(num): num = re.sub(r"\s+", "", str(num or "")) flags = [] if not num: return {"ok": False, "code": "", "flags": ["empty"]} if not AADHAAR_RE.match(num): flags.append("invalid format (need 12 digits)") return {"ok": False, "code": num, "flags": flags} if num.startswith(("0", "1")): flags.append("Aadhaar numbers cannot start with 0 or 1") if not _verhoeff_check([int(d) for d in num]): flags.append("Verhoeff checksum failed") return {"ok": not flags, "code": num, "masked": "XXXX-XXXX-" + num[-4:], "flags": flags or ["valid (Verhoeff checksum passed)"]} # ============================================================ # 2. PII redaction # ============================================================ # Reuse the field regexes from forensics ACC_RE = re.compile(r"\b\d{9,18}\b") IFSC_ALL = re.compile(r"\b[A-Z]{4}0[A-Z0-9]{6}\b") PAN_ALL = re.compile(r"\b[A-Z]{5}\d{4}[A-Z]\b") AAD_ALL = re.compile(r"\b\d{4}\s?\d{4}\s?\d{4}\b") def redact_text(text): """Mask PII fields in arbitrary text. Returns redacted text + list of found PII.""" found = {"ifsc": [], "pan": [], "aadhaar": [], "account": []} def _mask(s, last_n=4): return "X" * (len(s) - last_n) + s[-last_n:] for m in IFSC_ALL.findall(text): found["ifsc"].append(m) text = text.replace(m, _mask(m)) for m in PAN_ALL.findall(text): found["pan"].append(m) text = text.replace(m, _mask(m)) for m in AAD_ALL.findall(text): found["aadhaar"].append(m) text = text.replace(m, "XXXX-XXXX-" + m[-4:]) for m in ACC_RE.findall(text): if len(m) >= 10 and m not in found["aadhaar"]: found["account"].append(m) text = text.replace(m, _mask(m)) return text, found def redact_pdf(input_path, output_path): """Create a redacted copy of a PDF with PII boxes filled in black.""" src = fitz.open(input_path) out = fitz.open() found_all = {"ifsc": [], "pan": [], "aadhaar": [], "account": []} for page_num, page in enumerate(src): text_dict = page.get_text("dict") new_page = out.new_page(width=page.rect.width, height=page.rect.height) new_page.show_pdf_page(new_page.rect, src, page_num) # Search each PII pattern on the page and overlay black rectangles for pattern, key in [(IFSC_ALL, "ifsc"), (PAN_ALL, "pan"), (AAD_ALL, "aadhaar"), (ACC_RE, "account")]: text = page.get_text() for m in set(pattern.findall(text)): if key == "account" and (len(m) < 10 or m in found_all["aadhaar"]): continue found_all[key].append(m) # Find bounding boxes for this text on the page rects = page.search_for(m) for r in rects: new_page.draw_rect(r, color=(0, 0, 0), fill=(0, 0, 0), fill_opacity=1, overlay=True) out.save(output_path) out.close() src.close() return found_all def extract_pii_fields(path): """Convenience: pull all PII fields from a document (PDF or image).""" if str(path).lower().endswith(".pdf"): with fitz.open(path) as d: text = "\n".join(p.get_text() for p in d) else: try: import pytesseract from PIL import Image text = pytesseract.image_to_string(Image.open(path)) except Exception: text = "" fields = { "ifsc": IFSC_ALL.findall(text), "pan": PAN_ALL.findall(text), "aadhaar": AAD_ALL.findall(text), "accounts": [m for m in ACC_RE.findall(text) if len(m) >= 10], } fields["aadhaar"] = list(set(fields["aadhaar"])) return fields, text # ============================================================ # 3. RBI-style compliance report PDF # ============================================================ def build_compliance_report(forensic_report, source_path, kyc_results=None): """ Produces an RBI Master-Direction-style audit PDF with mandatory fields: - Customer identification - KYC verification status (per RBI KYC Master Direction 2016) - Fraud-screening status - Recommended risk weighting - Auditor sign-off block """ source_path = Path(source_path) s = getSampleStyleSheet() s.add(ParagraphStyle("RBITitle", parent=s["Title"], fontSize=18, textColor=colors.HexColor("#0c4a6e"))) s.add(ParagraphStyle("RBISub", parent=s["Normal"], fontSize=10, textColor=colors.dimgray, alignment=1)) s.add(ParagraphStyle("RBISection", parent=s["Heading2"], fontSize=12, textColor=colors.HexColor("#0c4a6e"), spaceAfter=4)) s.add(ParagraphStyle("Mono", parent=s["Normal"], fontName="Courier", fontSize=8, textColor=colors.dimgray)) buf = io.BytesIO() doc = SimpleDocTemplate(buf, pagesize=A4, leftMargin=2 * cm, rightMargin=2 * cm, topMargin=1.5 * cm, bottomMargin=1.5 * cm) story = [] # ----- Header ----- story.append(Paragraph("DOCSENTRY - REGULATORY COMPLIANCE AUDIT", s["RBITitle"])) story.append(Paragraph("Pursuant to RBI Master Direction on KYC (2016, as amended) " "and RBI Master Circular on Frauds (DBS.CO.CFMC.BC.No.1/23.04.001/2023-24)", s["RBISub"])) story.append(Spacer(1, 0.5 * cm)) # ----- Section 1: Document identification ----- story.append(Paragraph("1. Document Identification", s["RBISection"])) ident = [ ["Field", "Value"], ["Document filename", source_path.name], ["Document type", forensic_report.get("type", "-")], ["SHA-256 (full)", forensic_report.get("sha256", "-")], ["Audit timestamp (UTC)", forensic_report.get("analysed_at", "-")], ["Audit ID", "DS-" + forensic_report.get("sha256", "0" * 16)[:16].upper()], ] t = Table(ident, colWidths=[5 * cm, 12 * cm]) t.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#0c4a6e")), ("TEXTCOLOR", (0, 0), (-1, 0), colors.white), ("GRID", (0, 0), (-1, -1), 0.4, colors.grey), ("FONTSIZE", (0, 0), (-1, -1), 8), ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"), ])) story.append(t) story.append(Spacer(1, 0.4 * cm)) # ----- Section 2: KYC verification status ----- story.append(Paragraph("2. KYC Field Verification", s["RBISection"])) if not kyc_results: story.append(Paragraph("No KYC fields supplied for verification.", s["Normal"])) else: kyc_rows = [["Field", "Value", "Status", "Notes"]] for kind, result in kyc_results.items(): status = "PASS" if result.get("ok") else "FAIL" value = result.get("code", "-") if kind == "aadhaar" and result.get("masked"): value = result["masked"] notes = "; ".join(result.get("flags", [])) kyc_rows.append([kind.upper(), value, status, notes]) kt = Table(kyc_rows, colWidths=[2.5 * cm, 4 * cm, 1.5 * cm, 9 * cm]) kt.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#0c4a6e")), ("TEXTCOLOR", (0, 0), (-1, 0), colors.white), ("GRID", (0, 0), (-1, -1), 0.4, colors.grey), ("FONTSIZE", (0, 0), (-1, -1), 8), ])) story.append(kt) story.append(Spacer(1, 0.4 * cm)) # ----- Section 3: Fraud-screening status ----- story.append(Paragraph("3. Fraud / Tamper Screening", s["RBISection"])) band = forensic_report.get("risk_band", "UNKNOWN") score = forensic_report.get("risk_score", "-") band_colour = {"LOW": colors.HexColor("#16a34a"), "MEDIUM": colors.HexColor("#ca8a04"), "HIGH": colors.HexColor("#ea580c"), "CRITICAL": colors.HexColor("#dc2626")}.get(band, colors.grey) fraud_box = Table([ [Paragraph(f"{band}", s["Normal"]), Paragraph(f"Composite risk score: {score}
" f"Detection ensemble: Rule-based + Random Forest" f"{' + CNN (MobileNetV2)' if 'cnn_prediction' in forensic_report else ''}
" f"Recommended action: {forensic_report.get('recommended_action', '-')}", s["Normal"])] ], colWidths=[4 * cm, 13 * cm]) fraud_box.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (0, 0), band_colour), ("BACKGROUND", (1, 0), (1, 0), colors.HexColor("#f0f9ff")), ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), ("TOPPADDING", (0, 0), (-1, -1), 10), ("BOTTOMPADDING", (0, 0), (-1, -1), 10), ("LEFTPADDING", (0, 0), (-1, -1), 12), ("RIGHTPADDING", (0, 0), (-1, -1), 12), ])) story.append(fraud_box) story.append(Spacer(1, 0.3 * cm)) story.append(Paragraph("Detected anomalies:", s["Normal"])) for ev in forensic_report.get("evidence", []): story.append(Paragraph("• " + ev, s["Normal"])) story.append(Spacer(1, 0.4 * cm)) # ----- Section 4: Recommended risk weighting ----- story.append(Paragraph("4. Recommended Risk Treatment (per RBI guidelines)", s["RBISection"])) rw_map = {"LOW": ("Standard onboarding", "Risk weight 75% (retail)"), "MEDIUM": ("Request additional verification", "Risk weight 100%"), "HIGH": ("Mandatory manual review by fraud-risk team", "Risk weight 150% pending verification"), "CRITICAL": ("Decline / escalate to compliance", "Reject application; report under Suspicious Transaction Report (STR) workflow")} treatment, risk_weight = rw_map.get(band, ("-", "-")) rw_table = Table([ ["Recommended treatment", treatment], ["Suggested risk weight", risk_weight], ["Audit retention period", "10 years (per RBI Master Direction on KYC, Para 67)"], ], colWidths=[5 * cm, 12 * cm]) rw_table.setStyle(TableStyle([ ("GRID", (0, 0), (-1, -1), 0.4, colors.grey), ("BACKGROUND", (0, 0), (0, -1), colors.HexColor("#f0f9ff")), ("FONTSIZE", (0, 0), (-1, -1), 9), ])) story.append(rw_table) story.append(Spacer(1, 0.4 * cm)) # ----- Section 5: Auditor sign-off ----- story.append(Paragraph("5. Auditor Sign-Off", s["RBISection"])) signoff = Table([ ["Generated by", "DocSentry Automated Audit Engine v1.0"], ["Reviewer name", "_______________________________"], ["Designation", "_______________________________"], ["Date", "_______________________________"], ["Signature", "_______________________________"], ], colWidths=[5 * cm, 12 * cm]) signoff.setStyle(TableStyle([ ("GRID", (0, 0), (-1, -1), 0.4, colors.grey), ("FONTSIZE", (0, 0), (-1, -1), 9), ("BACKGROUND", (0, 0), (0, -1), colors.HexColor("#f0f9ff")), ])) story.append(signoff) story.append(Spacer(1, 0.4 * cm)) # ----- Disclaimer ----- story.append(Paragraph( "This document is an automated compliance report. The recommendations " "herein are advisory and do not substitute statutory verification or " "manual review by an authorised compliance officer. All audit records " "must be retained per RBI Master Direction requirements.", s["Mono"])) doc.build(story) buf.seek(0) return buf.read() if __name__ == "__main__": # Quick smoke tests print("IFSC SBIN0123456 :", validate_ifsc("SBIN0123456")) print("IFSC FAKE0000000 :", validate_ifsc("FAKE0000000")) print("PAN ABCDE1234F :", validate_pan("ABCDE1234F")) print("PAN bad :", validate_pan("ABCD12345")) print("Aadhaar valid :", validate_aadhaar("234567890124")) # not real print("Aadhaar starts 0 :", validate_aadhaar("012345678901")) text = "Name: RAMESH. Account: 78439336112. IFSC: SBIN0001234. PAN: ABCDE1234F" redacted, found = redact_text(text) print("Redacted:", redacted) print("Found:", found)