'code'

ageents/ledgerAgent.py

@@ -8,7 +8,7 @@ import os
 import uuid
 import json
 from datetime import datetime
-from typing import Optional, Dict, Any
+from typing import Optional
 from pydantic import BaseModel, Field
 from dotenv import load_dotenv
 from openai import AsyncOpenAI
@@ -62,32 +62,54 @@ class ManualEntryData(BaseModel):
 
 class ProcessedEntry(BaseModel):
     """Validated and processed entry"""
-    entry_type: str  # "expense" or "income"
+    entry_type: str
     category: str
     amount: float
     currency: str = "PKR"
     payment_method: str
     notes: str
     user_id: str
-    recorded_by: str  # "voice" or "manual"
+    recorded_by: str
     device_id: str
     confidence: float = 1.0
 
+class ValidationResult(BaseModel):
+    """Validation result response"""
+    valid: bool
+    errors: list[str] = Field(default_factory=list)
+    warnings: list[str] = Field(default_factory=list)
+    corrected_category: Optional[str] = None
+    corrected_amount: Optional[float] = None
+
+class SyncResult(BaseModel):
+    """Sync result response"""
+    status: str
+    entry_id: str
+    message: str
+    timestamp: str
+
 # ================== FUNCTION TOOLS ==================
 
 @function_tool()
-def validate_entry_data(entry: Dict[str, Any]) -> Dict[str, Any]:
+def validate_entry_data(
+    amount: float,
+    category: str,
+    payment_method: str,
+    user_id: str,
+) -> ValidationResult:
     """
-    Validate entry fields against business rules
-    Returns validation result with any corrections
+    Validate entry fields against business rules.
+    Returns validation result with corrections if needed.
     """
     errors = []
     warnings = []
+    corrected_category = category
+    corrected_amount = amount
     
     # Validate amount
-    if entry.get("amount", 0) <= 0:
+    if amount <= 0:
         errors.append("Amount must be greater than 0")
-    if entry.get("amount", 0) > 1000000:
+    if amount > 1000000:
         warnings.append("Large amount detected - verify correctness")
     
     # Validate category
@@ -95,41 +117,43 @@ def validate_entry_data(entry: Dict[str, Any]) -> Dict[str, Any]:
         "fertilizer", "seeds", "fuel", "labour", "pesticide",
         "equipment", "irrigation", "transport", "other"
     ]
-    if entry.get("category", "").lower() not in valid_categories:
-        errors.append(f"Invalid category. Valid options: {', '.join(valid_categories)}")
+    if category.lower() not in valid_categories:
+        errors.append(f"Invalid category. Valid: fertilizer, seeds, fuel, labour, pesticide, equipment, irrigation, transport, other")
+        corrected_category = "other"
     
     # Validate payment method
     valid_methods = ["cash", "bank", "mobile", "credit"]
-    if entry.get("payment_method", "").lower() not in valid_methods:
+    if payment_method.lower() not in valid_methods:
         warnings.append("Unknown payment method - setting to cash")
-        entry["payment_method"] = "cash"
     
     # Validate user_id
-    if not entry.get("user_id"):
+    if not user_id or len(user_id) == 0:
         errors.append("user_id is required")
     
-    return {
-        "valid": len(errors) == 0,
-        "errors": errors,
-        "warnings": warnings,
-        "corrected_entry": entry,
-    }
+    return ValidationResult(
+        valid=len(errors) == 0,
+        errors=errors,
+        warnings=warnings,
+        corrected_category=corrected_category,
+        corrected_amount=corrected_amount,
+    )
 
 @function_tool()
-def determine_entry_type(category: str, amount: float, notes: str) -> str:
+def determine_entry_type(
+    category: str,
+    amount: float,
+    notes: str,
+) -> str:
     """
-    Intelligent determination of entry type (income/expense)
-    Based on category and contextual clues from notes
+    Intelligent determination of entry type (income/expense).
+    Based on category and contextual clues from notes.
     """
-    # Default is expense
     entry_type = "expense"
     
-    # Income indicators in notes
-    income_keywords = ["sale", "sold", "income", "earned", "received", "بیچا", "آمدن"]
+    income_keywords = ["sale", "sold", "income", "earned", "received", "بیچا", "آمدن", "harvest"]
     if any(keyword in notes.lower() for keyword in income_keywords):
         entry_type = "income"
     
-    # Categories typically associated with income
     income_categories = ["sale", "income"]
     if category.lower() in income_categories:
         entry_type = "income"
@@ -148,10 +172,10 @@ def add_ledger_entry(
     recorded_by: str,
     device_id: str,
     confidence: float = 1.0,
-) -> EntryResponse:
+) -> str:
     """
     Add validated ledger entry to Firebase.
-    Automatically syncs to all connected clients.
+    Returns the entry ID.
     """
     try:
         entry_id = str(uuid.uuid4())
@@ -177,30 +201,19 @@ def add_ledger_entry(
             },
         }
         
-        # Save to Firebase under "entries" collection
+        # Save to Firebase
         ref = get_db_ref("entries")
         ref.child(entry_id).set(entry_data)
         
         print(f"✅ Entry saved to Firebase: {entry_id}")
+        return entry_id
         
-        return EntryResponse(
-            id=entry_id,
-            entryType=entry_type,
-            category=category,
-            amount=amount,
-            currency=currency,
-            paymentMethod=payment_method,
-            notes=notes,
-            createdAt=created_at,
-            recordedBy=recorded_by,
-            deviceId=device_id,
-        )
     except Exception as e:
         print(f"❌ Error saving entry: {str(e)}")
