from __future__ import annotations from datetime import date, datetime from decimal import Decimal, InvalidOperation from urllib.parse import quote from fastapi import APIRouter, Request from fastapi.responses import RedirectResponse, JSONResponse from sqlalchemy import select from app.core.db.common import CommonSessionLocal from app.core.security.csrf import get_or_create_csrf_token, validate_csrf from app.core.security.session_auth import get_current_user from app.core.templating import templates from app.modules.accounting.agent_bridge import request_agent_command from app.modules.accounting.cash_payment_models import CashPaymentRuleSetting from app.modules.accounting.ui import ( _accounting_storage_payload, _denied, _find_visible_client, _financial_year_for_date, _node_online, _require_partner, _visible_clients, ) from app.modules.core.audit.service import write_audit_log from app.modules.core.rbac.deps import get_user_permissions, get_user_roles from app.modules.documents.services import get_active_storage_node_for_branch router = APIRouter(prefix="/tools/accounting/cash-payments", tags=["accounting-cash-payment-ui"]) def _tenant_id(request: Request, user) -> int: return int(request.session.get("active_tenant_id") or getattr(user, "tenant_id", 0) or 0) def _setting(db, tenant_id: int) -> CashPaymentRuleSetting: row = db.execute( select(CashPaymentRuleSetting).where(CashPaymentRuleSetting.tenant_id == int(tenant_id)) ).scalar_one_or_none() if row: return row row = CashPaymentRuleSetting( tenant_id=int(tenant_id), cash_limit=10000.0, effective_from=date(2017, 4, 1), split_window_days=3, near_limit_percent=80.0, max_allocation_days=5, ) db.add(row) db.commit() db.refresh(row) return row def _parse_money(value: str, label: str) -> float: try: amount = Decimal(str(value or "").replace(",", "").strip()) except (InvalidOperation, ValueError): raise ValueError(f"{label} must be a valid amount.") if amount <= 0: raise ValueError(f"{label} must be greater than zero.") return float(amount.quantize(Decimal("0.01"))) def _allocation_plan(source_text: str, expected_total: float, cash_limit: float, max_days: int) -> dict: entries = [] errors = [] for line_no, raw in enumerate((source_text or "").splitlines(), start=1): text = raw.strip() if not text: continue parts = [part.strip() for part in text.split(",")] if len(parts) != 2: errors.append(f"Line {line_no}: use YYYY-MM-DD, amount.") continue try: paid_on = date.fromisoformat(parts[0]) amount = _parse_money(parts[1], f"Line {line_no} amount") except Exception as exc: errors.append(str(exc)) continue entries.append({"date": paid_on.isoformat(), "amount": amount}) supplied_total = round(sum(row["amount"] for row in entries), 2) by_date: dict[str, float] = {} for row in entries: by_date[row["date"]] = round(by_date.get(row["date"], 0.0) + row["amount"], 2) above_limit_dates = [ {"date": key, "amount": value} for key, value in sorted(by_date.items()) if value > float(cash_limit) + 0.009 ] dates = sorted(date.fromisoformat(key) for key in by_date) span_days = ((dates[-1] - dates[0]).days + 1) if dates else 0 if abs(supplied_total - round(float(expected_total), 2)) > 0.009: errors.append( f"Source-date total is {supplied_total:.2f}, but the accounting amount is {float(expected_total):.2f}." ) if above_limit_dates: errors.append("One or more actual payment dates exceed the configured daily review limit.") if span_days > int(max_days): errors.append( f"Actual source dates span {span_days} days, exceeding the configured {int(max_days)}-day allocation window." ) return { "entries": entries, "by_date": [{"date": key, "amount": value} for key, value in sorted(by_date.items())], "expected_total": round(float(expected_total), 2), "supplied_total": supplied_total, "cash_limit": round(float(cash_limit), 2), "span_days": span_days, "max_allocation_days": int(max_days), "above_limit_dates": above_limit_dates, "errors": errors, "status": "ready" if entries and not errors else "manual_review", "source_dates_only": True, "note": "The planner never creates or changes payment dates. Proposed entries use only the actual source dates entered by the user.", } def _render(request: Request, db, user, **context): base = { "request": request, "current_user": user, "current_user_roles": get_user_roles(db, user.id), "current_user_permissions": get_user_permissions(db, user.id), "csrf_token": get_or_create_csrf_token(request), } base.update(context) return templates.TemplateResponse( "modules/accounting/templates/accounting/cash_payment_review.html", base, ) @router.get("") def cash_payment_review( request: Request, client_id: int | None = None, tally_guid: str = "", date_from: str = "", date_to: str = "", job_id: str = "", cash_ledger_name: str = "", analyze: int = 0, saved: int = 0, error: str = "", ): db = CommonSessionLocal() try: user, response = _require_partner(request, db, "accounting.tally.view") if response: return response clients, scope = _visible_clients(db, request, user) selected_client = next((row for row in clients if client_id and int(row.id) == int(client_id)), None) tenant_id = _tenant_id(request, user) setting = _setting(db, tenant_id) node = get_active_storage_node_for_branch(db, scope.tenant_id, scope.branch_id) online = _node_online(node) today = date.today() fy = _financial_year_for_date(today) fy_start_year = int(fy.split("-", 1)[0]) start_text = date_from or date(fy_start_year, 4, 1).isoformat() end_text = date_to or today.isoformat() live_result = None review = None cache_job = None command_error = error or "" if selected_client and node and online: try: status_response = request_agent_command( node.node_code, "phase4_status", _accounting_storage_payload(selected_client, _financial_year_for_date(date.fromisoformat(start_text))), timeout_seconds=20, ) if status_response.get("ok"): live_result = status_response.get("result") or {} else: command_error = str(status_response.get("error") or "Local Agent status failed.") if job_id: status = request_agent_command( node.node_code, "accounting_cash_payment_cache_status", { **_accounting_storage_payload(selected_client, _financial_year_for_date(date.fromisoformat(start_text))), "job_id": str(job_id), }, timeout_seconds=20, ) if status.get("ok"): cache_job = (status.get("result") or {}).get("job") else: command_error = str(status.get("error") or "Could not read cash-payment cache status.") if analyze and job_id and cache_job and cache_job.get("status") == "completed": result = request_agent_command( node.node_code, "accounting_cash_payment_cache_analyze", { **_accounting_storage_payload(selected_client, _financial_year_for_date(date.fromisoformat(start_text))), "job_id": str(job_id), "cash_limit": float(setting.cash_limit), "split_window_days": int(setting.split_window_days), "near_limit_percent": float(setting.near_limit_percent), }, timeout_seconds=120, ) if result.get("ok"): review = (result.get("result") or {}).get("cash_payment_review") else: command_error = str(result.get("error") or "Cash payment SQLite analysis failed.") except Exception as exc: command_error = str(exc) return _render( request, db, user, title="Cash Payment Compliance Review", clients=clients, selected_client=selected_client, storage_node=node, agent_online=online, live_result=live_result, selected_tally_guid=str(tally_guid or ""), date_from=start_text, date_to=end_text, setting=setting, review=review, cache_job=cache_job, active_job_id=str(job_id or ""), selected_cash_ledger_name=str(cash_ledger_name or ""), saved=bool(saved), command_error=command_error, allocation_plan=None, allocation_input={"total_amount": "", "party_name": "", "expense_ledger": "", "source_payments": ""}, page_mode="compliance", ) finally: db.close() @router.post("/ledgers") async def cash_payment_ledgers(request: Request): """Return cached ledgers immediately or start a non-blocking Tally master job.""" form = await request.form() validate_csrf(request, str(form.get("csrf_token") or "")) db = CommonSessionLocal() try: user, response = _require_partner(request, db, "accounting.tally.view") if response: return JSONResponse({"ok": False, "error": "Access denied."}, status_code=403) clients, scope = _visible_clients(db, request, user) try: client_id = int(form.get("client_id") or 0) except Exception: client_id = 0 selected_client = next((row for row in clients if int(row.id) == client_id), None) if not selected_client: return JSONResponse({"ok": False, "error": "Select a valid client."}, status_code=400) tally_guid = str(form.get("tally_guid") or "").strip() date_from = str(form.get("date_from") or date.today().isoformat()).strip() force_refresh = str(form.get("force_refresh") or "").strip().lower() in {"1","true","yes","on"} if not tally_guid: return JSONResponse({"ok": False, "error": "Select a mapped Tally company."}, status_code=400) node = get_active_storage_node_for_branch(db, scope.tenant_id, scope.branch_id) if not node or not _node_online(node): return JSONResponse({"ok": False, "error": "ERP Local Agent is offline."}, status_code=409) try: fy = _financial_year_for_date(date.fromisoformat(date_from)) result = request_agent_command( node.node_code, "accounting_cash_payment_ledgers_start", { **_accounting_storage_payload(selected_client, fy), "tally_guid": tally_guid, "force_refresh": force_refresh, }, timeout_seconds=20, ) except Exception as exc: return JSONResponse({"ok": False, "error": str(exc)}, status_code=502) if not result.get("ok"): return JSONResponse({"ok": False, "error": str(result.get("error") or "Could not start Tally ledger sync.")}, status_code=409) return JSONResponse({"ok": True, **(result.get("result") or {})}) finally: db.close() @router.get("/ledgers/progress") def cash_payment_ledgers_progress(request: Request, client_id: int, job_id: str, date_from: str): db = CommonSessionLocal() try: user, response = _require_partner(request, db, "accounting.tally.view") if response: return JSONResponse({"ok": False, "error": "Access denied."}, status_code=403) clients, scope = _visible_clients(db, request, user) selected_client = next((row for row in clients if int(row.id) == int(client_id)), None) if not selected_client: return JSONResponse({"ok": False, "error": "Client not found."}, status_code=404) node = get_active_storage_node_for_branch(db, scope.tenant_id, scope.branch_id) if not node or not _node_online(node): return JSONResponse({"ok": False, "error": "ERP Local Agent is offline."}, status_code=409) try: fy = _financial_year_for_date(date.fromisoformat(date_from)) result = request_agent_command( node.node_code, "accounting_cash_payment_ledgers_status", { **_accounting_storage_payload(selected_client, fy), "job_id": str(job_id), }, timeout_seconds=15, ) except Exception as exc: return JSONResponse({"ok": False, "error": str(exc)}, status_code=502) if not result.get("ok"): return JSONResponse({"ok": False, "error": str(result.get("error") or "Could not read ledger-sync progress.")}, status_code=409) return JSONResponse({"ok": True, **(result.get("result") or {})}) finally: db.close() @router.post("/cache/start") async def start_cash_payment_cache(request: Request): form = await request.form() validate_csrf(request, str(form.get("csrf_token") or "")) db = CommonSessionLocal() try: user, response = _require_partner(request, db, "accounting.tally.view") if response: return JSONResponse({"ok": False, "error": "Access denied."}, status_code=403) clients, scope = _visible_clients(db, request, user) try: client_id = int(form.get("client_id") or 0) except Exception: client_id = 0 selected_client = next((row for row in clients if int(row.id) == client_id), None) if not selected_client: return JSONResponse({"ok": False, "error": "Select a valid client."}, status_code=400) tally_guid = str(form.get("tally_guid") or "").strip() cash_ledger_name = str(form.get("cash_ledger_name") or "").strip() date_from = str(form.get("date_from") or "").strip() date_to = str(form.get("date_to") or "").strip() try: parsed_from = date.fromisoformat(date_from) parsed_to = date.fromisoformat(date_to) if parsed_to < parsed_from: raise ValueError("To date cannot be before From date.") except Exception as exc: return JSONResponse({"ok": False, "error": str(exc)}, status_code=400) if not tally_guid: return JSONResponse({"ok": False, "error": "Select a mapped Tally company."}, status_code=400) if not cash_ledger_name: return JSONResponse({"ok": False, "error": "Load the Tally ledgers and confirm the Cash ledger."}, status_code=400) node = get_active_storage_node_for_branch(db, scope.tenant_id, scope.branch_id) if not node or not _node_online(node): return JSONResponse({"ok": False, "error": "ERP Local Agent is offline."}, status_code=409) setting = _setting(db, _tenant_id(request, user)) try: result = request_agent_command( node.node_code, "accounting_cash_payment_cache_start", { **_accounting_storage_payload(selected_client, _financial_year_for_date(parsed_from)), "tally_guid": tally_guid, "date_from": date_from, "date_to": date_to, "cash_ledger_name": cash_ledger_name, "cash_ledger_names": [cash_ledger_name], "cash_limit": float(setting.cash_limit), "split_window_days": int(setting.split_window_days), "near_limit_percent": float(setting.near_limit_percent), "tally_pause_seconds": 3.0, "requested_by_user_id": int(user.id), }, timeout_seconds=30, ) except Exception as exc: return JSONResponse({"ok": False, "error": str(exc)}, status_code=502) if not result.get("ok"): return JSONResponse({"ok": False, "error": str(result.get("error") or "Could not start Tally extraction.")}, status_code=409) return JSONResponse({"ok": True, **(result.get("result") or {})}) finally: db.close() @router.post("/cache/cancel") async def cancel_cash_payment_cache(request: Request): form = await request.form() validate_csrf(request, str(form.get("csrf_token") or "")) db = CommonSessionLocal() try: user, response = _require_partner(request, db, "accounting.tally.view") if response: return JSONResponse({"ok": False, "error": "Access denied."}, status_code=403) clients, scope = _visible_clients(db, request, user) try: client_id = int(form.get("client_id") or 0) except Exception: client_id = 0 selected_client = next((row for row in clients if int(row.id) == client_id), None) if not selected_client: return JSONResponse({"ok": False, "error": "Select a valid client."}, status_code=400) job_id = str(form.get("job_id") or "").strip() date_from = str(form.get("date_from") or "").strip() if not job_id: return JSONResponse({"ok": False, "error": "Extraction job id is required."}, status_code=400) try: parsed_from = date.fromisoformat(date_from) except Exception: return JSONResponse({"ok": False, "error": "Valid extraction From date is required."}, status_code=400) node = get_active_storage_node_for_branch(db, scope.tenant_id, scope.branch_id) if not node or not _node_online(node): return JSONResponse({"ok": False, "error": "ERP Local Agent is offline."}, status_code=409) result = request_agent_command( node.node_code, "accounting_cash_payment_cache_cancel", { **_accounting_storage_payload(selected_client, _financial_year_for_date(parsed_from)), "job_id": job_id, }, timeout_seconds=15, ) if not result.get("ok"): return JSONResponse({"ok": False, "error": str(result.get("error") or "Could not cancel extraction.")}, status_code=409) return JSONResponse({"ok": True, **(result.get("result") or {})}) except Exception as exc: return JSONResponse({"ok": False, "error": str(exc)}, status_code=502) finally: db.close() @router.get("/cache/progress") def cash_payment_cache_progress(request: Request, client_id: int, job_id: str, date_from: str): db = CommonSessionLocal() try: user, response = _require_partner(request, db, "accounting.tally.view") if response: return JSONResponse({"ok": False, "error": "Access denied."}, status_code=403) clients, scope = _visible_clients(db, request, user) selected_client = next((row for row in clients if int(row.id) == int(client_id)), None) if not selected_client: return JSONResponse({"ok": False, "error": "Client not found."}, status_code=404) node = get_active_storage_node_for_branch(db, scope.tenant_id, scope.branch_id) if not node or not _node_online(node): return JSONResponse({"ok": False, "error": "ERP Local Agent is offline."}, status_code=409) try: fy = _financial_year_for_date(date.fromisoformat(date_from)) result = request_agent_command( node.node_code, "accounting_cash_payment_cache_status", {**_accounting_storage_payload(selected_client, fy), "job_id": str(job_id)}, timeout_seconds=20, ) except Exception as exc: return JSONResponse({"ok": False, "error": str(exc)}, status_code=502) if not result.get("ok"): return JSONResponse({"ok": False, "error": str(result.get("error") or "Could not read extraction progress.")}, status_code=409) return JSONResponse({"ok": True, **(result.get("result") or {})}) finally: db.close() @router.get("/allocation") def cash_payment_allocation( request: Request, client_id: int | None = None, error: str = "", ): db = CommonSessionLocal() try: user, response = _require_partner(request, db, "accounting.tally.view") if response: return response clients, scope = _visible_clients(db, request, user) selected_client = next((row for row in clients if client_id and int(row.id) == int(client_id)), None) setting = _setting(db, _tenant_id(request, user)) node = get_active_storage_node_for_branch(db, scope.tenant_id, scope.branch_id) return _render( request, db, user, title="Cash Payment Entry Allocation", clients=clients, selected_client=selected_client, storage_node=node, agent_online=_node_online(node), live_result=None, selected_tally_guid="", date_from="", date_to="", setting=setting, review=None, cache_job=None, active_job_id="", saved=False, command_error=error or "", allocation_plan=None, allocation_input={"total_amount": "", "party_name": "", "expense_ledger": "", "source_payments": ""}, page_mode="allocation", ) finally: db.close() @router.post("/settings") async def save_cash_payment_settings(request: Request): form = await request.form() validate_csrf(request, str(form.get("csrf_token") or "")) db = CommonSessionLocal() try: user, response = _require_partner(request, db, "accounting.tally.view") if response: return response tenant_id = _tenant_id(request, user) setting = _setting(db, tenant_id) old = { "cash_limit": setting.cash_limit, "effective_from": setting.effective_from.isoformat() if setting.effective_from else None, "split_window_days": setting.split_window_days, "near_limit_percent": setting.near_limit_percent, "max_allocation_days": setting.max_allocation_days, } try: cash_limit = _parse_money(str(form.get("cash_limit") or ""), "Cash payment limit") effective_from = date.fromisoformat(str(form.get("effective_from") or "")) split_window_days = int(form.get("split_window_days") or 3) near_limit_percent = float(form.get("near_limit_percent") or 80) max_allocation_days = int(form.get("max_allocation_days") or 5) if not 1 <= split_window_days <= 31: raise ValueError("Split-payment review window must be between 1 and 31 days.") if not 1 <= near_limit_percent <= 100: raise ValueError("Near-limit percentage must be between 1 and 100.") if not 1 <= max_allocation_days <= 366: raise ValueError("Maximum source-date allocation window must be between 1 and 366 days.") except Exception as exc: return RedirectResponse(url="/tools/accounting/cash-payments?error=" + quote(str(exc)), status_code=303) setting.cash_limit = cash_limit setting.effective_from = effective_from setting.split_window_days = split_window_days setting.near_limit_percent = near_limit_percent setting.max_allocation_days = max_allocation_days setting.updated_by_user_id = int(user.id) setting.updated_at_utc = datetime.utcnow() db.commit() try: write_audit_log( db, actor=user, request=request, action="accounting.cash_payment_rule.updated", entity_type="CashPaymentRuleSetting", entity_id=int(setting.id), entity_name="Cash Payment Compliance Settings", target_tenant_id=int(tenant_id), details={ "before": old, "after": { "cash_limit": setting.cash_limit, "effective_from": setting.effective_from.isoformat(), "split_window_days": setting.split_window_days, "near_limit_percent": setting.near_limit_percent, "max_allocation_days": setting.max_allocation_days, }, }, ) except Exception: db.rollback() client_id = str(form.get("client_id") or "").strip() suffix = f"&client_id={client_id}" if client_id else "" return RedirectResponse(url=f"/tools/accounting/cash-payments?saved=1{suffix}", status_code=303) finally: db.close() @router.post("/allocation-plan") async def cash_payment_allocation_plan(request: Request): form = await request.form() validate_csrf(request, str(form.get("csrf_token") or "")) db = CommonSessionLocal() try: user, response = _require_partner(request, db, "accounting.tally.view") if response: return response clients, scope = _visible_clients(db, request, user) client_id = int(form.get("client_id") or 0) selected_client = next((row for row in clients if int(row.id) == client_id), None) if not selected_client: return _denied() setting = _setting(db, _tenant_id(request, user)) total_amount_text = str(form.get("total_amount") or "") party_name = str(form.get("party_name") or "").strip() expense_ledger = str(form.get("expense_ledger") or "").strip() source_payments = str(form.get("source_payments") or "") try: total_amount = _parse_money(total_amount_text, "Total cash payment") plan = _allocation_plan(source_payments, total_amount, float(setting.cash_limit), int(setting.max_allocation_days)) except Exception as exc: plan = {"entries": [], "errors": [str(exc)], "status": "manual_review", "source_dates_only": True} node = get_active_storage_node_for_branch(db, scope.tenant_id, scope.branch_id) return _render( request, db, user, title="Cash Payment Compliance Review", clients=clients, selected_client=selected_client, storage_node=node, agent_online=_node_online(node), live_result=None, selected_tally_guid="", date_from="", date_to="", setting=setting, review=None, cache_job=None, active_job_id="", saved=False, command_error="", allocation_plan=plan, allocation_input={ "total_amount": total_amount_text, "party_name": party_name, "expense_ledger": expense_ledger, "source_payments": source_payments, }, page_mode="allocation", ) finally: db.close()