| from __future__ import annotations |
|
|
| import io |
| import re |
| from typing import Any |
|
|
| import pandas as pd |
|
|
| from tools.financial_calculator import calculate_profit, calculate_profit_margin |
| from tools.text_parser import parse_business_input |
|
|
|
|
| DOCUMENT_TYPES = [ |
| "Income Statement", |
| "Cash Flow Record", |
| "Sales Record", |
| "Expense Record", |
| "Bank Statement", |
| "Mobile Money Statement", |
| "Receipts", |
| "Inventory/Stock Record", |
| "Loan or Debt Record", |
| "General Business Notes", |
| "Other", |
| ] |
|
|
| |
|
|
| _DOC_PARSE_SYSTEM = """You are a financial document parser for African small business records. |
| Your job is to extract clean financial totals from raw document content — without double-counting. |
| |
| CRITICAL RULES: |
| - Identify TOTALS vs LINE ITEMS vs SUBTOTALS. Only extract the final TOTAL revenue and TOTAL expenses. |
| - In multi-sheet Excel files, summary sheets often repeat totals already captured in detail sheets — do NOT add them together. |
| - If a document has both an Income Statement and a Cash Flow Statement, prefer the Income Statement for revenue and expenses. |
| - Negative numbers in cash flow statements represent outflows (expenses/payments) — treat their absolute value as expenses. |
| - For Income Statements: total_expenses = ALL costs combined (cost of goods + operating expenses + any other costs). Do NOT split them. |
| - Opening/closing balances, averages, and running totals are NOT revenue or expenses — ignore them. |
| - Do NOT sum all numbers blindly. Read labels carefully to understand what each number represents. |
| - If a value is unclear or cannot be reliably determined, return null — do NOT guess. |
| - Auto-detect the real document type regardless of what the user labeled it. |
| - Amounts are in Zambian Kwacha (K / ZMW) unless another currency is clearly stated. |
| - IMPORTANT: Every value in the JSON must be a plain pre-computed number. NEVER write arithmetic expressions like "18600 + 46980" — compute the sum yourself and write only the result. |
| - CRITICAL: Extract numbers ONLY from the DOCUMENT CONTENT provided by the user. NEVER reuse example numbers or labels from the schema/template. If the document does not contain a value, return null — do NOT invent or carry over numbers from prior conversations.""" |
|
|
| _DOC_PARSE_USER_TEMPLATE = """Filename: {filename} |
| User-labeled type: {doc_type} |
| |
| DOCUMENT CONTENT: |
| {content} |
| |
| Extract financial data and return ONLY a JSON object. No markdown fences, no extra text. |
| All numeric fields must be plain pre-computed numbers (never expressions like "1000 + 2000"). |
| |
| CRITICAL: The schema below uses <PLACEHOLDER> tokens to show the SHAPE of the JSON only. |
| You MUST replace every <PLACEHOLDER> with a value computed from the actual DOCUMENT CONTENT above. |
| NEVER copy these placeholder strings or any example values into your output — extract from the document. |
| |
| {{ |
| "detected_document_type": "<one of: Income Statement | Sales Record | Cash Flow Record | Expense Record | Bank Statement | Other>", |
| "currency": "<currency code from document, e.g. K>", |
| "period_description": "<human readable period from document, or null>", |
| "period_start": "<YYYY-MM-DD or null>", |
| "period_end": "<YYYY-MM-DD or null>", |
| "total_revenue": <number computed from document, or null>, |
| "total_expenses": <number computed from document, or null>, |
| "profit": <number computed from document, or null>, |
| "debt_balance": <number from document, or null>, |
| "expense_breakdown": [ |
| {{"category": "<expense category name from document>", "amount": <number from document>}} |
| ], |
| "product_breakdown": [ |
| {{"product": "<product name from document>", "units_sold": <number>, "unit_price": <number>, "revenue": <number>}} |
| ], |
| "monthly_breakdown": [{{"month": "<month label from document>", "revenue": <number>, "expenses": <number>, "profit": <number>}}], |
| "key_observations": ["<short factual observation grounded in the document>"], |
| "confidence": "<high | medium | low>", |
| "warnings": [] |
| }} |
| |
| Rules for detected_document_type: |
| - If the document has Product/Item columns with Units Sold and/or Unit Price → "Sales Record" |
| - If it has Revenue, Expenses, Profit rows → "Income Statement" |
| - If it has Cash In / Cash Out / Bank transactions → "Cash Flow Record" or "Bank Statement" |
| - Choose the most accurate type — do NOT default to "Income Statement". |
| |
| Rules for expense_breakdown: |
| - List each distinct expense LINE ITEM (not subtotals). Return [] if none visible. |
| - Amounts must be totals across the full period (not per month). |
| |
| Rules for product_breakdown: |
| - ONLY populate when detected_document_type is "Sales Record" or the document clearly lists products with quantities/prices. |
| - For each product row: "product" = name, "units_sold" = quantity, "unit_price" = price per unit, "revenue" = units × price. |
| - If revenue per product is stated directly, use that value. |
| - Return [] for all other document types. |
| |
| Now extract the actual data from the document above and return your JSON:""" |
|
|
|
|
| def _normalize_currency(raw: Any) -> str: |
| """Force any LLM-returned currency string down to a short symbol like 'K'. |
| Prevents 'Zambian Kwacha (K)' + amount = 'Zambian Kwacha (K)18,600' double-prefix. |
| """ |
| if not raw: |
| return "K" |
| s = str(raw).strip() |
| upper = s.upper() |
| if "ZMW" in upper or "KWACHA" in upper or upper == "K": |
| return "K" |
| if upper in {"USD", "US$", "$", "EUR", "€", "GBP", "£"}: |
| return "K" |
| |
| m = re.search(r"\(([^)]{1,3})\)", s) |
| if m: |
| return m.group(1) |
| return "K" |
|
|
|
|
| |
|
|
| def _extract_excel_as_text(file_bytes: bytes) -> tuple[str, int]: |
| """Read all sheets from an xlsx file and return as structured text + row count.""" |
| from openpyxl import load_workbook |
|
|
| wb = load_workbook(io.BytesIO(file_bytes), data_only=True) |
| sheets_text: list[str] = [] |
| total_rows = 0 |
|
|
| for sheet_name in wb.sheetnames: |
| ws = wb[sheet_name] |
| sheet_lines = [f"=== SHEET: {sheet_name} ==="] |
|
|
| for row in ws.iter_rows(values_only=True): |
| if all(cell is None or str(cell).strip() == "" for cell in row): |
| continue |
| row_text = " | ".join(str(cell) if cell is not None else "" for cell in row) |
| if row_text.strip(" |"): |
| sheet_lines.append(row_text) |
| total_rows += 1 |
|
|
| if len(sheet_lines) > 1: |
| sheets_text.append("\n".join(sheet_lines)) |
|
|
| return "\n\n".join(sheets_text), total_rows |
|
|
|
|
| def _parse_excel_by_sheets(file_bytes: bytes, filename: str = "") -> dict[str, Any] | None: |
| """ |
| Sheet-aware structured extraction for multi-sheet Excel workbooks. |
| Reads Income Statement / Cash Flow / Debt sheets independently to prevent mixing. |
| Returns a complete result dict on success, or None to fall through to the LLM path. |
| """ |
| try: |
| from openpyxl import load_workbook |
| except ImportError: |
| return None |
|
|
| try: |
| wb = load_workbook(io.BytesIO(file_bytes), data_only=True) |
| except Exception: |
| return None |
|
|
| names_lower = {s.lower().strip(): s for s in wb.sheetnames} |
|
|
| def find_sheet(*keywords: str): |
| for kw in keywords: |
| for nl, original in names_lower.items(): |
| if kw in nl: |
| return wb[original] |
| return None |
|
|
| ws_income = find_sheet("income statement", "income", "profit & loss", "p&l", "pnl") |
| ws_debt = find_sheet("debt", "loan", "liabilit", "payable") |
| ws_cf = find_sheet("cash flow", "cashflow") |
|
|
| |
| if ws_income is None and len(wb.sheetnames) == 1: |
| ws_income = wb[wb.sheetnames[0]] |
|
|
| if ws_income is None: |
| return None |
|
|
| |
| def to_num(v) -> float | None: |
| if v is None: |
| return None |
| if isinstance(v, (int, float)): |
| return float(v) |
| s = re.sub(r"[KZMWkzmw\s,]", "", str(v)) |
| if s.startswith("(") and s.endswith(")"): |
| s = "-" + s[1:-1] |
| try: |
| return float(s) |
| except (ValueError, TypeError): |
| return None |
|
|
| def label_matches(label: str, keywords: list[str]) -> bool: |
| nl = re.sub(r"[^a-z0-9/(). ]+", " ", label.lower()).strip() |
| for kw in keywords: |
| nk = re.sub(r"[^a-z0-9/(). ]+", " ", kw.lower()).strip() |
| if nk == nl or nk in nl: |
| return True |
| return False |
|
|
| def find_total_col_idx(ws) -> int | None: |
| """ |
| Scan the first 5 rows (skipping column 0, which holds row labels) to find |
| a column whose header contains 'total' or 'sum' but NOT 'average'. |
| Returns the 0-based column index, or None if not found. |
| """ |
| TOTAL_KWS = ("total", "sum", "full year", "6-month", "6 month", "period total") |
| SKIP_KWS = ("average", "avg", "mean") |
| for row in ws.iter_rows(max_row=5, values_only=True): |
| for idx, cell in enumerate(row): |
| if idx == 0 or cell is None: |
| continue |
| s = str(cell).strip().lower() |
| if any(kw in s for kw in TOTAL_KWS) and not any(kw in s for kw in SKIP_KWS): |
| return idx |
| return None |
|
|
| def find_row_value(ws, keywords: list[str], total_col_idx: int | None = None) -> float | None: |
| """ |
| Scan worksheet rows; return the period total of the best-matching row. |
| |
| Priority order: |
| 1. Cell in the explicitly identified Total column (most reliable). |
| 2. Sum all numeric cells in the row when the row has ≥ 4 numbers |
| (handles multi-month statements that have no separate Total column). |
| 3. Last numeric cell in the row (single-value fallback). |
| """ |
| candidates: list[tuple[float, bool]] = [] |
| for row in ws.iter_rows(values_only=True): |
| label = next((str(c).strip() for c in row if c is not None and str(c).strip()), None) |
| if not label or not label_matches(label, keywords): |
| continue |
|
|
| val: float | None = None |
|
|
| |
| if total_col_idx is not None and total_col_idx < len(row): |
| val = to_num(row[total_col_idx]) |
|
|
| |
| if val is None: |
| nums = [n for n in (to_num(c) for c in row) if n is not None] |
| if nums: |
| |
| |
| val = sum(nums) if len(nums) >= 4 else nums[-1] |
|
|
| if val is None: |
| continue |
| candidates.append((val, "total" in label.lower())) |
|
|
| if not candidates: |
| return None |
| |
| for val, is_total in candidates: |
| if is_total: |
| return val |
| return candidates[0][0] |
|
|
| def classify_income_row(label: str) -> str | None: |
| """Classify an income-statement line item as revenue/expense/ignore.""" |
| nl = re.sub(r"[^a-z0-9/(). ]+", " ", label.lower()).strip() |
| if not nl: |
| return None |
|
|
| |
| |
| if "total" in nl or "subtotal" in nl or "grand total" in nl: |
| return None |
|
|
| ignore_tokens = ( |
| "line item", |
| "key ratios", |
| "analysis", |
| "auditor", |
| "notes", |
| "margin", |
| "ratio", |
| "profit", |
| "loss", |
| "gross profit", |
| "operating profit", |
| "net profit", |
| "total revenue", |
| "total income", |
| "total sales", |
| "total expenses", |
| "total operating expenses", |
| "total other expenses", |
| "total cost of goods sold", |
| "total costs", |
| ) |
| if any(tok in nl for tok in ignore_tokens): |
| return None |
| if nl in {"revenue", "operating expenses"}: |
| return None |
|
|
| revenue_tokens = ("sales", "revenue", "income", "turnover", "receipts") |
| expense_tokens = ( |
| "expense", |
| "expenses", |
| "cost", |
| "rent", |
| "wage", |
| "salary", |
| "electricity", |
| "water", |
| "utility", |
| "transport", |
| "loan", |
| "repayment", |
| "airtime", |
| "marketing", |
| "maintenance", |
| "repair", |
| "interest", |
| "fee", |
| "freight", |
| "delivery", |
| "purchase", |
| "supplier", |
| ) |
| if any(tok in nl for tok in revenue_tokens): |
| return "revenue" |
| if any(tok in nl for tok in expense_tokens): |
| return "expense" |
| return None |
|
|
| def row_period_value(row, total_col_idx: int | None = None) -> float | None: |
| """Return one row's period value from total column or monthly cells.""" |
| value: float | None = None |
| if total_col_idx is not None and total_col_idx < len(row): |
| value = to_num(row[total_col_idx]) |
| if value is not None: |
| return value |
|
|
| nums = [n for n in (to_num(c) for c in row) if n is not None] |
| if not nums: |
| return None |
| return sum(nums) if len(nums) >= 4 else nums[-1] |
|
|
| def derive_totals_from_income_rows( |
| ws, total_col_idx: int | None = None |
| ) -> tuple[float | None, float | None, dict[str, float]]: |
| """ |
| Derive revenue/expenses from detailed line items when explicit total rows |
| are missing or formula cells are not available in data-only mode. |
| """ |
| revenue_sum = 0.0 |
| expense_sum = 0.0 |
| seen_revenue = False |
| seen_expense = False |
|
|
| for row in ws.iter_rows(values_only=True): |
| label = next((str(c).strip() for c in row if c is not None and str(c).strip()), None) |
| if not label: |
| continue |
| kind = classify_income_row(label) |
| if kind is None: |
| continue |
|
|
| val = row_period_value(row, total_col_idx) |
| if val is None: |
| continue |
| val = abs(val) |
| if kind == "revenue": |
| revenue_sum += val |
| seen_revenue = True |
| elif kind == "expense": |
| expense_sum += val |
| seen_expense = True |
|
|
| expense_breakdown: dict[str, float] = {} |
|
|
| def normalize_expense_label(label: str) -> str: |
| clean = re.sub(r"\s+", " ", label.strip()) |
| return clean.title() if clean.islower() else clean |
|
|
| for row in ws.iter_rows(values_only=True): |
| label = next((str(c).strip() for c in row if c is not None and str(c).strip()), None) |
| if not label: |
| continue |
| if classify_income_row(label) != "expense": |
| continue |
| val = row_period_value(row, total_col_idx) |
| if val is None: |
| continue |
| expense_breakdown[normalize_expense_label(label)] = round(abs(val), 2) |
|
|
| return ( |
| revenue_sum if seen_revenue else None, |
| expense_sum if seen_expense else None, |
| expense_breakdown, |
| ) |
|
|
| def infer_period_description(ws) -> str | None: |
| """Infer period label like 'Nov 2025 – Apr 2026' from any leading metadata |
| or column headers. Returns the chronological span (earliest – latest).""" |
| month_pattern = re.compile( |
| r"\b(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\s+\d{4}\b", |
| flags=re.IGNORECASE, |
| ) |
| month_order = { |
| m: i + 1 |
| for i, m in enumerate( |
| ["jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec"] |
| ) |
| } |
|
|
| def sort_key(token: str) -> tuple[int, int]: |
| mtxt, ytxt = token.split() |
| return int(ytxt), month_order.get(mtxt[:3].lower(), 0) |
|
|
| tokens: list[str] = [] |
| for row in ws.iter_rows(max_row=10, values_only=True): |
| for cell in row: |
| if cell is None: |
| continue |
| text = str(cell) |
| if not text.strip(): |
| continue |
| tokens.extend(m.group(0) for m in month_pattern.finditer(text)) |
|
|
| if not tokens: |
| return None |
| unique = sorted({t for t in tokens}, key=sort_key) |
| if len(unique) == 1: |
| return unique[0] |
| return f"{unique[0]} – {unique[-1]}" |
|
|
| def count_non_empty_rows(ws) -> int: |
| count = 0 |
| for row in ws.iter_rows(values_only=True): |
| if any(cell is not None and str(cell).strip() != "" for cell in row): |
| count += 1 |
| return count |
|
|
| |
| |
| |
| income_total_col = find_total_col_idx(ws_income) |
|
|
| revenue = find_row_value(ws_income, [ |
| "total revenue", "total sales", "total income", "gross revenue", |
| ], income_total_col) |
| expenses = find_row_value(ws_income, [ |
| "total expenses", "total costs", "total operating expenses", |
| "total expenditure", "total cost of sales", |
| ], income_total_col) |
| profit = find_row_value(ws_income, [ |
| "net profit / (loss)", "net profit/(loss)", "net profit", "net loss", "net income", |
| ], income_total_col) |
| if profit is None: |
| |
| profit = find_row_value(ws_income, [ |
| "profit / (loss)", "profit/(loss)", "operating profit", |
| ], income_total_col) |
|
|
| |
| derived_revenue, derived_expenses, derived_expense_breakdown = derive_totals_from_income_rows( |
| ws_income, income_total_col |
| ) |
| if revenue is None and derived_revenue is not None: |
| revenue = derived_revenue |
| if expenses is None and derived_expenses is not None: |
| expenses = derived_expenses |
|
|
| if revenue is None: |
| return None |
|
|
| revenue = abs(revenue) |
| if expenses is not None: |
| expenses = abs(expenses) |
| elif profit is not None: |
| expenses = revenue - profit |
| else: |
| expenses = 0.0 |
|
|
| if profit is None: |
| profit = revenue - expenses |
| |
|
|
| |
| |
| mismatch = abs((revenue - expenses) - profit) |
| tolerance = max(1.0, abs(revenue) * 0.01) |
| if mismatch > tolerance: |
| recomputed_expenses = revenue - profit |
| if recomputed_expenses >= 0: |
| expenses = recomputed_expenses |
|
|
| |
| debt = 0.0 |
| if ws_debt is not None: |
| total_debt = find_row_value(ws_debt, [ |
| "total debt", "total loans", "total liabilities", "total balance", |
| "total", "grand total", |
| ]) |
| if total_debt is not None: |
| debt = abs(total_debt) |
| else: |
| |
| HEADER_LABELS = { |
| "date", "lender", "creditor", "description", "type", |
| "debt type", "loan type", "amount", "balance", "notes", |
| } |
| for row in ws_debt.iter_rows(values_only=True): |
| label = next((str(c).strip() for c in row if c is not None and str(c).strip()), None) |
| if not label or label.lower().strip() in HEADER_LABELS: |
| continue |
| val = row_last_num(row) |
| if val is not None and val > 0: |
| debt += val |
|
|
| |
| cash_inflow, cash_outflow = revenue, expenses |
| if ws_cf is not None: |
| cf_in = find_row_value(ws_cf, [ |
| "total cash inflows", "total inflows", "total receipts", |
| ]) |
| cf_out = find_row_value(ws_cf, [ |
| "total cash outflows", "total outflows", "total payments", |
| ]) |
| if cf_in is not None: |
| cash_inflow = abs(cf_in) |
| if cf_out is not None: |
| cash_outflow = abs(cf_out) |
|
|
| margin = calculate_profit_margin(revenue, profit) |
| if derived_expense_breakdown: |
| biggest_cats = [ |
| c for c, _ in sorted(derived_expense_breakdown.items(), key=lambda x: x[1], reverse=True)[:3] |
| ] |
| else: |
| biggest_cats = [] |
| sheets_used = { |
| "income": ws_income.title, |
| "debt": ws_debt.title if ws_debt else None, |
| "cashflow": ws_cf.title if ws_cf else None, |
| } |
| period_description = infer_period_description(ws_income) |
| rows_analyzed = count_non_empty_rows(ws_income) |
|
|
| result = _blank_result("xlsx", filename, "Income Statement") |
| result.update({ |
| "rows_analyzed": rows_analyzed, |
| "period_description": period_description, |
| "date_range": period_description or "Unknown", |
| "revenue": round(revenue, 2), |
| "expenses": round(expenses, 2), |
| "profit": round(profit, 2), |
| "debt": round(debt, 2), |
| "cash_inflow": round(cash_inflow, 2), |
| "cash_outflow": round(cash_outflow, 2), |
| "profit_margin": round(margin, 2), |
| "confidence": "high", |
| "used_llm": False, |
| "expenses_breakdown": derived_expense_breakdown, |
| "biggest_expense_categories": biggest_cats, |
| "summary": ( |
| f"Structured multi-sheet extraction from '{ws_income.title}' " |
| f"(Revenue: K{revenue:,.2f}, Expenses: K{expenses:,.2f}, " |
| f"Profit: K{profit:,.2f}, Margin: {margin:.1f}%)." |
| ), |
| "warnings": [], |
| "metadata": {"sheets_used": sheets_used}, |
| }) |
| return result |
|
|
|
|
| def _extract_csv_as_text(file_bytes: bytes) -> tuple[str, int]: |
| frame = pd.read_csv(io.BytesIO(file_bytes)) |
| return frame.to_string(max_rows=300), len(frame) |
|
|
|
|
| def _extract_pdf_as_text(file_bytes: bytes) -> tuple[str, int]: |
| from pypdf import PdfReader |
| reader = PdfReader(io.BytesIO(file_bytes)) |
| pages = [page.extract_text() or "" for page in reader.pages] |
| return "\n\n".join(pages), len(pages) |
|
|
|
|
| def _truncate(text: str, max_chars: int = 10000) -> str: |
| if len(text) <= max_chars: |
| return text |
| return text[:8000] + "\n\n... [content truncated] ...\n\n" + text[-2000:] |
|
|
|
|
| |
|
|
| def _sanitize_llm_json(text: str) -> str: |
| """Replace arithmetic expressions like '18600 + 46980' with their computed value.""" |
| import re |
| def _eval_expr(m: re.Match) -> str: |
| try: |
| val = eval(m.group(0), {"__builtins__": {}}) |
| return str(val) |
| except Exception: |
| return m.group(0) |
| |
| return re.sub(r"\d[\d\s]*[\+\-\*\/][\d\s]+\d", _eval_expr, text) |
|
|
|
|
| def _llm_parse_document(raw_text: str, document_type: str, filename: str) -> dict[str, Any] | None: |
| """Call the LLM to intelligently extract financial data from raw document text. |
| |
| Raises exceptions so callers can surface the real error rather than silently falling back. |
| """ |
| from agents import get_chat_model, _extract_json_block |
| from langchain_core.messages import HumanMessage, SystemMessage |
|
|
| llm = get_chat_model() |
| if llm is None: |
| raise RuntimeError( |
| "LLM not configured (set LLM_PROVIDER=amd with AMD_BASE_URL/AMD_MODEL " |
| "or LLM_PROVIDER=openai with OPENAI_API_KEY, OPENAI_BASE_URL, MODEL_NAME)." |
| ) |
|
|
| user_content = _DOC_PARSE_USER_TEMPLATE.format( |
| filename=filename, |
| doc_type=document_type, |
| content=_truncate(raw_text), |
| ) |
|
|
| response = llm.invoke( |
| [SystemMessage(content=_DOC_PARSE_SYSTEM), HumanMessage(content=user_content)], |
| temperature=0.1, |
| max_tokens=1800, |
| ) |
| raw_response = getattr(response, "content", None) |
| if not isinstance(raw_response, str) or not raw_response.strip(): |
| raise ValueError("LLM returned an empty response.") |
|
|
| sanitized = _sanitize_llm_json(raw_response) |
| parsed = _extract_json_block(sanitized) |
| if parsed is None: |
| raise ValueError(f"Could not parse JSON from LLM response. First 300 chars: {sanitized[:300]!r}") |
| return parsed |
|
|
|
|
| |
|
|
| def _blank_result(source_type: str, filename: str = "", document_type: str = "Other") -> dict[str, Any]: |
| return { |
| "document_type": document_type, |
| "source_type": source_type, |
| "file_name": filename, |
| "rows_analyzed": 0, |
| "pages_analyzed": 0, |
| "characters_analyzed": 0, |
| "revenue": 0.0, |
| "expenses": 0.0, |
| "profit": 0.0, |
| "debt": 0.0, |
| "cash_inflow": 0.0, |
| "cash_outflow": 0.0, |
| "profit_margin": 0.0, |
| "date_range": "Unknown", |
| "period_description": None, |
| "detected_dates": [], |
| "detected_columns": {}, |
| "summary": "", |
| "warnings": [], |
| "preview_rows": [], |
| "expenses_breakdown": {}, |
| "biggest_expense_categories": [], |
| "confidence": "low", |
| "monthly_breakdown": None, |
| "key_observations": [], |
| "currency": "K", |
| "used_llm": False, |
| "metadata": {}, |
| |
| "product_breakdown": None, |
| "top_product": None, |
| "slowest_product": None, |
| } |
|
|
|
|
| def _map_llm_result( |
| llm: dict[str, Any], |
| source_type: str, |
| filename: str, |
| rows_analyzed: int = 0, |
| ) -> dict[str, Any]: |
| """Map LLM JSON output to the standard document result schema.""" |
| revenue = float(llm.get("total_revenue") or 0) |
| expenses = float(llm.get("total_expenses") or 0) |
| profit_raw = llm.get("profit") |
| profit = float(profit_raw) if profit_raw is not None else calculate_profit(revenue, expenses) |
| debt = float(llm.get("debt_balance") or 0) |
|
|
| detected_type = llm.get("detected_document_type") or "Other" |
| period_desc = llm.get("period_description") or None |
| currency = _normalize_currency(llm.get("currency")) |
|
|
| start = llm.get("period_start") |
| end = llm.get("period_end") |
| if start and end: |
| date_range = f"{start} to {end}" |
| elif period_desc: |
| date_range = period_desc |
| else: |
| date_range = "Unknown" |
|
|
| margin = calculate_profit_margin(revenue, profit) |
| confidence = llm.get("confidence", "low") |
| warnings = list(llm.get("warnings") or []) |
| observations = list(llm.get("key_observations") or []) |
|
|
| if confidence == "low": |
| warnings.insert(0, "Low confidence extraction — please verify figures manually before relying on them.") |
| elif confidence == "medium": |
| warnings.insert(0, "Partial extraction — some figures may need manual verification.") |
|
|
| |
| raw_breakdown_list = llm.get("expense_breakdown") or [] |
| expenses_breakdown: dict[str, float] = {} |
| for item in raw_breakdown_list: |
| if isinstance(item, dict): |
| cat = str(item.get("category") or "").strip() |
| amt = item.get("amount") |
| if cat and amt is not None: |
| try: |
| expenses_breakdown[cat] = round(float(amt), 2) |
| except (TypeError, ValueError): |
| pass |
|
|
| |
| if expenses_breakdown: |
| biggest_cats = [c for c, _ in sorted(expenses_breakdown.items(), key=lambda x: x[1], reverse=True)[:3]] |
| else: |
| biggest_cats = [] |
|
|
| |
| raw_products = llm.get("product_breakdown") or [] |
| product_breakdown: list[dict] | None = None |
| top_product: str | None = None |
| slowest_product: str | None = None |
|
|
| if raw_products and isinstance(raw_products, list): |
| cleaned: list[dict] = [] |
| for item in raw_products: |
| if not isinstance(item, dict): |
| continue |
| name = str(item.get("product") or "").strip() |
| if not name: |
| continue |
| units = float(item.get("units_sold") or 0) |
| price = float(item.get("unit_price") or 0) |
| |
| rev_stated = item.get("revenue") |
| rev = float(rev_stated) if rev_stated is not None else round(units * price, 2) |
| cleaned.append({"product": name, "units_sold": units, "unit_price": price, "revenue": rev}) |
| if cleaned: |
| by_rev = sorted(cleaned, key=lambda x: x["revenue"], reverse=True) |
| product_breakdown = by_rev |
| top_product = by_rev[0]["product"] |
| slowest_product = by_rev[-1]["product"] if len(by_rev) > 1 else None |
|
|
| summary = ( |
| f"LLM-extracted {detected_type.lower()} covering {period_desc or 'an unknown period'}. " |
| f"Revenue: {currency}{revenue:,.2f}, Expenses: {currency}{expenses:,.2f}, " |
| f"Profit: {currency}{profit:,.2f} ({margin:.1f}% margin)." |
| ) |
|
|
| result = _blank_result(source_type, filename, detected_type) |
| result.update({ |
| "rows_analyzed": rows_analyzed, |
| "revenue": round(revenue, 2), |
| "expenses": round(expenses, 2), |
| "profit": round(profit, 2), |
| "debt": round(debt, 2), |
| "cash_inflow": round(revenue, 2), |
| "cash_outflow": round(expenses, 2), |
| "profit_margin": round(margin, 2), |
| "date_range": date_range, |
| "period_description": period_desc, |
| "summary": summary, |
| "warnings": warnings, |
| "expenses_breakdown": expenses_breakdown, |
| "biggest_expense_categories": biggest_cats, |
| "confidence": confidence, |
| "monthly_breakdown": llm.get("monthly_breakdown"), |
| "key_observations": observations, |
| "currency": currency, |
| "used_llm": True, |
| "product_breakdown": product_breakdown, |
| "top_product": top_product, |
| "slowest_product": slowest_product, |
| }) |
| return result |
|
|
|
|
| |
|
|
| def _normalize_name(value: str) -> str: |
| return re.sub(r"[^a-z0-9]+", " ", str(value).strip().lower()).strip() |
|
|
|
|
| def _find_matching_column(columns: list[str], keywords: list[str]) -> str | None: |
| normalized_map = {column: _normalize_name(column) for column in columns} |
| for keyword in keywords: |
| nk = _normalize_name(keyword) |
| for original, normalized in normalized_map.items(): |
| if normalized == nk: |
| return original |
| for keyword in keywords: |
| nk = _normalize_name(keyword) |
| for original, normalized in normalized_map.items(): |
| if nk in normalized: |
| return original |
| return None |
|
|
|
|
| def _to_numeric(series: pd.Series) -> pd.Series: |
| cleaned = ( |
| series.astype(str) |
| .str.replace(",", "", regex=False) |
| .str.replace(r"[^0-9.\-]", "", regex=True) |
| .replace("", pd.NA) |
| ) |
| return pd.to_numeric(cleaned, errors="coerce") |
|
|
|
|
| def _extract_date_range(frame: pd.DataFrame, date_column: str | None) -> tuple[str, list[str]]: |
| if not date_column: |
| return "Unknown", [] |
| dates = pd.to_datetime(frame[date_column], errors="coerce").dropna() |
| if dates.empty: |
| return "Unknown", [] |
| start = dates.min().date().isoformat() |
| end = dates.max().date().isoformat() |
| sample_dates = sorted({d.date().isoformat() for d in dates.head(5)}) |
| return f"{start} to {end}", sample_dates |
|
|
|
|
| def _document_type_columns(frame: pd.DataFrame, document_type: str) -> dict[str, str | None]: |
| columns = frame.columns.tolist() |
| return { |
| "date": _find_matching_column(columns, ["date", "period", "day", "month"]), |
| "description": _find_matching_column(columns, ["description", "details", "notes"]), |
| "product": _find_matching_column(columns, ["product", "item", "product name", "item name", "service", "goods"]), |
| "category": _find_matching_column(columns, ["category", "type", "transaction type", "class"]), |
| "amount": _find_matching_column(columns, ["amount", "value", "total", "price"]), |
| "revenue": _find_matching_column(columns, ["revenue", "sales", "income", "total sales"]), |
| "expense": _find_matching_column(columns, ["expense", "expenses", "cost", "operating expense", "cash out"]), |
| "profit": _find_matching_column(columns, ["profit", "net profit", "gross profit"]), |
| "debt": _find_matching_column(columns, ["debt", "loan", "balance", "repayment"]), |
| "cash_in": _find_matching_column(columns, ["cash in", "money in", "inflow", "credit"]), |
| "cash_out": _find_matching_column(columns, ["cash out", "money out", "outflow", "debit"]), |
| "quantity": _find_matching_column(columns, ["quantity", "qty", "units", "units sold", "qty sold"]), |
| "unit_price": _find_matching_column(columns, ["unit price", "price per unit", "rate", "price"]), |
| } |
|
|
|
|
| def _detect_is_sales_record(frame: pd.DataFrame) -> bool: |
| """Return True when the DataFrame looks like a product sales breakdown. |
| |
| Signals: has a product/item name column AND at least one of units-sold or |
| unit-price columns — the hallmark of a sales-by-product table rather than a |
| P&L or cash-flow statement. |
| """ |
| cols_lower = [c.lower().strip() for c in frame.columns] |
| product_hit = any( |
| kw in col |
| for col in cols_lower |
| for kw in ("product", "item name", "item", "goods", "service") |
| ) |
| unit_hit = any( |
| kw in col |
| for col in cols_lower |
| for kw in ("units sold", "qty sold", "quantity sold", "units", "qty") |
| ) |
| price_hit = any( |
| kw in col |
| for col in cols_lower |
| for kw in ("unit price", "price per unit", "price", "rate") |
| ) |
| revenue_expense_hit = any( |
| kw in col |
| for col in cols_lower |
| for kw in ("revenue", "expenses", "profit", "net profit") |
| ) |
| |
| |
| return product_hit and (unit_hit or price_hit) and not revenue_expense_hit |
|
|
|
|
| REVENUE_KEYWORDS = ["revenue", "sales", "sale", "income", "money in", "inflow", "customer payment", "credit", "deposit"] |
| EXPENSE_KEYWORDS = ["expense", "expenses", "cost", "costs", "purchase", "purchases", "rent", "stock", |
| "transport", "supplier", "money out", "outflow", "debit", "fees", "utilities", "wages"] |
| DEBT_KEYWORDS = ["debt", "loan", "borrowed", "payable", "supplier debt", "owed", "repayment", "balance due"] |
|
|
|
|
| def _classify_from_text(text: str) -> str: |
| lowered = text.lower() |
| if any(k in lowered for k in DEBT_KEYWORDS): |
| return "debt" |
| if any(k in lowered for k in REVENUE_KEYWORDS): |
| return "revenue" |
| if any(k in lowered for k in EXPENSE_KEYWORDS): |
| return "expense" |
| return "unknown" |
|
|
|
|
| def classify_transaction_row(row: dict[str, Any], document_type: str) -> str: |
| text = " ".join(str(v) for v in row.values() if v is not None and str(v).strip()) |
| base = _classify_from_text(text) |
| document_type = document_type or "Other" |
| if document_type == "Sales Record": |
| return "revenue" |
| if document_type == "Expense Record": |
| return "expense" |
| if document_type == "Receipts": |
| return "expense" if "sales receipt" not in text.lower() else "revenue" |
| if document_type == "Loan or Debt Record": |
| return "debt" |
| if document_type == "Inventory/Stock Record": |
| return "expense" |
| if document_type == "Cash Flow Record": |
| return base |
| if document_type in {"Bank Statement", "Mobile Money Statement"} and base == "unknown": |
| tx_type = str(row.get("Transaction Type", row.get("Type", ""))).lower() |
| if "credit" in tx_type or "deposit" in tx_type: |
| return "revenue" |
| if "debit" in tx_type or "withdraw" in tx_type or "cash-out" in tx_type: |
| return "expense" |
| return base |
|
|
|
|
| def _keyword_parse_frame(data: pd.DataFrame, document_type: str) -> dict[str, Any]: |
| """Original keyword-based parser — used as fallback when LLM is unavailable.""" |
| result = _blank_result("table", document_type=document_type) |
| if data.empty: |
| result["summary"] = "The uploaded file did not contain any rows to analyze." |
| result["warnings"].append("No rows were found in the uploaded document.") |
| return result |
|
|
| frame = data.copy() |
| frame.columns = [str(c).strip() for c in frame.columns] |
| detected = _document_type_columns(frame, document_type) |
| result["detected_columns"] = {k: v for k, v in detected.items() if v} |
| result["rows_analyzed"] = int(len(frame)) |
| result["preview_rows"] = frame.head(5).fillna("").to_dict(orient="records") |
|
|
| date_range, detected_dates = _extract_date_range(frame, detected["date"]) |
| result["date_range"] = date_range |
| result["detected_dates"] = detected_dates |
|
|
| revenue = expenses = debt = cash_inflow = cash_outflow = 0.0 |
| warnings: list[str] = [] |
|
|
| if detected["revenue"] and detected["revenue"] != detected["cash_in"]: |
| revenue += float(_to_numeric(frame[detected["revenue"]]).fillna(0).sum()) |
| if detected["expense"] and detected["expense"] != detected["cash_out"]: |
| expenses += float(_to_numeric(frame[detected["expense"]]).fillna(0).sum()) |
| if detected["debt"]: |
| debt += float(_to_numeric(frame[detected["debt"]]).fillna(0).sum()) |
| if detected["cash_in"]: |
| cash_inflow += float(_to_numeric(frame[detected["cash_in"]]).fillna(0).sum()) |
| revenue += cash_inflow if document_type in {"Cash Flow Record", "Bank Statement", "Mobile Money Statement"} else 0 |
| if detected["cash_out"]: |
| cash_outflow += float(_to_numeric(frame[detected["cash_out"]]).fillna(0).sum()) |
| expenses += cash_outflow if document_type in {"Cash Flow Record", "Bank Statement", "Mobile Money Statement"} else 0 |
|
|
| amount_column = detected["amount"] |
| if document_type == "Expense Record" and amount_column and not detected["expense"]: |
| expenses += float(_to_numeric(frame[amount_column]).fillna(0).sum()) |
| elif document_type == "Loan or Debt Record" and amount_column and not detected["debt"]: |
| debt += float(_to_numeric(frame[amount_column]).fillna(0).sum()) |
|
|
| unclassified = 0 |
| expense_breakdown: dict[str, float] = {} |
| if amount_column: |
| amounts = _to_numeric(frame[amount_column]).fillna(0) |
| for index, amount in amounts.items(): |
| if amount == 0: |
| continue |
| row_dict = frame.loc[index].to_dict() |
| classification = classify_transaction_row(row_dict, document_type) |
| if document_type in {"Income Statement", "Other"} and classification == "revenue" and not detected["revenue"]: |
| revenue += abs(float(amount)) |
| elif document_type in {"Income Statement", "Other"} and classification == "expense" and not detected["expense"]: |
| expenses += abs(float(amount)) |
| elif classification == "debt" and not detected["debt"]: |
| debt += abs(float(amount)) |
| elif document_type in {"Bank Statement", "Mobile Money Statement", "Cash Flow Record"}: |
| if classification == "revenue" and not detected["cash_in"]: |
| cash_inflow += abs(float(amount)) |
| revenue += abs(float(amount)) |
| elif classification == "expense" and not detected["cash_out"]: |
| cash_outflow += abs(float(amount)) |
| expenses += abs(float(amount)) |
| elif classification == "debt" and not detected["debt"]: |
| debt += abs(float(amount)) |
| elif document_type == "Sales Record" and not detected["revenue"]: |
| revenue += abs(float(amount)) |
| elif document_type in {"Expense Record", "Receipts", "Inventory/Stock Record"} and not detected["expense"]: |
| expenses += abs(float(amount)) |
| else: |
| unclassified += 1 |
|
|
| category_column = detected["category"] |
| if classification == "expense" and category_column and pd.notna(frame.at[index, category_column]): |
| cat = str(frame.at[index, category_column]).strip() |
| expense_breakdown[cat] = expense_breakdown.get(cat, 0.0) + abs(float(amount)) |
|
|
| if unclassified: |
| warnings.append(f"{unclassified} row(s) could not be classified for this document type.") |
|
|
| if document_type == "Sales Record" and revenue == 0 and detected["quantity"] and detected["unit_price"]: |
| revenue = float((_to_numeric(frame[detected["quantity"]]) * _to_numeric(frame[detected["unit_price"]])).fillna(0).sum()) |
|
|
| if detected["profit"]: |
| profit = float(_to_numeric(frame[detected["profit"]]).fillna(0).sum()) |
| warnings.append("Profit was read directly from the document.") |
| else: |
| profit = calculate_profit(revenue, expenses) |
|
|
| margin = calculate_profit_margin(revenue, profit) |
| biggest = [n for n, _ in sorted(expense_breakdown.items(), key=lambda x: x[1], reverse=True)[:3]] |
|
|
| |
| product_breakdown: list[dict] | None = None |
| top_product: str | None = None |
| slowest_product: str | None = None |
|
|
| if document_type == "Sales Record" and detected.get("product"): |
| prod_col = detected["product"] |
| units_col = detected.get("quantity") |
| price_col = detected.get("unit_price") |
| amt_col = detected.get("amount") |
|
|
| rows: list[dict] = [] |
| for _, row in frame.iterrows(): |
| name = str(row.get(prod_col, "")).strip() |
| if not name or name.lower() in ("nan", "none", "total", "grand total"): |
| continue |
| units = ( |
| float(_to_numeric(pd.Series([row.get(units_col)])).fillna(0).iloc[0]) |
| if units_col else 0.0 |
| ) |
| price = ( |
| float(_to_numeric(pd.Series([row.get(price_col)])).fillna(0).iloc[0]) |
| if price_col else 0.0 |
| ) |
| rev = units * price if price > 0 and units > 0 else ( |
| abs(float(_to_numeric(pd.Series([row.get(amt_col)])).fillna(0).iloc[0])) |
| if amt_col else 0.0 |
| ) |
| if rev > 0 or units > 0: |
| rows.append({"product": name, "units_sold": units, "unit_price": price, "revenue": rev}) |
|
|
| if rows: |
| by_rev = sorted(rows, key=lambda x: x["revenue"], reverse=True) |
| product_breakdown = by_rev |
| top_product = by_rev[0]["product"] |
| slowest_product = by_rev[-1]["product"] if len(by_rev) > 1 else None |
|
|
| |
| if revenue == 0: |
| revenue = round(sum(r["revenue"] for r in rows), 2) |
| profit = calculate_profit(revenue, expenses) |
| margin = calculate_profit_margin(revenue, profit) |
|
|
| result.update({ |
| "revenue": round(revenue, 2), |
| "expenses": round(expenses, 2), |
| "profit": round(profit, 2), |
| "debt": round(debt, 2), |
| "cash_inflow": round(cash_inflow or revenue, 2), |
| "cash_outflow": round(cash_outflow or expenses, 2), |
| "profit_margin": round(margin, 2), |
| "warnings": warnings, |
| "biggest_expense_categories": biggest, |
| "confidence": "low", |
| "used_llm": False, |
| "product_breakdown": product_breakdown, |
| "top_product": top_product, |
| "slowest_product": slowest_product, |
| }) |
| result["summary"] = f"Keyword-based fallback parser used for this {document_type.lower()}. LLM unavailable." |
| return result |
|
|
|
|
| |
|
|
| def _classify_llm_error(exc: Exception) -> str: |
| """Return a user-friendly message for known LLM failure modes.""" |
| msg = str(exc) |
| if "429" in msg or "rate_limit" in msg.lower() or "Rate limit" in msg: |
| import re |
| wait = re.search(r"try again in (\S+)", msg) |
| wait_str = f" Please try again in {wait.group(1)}." if wait else " Please try again later." |
| return ( |
| f"AI daily usage limit reached.{wait_str} " |
| "Document parsing will use the keyword-based fallback method." |
| ) |
| if "401" in msg or "auth" in msg.lower() or "api_key" in msg.lower(): |
| return "AI service key invalid or missing. Check your API key configuration." |
| if "timeout" in msg.lower() or "connect" in msg.lower(): |
| return "LLM request timed out. Check your internet connection and try again." |
| return f"LLM error: {msg[:200]}" |
|
|
|
|
| def parse_excel_file(uploaded_file: Any, document_type: str = "Other") -> dict[str, Any]: |
| file_bytes = uploaded_file.getvalue() |
| filename = uploaded_file.name |
| extraction_error: str | None = None |
| llm_error: str | None = None |
|
|
| |
| try: |
| structured = _parse_excel_by_sheets(file_bytes, filename) |
| if structured is not None: |
| return structured |
| except Exception: |
| pass |
|
|
| |
| raw_text: str = "" |
| rows: int = 0 |
| try: |
| raw_text, rows = _extract_excel_as_text(file_bytes) |
| except Exception as exc: |
| extraction_error = str(exc) |
|
|
| |
| try: |
| _xl_frame = pd.read_excel(io.BytesIO(file_bytes), nrows=5) |
| if document_type in ("Income Statement", "Other") and _detect_is_sales_record(_xl_frame): |
| document_type = "Sales Record" |
| except Exception: |
| pass |
|
|
| |
| if raw_text.strip(): |
| print("=== RAW EXCEL CONTENT BEING SENT TO LLM ===") |
| print(raw_text[:2000]) |
| print("=== END RAW CONTENT ===") |
| try: |
| llm = _llm_parse_document(raw_text, document_type, filename) |
| if llm: |
| return _map_llm_result(llm, "xlsx", filename, rows) |
| except Exception as exc: |
| llm_error = _classify_llm_error(exc) |
|
|
| |
| try: |
| frame = pd.read_excel(io.BytesIO(file_bytes)) |
| if document_type in ("Income Statement", "Other") and _detect_is_sales_record(frame): |
| document_type = "Sales Record" |
| result = _keyword_parse_frame(frame, document_type) |
| result["source_type"] = "xlsx" |
| result["file_name"] = filename |
| msgs: list[str] = [] |
| if llm_error: |
| msgs.append(llm_error) |
| if extraction_error: |
| msgs.append(f"Sheet extraction error: {extraction_error}") |
| if not llm_error and not extraction_error: |
| msgs.append("Keyword-based fallback used — numbers may not be accurate.") |
| result["warnings"] = msgs + result.get("warnings", []) |
| return result |
| except Exception as exc: |
| result = _blank_result("xlsx", filename, document_type) |
| if llm_error: |
| result["warnings"].append(llm_error) |
| if extraction_error: |
| result["warnings"].append(f"Sheet extraction error: {extraction_error}") |
| result["warnings"].append(f"Could not read Excel file: {exc}") |
| return result |
|
|
|
|
| def parse_csv_file(uploaded_file: Any, document_type: str = "Other") -> dict[str, Any]: |
| file_bytes = uploaded_file.getvalue() |
| filename = uploaded_file.name |
|
|
| try: |
| frame = pd.read_csv(io.BytesIO(file_bytes)) |
| |
| if document_type in ("Income Statement", "Other") and _detect_is_sales_record(frame): |
| document_type = "Sales Record" |
| raw_text = frame.to_string(max_rows=300) |
| llm = _llm_parse_document(raw_text, document_type, filename) |
| if llm: |
| return _map_llm_result(llm, "csv", filename, len(frame)) |
| except Exception: |
| pass |
|
|
| try: |
| frame = pd.read_csv(io.BytesIO(file_bytes)) |
| if document_type in ("Income Statement", "Other") and _detect_is_sales_record(frame): |
| document_type = "Sales Record" |
| result = _keyword_parse_frame(frame, document_type) |
| result["source_type"] = "csv" |
| result["file_name"] = filename |
| result["warnings"].insert(0, "LLM parsing unavailable — used keyword-based fallback.") |
| return result |
| except Exception as exc: |
| result = _blank_result("csv", filename, document_type) |
| result["warnings"].append(f"Could not read CSV file: {exc}") |
| return result |
|
|
|
|
| def parse_text_file(uploaded_file: Any, document_type: str = "Other") -> dict[str, Any]: |
| content = uploaded_file.getvalue().decode("utf-8", errors="ignore") |
| filename = uploaded_file.name |
|
|
| try: |
| llm = _llm_parse_document(content, document_type, filename) |
| if llm: |
| return _map_llm_result(llm, "txt", filename, content.count("\n")) |
| except Exception: |
| pass |
|
|
| parsed = parse_business_input(content) |
| result = _blank_result("txt", filename, document_type) |
| result.update({ |
| "characters_analyzed": len(content), |
| "revenue": parsed["revenue"], |
| "expenses": parsed["expenses"], |
| "profit": round(parsed["revenue"] - parsed["expenses"], 2), |
| "debt": parsed["debt"], |
| "cash_inflow": parsed["revenue"], |
| "cash_outflow": parsed["expenses"], |
| "summary": f"Text file interpreted as {document_type.lower()} using keyword extraction.", |
| "metadata": { |
| "location": parsed.get("location"), |
| "business_type": parsed.get("business_type"), |
| }, |
| }) |
| return result |
|
|
|
|
| def parse_pdf_file(uploaded_file: Any, document_type: str = "Other") -> dict[str, Any]: |
| result = _blank_result("pdf", uploaded_file.name, document_type) |
| result["warnings"].append("PDF support is experimental. For best results, upload CSV or Excel.") |
| try: |
| from pypdf import PdfReader |
| except Exception: |
| result["summary"] = "PDF parsing unavailable — pypdf not installed." |
| result["warnings"].append("Install pypdf to enable PDF text extraction.") |
| return result |
|
|
| try: |
| raw_text, page_count = _extract_pdf_as_text(uploaded_file.getvalue()) |
| result["pages_analyzed"] = page_count |
| result["characters_analyzed"] = len(raw_text) |
|
|
| llm = _llm_parse_document(raw_text, document_type, uploaded_file.name) |
| if llm: |
| mapped = _map_llm_result(llm, "pdf", uploaded_file.name, page_count) |
| mapped["pages_analyzed"] = page_count |
| mapped["characters_analyzed"] = len(raw_text) |
| mapped["warnings"] = result["warnings"] + mapped["warnings"] |
| return mapped |
|
|
| |
| parsed = parse_business_input(raw_text) |
| result.update({ |
| "revenue": parsed["revenue"], |
| "expenses": parsed["expenses"], |
| "profit": round(parsed["revenue"] - parsed["expenses"], 2), |
| "debt": parsed["debt"], |
| "cash_inflow": parsed["revenue"], |
| "cash_outflow": parsed["expenses"], |
| "summary": f"PDF read as text and interpreted as {document_type.lower()}.", |
| "metadata": { |
| "location": parsed.get("location"), |
| "business_type": parsed.get("business_type"), |
| }, |
| }) |
| except Exception as exc: |
| result["warnings"].append(f"PDF extraction failed: {exc}") |
| return result |
|
|
|
|
| def parse_uploaded_file(uploaded_file: Any, document_type: str = "Other") -> dict[str, Any]: |
| suffix = uploaded_file.name.lower().rsplit(".", maxsplit=1)[-1] if "." in uploaded_file.name else "" |
| if suffix == "csv": |
| return parse_csv_file(uploaded_file, document_type) |
| if suffix in {"xlsx", "xlsm"}: |
| return parse_excel_file(uploaded_file, document_type) |
| if suffix == "txt": |
| return parse_text_file(uploaded_file, document_type) |
| if suffix == "pdf": |
| return parse_pdf_file(uploaded_file, document_type) |
|
|
| result = _blank_result(suffix or "unknown", uploaded_file.name, document_type) |
| result["summary"] = "The uploaded file format is not supported for document analysis." |
| result["warnings"].append("Please upload a CSV, Excel (.xlsx), TXT, or PDF file.") |
| return result |
|
|
|
|
| def summarize_document_records(data: dict[str, Any], document_type: str | None = None) -> str: |
| active_type = document_type or data.get("document_type", "Other") |
| period = data.get("period_description") or data.get("date_range") or "unknown period" |
| obs = (" Observations: " + "; ".join(data.get("key_observations", [])[:2])) if data.get("key_observations") else "" |
| return ( |
| f"{data.get('summary', '').strip()} " |
| f"Document type: {active_type}. Period: {period}. " |
| f"Revenue: {data.get('revenue', 0):,.2f}. " |
| f"Expenses: {data.get('expenses', 0):,.2f}. " |
| f"Profit: {data.get('profit', 0):,.2f}. " |
| f"Debt: {data.get('debt', 0):,.2f}." |
| f"{obs}" |
| ).strip() |
|
|
