Spaces:
Sleeping
Sleeping
| """ | |
| integrations/document_parser.py | |
| SentinelFin Document Intelligence Engine | |
| Parses AML-relevant documents into structured transaction data. | |
| Supported formats: | |
| - PDF (bank statements, SWIFT confirmations, corporate docs) | |
| - CSV (transaction monitoring exports, core banking exports) | |
| - TXT (SWIFT MT103/MT202 messages, correspondent bank messages) | |
| - JSON (API exports, structured transaction feeds) | |
| When raw text is extracted, Llama 3.1 via AWS Bedrock performs intelligent | |
| entity and transaction extraction — exactly how UiPath Document Understanding | |
| works in production (ML-based field extraction → structured output). | |
| """ | |
| import io | |
| import csv | |
| import json | |
| import re | |
| import datetime | |
| from typing import Optional | |
| import pdfplumber | |
| # ── SWIFT MT103 Parser ──────────────────────────────────────────────────────── | |
| def parse_swift_mt103(text: str) -> list[dict]: | |
| """ | |
| Parse SWIFT MT103 (Customer Credit Transfer) and MT202 (Bank Transfer) | |
| messages into structured transaction objects. | |
| The MT103 format is used by SWIFT for international wire transfers. | |
| Real banks use this format for every correspondent banking transaction. | |
| """ | |
| transactions = [] | |
| messages = re.split(r'\{1:', text) | |
| for block in messages: | |
| if not block.strip(): | |
| continue | |
| txn = { | |
| "txn_id": "", | |
| "date": datetime.datetime.utcnow().isoformat() + "Z", | |
| "amount": 0, | |
| "type": "WIRE_TRANSFER", | |
| "originator": {}, | |
| "beneficiary": {}, | |
| "purpose": "", | |
| "swift_raw": block[:300], | |
| } | |
| # :20: Transaction reference | |
| ref = re.search(r':20:([^\r\n:]+)', block) | |
| if ref: | |
| txn["txn_id"] = ref.group(1).strip() | |
| # :32A: Value date, currency, amount (e.g., 260610USD250000,) | |
| val_date = re.search(r':32A:(\d{6})([A-Z]{3})([\d,]+)', block) | |
| if val_date: | |
| d = val_date.group(1) | |
| try: | |
| txn["date"] = f"20{d[:2]}-{d[2:4]}-{d[4:6]}T09:00:00Z" | |
| except Exception: | |
| pass | |
| txn["currency"] = val_date.group(2) | |
| try: | |
| txn["amount"] = int(float(val_date.group(3).replace(",", "."))) | |
| except Exception: | |
| pass | |
| # :50K: / :50A: Ordering customer (originator) | |
| orig = re.search(r':50[AKF]:(/([^\r\n]+))?\r?\n([^\r\n:]+)?\r?\n?([^\r\n:]+)?', block) | |
| if orig: | |
| txn["originator"] = { | |
| "account": (orig.group(2) or "").strip(), | |
| "name": (orig.group(3) or "Unknown").strip(), | |
| "country": (orig.group(4) or "").strip(), | |
| } | |
| # :59: / :59A: Beneficiary customer | |
| bene = re.search(r':59[A-Z]?:(/([^\r\n]+))?\r?\n([^\r\n:]+)?\r?\n?([^\r\n:]+)?', block) | |
| if bene: | |
| txn["beneficiary"] = { | |
| "account": (bene.group(2) or "").strip(), | |
| "name": (bene.group(3) or "Unknown").strip(), | |
| "country": (bene.group(4) or "").strip(), | |
| } | |
| # :52A: / :57A: Correspondent bank | |
| corr = re.search(r':52[AD]:([^\r\n:]+)', block) | |
| if corr: | |
| txn["correspondent_bank"] = corr.group(1).strip() | |
| # :70: Remittance information (purpose) | |
| remit = re.search(r':70:([^\r\n:]+)', block) | |
| if remit: | |
| txn["purpose"] = remit.group(1).strip() | |
| if txn["txn_id"] or txn["amount"] > 0: | |
| if not txn["txn_id"]: | |
| txn["txn_id"] = f"SWIFT-{len(transactions)+1:03d}" | |
| transactions.append(txn) | |
| return transactions | |
| # ── CSV / Excel Parser ──────────────────────────────────────────────────────── | |
| def parse_csv_transactions(content: str) -> list[dict]: | |
| """ | |
| Parse CSV exports from: | |
| - Core banking systems (Temenos, FIS, Fiserv) | |
| - Transaction monitoring systems (NICE Actimize, FICO, Oracle FCCM) | |
| - Payment rails exports | |
| """ | |
| transactions = [] | |
| reader = csv.DictReader(io.StringIO(content)) | |
| headers = [h.lower().strip() for h in (reader.fieldnames or [])] | |
| def find_col(candidates: list[str], row: dict) -> str: | |
| for c in candidates: | |
| for k in row: | |
| if c in k.lower(): | |
| return row[k] or "" | |
| return "" | |
| for i, row in enumerate(reader): | |
| amount_str = find_col(["amount", "value", "sum", "usd", "eur"], row) | |
| try: | |
| amount = int(float(re.sub(r'[^\d.]', '', amount_str or "0"))) | |
| except Exception: | |
| amount = 0 | |
| txn = { | |
| "txn_id": find_col(["ref", "id", "txn", "transaction"], row) or f"CSV-{i+1:03d}", | |
| "date": find_col(["date", "time", "timestamp", "value_date"], row) or datetime.datetime.utcnow().isoformat() + "Z", | |
| "amount": amount, | |
| "type": find_col(["type", "kind", "direction", "category"], row) or "WIRE", | |
| "originator": { | |
| "name": find_col(["sender", "from", "originator", "payer", "debtor"], row), | |
| "account": find_col(["sender_acc", "from_acc", "debit_acc", "iban_from"], row), | |
| "country": find_col(["sender_country", "from_country", "origin"], row), | |
| }, | |
| "beneficiary": { | |
| "name": find_col(["receiver", "to", "beneficiary", "payee", "creditor"], row), | |
| "account": find_col(["receiver_acc", "to_acc", "credit_acc", "iban_to"], row), | |
| "country": find_col(["receiver_country", "to_country", "destination"], row), | |
| }, | |
| "purpose": find_col(["purpose", "description", "narrative", "details", "remit"], row), | |
| } | |
| # Normalize date | |
| for fmt in ["%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y", "%d-%m-%Y"]: | |
| try: | |
| dt = datetime.datetime.strptime(txn["date"][:10], fmt) | |
| txn["date"] = dt.isoformat() + "Z" | |
| break | |
| except Exception: | |
| pass | |
| if amount > 0 or txn["originator"]["name"]: | |
| transactions.append(txn) | |
| return transactions | |
| # ── PDF Parser ──────────────────────────────────────────────────────────────── | |
| def parse_pdf(file_bytes: bytes) -> tuple[str, list[dict]]: | |
| """ | |
| Extract text and attempt to parse transactions from PDF. | |
| Supports: | |
| - Bank statement PDFs | |
| - SWIFT MT confirmation letters | |
| - Corporate documents (ownership, registration) | |
| Returns (full_text, transactions_list) | |
| """ | |
| full_text = "" | |
| transactions = [] | |
| try: | |
| with pdfplumber.open(io.BytesIO(file_bytes)) as pdf: | |
| for page in pdf.pages: | |
| text = page.extract_text() or "" | |
| full_text += text + "\n" | |
| # Try to parse tables from each page | |
| for table in (page.extract_tables() or []): | |
| if not table: | |
| continue | |
| headers = [str(c).lower().strip() if c else "" for c in table[0]] | |
| for row in table[1:]: | |
| if not any(row): | |
| continue | |
| row_dict = {h: str(v).strip() if v else "" for h, v in zip(headers, row)} | |
| # Check if this looks like a transaction row | |
| amount_keys = [k for k in row_dict if any(x in k for x in ["amount", "value", "debit", "credit", "usd", "eur"])] | |
| if amount_keys: | |
| amount_str = row_dict.get(amount_keys[0], "0") | |
| try: | |
| amount = int(float(re.sub(r'[^\d.]', '', amount_str or "0"))) | |
| except Exception: | |
| amount = 0 | |
| if amount > 0: | |
| transactions.append({ | |
| "txn_id": row_dict.get("ref", row_dict.get("id", f"PDF-{len(transactions)+1:03d}")), | |
| "date": row_dict.get("date", row_dict.get("value date", datetime.datetime.utcnow().isoformat() + "Z")), | |
| "amount": amount, | |
| "type": row_dict.get("type", "WIRE"), | |
| "originator": {"name": row_dict.get("from", row_dict.get("sender", "")) }, | |
| "beneficiary": {"name": row_dict.get("to", row_dict.get("receiver", ""))}, | |
| "source": "PDF_TABLE", | |
| }) | |
| except Exception as e: | |
| full_text = f"[PDF parse error: {e}]" | |
| # Also try SWIFT format in the text | |
| if "{1:" in full_text or ":20:" in full_text: | |
| swift_txns = parse_swift_mt103(full_text) | |
| transactions.extend(swift_txns) | |
| return full_text, transactions | |
| # ── AI-Powered Extraction (Llama 3.1 fallback) ─────────────────────────────── | |
| def extract_with_llm(document_text: str, llm_client) -> list[dict]: | |
| """ | |
| When structured parsing fails, use Llama 3.1 to intelligently extract | |
| transaction data from unstructured text. This mirrors how UiPath's | |
| Document Understanding ML models work — reading any document and | |
| producing structured JSON output. | |
| """ | |
| prompt = f"""You are an expert AML (Anti-Money Laundering) document analyst. | |
| Extract ALL financial transactions from the following document. | |
| Return ONLY valid JSON array, no other text. | |
| Each transaction must follow this exact schema: | |
| {{ | |
| "txn_id": "unique reference (string)", | |
| "date": "ISO 8601 date string", | |
| "amount": numeric dollar amount (integer, no commas), | |
| "type": "WIRE_IN" or "WIRE_OUT" or "WIRE_TRANSFER" or "CRYPTO" or "CASH", | |
| "originator": {{"name": "sender name", "country": "country", "account": "account number"}}, | |
| "beneficiary": {{"name": "recipient name", "country": "country", "account": "account number"}}, | |
| "purpose": "stated purpose/description" | |
| }} | |
| If you cannot find a specific field, use empty string "". | |
| Return [] if no transactions are found. | |
| DOCUMENT: | |
| {document_text[:8000]} | |
| TRANSACTIONS JSON:""" | |
| try: | |
| response = llm_client.converse( | |
| modelId="meta.llama3-70b-instruct-v1:0", | |
| messages=[{"role": "user", "content": [{"text": prompt}]}], | |
| inferenceConfig={"maxTokens": 2048, "temperature": 0.0}, | |
| ) | |
| raw = response["output"]["message"]["content"][0]["text"].strip() | |
| # Extract JSON array | |
| json_match = re.search(r'\[.*\]', raw, re.DOTALL) | |
| if json_match: | |
| return json.loads(json_match.group(0)) | |
| except Exception as e: | |
| print(f"[DOC PARSER] LLM extraction failed: {e}") | |
| return [] | |
| # ── Main Parser Entry Point ─────────────────────────────────────────────────── | |
| def parse_document(filename: str, file_bytes: bytes, llm_client=None) -> dict: | |
| """ | |
| Main entry point for document parsing. | |
| Routes to the appropriate parser based on file extension. | |
| Returns: | |
| { | |
| "filename": str, | |
| "format": str, | |
| "raw_text": str, | |
| "transactions": list[dict], | |
| "entity_hints": list[str], | |
| "parse_method": str, | |
| "confidence": float, | |
| } | |
| """ | |
| ext = filename.rsplit(".", 1)[-1].lower() | |
| raw_text = "" | |
| transactions = [] | |
| parse_method = "unknown" | |
| print(f"[DOC PARSER] Processing '{filename}' (type: {ext}, size: {len(file_bytes)} bytes)") | |
| if ext == "pdf": | |
| raw_text, transactions = parse_pdf(file_bytes) | |
| parse_method = "PDF+SWIFT" if transactions else "PDF_TEXT" | |
| elif ext in ("csv", "tsv"): | |
| raw_text = file_bytes.decode("utf-8", errors="replace") | |
| transactions = parse_csv_transactions(raw_text) | |
| parse_method = "CSV" | |
| elif ext in ("txt", "swift", "mt103", "mt202"): | |
| raw_text = file_bytes.decode("utf-8", errors="replace") | |
| # Try SWIFT first | |
| transactions = parse_swift_mt103(raw_text) | |
| parse_method = "SWIFT_MT103" if transactions else "TEXT" | |
| elif ext == "json": | |
| raw_text = file_bytes.decode("utf-8", errors="replace") | |
| try: | |
| data = json.loads(raw_text) | |
| if isinstance(data, list): | |
| transactions = data | |
| elif isinstance(data, dict): | |
| transactions = data.get("transactions", [data]) | |
| parse_method = "JSON" | |
| except Exception: | |
| parse_method = "JSON_INVALID" | |
| # If no transactions found, use LLM extraction | |
| if not transactions and raw_text and llm_client: | |
| print(f"[DOC PARSER] Structured parse found 0 transactions. Trying LLM extraction...") | |
| transactions = extract_with_llm(raw_text, llm_client) | |
| parse_method = f"{parse_method}+LLM" | |
| # Extract entity hints from raw text | |
| entity_hints = extract_entity_hints(raw_text) | |
| confidence = min(1.0, len(transactions) * 0.2 + 0.3) if transactions else 0.1 | |
| print(f"[DOC PARSER] Extracted {len(transactions)} transactions via {parse_method} (confidence: {confidence:.0%})") | |
| return { | |
| "filename": filename, | |
| "format": ext.upper(), | |
| "raw_text": raw_text[:2000], # Preview only | |
| "transactions": transactions, | |
| "entity_hints": entity_hints, | |
| "parse_method": parse_method, | |
| "confidence": confidence, | |
| "tx_count": len(transactions), | |
| } | |
| def extract_entity_hints(text: str) -> list[str]: | |
| """Extract probable entity names and account numbers from raw text.""" | |
| hints = [] | |
| # Look for IBAN-like patterns | |
| ibans = re.findall(r'\b[A-Z]{2}\d{2}[A-Z0-9]{4,30}\b', text) | |
| hints.extend(ibans[:5]) | |
| # Look for capitalized company names | |
| companies = re.findall(r'\b([A-Z][a-z]+ (?:Corp|Ltd|LLC|GmbH|Holdings|Trading|Finance|Bank|Co|Inc)[,.]?)\b', text) | |
| hints.extend(companies[:5]) | |
| return list(set(hints))[:10] | |
| # ── Sample Documents for Demo ───────────────────────────────────────────────── | |
| SAMPLE_SWIFT_MT103 = """{1:F01DEUTDEBBAXXX0000000000}{2:I103CITIUS33XXXXN}{3:{108:MT103}}{4: | |
| :20:CAS2026007REF | |
| :23B:CRED | |
| :32A:260610USD250000, | |
| :50K:/CY-999123 | |
| UNKNOWN ENTITY A | |
| OFFSHORE DRIVE 1 | |
| GEORGE TOWN, CAYMAN ISLANDS | |
| :52A:BOFAUS3N | |
| :57A:CITIUS33 | |
| :59:/PA-888456 | |
| SHELL CORP B | |
| PANAMA CITY, PANAMA | |
| :70:INVESTMENT PORTFOLIO MANAGEMENT | |
| :71A:SHA | |
| -}{5:{CHK:ABCDEF123456}} | |
| {1:F01DEUTDEBBAXXX0000000001}{2:I103CITIUS33XXXXN}{4: | |
| :20:CAS2026007REF2 | |
| :23B:CRED | |
| :32A:260611USD120000, | |
| :50K:/ACC-554433 | |
| TRANSIT ACCOUNT | |
| NEW YORK, USA | |
| :59:/CP-777789 | |
| HOLDINGS C | |
| LIMASSOL, CYPRUS | |
| :70:CONSULTING FEES | |
| :71A:SHA | |
| -} | |
| """ | |
| SAMPLE_CSV = """ref,date,amount,type,sender,sender_account,sender_country,receiver,receiver_account,receiver_country,purpose | |
| TXN-301,2026-06-08,890000,WIRE_IN,Aryan Trading Co,IR-445566,Iran,Global Import LLC,AE-112233,UAE,Trade goods payment | |
| TXN-302,2026-06-09,870000,WIRE_OUT,Global Import LLC,AE-112233,UAE,Hezbollah Finance,LB-998877,Lebanon,Investment | |
| TXN-303,2026-06-10,850000,WIRE_OUT,Hezbollah Finance,LB-998877,Lebanon,Offshore Shell X,VG-556644,BVI,Asset management | |
| TXN-304,2026-06-11,820000,WIRE_OUT,Offshore Shell X,VG-556644,BVI,Final Corp Z,CH-334455,Switzerland,Portfolio rebalancing | |
| """ | |
| SAMPLE_JSON = json.dumps({ | |
| "alert_type": "SUSPECTED_LAYERING", | |
| "alert_id": "CAS-2026-CRYPTO", | |
| "source_system": "NICE Actimize SAM", | |
| "transactions": [ | |
| {"txn_id": "CRYPTO-001", "date": "2026-06-14T03:00:00Z", "amount": 2100000, "type": "CRYPTO_IN", | |
| "originator": {"name": "Tornado Cash Mixer", "country": "Decentralized", "account": "0xAB...FF11"}, | |
| "beneficiary": {"name": "OTC Desk Alpha", "country": "Hong Kong", "account": "HK-OTC-001"}}, | |
| {"txn_id": "CRYPTO-002", "date": "2026-06-14T06:30:00Z", "amount": 2050000, "type": "WIRE_OUT", | |
| "originator": {"name": "OTC Desk Alpha", "country": "Hong Kong", "account": "HK-OTC-001"}, | |
| "beneficiary": {"name": "PJSC Gazprombank", "country": "Russia", "account": "RU-GAZP-999"}}, | |
| ] | |
| }, indent=2) | |