212 lines
8.6 KiB
Python
212 lines
8.6 KiB
Python
# modules/users/repositories.py
|
|
from typing import Optional, List, Dict, Any
|
|
from datetime import datetime
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import and_, or_
|
|
|
|
from app.core.baseRepository import BaseRepository
|
|
from app.modules.users.models import Users, TipoUsuario, Rol
|
|
from app.core.security import get_password_hash, verify_password
|
|
|
|
|
|
class UserRepository(BaseRepository[Users]):
|
|
"""Repositorio específico para Users"""
|
|
|
|
def __init__(self, db: Session):
|
|
super().__init__(db, Users)
|
|
|
|
# ============================================
|
|
# MÉTODOS DE AUTENTICACIÓN
|
|
# ============================================
|
|
|
|
def get_by_email(self, email: str, include_deleted: bool = False) -> Optional[Users]:
|
|
"""Obtiene usuario por email"""
|
|
return self._base_query(include_deleted).filter(Users.email == email).first()
|
|
|
|
def get_by_firma(self, firma: str, include_deleted: bool = False) -> Optional[Users]:
|
|
"""Obtiene usuario por firma digital"""
|
|
return self._base_query(include_deleted).filter(Users.firma == firma).first()
|
|
|
|
def authenticate(self, email: str, password: str) -> Optional[Users]:
|
|
"""Autentica usuario por email y contraseña"""
|
|
user = self.get_by_email(email)
|
|
if not user:
|
|
return None
|
|
if not user.is_active or user.deleted_at is not None:
|
|
return None
|
|
if not verify_password(password, user.password):
|
|
return None
|
|
return user
|
|
|
|
def update_last_login(self, user_id: int, ip_address: str) -> Optional[Users]:
|
|
"""Actualiza timestamp e IP del último login"""
|
|
user = self.get_by_id(user_id)
|
|
if user:
|
|
user.last_login = datetime.utcnow()
|
|
user.last_ip = ip_address
|
|
self.db.commit()
|
|
self.db.refresh(user)
|
|
return user
|
|
|
|
def change_password(self, user_id: int, new_password: str, updated_by: int = None) -> bool:
|
|
"""Cambia la contraseña del usuario"""
|
|
user = self.get_by_id(user_id)
|
|
if not user:
|
|
return False
|
|
|
|
user.password = get_password_hash(new_password)
|
|
if updated_by:
|
|
user.updated_by = updated_by
|
|
|
|
self.db.commit()
|
|
return True
|
|
|
|
# ============================================
|
|
# MÉTODOS POR TIPO DE USUARIO
|
|
# ============================================
|
|
|
|
def get_by_tipo_usuario(self, tipo_usuario: str, include_deleted: bool = False) -> List[Users]:
|
|
"""Obtiene usuarios por tipo (root, admin, admin_licencias, user)"""
|
|
return self._base_query(include_deleted).filter(Users.tipo_usuario == tipo_usuario).all()
|
|
|
|
def get_root(self, include_deleted: bool = False) -> List[Users]:
|
|
"""Usuarios tipo root"""
|
|
return self.get_by_tipo_usuario(TipoUsuario.ROOT, include_deleted)
|
|
|
|
def get_admin(self, include_deleted: bool = False) -> List[Users]:
|
|
"""Usuarios tipo admin"""
|
|
return self.get_by_tipo_usuario(TipoUsuario.ADMIN, include_deleted)
|
|
|
|
def get_admin_licencias(self, include_deleted: bool = False) -> List[Users]:
|
|
"""Usuarios tipo admin_licencias"""
|
|
return self.get_by_tipo_usuario(TipoUsuario.ADMIN_LICENCIAS, include_deleted)
|
|
|
|
def get_regular_users(self, include_deleted: bool = False) -> List[Users]:
|
|
"""Usuarios tipo user (regulares)"""
|
|
return self.get_by_tipo_usuario(TipoUsuario.USER, include_deleted)
|
|
|
|
# ============================================
|
|
# MÉTODOS POR ROL OPERATIVO
|
|
# ============================================
|
|
|
|
def get_by_rol_operativo(self, rol: str, include_deleted: bool = False) -> List[Users]:
|
|
"""Obtiene usuarios por rol operativo"""
|
|
return self._base_query(include_deleted).filter(Users.rol_operativo == rol).all()
|
|
|
|
def get_by_permiso_diot(self, include_deleted: bool = False) -> List[Users]:
|
|
"""Obtiene usuarios que tienen permiso DIOT"""
|
|
return self._base_query(include_deleted).filter(Users.permite_diot == True).all()
|
|
|
|
# ============================================
|
|
# MÉTODOS POR ASOCIACIONES
|
|
# ============================================
|
|
|
|
def get_by_enterprise(self, enterprise_id: int, include_deleted: bool = False) -> List[Users]:
|
|
"""Obtiene usuarios de una empresa específica"""
|
|
return self._base_query(include_deleted).filter(Users.enterprise_id == enterprise_id).all()
|
|
|
|
def get_by_sucursal(self, sucursal_id: int, include_deleted: bool = False) -> List[Users]:
|
|
"""Obtiene usuarios de una sucursal específica"""
|
|
return self._base_query(include_deleted).filter(Users.sucursales_id == sucursal_id).all()
|
|
|
|
def get_by_licencia(self, licencia_id: int, include_deleted: bool = False) -> List[Users]:
|
|
"""Obtiene usuarios asociados a una licencia"""
|
|
return self._base_query(include_deleted).filter(Users.licencia_id == licencia_id).all()
|
|
|
|
def get_by_perfil(self, perfil_id: int, include_deleted: bool = False) -> List[Users]:
|
|
"""Obtiene usuarios por perfil"""
|
|
return self._base_query(include_deleted).filter(Users.perfil_id == perfil_id).all()
|
|
|
|
# ============================================
|
|
# MÉTODOS DE ESTADO
|
|
# ============================================
|
|
|
|
def get_active(self, skip: int = 0, limit: int = 100) -> List[Users]:
|
|
"""Obtiene usuarios activos (no eliminados)"""
|
|
return self.get_all(skip=skip, limit=limit, include_deleted=False)
|
|
|
|
def get_inactive(self, skip: int = 0, limit: int = 100) -> List[Users]:
|
|
"""Obtiene usuarios inactivos o eliminados"""
|
|
return self._base_query(include_deleted=True).filter(
|
|
or_(
|
|
Users.is_active == False,
|
|
Users.deleted_at.isnot(None)
|
|
)
|
|
).offset(skip).limit(limit).all()
|
|
|
|
def activate(self, user_id: int, updated_by: int = None) -> Optional[Users]:
|
|
"""Activa un usuario inactivo"""
|
|
user = self.get_by_id(user_id, include_deleted=True)
|
|
if not user:
|
|
return None
|
|
|
|
user.is_active = True
|
|
if user.deleted_at:
|
|
user.deleted_at = None
|
|
if updated_by:
|
|
user.updated_by = updated_by
|
|
|
|
self.db.commit()
|
|
self.db.refresh(user)
|
|
return user
|
|
|
|
def deactivate(self, user_id: int, updated_by: int = None) -> Optional[Users]:
|
|
"""Desactiva un usuario (sin soft delete)"""
|
|
user = self.get_by_id(user_id)
|
|
if not user:
|
|
return None
|
|
|
|
user.is_active = False
|
|
if updated_by:
|
|
user.updated_by = updated_by
|
|
|
|
self.db.commit()
|
|
self.db.refresh(user)
|
|
return user
|
|
|
|
# ============================================
|
|
# MÉTODOS DE BÚSQUEDA
|
|
# ============================================
|
|
|
|
def search(self, term: str, skip: int = 0, limit: int = 100) -> List[Users]:
|
|
"""Búsqueda por nombre, email o RFC"""
|
|
search_term = f"%{term}%"
|
|
return self._base_query().filter(
|
|
or_(
|
|
Users.name.ilike(search_term),
|
|
Users.last_name.ilike(search_term),
|
|
Users.email.ilike(search_term),
|
|
Users.rfc.ilike(search_term)
|
|
)
|
|
).offset(skip).limit(limit).all()
|
|
|
|
def count_by_enterprise(self, enterprise_id: int) -> int:
|
|
"""Cuenta usuarios por empresa"""
|
|
return self._base_query().filter(Users.enterprise_id == enterprise_id).count()
|
|
|
|
def count_by_tipo(self, tipo_usuario: str) -> int:
|
|
"""Cuenta usuarios por tipo"""
|
|
return self._base_query().filter(Users.tipo_usuario == tipo_usuario).count()
|
|
|
|
# ============================================
|
|
# SOBRESCRITURA DE MÉTODOS BASE CON TRAZABILIDAD
|
|
# ============================================
|
|
|
|
def create(self, data: Dict[str, Any], created_by: int = None) -> Users:
|
|
"""Crea usuario con hash de contraseña"""
|
|
create_data = data.copy()
|
|
|
|
# Hashear contraseña si existe
|
|
if "password" in create_data:
|
|
raw_password = create_data["password"]
|
|
print(f"HASHEANDO PASSWORD: {raw_password} (longitud: {len(raw_password)})")
|
|
create_data["password"] = get_password_hash(raw_password)
|
|
print(f"HASH GENERADO: {create_data['password'][:20]}...")
|
|
|
|
return super().create(create_data, created_by)
|
|
|
|
def update(self, id: int, data: Dict[str, Any], updated_by: int = None, include_deleted: bool = False) -> Optional[Users]:
|
|
"""Actualiza usuario, hashea contraseña si viene"""
|
|
if "password" in data:
|
|
data["password"] = get_password_hash(data["password"])
|
|
return super().update(id, data, updated_by, include_deleted) |