67 lines
2.6 KiB
Python
67 lines
2.6 KiB
Python
|
# routes/selic.py
|
||
|
|
import requests
|
||
|
|
from fastapi import APIRouter, Query, Depends
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
from sqlalchemy import text
|
||
|
|
from database import get_session
|
||
|
|
from models import SelicMensal
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
|
|
||
|
|
router = APIRouter()
|
||
|
|
|
||
|
|
BCB_API_URL = "https://api.bcb.gov.br/dados/serie/bcdata.sgs.4390/dados"
|
||
|
|
|
||
|
|
# 🔁 Função reutilizável para startup ou API
|
||
|
|
async def atualizar_selic_com_base_na_competencia(db: AsyncSession, a_partir_de: str = None):
|
||
|
|
result = await db.execute(text("SELECT MIN(referencia_competencia) FROM faturas.faturas"))
|
||
|
|
menor_comp = result.scalar()
|
||
|
|
if not menor_comp:
|
||
|
|
return {"message": "Nenhuma fatura encontrada na base."}
|
||
|
|
|
||
|
|
inicio = datetime.strptime(a_partir_de, "%m/%Y") if a_partir_de else datetime.strptime(menor_comp, "%m/%Y")
|
||
|
|
|
||
|
|
result_ultima = await db.execute(text("SELECT MAX(mes) FROM faturas.selic_mensal"))
|
||
|
|
ultima = result_ultima.scalar()
|
||
|
|
fim = datetime.today() if not ultima else max(datetime.today(), ultima + timedelta(days=31))
|
||
|
|
|
||
|
|
resultados = []
|
||
|
|
atual = inicio
|
||
|
|
while atual <= fim:
|
||
|
|
mes_ref = atual.replace(day=1)
|
||
|
|
|
||
|
|
existe = await db.execute(
|
||
|
|
text("SELECT 1 FROM faturas.selic_mensal WHERE mes = :mes"),
|
||
|
|
{"mes": mes_ref}
|
||
|
|
)
|
||
|
|
if existe.scalar():
|
||
|
|
atual += timedelta(days=32)
|
||
|
|
atual = atual.replace(day=1)
|
||
|
|
continue
|
||
|
|
|
||
|
|
url = f"{BCB_API_URL}?formato=json&dataInicial={mes_ref.strftime('%d/%m/%Y')}&dataFinal={mes_ref.strftime('%d/%m/%Y')}"
|
||
|
|
r = requests.get(url, timeout=10)
|
||
|
|
if not r.ok:
|
||
|
|
atual += timedelta(days=32)
|
||
|
|
atual = atual.replace(day=1)
|
||
|
|
continue
|
||
|
|
|
||
|
|
dados = r.json()
|
||
|
|
if dados:
|
||
|
|
valor = float(dados[0]['valor'].replace(',', '.')) / 100
|
||
|
|
db.add(SelicMensal(mes=mes_ref, fator=valor))
|
||
|
|
resultados.append({"mes": mes_ref.strftime("%m/%Y"), "fator": valor})
|
||
|
|
|
||
|
|
atual += timedelta(days=32)
|
||
|
|
atual = atual.replace(day=1)
|
||
|
|
|
||
|
|
await db.commit()
|
||
|
|
return {"message": f"Fatores SELIC atualizados com sucesso.", "novos_registros": resultados}
|
||
|
|
|
||
|
|
# 🛠️ Rota opcional reutilizando a função
|
||
|
|
@router.post("/atualizar-selic")
|
||
|
|
async def atualizar_selic(
|
||
|
|
db: AsyncSession = Depends(get_session),
|
||
|
|
a_partir_de: str = Query(None, description="Opcional: formato MM/AAAA para forçar atualização a partir de determinada data")
|
||
|
|
):
|
||
|
|
return await atualizar_selic_com_base_na_competencia(db=db, a_partir_de=a_partir_de)
|