A R R R Associates
352 lines
12 KiB
Python
352 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from urllib.parse import urlencode
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import RedirectResponse
|
|
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.templating import templates
|
|
from app.modules.accounting.automatic_voucher_service import (
|
|
create_bank_draft,
|
|
create_eway_invoice_draft,
|
|
)
|
|
from app.modules.accounting.chart_models import AccountingChartLedger
|
|
from app.modules.accounting.chart_service import effective_role
|
|
from app.modules.accounting.source_adapter_service import (
|
|
bank_rows,
|
|
discover_sources,
|
|
eway_rows,
|
|
group_eway_invoices,
|
|
)
|
|
from app.modules.accounting.ui import _find_visible_client, _require_partner, _visible_clients
|
|
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
|
|
|
|
router = APIRouter(
|
|
prefix="/tools/accounting/voucher-engine/automatic",
|
|
tags=["accounting-automatic-voucher-source-ui"],
|
|
)
|
|
|
|
|
|
def _go(client_id=0, message="", error="", **extra):
|
|
params = {}
|
|
if client_id:
|
|
params["client_id"] = client_id
|
|
if message:
|
|
params["message"] = message[:350]
|
|
if error:
|
|
params["error"] = error[:350]
|
|
for key, value in extra.items():
|
|
if value not in ("", None):
|
|
params[key] = value
|
|
return RedirectResponse(
|
|
"/tools/accounting/voucher-engine/automatic"
|
|
+ ("?" + urlencode(params) if params else ""),
|
|
status_code=303,
|
|
)
|
|
|
|
|
|
def _company_options(db, tenant_id, client_id):
|
|
rows = list(
|
|
db.execute(
|
|
select(AccountingChartLedger.tally_guid, AccountingChartLedger.company_name)
|
|
.where(
|
|
AccountingChartLedger.tenant_id == int(tenant_id),
|
|
AccountingChartLedger.client_id == int(client_id),
|
|
)
|
|
.distinct()
|
|
.order_by(AccountingChartLedger.company_name)
|
|
).all()
|
|
)
|
|
return [{"guid": guid, "name": name or guid} for guid, name in rows if guid]
|
|
|
|
|
|
def _company_name(options, tally_guid):
|
|
return next((row["name"] for row in options if row["guid"] == tally_guid), "")
|
|
|
|
|
|
def _source_table(valid, requested):
|
|
names = {row.table_name for row in valid}
|
|
if requested in names:
|
|
return requested
|
|
return valid[0].table_name if valid else ""
|
|
|
|
|
|
@router.get("")
|
|
def page(
|
|
request: Request,
|
|
client_id: int | None = None,
|
|
tally_guid: str = "",
|
|
bank_source: str = "",
|
|
eway_source: str = "",
|
|
direction: str = "",
|
|
from_date: str = "",
|
|
to_date: str = "",
|
|
message: str = "",
|
|
error: str = "",
|
|
):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, denied = _require_partner(request, db, "accounting.learning.view")
|
|
if denied:
|
|
return denied
|
|
|
|
clients, scope = _visible_clients(db, request, user)
|
|
selected = next(
|
|
(c for c in clients if client_id and int(c.id) == int(client_id)),
|
|
None,
|
|
)
|
|
|
|
company_options = []
|
|
sources = {"bank": [], "eway": []}
|
|
selected_bank_source = ""
|
|
selected_eway_source = ""
|
|
banks = []
|
|
eway_invoices = []
|
|
bank_ledgers = []
|
|
|
|
if selected:
|
|
company_options = _company_options(db, scope.tenant_id, selected.id)
|
|
if not tally_guid and len(company_options) == 1:
|
|
tally_guid = company_options[0]["guid"]
|
|
if tally_guid not in {row["guid"] for row in company_options}:
|
|
tally_guid = ""
|
|
|
|
if tally_guid:
|
|
bank_ledgers = list(
|
|
db.execute(
|
|
select(AccountingChartLedger)
|
|
.where(
|
|
AccountingChartLedger.tenant_id == int(scope.tenant_id),
|
|
AccountingChartLedger.client_id == int(selected.id),
|
|
AccountingChartLedger.tally_guid == tally_guid,
|
|
)
|
|
.order_by(AccountingChartLedger.name)
|
|
).scalars().all()
|
|
)
|
|
bank_ledgers = [
|
|
row for row in bank_ledgers
|
|
if effective_role(row) in {"BANK", "CASH"}
|
|
]
|
|
|
|
sources = discover_sources(db)
|
|
selected_bank_source = _source_table(sources["bank"], bank_source)
|
|
selected_eway_source = _source_table(sources["eway"], eway_source)
|
|
|
|
if selected_bank_source:
|
|
banks = bank_rows(
|
|
db,
|
|
table_name=selected_bank_source,
|
|
client_id=selected.id,
|
|
from_date=from_date,
|
|
to_date=to_date,
|
|
limit=300,
|
|
)
|
|
|
|
if selected_eway_source:
|
|
items = eway_rows(
|
|
db,
|
|
table_name=selected_eway_source,
|
|
client_id=selected.id,
|
|
direction=direction,
|
|
from_date=from_date,
|
|
to_date=to_date,
|
|
limit=5000,
|
|
)
|
|
eway_invoices = group_eway_invoices(items)
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/accounting/templates/accounting/automatic_voucher_sources.html",
|
|
{
|
|
"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),
|
|
"title": "Automatic Accounting Preparation",
|
|
"clients": clients,
|
|
"selected_client": selected,
|
|
"selected_tally_guid": tally_guid,
|
|
"company_options": company_options,
|
|
"sources": sources,
|
|
"selected_bank_source": selected_bank_source,
|
|
"selected_eway_source": selected_eway_source,
|
|
"bank_rows": banks,
|
|
"eway_invoices": eway_invoices,
|
|
"bank_ledgers": bank_ledgers,
|
|
"direction": direction,
|
|
"from_date": from_date,
|
|
"to_date": to_date,
|
|
"message": message,
|
|
"error": error,
|
|
},
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/bank/draft")
|
|
def bank_draft(
|
|
request: Request,
|
|
client_id: int = Form(...),
|
|
tally_guid: str = Form(...),
|
|
source_table: str = Form(...),
|
|
source_id: str = Form(...),
|
|
bank_ledger_name: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, denied = _require_partner(request, db, "accounting.learning.manage")
|
|
if denied:
|
|
return denied
|
|
|
|
client, _clients, scope = _find_visible_client(db, request, user, client_id)
|
|
if not client:
|
|
raise ValueError("Client is not visible.")
|
|
|
|
sources = discover_sources(db)["bank"]
|
|
if source_table not in {row.table_name for row in sources}:
|
|
raise ValueError("Selected bank source is not an approved discovered source.")
|
|
|
|
source = next(
|
|
(
|
|
row
|
|
for row in bank_rows(
|
|
db,
|
|
table_name=source_table,
|
|
client_id=client.id,
|
|
limit=2000,
|
|
)
|
|
if str(row.get("source_id") or "") == str(source_id)
|
|
),
|
|
None,
|
|
)
|
|
if source is None:
|
|
raise ValueError("Bank transaction could not be reloaded from the source.")
|
|
|
|
options = _company_options(db, scope.tenant_id, client.id)
|
|
company_name = _company_name(options, tally_guid)
|
|
if not company_name:
|
|
raise ValueError("Selected Tally company is not available in Chart of Accounts.")
|
|
|
|
draft = create_bank_draft(
|
|
db,
|
|
tenant_id=scope.tenant_id,
|
|
client_id=client.id,
|
|
tally_guid=tally_guid,
|
|
company_name=company_name,
|
|
source_row=source,
|
|
bank_ledger_name=bank_ledger_name,
|
|
user_id=user.id,
|
|
)
|
|
return _go(
|
|
client.id,
|
|
tally_guid=tally_guid,
|
|
bank_source=source_table,
|
|
message=f"{draft.voucher_type} draft #{draft.id} created automatically from bank transaction.",
|
|
)
|
|
except Exception as exc:
|
|
db.rollback()
|
|
return _go(
|
|
client_id,
|
|
tally_guid=tally_guid,
|
|
bank_source=source_table,
|
|
error=str(exc),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/eway/draft")
|
|
def eway_draft(
|
|
request: Request,
|
|
client_id: int = Form(...),
|
|
tally_guid: str = Form(...),
|
|
source_table: str = Form(...),
|
|
direction: str = Form(...),
|
|
eway_bill_no: str = Form(""),
|
|
invoice_no: str = Form(""),
|
|
invoice_date: str = Form(""),
|
|
party_gstin: str = Form(""),
|
|
party_name: str = Form(""),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, denied = _require_partner(request, db, "accounting.learning.manage")
|
|
if denied:
|
|
return denied
|
|
|
|
client, _clients, scope = _find_visible_client(db, request, user, client_id)
|
|
if not client:
|
|
raise ValueError("Client is not visible.")
|
|
|
|
sources = discover_sources(db)["eway"]
|
|
if source_table not in {row.table_name for row in sources}:
|
|
raise ValueError("Selected E-Way Bill source is not an approved discovered source.")
|
|
|
|
invoices = group_eway_invoices(
|
|
eway_rows(
|
|
db,
|
|
table_name=source_table,
|
|
client_id=client.id,
|
|
direction=direction,
|
|
limit=10000,
|
|
)
|
|
)
|
|
invoice = next(
|
|
(
|
|
row
|
|
for row in invoices
|
|
if (row.get("eway_bill_no") or "") == eway_bill_no
|
|
and (row.get("invoice_no") or "") == invoice_no
|
|
and (row.get("invoice_date") or "") == invoice_date
|
|
and (row.get("party_gstin") or "") == party_gstin
|
|
and (row.get("party_name") or "") == party_name
|
|
),
|
|
None,
|
|
)
|
|
if invoice is None:
|
|
raise ValueError("E-Way Bill invoice/items could not be reloaded from the source.")
|
|
|
|
options = _company_options(db, scope.tenant_id, client.id)
|
|
company_name = _company_name(options, tally_guid)
|
|
if not company_name:
|
|
raise ValueError("Selected Tally company is not available in Chart of Accounts.")
|
|
|
|
draft = create_eway_invoice_draft(
|
|
db,
|
|
tenant_id=scope.tenant_id,
|
|
client_id=client.id,
|
|
tally_guid=tally_guid,
|
|
company_name=company_name,
|
|
invoice=invoice,
|
|
user_id=user.id,
|
|
)
|
|
return _go(
|
|
client.id,
|
|
tally_guid=tally_guid,
|
|
eway_source=source_table,
|
|
direction=direction,
|
|
message=(
|
|
f"{draft.voucher_type} draft #{draft.id} created automatically "
|
|
f"with {len(draft.items)} stock-item line(s)."
|
|
),
|
|
)
|
|
except Exception as exc:
|
|
db.rollback()
|
|
return _go(
|
|
client_id,
|
|
tally_guid=tally_guid,
|
|
eway_source=source_table,
|
|
direction=direction,
|
|
error=str(exc),
|
|
)
|
|
finally:
|
|
db.close()
|