264 lines
8.2 KiB
Python
264 lines
8.2 KiB
Python
"""CRUD de documentos fiscais confirmados."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Form, Query, Request
|
|
from starlette.responses import RedirectResponse
|
|
|
|
from .. import auth
|
|
from .. import database as db
|
|
from ..dates import current_competencia
|
|
from ..storage import money
|
|
from ..templating import flash, render
|
|
|
|
router = APIRouter()
|
|
|
|
PAGE_SIZE = 10
|
|
|
|
|
|
def _page_window(page: int, total_pages: int, span: int = 2) -> list[int | None]:
|
|
"""Monta a lista de números de página com None para reticências."""
|
|
pages = {1, total_pages, *range(page - span, page + span + 1)}
|
|
pages = sorted(p for p in pages if 1 <= p <= total_pages)
|
|
windowed: list[int | None] = []
|
|
prev = None
|
|
for p in pages:
|
|
if prev is not None and p - prev > 1:
|
|
windowed.append(None)
|
|
windowed.append(p)
|
|
prev = p
|
|
return windowed
|
|
|
|
|
|
def _parse_money(value: str) -> float | None:
|
|
text = (value or "").strip().replace("R$", "").replace(" ", "")
|
|
if not text:
|
|
return None
|
|
if "," in text and "." in text:
|
|
text = text.replace(".", "").replace(",", ".")
|
|
elif "," in text:
|
|
text = text.replace(",", ".")
|
|
try:
|
|
return round(float(text), 2)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _parse_competencia_filter(mes: str, ano: str) -> tuple[int, int] | None:
|
|
"""(mês, ano) a partir dos `<select>` de filtro "De"/"Até", ou None se o
|
|
par não foi informado (filtro de competência é opcional na listagem)."""
|
|
mes = (mes or "").strip()
|
|
ano = (ano or "").strip()
|
|
if not mes or not ano:
|
|
return None
|
|
try:
|
|
mes_val, ano_val = int(mes), int(ano)
|
|
except ValueError:
|
|
return None
|
|
if not (1 <= mes_val <= 12):
|
|
return None
|
|
return mes_val, ano_val
|
|
|
|
|
|
@router.get("/documents")
|
|
def list_documents(
|
|
request: Request,
|
|
start_mes: str = Query(""),
|
|
start_ano: str = Query(""),
|
|
end_mes: str = Query(""),
|
|
end_ano: str = Query(""),
|
|
supplier: str = Query(""),
|
|
category: str = Query(""),
|
|
sort: str = Query("competencia"),
|
|
order: str = Query("desc"),
|
|
page: int = Query(1, ge=1),
|
|
):
|
|
if sort not in db.FISCAL_SORT_COLUMNS:
|
|
sort = "competencia"
|
|
order = "asc" if order.lower() == "asc" else "desc"
|
|
start = _parse_competencia_filter(start_mes, start_ano)
|
|
end = _parse_competencia_filter(end_mes, end_ano)
|
|
with db.session() as conn:
|
|
filter_kwargs = dict(
|
|
start=start,
|
|
end=end,
|
|
supplier=supplier or None,
|
|
category=category or None,
|
|
)
|
|
total_count, total = db.fiscal_summary(conn, **filter_kwargs)
|
|
total_pages = max(1, -(-total_count // PAGE_SIZE))
|
|
page = min(page, total_pages)
|
|
rows = [
|
|
dict(r)
|
|
for r in db.list_fiscal(
|
|
conn,
|
|
**filter_kwargs,
|
|
sort=sort,
|
|
order=order,
|
|
limit=PAGE_SIZE,
|
|
offset=(page - 1) * PAGE_SIZE,
|
|
)
|
|
]
|
|
categorias = [dict(r) for r in db.list_categorias_por_nome(conn)]
|
|
return render(
|
|
request,
|
|
"documents_list.html",
|
|
rows=rows,
|
|
total=total,
|
|
total_count=total_count,
|
|
page=page,
|
|
total_pages=total_pages,
|
|
page_numbers=_page_window(page, total_pages),
|
|
filters={
|
|
"start_mes": start_mes,
|
|
"start_ano": start_ano,
|
|
"end_mes": end_mes,
|
|
"end_ano": end_ano,
|
|
"supplier": supplier,
|
|
"category": category,
|
|
},
|
|
sort=sort,
|
|
order=order,
|
|
categorias=categorias,
|
|
money=money,
|
|
)
|
|
|
|
|
|
def _parse_mes_ano(mes: str, ano: str) -> tuple[int, int]:
|
|
"""(mês, ano) a partir dos `<select>` do formulário, com o mesmo fallback
|
|
"mês/ano corrente" que o antigo `<input type="date">` tinha para valor em
|
|
branco."""
|
|
fallback_mes, fallback_ano = current_competencia()
|
|
try:
|
|
mes_val = int(mes)
|
|
except (TypeError, ValueError):
|
|
mes_val = None
|
|
if mes_val is None or not (1 <= mes_val <= 12):
|
|
mes_val = fallback_mes
|
|
try:
|
|
ano_val = int(ano)
|
|
except (TypeError, ValueError):
|
|
ano_val = fallback_ano
|
|
return mes_val, ano_val
|
|
|
|
|
|
def _parse_categoria_id(value: str) -> int | None:
|
|
value = (value or "").strip()
|
|
if not value:
|
|
return None
|
|
try:
|
|
return int(value)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
@router.get("/documents/new")
|
|
def new_form(request: Request):
|
|
with db.session() as conn:
|
|
categorias = [dict(r) for r in db.list_categorias_por_nome(conn)]
|
|
default_mes, default_ano = current_competencia()
|
|
return render(
|
|
request,
|
|
"document_form.html",
|
|
doc=None,
|
|
mode="new",
|
|
categorias=categorias,
|
|
default_mes=default_mes,
|
|
default_ano=default_ano,
|
|
)
|
|
|
|
|
|
@router.post("/documents/new")
|
|
def create_document(
|
|
request: Request,
|
|
csrf_token: str = Form(""),
|
|
mes: str = Form(""),
|
|
ano: str = Form(""),
|
|
supplier_name: str = Form(""),
|
|
total_paid: str = Form(""),
|
|
categoria_id: str = Form(""),
|
|
):
|
|
if not auth.check_csrf(request, csrf_token):
|
|
flash(request, "Sessão expirada.", "error")
|
|
return RedirectResponse("/documents/new", status_code=303)
|
|
|
|
total = _parse_money(total_paid)
|
|
supplier = supplier_name.strip()
|
|
if not supplier or total is None:
|
|
flash(request, "Informe fornecedor e valor válidos.", "error")
|
|
return RedirectResponse("/documents/new", status_code=303)
|
|
|
|
mes_val, ano_val = _parse_mes_ano(mes, ano)
|
|
with db.session() as conn:
|
|
db.create_fiscal(
|
|
conn,
|
|
mes=mes_val,
|
|
ano=ano_val,
|
|
supplier_name=supplier,
|
|
total_paid=total,
|
|
categoria_id=_parse_categoria_id(categoria_id),
|
|
)
|
|
flash(request, "Documento lançado.", "success")
|
|
return RedirectResponse("/documents", status_code=303)
|
|
|
|
|
|
@router.get("/documents/{doc_id}/edit")
|
|
def edit_form(request: Request, doc_id: int):
|
|
with db.session() as conn:
|
|
doc = db.get_fiscal(conn, doc_id)
|
|
categorias = [dict(r) for r in db.list_categorias_por_nome(conn)]
|
|
if doc is None:
|
|
flash(request, "Documento não encontrado.", "error")
|
|
return RedirectResponse("/documents", status_code=303)
|
|
return render(request, "document_form.html", doc=dict(doc), mode="edit", categorias=categorias)
|
|
|
|
|
|
@router.post("/documents/{doc_id}/edit")
|
|
def update_document(
|
|
request: Request,
|
|
doc_id: int,
|
|
csrf_token: str = Form(""),
|
|
mes: str = Form(""),
|
|
ano: str = Form(""),
|
|
supplier_name: str = Form(""),
|
|
total_paid: str = Form(""),
|
|
categoria_id: str = Form(""),
|
|
):
|
|
if not auth.check_csrf(request, csrf_token):
|
|
flash(request, "Sessão expirada.", "error")
|
|
return RedirectResponse(f"/documents/{doc_id}/edit", status_code=303)
|
|
|
|
total = _parse_money(total_paid)
|
|
supplier = supplier_name.strip()
|
|
if not supplier or total is None:
|
|
flash(request, "Informe fornecedor e valor válidos.", "error")
|
|
return RedirectResponse(f"/documents/{doc_id}/edit", status_code=303)
|
|
|
|
mes_val, ano_val = _parse_mes_ano(mes, ano)
|
|
with db.session() as conn:
|
|
if db.get_fiscal(conn, doc_id) is None:
|
|
flash(request, "Documento não encontrado.", "error")
|
|
return RedirectResponse("/documents", status_code=303)
|
|
db.update_fiscal(
|
|
conn,
|
|
doc_id,
|
|
mes=mes_val,
|
|
ano=ano_val,
|
|
supplier_name=supplier,
|
|
total_paid=total,
|
|
categoria_id=_parse_categoria_id(categoria_id),
|
|
)
|
|
flash(request, "Documento atualizado.", "success")
|
|
return RedirectResponse("/documents", status_code=303)
|
|
|
|
|
|
@router.post("/documents/{doc_id}/delete")
|
|
def delete_document(request: Request, doc_id: int, csrf_token: str = Form("")):
|
|
if not auth.check_csrf(request, csrf_token):
|
|
flash(request, "Sessão expirada.", "error")
|
|
return RedirectResponse("/documents", status_code=303)
|
|
with db.session() as conn:
|
|
db.delete_fiscal(conn, doc_id)
|
|
flash(request, "Documento excluído.", "info")
|
|
return RedirectResponse("/documents", status_code=303)
|