# -*- coding: utf-8 -*- from odoo import models, fields, api, _ from odoo.exceptions import UserError, ValidationError import logging _logger = logging.getLogger(__name__) class LimsTest(models.Model): _name = 'lims.test' _description = 'Prueba de Laboratorio' _inherit = ['mail.thread', 'mail.activity.mixin'] _rec_name = 'name' _order = 'create_date desc' name = fields.Char( string='Código de Prueba', required=True, readonly=True, copy=False, default='Nuevo' ) sale_order_line_id = fields.Many2one( 'sale.order.line', string='Línea de Orden', required=True, ondelete='restrict' ) sale_order_id = fields.Many2one( 'sale.order', string='Orden de Venta', related='sale_order_line_id.order_id', store=True, readonly=True ) patient_id = fields.Many2one( 'res.partner', string='Paciente', related='sale_order_line_id.order_id.partner_id', store=True, readonly=True ) product_id = fields.Many2one( 'product.product', string='Análisis', related='sale_order_line_id.product_id', store=True, readonly=True ) sample_id = fields.Many2one( 'stock.lot', string='Muestra', domain="[('is_lab_sample', '=', True), ('patient_id', '=', patient_id), ('state', 'in', ['collected', 'in_analysis'])]", tracking=True ) sample_state = fields.Selection( related='sample_id.state', string='Estado de Muestra', readonly=True ) state = fields.Selection([ ('draft', 'Borrador'), ('in_process', 'En Proceso'), ('result_entered', 'Resultado Ingresado'), ('validated', 'Validado'), ('cancelled', 'Cancelado') ], string='Estado', default='draft', tracking=True) validator_id = fields.Many2one( 'res.users', string='Validador', readonly=True, tracking=True ) validation_date = fields.Datetime( string='Fecha de Validación', readonly=True, tracking=True ) technician_id = fields.Many2one( 'res.users', string='Técnico', default=lambda self: self.env.user, tracking=True ) require_validation = fields.Boolean( string='Requiere Validación', compute='_compute_require_validation', store=True ) result_ids = fields.One2many( 'lims.result', 'test_id', string='Resultados' ) notes = fields.Text( string='Observaciones' ) company_id = fields.Many2one( 'res.company', string='Compañía', required=True, default=lambda self: self.env.company ) # Campos para dashboards demográficos patient_gender = fields.Selection( related='patient_id.gender', string='Género del Paciente', store=True, readonly=True ) patient_age_range = fields.Selection( related='patient_id.age_range', string='Rango de Edad', store=True, readonly=True ) @api.depends('company_id') def _compute_require_validation(self): """Calcula si la prueba requiere validación basado en configuración.""" IrConfig = self.env['ir.config_parameter'].sudo() require_validation = IrConfig.get_param('lims_management.require_validation', 'True') for record in self: record.require_validation = require_validation == 'True' @api.onchange('sale_order_line_id') def _onchange_sale_order_line(self): """Update sample domain when order line changes""" if self.sale_order_line_id: # Try to find a suitable sample from the order order = self.sale_order_line_id.order_id product = self.sale_order_line_id.product_id if order.is_lab_request and product.required_sample_type_id: # Find samples for this patient with the required sample type suitable_samples = self.env['stock.lot'].search([ ('is_lab_sample', '=', True), ('patient_id', '=', order.partner_id.id), ('sample_type_product_id', '=', product.required_sample_type_id.id), ('state', 'in', ['collected', 'in_analysis']) ]) if suitable_samples: # If only one sample, select it automatically if len(suitable_samples) == 1: self.sample_id = suitable_samples[0] # Update domain to show only suitable samples return { 'domain': { 'sample_id': [ ('id', 'in', suitable_samples.ids) ] } } @api.model_create_multi def create(self, vals_list): """Genera código único al crear.""" for vals in vals_list: if vals.get('name', 'Nuevo') == 'Nuevo': vals['name'] = self.env['ir.sequence'].next_by_code('lims.test') or 'Nuevo' tests = super().create(vals_list) # Generar resultados automáticamente tests._generate_test_results() return tests def _generate_test_results(self): """Genera automáticamente las líneas de resultado basadas en los parámetros configurados del análisis.""" for test in self: if test.result_ids: # Si ya tiene resultados, no generar nuevos continue # Obtener el product.template del análisis product_tmpl = test.product_id.product_tmpl_id # Buscar los parámetros configurados para este análisis template_parameters = self.env['product.template.parameter'].search([ ('product_tmpl_id', '=', product_tmpl.id) ], order='sequence, id') # Crear una línea de resultado por cada parámetro for param_config in template_parameters: # Preparar las notas/instrucciones notes = param_config.instructions or '' # Si es un parámetro de tipo selection, agregar instrucciones de autocompletado if param_config.parameter_value_type == 'selection': selection_values = param_config.parameter_id.selection_values if selection_values: options = [v.strip() for v in selection_values.split(',')] if options: # Generar instrucciones automáticas auto_instructions = "Opciones: " + ", ".join(options) + ". " auto_instructions += "Puede escribir las iniciales o parte del texto. " # Agregar ejemplos específicos examples = [] for opt in options[:3]: # Mostrar ejemplos para las primeras 3 opciones if opt: initial = opt[0].upper() examples.append(f"{initial}={opt}") if examples: auto_instructions += "Ej: " + ", ".join(examples) # Combinar con instrucciones existentes if notes: notes = auto_instructions + "\n" + notes else: notes = auto_instructions result_vals = { 'test_id': test.id, 'parameter_id': param_config.parameter_id.id, 'sequence': param_config.sequence, 'notes': notes } # Inicializar valores según el tipo if param_config.parameter_value_type == 'boolean': result_vals['value_boolean'] = False self.env['lims.result'].create(result_vals) if template_parameters: _logger.info(f"Generados {len(template_parameters)} resultados para la prueba {test.name}") else: _logger.warning(f"No se encontraron parámetros configurados para el análisis {product_tmpl.name}") def action_start_process(self): """Inicia el proceso de análisis.""" self.ensure_one() # Verificar permisos: solo técnicos y administradores if not (self.env.user.has_group('lims_management.group_lims_technician') or self.env.user.has_group('lims_management.group_lims_admin')): raise UserError(_('No tiene permisos para iniciar el proceso de análisis. Solo técnicos y administradores pueden realizar esta acción.')) if self.state != 'draft': raise UserError(_('Solo se pueden procesar pruebas en estado borrador.')) if not self.sample_id: raise UserError(_('Debe asignar una muestra antes de iniciar el proceso.')) self.write({ 'state': 'in_process', 'technician_id': self.env.user.id }) # Log en el chatter self.message_post( body=_('Prueba iniciada por %s') % self.env.user.name, subject=_('Proceso Iniciado'), message_type='notification' ) # Actualizar estado de la muestra si es necesario if self.sample_id and self.sample_id.state == 'collected': self.sample_id.write({'state': 'in_process'}) self.sample_id.message_post( body=_('Muestra en análisis para la prueba %s') % self.name, subject=_('Estado actualizado'), message_type='notification' ) return True def action_enter_results(self): """Marca como resultados ingresados.""" self.ensure_one() # Verificar permisos: solo técnicos y administradores if not (self.env.user.has_group('lims_management.group_lims_technician') or self.env.user.has_group('lims_management.group_lims_admin')): raise UserError(_('No tiene permisos para ingresar resultados. Solo técnicos y administradores pueden realizar esta acción.')) if self.state != 'in_process': raise UserError(_('Solo se pueden ingresar resultados en pruebas en proceso.')) if not self.result_ids: raise UserError(_('Debe ingresar al menos un resultado.')) # Verificar que todos los resultados tengan valores ingresados empty_results = self.result_ids.filtered( lambda r: not r.value_text and not r.value_numeric and not r.value_selection and not r.value_boolean and r.parameter_id.value_type != 'boolean' ) if empty_results: params = ', '.join(empty_results.mapped('parameter_id.name')) raise UserError(_('Los siguientes parámetros no tienen resultados ingresados: %s') % params) # Si no requiere validación, pasar directamente a validado if not self.require_validation: self.write({ 'state': 'validated', 'validator_id': self.env.user.id, 'validation_date': fields.Datetime.now() }) self.message_post( body=_('Resultados ingresados y auto-validados por %s') % self.env.user.name, subject=_('Resultados Validados'), message_type='notification' ) else: self.state = 'result_entered' self.message_post( body=_('Resultados ingresados por %s') % self.env.user.name, subject=_('Resultados Ingresados'), message_type='notification' ) return True def action_validate(self): """Valida los resultados (solo administradores).""" self.ensure_one() # Verificar permisos: solo administradores if not self.env.user.has_group('lims_management.group_lims_admin'): raise UserError(_('No tiene permisos para validar resultados. Solo administradores pueden realizar esta acción.')) if self.state != 'result_entered': raise UserError(_('Solo se pueden validar pruebas con resultados ingresados.')) # Verificar que todos los resultados críticos tengan observaciones si están fuera de rango critical_results = [] for result in self.result_ids: if result.is_critical: # Usar el campo is_critical del resultado, no del parámetro if not result.notes: critical_results.append(result.parameter_id.name) if critical_results: raise UserError(_('Los siguientes parámetros críticos están fuera de rango y requieren observaciones: %s') % ', '.join(critical_results)) self.write({ 'state': 'validated', 'validator_id': self.env.user.id, 'validation_date': fields.Datetime.now() }) # Log en el chatter con más detalles out_of_range_count = len(self.result_ids.filtered('is_out_of_range')) body = _('Resultados validados por %s') % self.env.user.name if out_of_range_count: body += _('
%d parámetros fuera de rango') % out_of_range_count self.message_post( body=body, subject=_('Resultados Validados'), message_type='notification' ) # Actualizar estado de la muestra si todas las pruebas están validadas if self.sample_id: all_tests = self.env['lims.test'].search([ ('sample_id', '=', self.sample_id.id), ('state', '!=', 'cancelled') ]) if all(test.state == 'validated' for test in all_tests): self.sample_id.write({'state': 'analyzed'}) self.sample_id.message_post( body=_('Todas las pruebas de la muestra han sido validadas'), subject=_('Análisis completado'), message_type='notification' ) return True def action_cancel(self): """Cancela la prueba.""" self.ensure_one() # Verificar permisos: técnicos y administradores pueden cancelar if not (self.env.user.has_group('lims_management.group_lims_technician') or self.env.user.has_group('lims_management.group_lims_admin')): raise UserError(_('No tiene permisos para cancelar pruebas. Solo técnicos y administradores pueden realizar esta acción.')) if self.state == 'validated': # Solo administradores pueden cancelar pruebas validadas if not self.env.user.has_group('lims_management.group_lims_admin'): raise UserError(_('No se pueden cancelar pruebas validadas. Solo administradores pueden realizar esta acción.')) old_state = self.state self.state = 'cancelled' # Log en el chatter con el estado anterior self.message_post( body=_('Prueba cancelada por %s (estado anterior: %s)') % (self.env.user.name, dict(self._fields['state'].selection).get(old_state)), subject=_('Prueba Cancelada'), message_type='notification' ) return True def action_regenerate_results(self): """Regenera los resultados basados en la configuración actual del análisis.""" self.ensure_one() # Verificar permisos: solo técnicos y administradores if not (self.env.user.has_group('lims_management.group_lims_technician') or self.env.user.has_group('lims_management.group_lims_admin')): raise UserError(_('No tiene permisos para regenerar resultados. Solo técnicos y administradores pueden realizar esta acción.')) if self.state not in ['draft', 'in_process']: raise UserError(_('Solo se pueden regenerar resultados en pruebas en borrador o en proceso.')) # Confirmar con el usuario if self.result_ids: # En producción, aquí se mostraría un wizard de confirmación # Por ahora, eliminamos los resultados existentes self.result_ids.unlink() # Regenerar self._generate_test_results() self.message_post( body=_('Resultados regenerados por %s') % self.env.user.name, subject=_('Resultados Regenerados'), message_type='notification' ) return True def action_draft(self): """Regresa a borrador.""" self.ensure_one() # Verificar permisos: solo administradores pueden regresar a borrador if not self.env.user.has_group('lims_management.group_lims_admin'): raise UserError(_('No tiene permisos para regresar pruebas a borrador. Solo administradores pueden realizar esta acción.')) if self.state not in ['cancelled']: raise UserError(_('Solo se pueden regresar a borrador pruebas canceladas.')) self.state = 'draft' self.message_post( body=_('Prueba regresada a borrador por %s') % self.env.user.name, subject=_('Estado Restaurado'), message_type='notification' ) return True @api.constrains('state') def _check_state_transition(self): """Valida que las transiciones de estado sean válidas""" for record in self: # Definir transiciones válidas valid_transitions = { 'draft': ['in_process', 'cancelled'], 'in_process': ['result_entered', 'cancelled'], 'result_entered': ['validated', 'cancelled'], 'validated': ['cancelled'], # Solo admin puede cancelar validados 'cancelled': ['draft'] # Solo admin puede regresar a draft } # Si es un registro nuevo, no hay transición que validar if not record._origin.id: continue old_state = record._origin.state new_state = record.state # Si el estado no cambió, no hay nada que validar if old_state == new_state: continue # Verificar si la transición es válida if old_state in valid_transitions: if new_state not in valid_transitions[old_state]: raise ValidationError( _('Transición de estado no válida: No se puede cambiar de "%s" a "%s"') % (dict(self._fields['state'].selection).get(old_state), dict(self._fields['state'].selection).get(new_state)) ) @api.constrains('sample_id', 'state') def _check_sample_state(self): """Valida que la muestra esté en un estado apropiado para la prueba""" for record in self: if record.sample_id and record.state in ['in_process', 'result_entered']: # La muestra debe estar al menos recolectada if record.sample_id.state in ['pending_collection', 'cancelled']: raise ValidationError( _('No se puede procesar una prueba con una muestra en estado "%s"') % dict(record.sample_id._fields['state'].selection).get(record.sample_id.state) ) @api.model def create(self, vals): """Override create para validaciones adicionales""" # Si se está creando con un estado diferente a draft, verificar permisos if vals.get('state') and vals['state'] != 'draft': if not self.env.user.has_group('lims_management.group_lims_admin'): raise UserError(_('Solo administradores pueden crear pruebas en estado diferente a borrador')) return super().create(vals) def write(self, vals): """Override write para auditoría adicional""" # Si se está cambiando el estado, registrar más detalles if 'state' in vals: for record in self: old_state = record.state # El write real se hace en el super() result = super().write(vals) # Registrar cambios importantes después del write if 'sample_id' in vals: for record in self: if vals.get('sample_id'): sample = self.env['stock.lot'].browse(vals['sample_id']) record.message_post( body=_('Muestra asignada: %s') % sample.name, subject=_('Muestra Asignada'), message_type='notification' ) return result