Spaces:
Running
Running
File size: 5,237 Bytes
35c6a2c c84b9d8 35c6a2c c84b9d8 35c6a2c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 |
import sys
from models.MovimientosM import Movimiento
from fastapi import HTTPException
from schemas.MovimientosSh import *
from sqlalchemy.orm import Session
from datetime import date,datetime,timedelta
from sqlalchemy import func,between
from core.utils import get_current_date
def create_new_movimiento(movi:MovimientosCreate,id_usuario:str,db:Session):
db_movimiento = Movimiento(
id_usuario = id_usuario,
cuenta = movi.cuenta,
tipo_movimiento = movi.tipo_movimiento,
fecha = movi.fecha,
descripcion = movi.descripcion,
)
try:
db.add(db_movimiento)
db.commit()
db.refresh(db_movimiento)
db.close()
return db_movimiento
except Exception as e :
db.rollback()
# Imprimir el error en la consola
print(f"Error al crear un usuario: {str(e)}",file=sys.stderr)
raise HTTPException(status_code=500,detail=f"Error al crear un movimiento: {str(e)} ")
def get_weekly_expenses(db: Session, user_id: str,tipo:str):
today = datetime.today()
if tipo == "SEMANA":
start_week = today - timedelta(days=today.weekday())
end_week = start_week + timedelta(days=6)
if tipo == "MES":
start_week = today.replace(day=1)
next_month = start_week.replace(month=start_week.month + 1, day=1)
end_week = next_month - timedelta(days=1)
if tipo == "HOY":
start_week = today.replace(hour=0, minute=0, second=0, microsecond=0)
# Obtener el último momento del día (23:59:59)
end_week = today.replace(hour=23, minute=59, second=59, microsecond=999999)
expenses = db.query(Movimiento.descripcion, Movimiento.cuenta,Movimiento.id_movimientos).\
filter(Movimiento.id_usuario == user_id,
Movimiento.tipo_movimiento == 'GASTO',
Movimiento.estado_gasto == True,
between(Movimiento.fecha, start_week.date(), end_week.date())).\
all()
total_expenses = sum(exp[1] for exp in expenses)
details = [{"descripcion": exp[0], "cuenta": float(exp[1]),"id_movimientos":int(exp[2])} for exp in expenses]
return total_expenses,details
def get_consulted_expenses(db: Session, user_id: str,start_week : date,end_week:date):
expenses = db.query(Movimiento.descripcion, Movimiento.cuenta).\
filter(Movimiento.id_usuario == user_id,
Movimiento.tipo_movimiento == 'GASTO',
Movimiento.estado_gasto == True,
between(Movimiento.fecha, start_week, end_week)).\
all()
total_expenses = sum(exp[1] for exp in expenses) # Sumando las cuentas (índice 1)
# Convertir los detalles de gastos a una lista de diccionarios
details = [{"descripcion": exp[0], "cuenta": float(exp[1])} for exp in expenses]
return total_expenses,details
def buscar_movimientos(id:int,db:Session):
movimiento_db = db.query(Movimiento).filter(Movimiento.id_movimientos == id).first()
return movimiento_db
def money_account(db:Session,id_usuario:str,fecha_inicio:date = None,fecha_fin:date = None):
if fecha_inicio is None or fecha_fin is None:
fecha_inicio = get_current_date()
fecha_fin = get_current_date()
ingresos = db.query(func.sum(Movimiento.cuenta)).filter(
Movimiento.tipo_movimiento == 'INGRESO',
between(Movimiento.fecha,fecha_inicio,fecha_fin),
Movimiento.id_usuario == id_usuario,
Movimiento.estado_gasto == 1
).scalar() or 0
gastos = db.query(func.sum(Movimiento.cuenta)).filter(
Movimiento.tipo_movimiento == 'GASTO',
between(Movimiento.fecha,fecha_inicio,fecha_fin),
Movimiento.id_usuario == id_usuario,
Movimiento.estado_gasto == 1
).scalar() or 0
resultado = ingresos - gastos;
return resultado
def update_movimiento_crud(movi:MovimientosUpdate,db:Session):
movimiento_db = buscar_movimientos(movi.id_movimientos,db)
if movimiento_db:
try:
movimiento_db.cuenta = movi.cuenta
movimiento_db.tipo_movimiento = movi.tipo_movimiento
movimiento_db.descripcion = movi.descripcion
db.add(movimiento_db)
db.commit()
db.refresh(movimiento_db)
return movimiento_db
except Exception as e :
db.rollback()
print(f"Error al actualizar movimiento: {str(e)}",file=sys.stderr)
raise HTTPException(status_code=500,detail=f"Error al actualizar un movimiento: {str(e)} ")
return None
def update_movimientos_status_crud(movi:MovimientosUpdateStatus, db:Session):
movimientos_db = buscar_movimientos(movi.id_movimientos, db)
if movimientos_db:
try:
movimientos_db.estado_gasto = movi.estado_gasto
db.add(movimientos_db)
db.commit()
return True
except Exception as e :
db.rollback()
print(f"Error al update un movimiento {str(e)}",file=sys.stderr)
raise HTTPException(status_code=500,detail=f"Error al update movimiento: {str(e)} ")
return False |