from features.base_feature import BaseFeature from tools.calculator import CalculatorTool _calc = CalculatorTool() class ExpensesFeature(BaseFeature): async def get_context(self, preferences_a: dict, preferences_b: dict, user_a_id: int = None, user_b_id: int = None) -> str: """ Pre-calculate expense totals using the safe Calculator tool. Inject exact figures so the LLM never does arithmetic. """ raw_a = preferences_a.get("raw_details", {}) raw_b = preferences_b.get("raw_details", {}) # Collect line items from both parties items = {} for raw in (raw_a, raw_b): expenses = raw.get("expenses") or raw.get("items") or raw.get("line_items") or [] if isinstance(expenses, list): for item in expenses: if isinstance(item, dict): name = item.get("name") or item.get("item") or "item" amount = item.get("amount") or item.get("cost") or item.get("price") or 0 try: amount = float(amount) except (TypeError, ValueError): amount = 0 if amount > 0: items[name] = items.get(name, 0) + amount lines = ["EXPENSE SPLITTING DOMAIN RULES:"] lines.append("• Use ONLY the pre-calculated amounts below. NEVER estimate or round differently.") lines.append("• Equal splits (50-50) are the default. Unequal splits need explicit justification.") lines.append("• After reaching agreement, include a 'settlement' key with who pays whom and how much.") lines.append("• Use the calculator results below — do NOT re-calculate with different numbers.") if items: total = sum(items.values()) lines.append(f"\nLine items (pre-verified by Calculator tool):") for name, amount in items.items(): half = amount / 2 lines.append(f" • {name}: ₹{amount:,.0f} → 50-50 split = ₹{half:,.2f} each") lines.append(f"\nTotal: ₹{total:,.0f} → 50-50 = ₹{total/2:,.2f} each") else: lines.append("\nNo line items found — extract amounts from the preferences and calculate fair splits.") # UPI info upi_a = raw_a.get("upi_id") or raw_a.get("upi") upi_b = raw_b.get("upi_id") or raw_b.get("upi") if upi_a: lines.append(f"\nParty A UPI ID: {upi_a}") if upi_b: lines.append(f"\nParty B UPI ID: {upi_b}") if upi_a or upi_b: lines.append("Include the relevant UPI ID in the settlement details of your proposal.") return "\n".join(lines) def format_resolution( self, resolution: dict, preferences_a: dict, preferences_b: dict ) -> str: status = resolution.get("status", "resolved") final = resolution.get("final_proposal", {}) details = final.get("details", {}) rounds = resolution.get("rounds_taken", "?") summary = resolution.get("summary", "Agreement reached") if status == "escalated": return ( f"⚠️ *Expenses — Human Review Needed*\n\n" f"_{summary}_\n\n" f"Agents couldn't fully agree in {rounds} round(s). " f"Please review the proposed split above." ) # Build breakdown table line_items = details.get("line_items") or details.get("items") or [] raw_settlement = details.get("settlement") or {} # Guard: settlement may be a string summary instead of a dict settlement = raw_settlement if isinstance(raw_settlement, dict) else {} payer = settlement.get("payer") or settlement.get("from") or "" payee = settlement.get("payee") or settlement.get("to") or "" amount = (settlement.get("amount") or details.get("amount") or details.get("total_owed") or (str(raw_settlement) if isinstance(raw_settlement, str) else "")) lines = ["💰 *Expenses Settled!*\n"] if line_items and isinstance(line_items, list): lines.append("📊 *Breakdown:*") for item in line_items: if isinstance(item, dict): name = item.get("name") or item.get("item", "Item") cost = item.get("amount") or item.get("cost") or "" split = item.get("split") or item.get("ratio") or "50-50" a_pays = item.get("party_a") or item.get("a_pays") or "" b_pays = item.get("party_b") or item.get("b_pays") or "" if a_pays and b_pays: lines.append(f" • {name} (₹{cost}) — {split} → A: ₹{a_pays} / B: ₹{b_pays}") else: lines.append(f" • {name}: {split} split") lines.append("") if payer and amount: lines.append(f"💸 *{payer} owes {payee}: ₹{amount}*") elif amount: lines.append(f"💸 *Settlement amount: ₹{amount}*") lines.append(f"\n⏱ Agreed in {rounds} round(s)") if summary and summary != "Agreement reached": lines.append(f"_{summary}_") return "\n".join(lines)