From e81745b63834300be8e1cea31ce3484e4a793336 Mon Sep 17 00:00:00 2001 From: Iyeoluwa Akinrinola Date: Thu, 3 Jul 2025 00:36:55 +0100 Subject: [PATCH] Add /transactions/import/image endpoint to extract transactions from images using AI --- document_processor.py | 129 +++++++++++++++++++++++++++++++++++++++++- main.py | 94 ++++++++++++++++++++++++++++++ 2 files changed, 222 insertions(+), 1 deletion(-) diff --git a/document_processor.py b/document_processor.py index e34b282..87337c2 100644 --- a/document_processor.py +++ b/document_processor.py @@ -201,4 +201,131 @@ class DocumentProcessor: return file_path except Exception as e: - raise Exception(f"File save error: {str(e)}") \ No newline at end of file + raise Exception(f"Failed to save file: {str(e)}") + + async def extract_transactions_from_image(self, image_path: str) -> Dict[str, Any]: + """Extract multiple transactions from an image (bank statement, credit card statement, etc.)""" + try: + # Encode image to base64 + base64_image = self._encode_image(image_path) + + # Create Groq vision prompt for transaction extraction + prompt = """ + Analyze this financial document image (bank statement, credit card statement, etc.) and extract ALL transactions in JSON format. + + Look for transaction lists, payment records, or any financial entries that show: + - Date + - Amount (positive or negative) + - Vendor/Description/Payee name + - Any additional notes or memo + + Return the transactions as a JSON array: + { + "extraction_success": true, + "transactions": [ + { + "date": "YYYY-MM-DD", + "amount": 0.00, + "vendor": "Vendor name", + "memo": "Additional notes" + }, + { + "date": "YYYY-MM-DD", + "amount": -0.00, + "vendor": "Another vendor", + "memo": "Payment or charge description" + } + ] + } + + Rules: + - Extract ALL visible transactions + - Include both positive (credits) and negative (debits) amounts + - Use the actual date format from the document + - Vendor should be the merchant/payee name + - Memo can include transaction type, reference numbers, etc. + - If no transactions found, return empty array but set extraction_success to true + + Return only valid JSON. + """ + + # Call Groq vision API + response = self.client.chat.completions.create( + messages=[ + { + "role": "user", + "content": [ + {"type": "text", "text": prompt}, + { + "type": "image_url", + "image_url": { + "url": f"data:image/jpeg;base64,{base64_image}", + }, + }, + ], + } + ], + model=self.model, + max_tokens=2000, # Higher token limit for multiple transactions + temperature=0.1 + ) + + # Parse response + result_text = response.choices[0].message.content.strip() + return self._parse_transaction_extraction_result(result_text) + + except Exception as e: + return { + "extraction_success": False, + "error": f"Transaction extraction error: {str(e)}", + "transactions": [] + } + + def _parse_transaction_extraction_result(self, result_text: str) -> Dict[str, Any]: + """Parse Groq response for transaction extraction""" + try: + import json + import re + + # Find JSON in response + json_match = re.search(r'\{.*\}', result_text, re.DOTALL) + if json_match: + json_str = json_match.group() + data = json.loads(json_str) + + # Validate and clean data + transactions = data.get("transactions", []) + cleaned_transactions = [] + + for txn in transactions: + try: + # Clean and validate each transaction + cleaned_txn = { + "date": str(txn.get("date", "")).strip(), + "amount": float(str(txn.get("amount", 0)).replace('$', '').replace(',', '')), + "vendor": str(txn.get("vendor", "")).strip(), + "memo": str(txn.get("memo", "")).strip() + } + cleaned_transactions.append(cleaned_txn) + except Exception as e: + # Skip invalid transactions + continue + + return { + "extraction_success": data.get("extraction_success", True), + "transactions": cleaned_transactions, + "total_transactions": len(cleaned_transactions) + } + else: + return { + "extraction_success": False, + "error": "Could not parse JSON from AI response", + "transactions": [] + } + + except Exception as e: + return { + "extraction_success": False, + "error": f"JSON parsing error: {str(e)}", + "transactions": [] + } \ No newline at end of file diff --git a/main.py b/main.py index 9033043..ef12860 100644 --- a/main.py +++ b/main.py @@ -155,6 +155,100 @@ async def import_quickbooks_transactions_csv(file: UploadFile = File(...)): except Exception as e: raise HTTPException(status_code=500, detail=str(e)) +@app.post("/transactions/import/image", response_model=QuickBooksImportResponse) +async def import_transactions_from_image(file: UploadFile = File(...)): + """ + Import transactions from an image (bank statement, credit card statement, etc.) using AI extraction. + + This endpoint uses AI to extract transaction data from images like: + - Bank statements + - Credit card statements + - Transaction lists + - Financial documents + """ + try: + # Validate file type + allowed_types = ['jpg', 'jpeg', 'png', 'gif', 'bmp', 'pdf'] + file_extension = file.filename.split('.')[-1].lower() + + if file_extension not in allowed_types: + raise HTTPException(status_code=400, detail=f"Unsupported file type. Allowed: {allowed_types}") + + # Read file content + file_content = await file.read() + + # Save file temporarily + file_path = await document_processor.save_uploaded_file(file_content, file.filename) + + # Use AI to extract transactions from the image + extraction_result = await document_processor.extract_transactions_from_image(file_path) + + if not extraction_result.get("extraction_success", False): + raise HTTPException( + status_code=400, + detail=f"Failed to extract transactions from image: {extraction_result.get('error', 'Unknown error')}" + ) + + # Parse extracted transactions + transactions = [] + errors = [] + + extracted_transactions = extraction_result.get("transactions", []) + + for idx, txn in enumerate(extracted_transactions): + try: + # Generate unique ID + txn_id = f"img_{file.filename}_{idx+1}" + + # Parse date + txn_date = txn.get("date", "") + if not txn_date: + raise ValueError("No date found in transaction") + + # Parse amount + amount_str = str(txn.get("amount", "0")) + amount = float(amount_str.replace('$', '').replace(',', '').strip()) + + # Get vendor/description + payee_name = txn.get("vendor", txn.get("description", "Unknown")) + + # Get memo/notes + memo = txn.get("memo", txn.get("notes", "")) + + transactions.append({ + "id": txn_id, + "txn_date": txn_date, + "amount": amount, + "payee_name": payee_name, + "memo": memo + }) + + except Exception as e: + errors.append(f"Transaction {idx+1}: {str(e)}") + + if not transactions: + raise HTTPException( + status_code=400, + detail="No valid transactions could be extracted from the image" + ) + + # Store transactions globally for auto-matching + global stored_transactions + stored_transactions = transactions + + # Use the same logic as the JSON import endpoint + request_obj = QuickBooksImportRequest(transactions=transactions) + response = await import_quickbooks_transactions(request_obj) + + # Attach errors from image parsing + if hasattr(response, 'errors'): + response.errors.extend(errors) + + return response + + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + # ============================================================================ # RECEIPT PROCESSING ENDPOINTS # ============================================================================