Add /transactions/import/image endpoint to extract transactions from images using AI
This commit is contained in:
Iyeoluwa Akinrinola
+128
-1
@@ -201,4 +201,131 @@ class DocumentProcessor:
|
||||
return file_path
|
||||
|
||||
except Exception as e:
|
||||
raise Exception(f"File save error: {str(e)}")
|
||||
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": []
|
||||
}
|
||||
@@ -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
|
||||
# ============================================================================
|
||||
|
||||
Reference in New Issue
Block a user