-        raise
+        raise ValueError(f"Failed to save entry: {str(e)}")
 
 @function_tool()
-def sync_entry_to_frontend(entry_id: str, user_id: str) -> Dict[str, Any]:
+def sync_entry_to_frontend(entry_id: str, user_id: str) -> SyncResult:
     """
     Trigger frontend sync for new entry.
     Frontend listens for "expense-added" event.
@@ -210,23 +223,21 @@ def sync_entry_to_frontend(entry_id: str, user_id: str) -> Dict[str, Any]:
         entry = ref.get()
         
         if not entry:
-            return {"status": "error", "message": "Entry not found"}
+            raise ValueError("Entry not found after save")
         
-        # Frontend will receive this via real-time listener
-        sync_data = {
-            "status": "synced",
-            "entry_id": entry_id,
-            "user_id": user_id,
-            "timestamp": datetime.utcnow().isoformat(),
-            "event": "expense-added",
-        }
+        sync_data = SyncResult(
+            status="synced",
+            entry_id=entry_id,
+            message=f"Entry {entry_id} synced successfully",
+            timestamp=datetime.utcnow().isoformat(),
+        )
         
         print(f"🔄 Entry {entry_id} marked for frontend sync")
-        return {"status": "success", "data": sync_data}
+        return sync_data
         
     except Exception as e:
         print(f"❌ Sync error: {str(e)}")
-        return {"status": "error", "message": str(e)}
+        raise ValueError(f"Sync failed: {str(e)}")
 
 # ================== AGENT DEFINITION ==================
 
@@ -248,6 +259,7 @@ When processing entries:
 - Extract clear, concise notes from the original input
 - Determine entry type (expense vs income) based on category and context
 - Use the add_ledger_entry tool to persist to database
+- Use sync_entry_to_frontend to trigger frontend updates
 
 For Urdu/Roman Urdu entries:
 - "khad" = fertilizer, "beej" = seeds, "dawa" = pesticide, "mazdoor" = labour
@@ -264,10 +276,10 @@ Always confirm successful entry addition and provide user feedback.""",
 
 # ================== PROCESSING FUNCTIONS ==================
 
-async def process_voice_entry(voice_data: VoiceEntryData) -> EntryResponse:
+async def process_voice_entry(voice_data: VoiceEntryData) -> str:
     """
     Process a voice-parsed entry through the agent.
-    Called when user confirms voice entry from frontend.
+    Returns entry ID on success.
     """
     print(f"\n🎙️ Processing voice entry: {voice_data.transcript}")
     
@@ -288,25 +300,26 @@ Device: {voice_data.device_id}
 Recorded: {voice_data.recorded_at}
 
 Steps:
-1. Validate the entry data
-2. Determine if this is income or expense
+1. Validate the entry data using validate_entry_data
+2. Determine if this is income or expense using determine_entry_type
 3. Add to ledger via add_ledger_entry
-4. Confirm successful addition
+4. Sync to frontend using sync_entry_to_frontend
+5. Confirm successful addition
 
-Respond in Urdu/English mix as appropriate."""
+Respond in Urdu/English mix as appropriate. Be brief and direct."""
     
     try:
         result = await runner.run(prompt)
         print(f"✅ Voice entry processed successfully")
-        return result
+        return str(result)
     except Exception as e:
         print(f"❌ Error processing voice entry: {str(e)}")
         raise
 
-async def process_manual_entry(manual_data: ManualEntryData) -> EntryResponse:
+async def process_manual_entry(manual_data: ManualEntryData) -> str:
     """
     Process a manual form entry through the agent.
-    Called when user submits the manual expense form.
+    Returns entry ID on success.
     """
     print(f"\n📝 Processing manual entry: {manual_data.category}")
     
@@ -326,24 +339,25 @@ User ID: {manual_data.user_id}
 Device: {manual_data.device_id}
 
 Steps:
-1. Validate all entry fields
-2. Determine entry type (should be "expense" for manual entries)
-3. Add to ledger via add_ledger_entry tool
-4. Confirm and sync to frontend
+1. Validate all entry fields using validate_entry_data
+2. Determine entry type using determine_entry_type (should be "expense")
+3. Add to ledger via add_ledger_entry
+4. Sync to frontend using sync_entry_to_frontend
+5. Confirm successful addition
 
-Respond in Urdu/English mix as appropriate."""
+Respond briefly and directly."""
     
     try:
         result = await runner.run(prompt)
         print(f"✅ Manual entry processed successfully")
-        return result
+        return str(result)
     except Exception as e:
         print(f"❌ Error processing manual entry: {str(e)}")
         raise
 
-async def process_batch_entries(entries: list[Dict[str, Any]], user_id: str) -> list[EntryResponse]:
+async def process_batch_entries(entries: list, user_id: str) -> list:
     """
-    Process multiple entries in batch (e.g., from offline sync).
+    Process multiple entries in batch.
     """
     print(f"\n📦 Processing batch of {len(entries)} entries for user {user_id}")
     
@@ -356,7 +370,7 @@ async def process_batch_entries(entries: list[Dict[str, Any]], user_id: str) ->
                 result = await process_manual_entry(ManualEntryData(**entry))
             results.append(result)
         except Exception as e:
-            print(f"⚠️ Error in batch processing: {str(e)}")
+            print(f"⚠️ Error in batch: {str(e)}")
             continue
     
     return results