# 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)