| from __future__ import annotations |
|
|
| import re |
| from typing import Any |
|
|
| from tools.financial_calculator import calculate_profit, format_currency |
| from tools.financial_engine import build_forecast_brief, detect_forecast_intent |
|
|
|
|
| def build_followup_visual(question: str, report: dict[str, Any]) -> dict[str, Any] | None: |
| question_lower = question.lower().strip() |
| cashflow = report["cashflow"] |
| document = report.get("document_analysis") or {} |
| transaction = report.get("transaction_summary") or {} |
| loan = report["loan"] |
|
|
| if "where is most of my money going" in question_lower or "where is my money going" in question_lower: |
| items: list[dict[str, str]] = [] |
| for category in document.get("biggest_expense_categories", [])[:3]: |
| items.append({"label": "Expense category", "value": category}) |
| if transaction.get("biggest_expense_category"): |
| items.append({"label": "Largest transaction expense", "value": str(transaction["biggest_expense_category"])}) |
| if not items: |
| items.append({"label": "Expense ratio", "value": f"{cashflow['expense_ratio']:.1f}%"}) |
| return {"title": "Expense Focus", "items": items} |
|
|
| if "uploaded document" in question_lower or "document show" in question_lower or "income statement" in question_lower: |
| return { |
| "title": "Document Snapshot", |
| "items": [ |
| {"label": "Revenue", "value": format_currency(document.get("revenue", cashflow["revenue"]))}, |
| {"label": "Expenses", "value": format_currency(document.get("expenses", cashflow["expenses"]))}, |
| {"label": "Profit", "value": format_currency(document.get("profit", cashflow["profit"]))}, |
| {"label": "Profit Margin", "value": f"{document.get('profit_margin', cashflow['profit_margin']):.1f}%"}, |
| ], |
| } |
|
|
| if "sales drop" in question_lower or "drop next month" in question_lower: |
| revenue = cashflow["revenue"] |
| expenses = cashflow["expenses"] |
| ten_drop_profit = calculate_profit(revenue * 0.9, expenses) |
| twenty_drop_profit = calculate_profit(revenue * 0.8, expenses) |
| return { |
| "title": "Sales Drop Scenario", |
| "items": [ |
| {"label": "Current Profit", "value": format_currency(cashflow["profit"])}, |
| {"label": "Profit if Sales Drop 10%", "value": format_currency(ten_drop_profit)}, |
| {"label": "Profit if Sales Drop 20%", "value": format_currency(twenty_drop_profit)}, |
| ], |
| } |
|
|
| if "borrow" in question_lower or "loan" in question_lower: |
| match = re.search(r"(?:k|zmw)?\s*([\d,]+(?:\.\d+)?)", question_lower) |
| requested = float(match.group(1).replace(",", "")) if match else loan.get("requested_loan_amount", 0.0) |
| if requested > 0: |
| risk_level = "Higher than safer range" if requested > loan["suggested_loan_amount"] else "Within safer range" |
| return { |
| "title": "Borrowing Scenario", |
| "items": [ |
| {"label": "Current Profit", "value": format_currency(cashflow["profit"])}, |
| {"label": "Requested Loan", "value": format_currency(requested)}, |
| {"label": "Safer Level", "value": format_currency(loan["suggested_loan_amount"])}, |
| {"label": "Risk View", "value": risk_level}, |
| ], |
| } |
|
|
| if "cash flow" in question_lower: |
| return { |
| "title": "Cash Flow Snapshot", |
| "items": [ |
| {"label": "Revenue", "value": format_currency(cashflow["revenue"])}, |
| {"label": "Expenses", "value": format_currency(cashflow["expenses"])}, |
| {"label": "Profit", "value": format_currency(cashflow["profit"])}, |
| {"label": "Status", "value": cashflow["cash_flow_status"]}, |
| ], |
| } |
|
|
| return None |
|
|
|
|
| _CHART_TRIGGERS = [ |
| "graph", "chart", "plot", "visualize", "visualise", |
| "pie chart", "bar chart", "line graph", "line chart", |
| "show me a graph", "show me a chart", "show on a graph", "on a graph", |
| "show graph", "show chart", |
| ] |
|
|
|
|
| _CHART_AFFIRM = ("yes", "yeah", "yep", "sure", "ok", "okay", "please", "go ahead", "show me", "show it") |
|
|
|
|
| def _is_short_affirmation(q: str) -> bool: |
| if len(q.split()) > 6: |
| return False |
| return any(q == w or q.startswith(w + " ") or q.startswith(w + ",") or q.startswith(w + ".") for w in _CHART_AFFIRM) |
|
|
|
|
| def _last_user_forecast_intent(messages: list[dict] | None, current_question: str) -> dict[str, Any] | None: |
| if not messages: |
| return None |
| for prev in reversed(messages): |
| if prev.get("role") != "user": |
| continue |
| content = prev.get("content", "") |
| if content == current_question: |
| continue |
| intent = detect_forecast_intent(content) |
| if intent: |
| return intent |
| return None |
|
|
|
|
| def build_chart_data( |
| question: str, |
| report: dict[str, Any], |
| messages: list[dict] | None = None, |
| ) -> dict[str, Any] | None: |
| """Return chart data when the user asks for a graph/chart/plot.""" |
| q = question.lower().strip() |
|
|
| has_chart_kw = any(kw in q for kw in _CHART_TRIGGERS) |
| is_affirmation = _is_short_affirmation(q) |
|
|
| if not has_chart_kw and not is_affirmation: |
| return None |
|
|
| cashflow = report["cashflow"] |
| parsed = report.get("parsed_data", {}) |
| document = report.get("document_analysis") or {} |
| loan = report.get("loan", {}) |
|
|
| |
| |
| forecast_intent = detect_forecast_intent(question) if has_chart_kw else None |
| if forecast_intent is None and is_affirmation: |
| forecast_intent = _last_user_forecast_intent(messages, question) |
|
|
| |
| if not has_chart_kw and not forecast_intent: |
| return None |
|
|
| if forecast_intent: |
| brief = build_forecast_brief(report, forecast_intent["months"]) |
| if brief: |
| base = brief["base_case"]["months"] |
| opt = brief["optimistic"]["months"] |
| pess = brief["pessimistic"]["months"] |
| return { |
| "chart_type": "line", |
| "title": f"Forecast — Next {brief['months']} Months", |
| "months": [m["month_name"] for m in base], |
| "series": [ |
| {"name": "Base Case Profit", "values": [m["profit"] for m in base], "color": "#2563EB"}, |
| {"name": "Optimistic Profit", "values": [m["profit"] for m in opt], "color": "#10B981"}, |
| {"name": "Pessimistic Profit", "values": [m["profit"] for m in pess], "color": "#EF4444"}, |
| ], |
| } |
|
|
| |
| monthly = document.get("monthly_breakdown") or [] |
| if monthly and any(kw in q for kw in ["month", "trend", "over time", "history", "line"]): |
| months = [str(e.get("month", "")) for e in monthly] |
| revenues = [float(e.get("revenue") or e.get("Revenue") or 0) for e in monthly] |
| expenses_vals = [float(e.get("expenses") or e.get("Expenses") or 0) for e in monthly] |
| profits = [float(e.get("profit") or e.get("Profit") or 0) for e in monthly] |
| if any(revenues) or any(profits): |
| return { |
| "chart_type": "line", |
| "title": "Monthly Performance Trend", |
| "months": months, |
| "series": [ |
| {"name": "Revenue", "values": revenues, "color": "#10B981"}, |
| {"name": "Expenses", "values": expenses_vals, "color": "#EF4444"}, |
| {"name": "Profit", "values": profits, "color": "#2563EB"}, |
| ], |
| } |
|
|
| |
| if any(kw in q for kw in ["expense", "spending", "cost", "breakdown", "money going", "where"]): |
| breakdown = parsed.get("expenses_breakdown") or {} |
| items = [(k, float(v or 0)) for k, v in breakdown.items() if float(v or 0) > 0] |
| if items: |
| return { |
| "chart_type": "pie", |
| "title": "Expense Breakdown", |
| "labels": [k.replace("_", " ").title() for k, _ in items], |
| "values": [v for _, v in items], |
| } |
|
|
| |
| if any(kw in q for kw in ["loan", "borrow", "credit"]): |
| safe_loan = float(loan.get("suggested_loan_amount") or 0) |
| profit = float(cashflow.get("profit", 0)) |
| return { |
| "chart_type": "bar", |
| "title": "Loan Overview", |
| "labels": ["Monthly Profit", "Safe Loan Amount"], |
| "values": [profit, safe_loan], |
| "colors": ["#10B981", "#2563EB"], |
| } |
|
|
| |
| return { |
| "chart_type": "bar", |
| "title": "Cash Flow Overview", |
| "labels": ["Revenue", "Expenses", "Profit"], |
| "values": [ |
| float(cashflow.get("revenue", 0)), |
| float(cashflow.get("expenses", 0)), |
| float(cashflow.get("profit", 0)), |
| ], |
| "colors": ["#10B981", "#EF4444", "#2563EB"], |
| } |
|
|
