85 lines
3.5 KiB
Python
Executable File
85 lines
3.5 KiB
Python
Executable File
# app/relatorios.py
|
|
from fastapi import APIRouter, Depends, Query
|
|
from fastapi.responses import StreamingResponse
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.future import select
|
|
from database import get_session
|
|
from models import Fatura, ParametrosFormula, AliquotaUF
|
|
from io import BytesIO
|
|
import pandas as pd
|
|
from datetime import datetime
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def calcular_pis_cofins_corretos(base, icms, aliquota):
|
|
try:
|
|
return round((base - (base - icms)) * aliquota, 5)
|
|
except:
|
|
return 0.0
|
|
|
|
|
|
@router.get("/relatorio-exclusao-icms")
|
|
async def relatorio_exclusao_icms(cliente: str = Query(None), db: AsyncSession = Depends(get_session)):
|
|
faturas = db.query(Fatura).all()
|
|
dados = []
|
|
for f in faturas:
|
|
if f.base_pis == f.base_icms == f.base_cofins:
|
|
pis_corr = calcular_pis_cofins_corretos(f.base_pis, f.valor_icms, f.aliq_pis)
|
|
cofins_corr = calcular_pis_cofins_corretos(f.base_cofins, f.valor_icms, f.aliq_cofins)
|
|
dados.append({
|
|
"Classificacao": f.classificacao,
|
|
"Nome": f.nome,
|
|
"UC": f.uc,
|
|
"Competencia": f.referencia,
|
|
"Valor Total": f.valor_total,
|
|
"Alíquota PIS": f.aliq_pis,
|
|
"Alíquota ICMS": f.aliq_icms,
|
|
"Alíquota COFINS": f.aliq_cofins,
|
|
"Valor PIS": f.valor_pis,
|
|
"Valor ICMS": f.valor_icms,
|
|
"Valor COFINS": f.valor_cofins,
|
|
"Base PIS": f.base_pis,
|
|
"Base ICMS": f.base_icms,
|
|
"PIS Corrigido": pis_corr,
|
|
"COFINS Corrigido": cofins_corr,
|
|
"Arquivo": f.arquivo
|
|
})
|
|
|
|
df = pd.DataFrame(dados)
|
|
excel_file = BytesIO()
|
|
df.to_excel(excel_file, index=False)
|
|
excel_file.seek(0)
|
|
return StreamingResponse(excel_file, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": "attachment; filename=relatorio_exclusao_icms.xlsx"})
|
|
|
|
|
|
@router.get("/relatorio-aliquota-incorreta")
|
|
async def relatorio_icms_errado(cliente: str = Query(None), db: AsyncSession = Depends(get_session)):
|
|
result = await db.execute(select(Fatura))
|
|
faturas = result.scalars().all()
|
|
dados = []
|
|
for f in faturas:
|
|
aliq_registrada = db.query(AliquotaUF).filter_by(uf=f.estado, exercicio=f.referencia[-4:]).first()
|
|
if aliq_registrada and abs(f.aliq_icms - aliq_registrada.aliquota) > 0.001:
|
|
icms_corr = round((f.base_icms * aliq_registrada.aliquota), 5)
|
|
dados.append({
|
|
"Classificacao": f.classificacao,
|
|
"Nome": f.nome,
|
|
"UC": f.uc,
|
|
"Competencia": f.referencia,
|
|
"Valor Total": f.valor_total,
|
|
"Alíquota ICMS (Fatura)": f.aliq_icms,
|
|
"Alíquota ICMS (Correta)": aliq_registrada.aliquota,
|
|
"Base ICMS": f.base_icms,
|
|
"Valor ICMS": f.valor_icms,
|
|
"ICMS Corrigido": icms_corr,
|
|
"Arquivo": f.arquivo
|
|
})
|
|
|
|
df = pd.DataFrame(dados)
|
|
excel_file = BytesIO()
|
|
df.to_excel(excel_file, index=False)
|
|
excel_file.seek(0)
|
|
return StreamingResponse(excel_file, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": "attachment; filename=relatorio_icms_errado.xlsx"})
|