Spaces:
Sleeping
Sleeping
| """ | |
| 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("<i>No KYC fields supplied for verification.</i>", 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"<para alignment='center'><font size=18 color='white'><b>{band}</b></font></para>", | |
| s["Normal"]), | |
| Paragraph(f"<b>Composite risk score:</b> {score}<br/>" | |
| f"<b>Detection ensemble:</b> Rule-based + Random Forest" | |
| f"{' + CNN (MobileNetV2)' if 'cnn_prediction' in forensic_report else ''}<br/>" | |
| f"<b>Recommended action:</b> {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( | |
| "<i>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.</i>", 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) | |