497 lines
22 KiB
Python
497 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""API autenticada de formularios digitales (panel de la clínica).
|
|
|
|
Rutas bajo `/skeen/frontend/v1/formularios/*`. Mismo guard doble que el resto
|
|
del frontend: `@require_role` para el rango del rol y `@require_perm` para la
|
|
acción concreta del catálogo (`formularios.ver|crear|editar|enviar`).
|
|
|
|
Los binarios (firma y fotos) se sirven por rutas propias que aceptan el token
|
|
de sesión por header o por query string, para poder usarlos en `<img src>`.
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
import logging
|
|
|
|
from odoo import fields, http
|
|
from odoo.http import request, Response
|
|
|
|
from odoo.addons.skeen_whatsapp.controllers.auth import (
|
|
_json, current_user, require_role, require_perm,
|
|
verify_token as verify_session_token,
|
|
)
|
|
from odoo.addons.skeen_citas.clinic_settings import get_notificaciones, url_publica
|
|
|
|
from .comun import (
|
|
firmar_token, plantilla_dict, respuesta_dict, adjunto_dict,
|
|
limpiar_respuestas, validar_respuestas,
|
|
parse_data_url, MAX_FOTO_BYTES, MAX_FIRMA_BYTES, MIMES_FOTO,
|
|
)
|
|
from ..models.formulario_plantilla import normalizar_campos
|
|
|
|
PREFIX = '/skeen/frontend/v1/formularios'
|
|
|
|
|
|
def _body():
|
|
try:
|
|
return json.loads(request.httprequest.get_data(as_text=True) or '{}')
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _error(message, status=400):
|
|
return _json({'status': 'error', 'message': message}, status)
|
|
|
|
|
|
def _enlace_publico(env, token):
|
|
base = url_publica(env, '')
|
|
ruta = '/formulario/%s' % token
|
|
return (base + ruta) if base else ruta
|
|
|
|
|
|
def _kind_de_campo(plantilla, clave):
|
|
for campo in plantilla.campos_lista():
|
|
if campo.get('clave') == clave:
|
|
return campo.get('momento') or 'otro'
|
|
return 'otro'
|
|
|
|
|
|
class SkeenFormulariosController(http.Controller):
|
|
|
|
# ------------------------------------------------------------------
|
|
# Plantillas
|
|
# ------------------------------------------------------------------
|
|
@http.route(PREFIX + '/plantillas', type='http', auth='none',
|
|
methods=['GET', 'OPTIONS'], csrf=False)
|
|
@require_role('lectura')
|
|
@require_perm('formularios.ver')
|
|
def listar_plantillas(self, **kw):
|
|
try:
|
|
incluir_inactivas = str(kw.get('include_inactive') or '') in ('1', 'true', 'True')
|
|
dominio = [] if incluir_inactivas else [('activo', '=', True)]
|
|
if kw.get('tipo'):
|
|
dominio.append(('tipo', '=', kw['tipo']))
|
|
plantillas = request.env['skeen.formulario.plantilla'].sudo().search(
|
|
dominio, order='secuencia, name, id')
|
|
return _json({
|
|
'status': 'success',
|
|
'plantillas': [plantilla_dict(p) for p in plantillas],
|
|
'total': len(plantillas),
|
|
})
|
|
except Exception as e:
|
|
return _error(str(e), 500)
|
|
|
|
@http.route(PREFIX + '/plantillas/<int:plantilla_id>', type='http', auth='none',
|
|
methods=['GET', 'OPTIONS'], csrf=False)
|
|
@require_role('lectura')
|
|
@require_perm('formularios.ver')
|
|
def get_plantilla(self, plantilla_id, **kw):
|
|
plantilla = request.env['skeen.formulario.plantilla'].sudo().browse(plantilla_id)
|
|
if not plantilla.exists():
|
|
return _error('Plantilla no encontrada', 404)
|
|
return _json({'status': 'success', 'plantilla': plantilla_dict(plantilla)})
|
|
|
|
@http.route(PREFIX + '/plantillas', type='http', auth='none',
|
|
methods=['POST'], csrf=False)
|
|
@require_role('recepcion')
|
|
@require_perm('formularios.crear')
|
|
def crear_plantilla(self, **kw):
|
|
data = _body()
|
|
try:
|
|
campos = normalizar_campos(data.get('campos') or [])
|
|
vals = self._vals_plantilla(data, campos, crear=True)
|
|
plantilla = request.env['skeen.formulario.plantilla'].sudo().create(vals)
|
|
self._set_m2m(plantilla, data)
|
|
return _json({'status': 'success', 'plantilla': plantilla_dict(plantilla)}, 201)
|
|
except ValueError as e:
|
|
return _error(str(e), 400)
|
|
except Exception as e:
|
|
request.env.cr.rollback()
|
|
return _error(str(e), 500)
|
|
|
|
@http.route(PREFIX + '/plantillas/<int:plantilla_id>', type='http', auth='none',
|
|
methods=['PUT', 'OPTIONS'], csrf=False)
|
|
@require_role('recepcion')
|
|
@require_perm('formularios.editar')
|
|
def actualizar_plantilla(self, plantilla_id, **kw):
|
|
plantilla = request.env['skeen.formulario.plantilla'].sudo().browse(plantilla_id)
|
|
if not plantilla.exists():
|
|
return _error('Plantilla no encontrada', 404)
|
|
data = _body()
|
|
try:
|
|
campos = None
|
|
if 'campos' in data:
|
|
campos = normalizar_campos(data.get('campos') or [])
|
|
vals = self._vals_plantilla(data, campos, crear=False)
|
|
# Si la plantilla ya se usó, cualquier cambio de definición abre una
|
|
# versión nueva: las respuestas viejas conservan su snapshot.
|
|
definicion = {'campos', 'texto_legal', 'requiere_firma', 'requiere_foto',
|
|
'name', 'tipo', 'descripcion'}
|
|
if plantilla.tiene_respuestas() and (definicion & set(data.keys())):
|
|
vals['version'] = plantilla.version + 1
|
|
plantilla.write(vals)
|
|
self._set_m2m(plantilla, data)
|
|
return _json({'status': 'success', 'plantilla': plantilla_dict(plantilla)})
|
|
except ValueError as e:
|
|
return _error(str(e), 400)
|
|
except Exception as e:
|
|
request.env.cr.rollback()
|
|
return _error(str(e), 500)
|
|
|
|
@http.route(PREFIX + '/plantillas/<int:plantilla_id>', type='http', auth='none',
|
|
methods=['DELETE'], csrf=False)
|
|
@require_role('recepcion')
|
|
@require_perm('formularios.editar')
|
|
def borrar_plantilla(self, plantilla_id, **kw):
|
|
plantilla = request.env['skeen.formulario.plantilla'].sudo().browse(plantilla_id)
|
|
if not plantilla.exists():
|
|
return _error('Plantilla no encontrada', 404)
|
|
try:
|
|
if plantilla.tiene_respuestas():
|
|
# Con respuestas asociadas solo se desactiva: se conserva el
|
|
# histórico del expediente y deja de ofrecerse para envíos.
|
|
plantilla.write({'activo': False})
|
|
return _json({'status': 'success', 'desactivada': True,
|
|
'message': 'La plantilla tiene respuestas: se desactivó en lugar de borrarse.',
|
|
'plantilla': plantilla_dict(plantilla)})
|
|
plantilla.unlink()
|
|
return _json({'status': 'success', 'desactivada': False,
|
|
'message': 'Plantilla eliminada'})
|
|
except Exception as e:
|
|
request.env.cr.rollback()
|
|
return _error(str(e), 500)
|
|
|
|
def _vals_plantilla(self, data, campos, crear):
|
|
vals = {}
|
|
if 'name' in data or crear:
|
|
name = str(data.get('name') or '').strip()
|
|
if not name:
|
|
raise ValueError('El nombre de la plantilla es obligatorio')
|
|
vals['name'] = name[:200]
|
|
if 'tipo' in data or crear:
|
|
tipo = str(data.get('tipo') or 'consentimiento').strip()
|
|
if tipo not in dict(request.env['skeen.formulario.plantilla']._fields['tipo'].selection):
|
|
raise ValueError('Tipo de plantilla no válido')
|
|
vals['tipo'] = tipo
|
|
if 'descripcion' in data:
|
|
vals['descripcion'] = str(data.get('descripcion') or '')[:2000]
|
|
if campos is not None:
|
|
vals['campos'] = json.dumps(campos, ensure_ascii=False)
|
|
if 'texto_legal' in data:
|
|
vals['texto_legal'] = str(data.get('texto_legal') or '')
|
|
for bandera in ('requiere_firma', 'requiere_foto', 'activo'):
|
|
if bandera in data:
|
|
vals[bandera] = bool(data[bandera])
|
|
if 'duracion_estimada_min' in data:
|
|
try:
|
|
vals['duracion_estimada_min'] = max(0, int(data['duracion_estimada_min']))
|
|
except (TypeError, ValueError):
|
|
raise ValueError('La duración estimada debe ser un número entero')
|
|
if 'secuencia' in data:
|
|
try:
|
|
vals['secuencia'] = int(data['secuencia'])
|
|
except (TypeError, ValueError):
|
|
raise ValueError('La secuencia debe ser un número entero')
|
|
return vals
|
|
|
|
def _set_m2m(self, plantilla, data):
|
|
Sucursal = request.env['skeen.sucursal'].sudo()
|
|
Servicio = request.env['skeen.servicio'].sudo()
|
|
if 'sucursal_ids' in data:
|
|
ids = [int(i) for i in (data.get('sucursal_ids') or []) if str(i).isdigit()]
|
|
plantilla.sucursales = [(6, 0, Sucursal.search([('id', 'in', ids)]).ids)]
|
|
if 'servicio_ids' in data:
|
|
ids = [int(i) for i in (data.get('servicio_ids') or []) if str(i).isdigit()]
|
|
plantilla.servicios = [(6, 0, Servicio.search([('id', 'in', ids)]).ids)]
|
|
|
|
# ------------------------------------------------------------------
|
|
# Respuestas
|
|
# ------------------------------------------------------------------
|
|
@http.route(PREFIX + '/respuestas', type='http', auth='none',
|
|
methods=['GET', 'OPTIONS'], csrf=False)
|
|
@require_role('lectura')
|
|
@require_perm('formularios.ver')
|
|
def listar_respuestas(self, **kw):
|
|
try:
|
|
Respuesta = request.env['skeen.formulario.respuesta'].sudo()
|
|
dominio = []
|
|
if kw.get('partner_id'):
|
|
dominio.append(('partner_id', '=', int(kw['partner_id'])))
|
|
if kw.get('plantilla_id'):
|
|
dominio.append(('plantilla_id', '=', int(kw['plantilla_id'])))
|
|
if kw.get('estado'):
|
|
dominio.append(('estado', '=', kw['estado']))
|
|
if kw.get('start'):
|
|
dominio.append(('create_date', '>=', '%s 00:00:00' % kw['start']))
|
|
if kw.get('end'):
|
|
dominio.append(('create_date', '<=', '%s 23:59:59' % kw['end']))
|
|
try:
|
|
page = max(1, int(kw.get('page') or 1))
|
|
except (TypeError, ValueError):
|
|
page = 1
|
|
try:
|
|
page_size = min(200, max(1, int(kw.get('page_size') or 50)))
|
|
except (TypeError, ValueError):
|
|
page_size = 50
|
|
total = Respuesta.search_count(dominio)
|
|
offset = (page - 1) * page_size
|
|
registros = Respuesta.search(dominio, order='create_date desc, id desc',
|
|
limit=page_size, offset=offset)
|
|
resumen = {}
|
|
for estado, _label in Respuesta._fields['estado'].selection:
|
|
resumen[estado] = Respuesta.search_count(dominio + [('estado', '=', estado)])
|
|
return _json({
|
|
'status': 'success',
|
|
'respuestas': [respuesta_dict(r) for r in registros],
|
|
'total': total,
|
|
'page': page,
|
|
'page_size': page_size,
|
|
'total_pages': (total + page_size - 1) // page_size if page_size else 0,
|
|
'resumen': resumen,
|
|
})
|
|
except Exception as e:
|
|
return _error(str(e), 500)
|
|
|
|
@http.route(PREFIX + '/respuestas/<int:respuesta_id>', type='http', auth='none',
|
|
methods=['GET', 'OPTIONS'], csrf=False)
|
|
@require_role('lectura')
|
|
@require_perm('formularios.ver')
|
|
def get_respuesta(self, respuesta_id, **kw):
|
|
respuesta = request.env['skeen.formulario.respuesta'].sudo().browse(respuesta_id)
|
|
if not respuesta.exists():
|
|
return _error('Respuesta no encontrada', 404)
|
|
return _json({'status': 'success', 'respuesta': respuesta_dict(respuesta, detalle=True)})
|
|
|
|
@http.route(PREFIX + '/fotos', type='http', auth='none',
|
|
methods=['GET', 'OPTIONS'], csrf=False)
|
|
@require_role('lectura')
|
|
@require_perm('formularios.ver')
|
|
def listar_fotos(self, **kw):
|
|
"""Fotos y archivos agrupables por fecha para la galería del expediente."""
|
|
try:
|
|
dominio = []
|
|
if kw.get('partner_id'):
|
|
dominio.append(('partner_id', '=', int(kw['partner_id'])))
|
|
if kw.get('respuesta_id'):
|
|
dominio.append(('respuesta_id', '=', int(kw['respuesta_id'])))
|
|
adjuntos = request.env['skeen.formulario.adjunto'].sudo().search(
|
|
dominio, order='create_date desc, id desc', limit=200)
|
|
return _json({
|
|
'status': 'success',
|
|
'total': len(adjuntos),
|
|
'fotos': [{
|
|
**adjunto_dict(a),
|
|
'respuesta_id': a.respuesta_id.id,
|
|
'respuesta_name': a.respuesta_id.name,
|
|
'plantilla': a.respuesta_id.plantilla_id.name,
|
|
'cita_id': a.cita_id.id if a.cita_id else None,
|
|
} for a in adjuntos],
|
|
})
|
|
except Exception as e:
|
|
return _error(str(e), 500)
|
|
|
|
@http.route(PREFIX + '/enviar', type='http', auth='none',
|
|
methods=['POST', 'OPTIONS'], csrf=False)
|
|
@require_role('recepcion')
|
|
@require_perm('formularios.enviar')
|
|
def enviar_formulario(self, **kw):
|
|
data = _body()
|
|
usuario = getattr(request, 'frontend_user', None)
|
|
try:
|
|
Plantilla = request.env['skeen.formulario.plantilla'].sudo()
|
|
plantilla = Plantilla.browse(int(data.get('plantilla_id') or 0))
|
|
if not plantilla.exists() or not plantilla.activo:
|
|
return _error('La plantilla no existe o está desactivada', 404)
|
|
partner = request.env['res.partner'].sudo().browse(int(data.get('partner_id') or 0))
|
|
if not partner.exists() or not partner.is_patient:
|
|
return _error('El paciente no existe', 404)
|
|
canal = str(data.get('canal') or 'enlace').strip().lower()
|
|
if canal not in ('whatsapp', 'email', 'enlace'):
|
|
return _error('Canal no válido (whatsapp, email o enlace)')
|
|
cita = None
|
|
if data.get('cita_id'):
|
|
cita = request.env['skeen.cita'].sudo().browse(int(data['cita_id']))
|
|
if not cita.exists() or cita.partner_id.id != partner.id:
|
|
cita = None
|
|
|
|
Respuesta = request.env['skeen.formulario.respuesta'].sudo()
|
|
respuesta = Respuesta.create({
|
|
'plantilla_id': plantilla.id,
|
|
'plantilla_version': plantilla.version,
|
|
'partner_id': partner.id,
|
|
'cita_id': cita.id if cita else False,
|
|
'estado': 'enviado',
|
|
'enviado_at': fields.Datetime.now(),
|
|
'canal_envio': canal,
|
|
'respuestas': '{}',
|
|
})
|
|
token = firmar_token(request.env, respuesta)
|
|
respuesta.write({'token': token})
|
|
enlace = _enlace_publico(request.env, token)
|
|
|
|
notificado = self._encolar_envio(respuesta, canal, enlace)
|
|
return _json({
|
|
'status': 'success',
|
|
'respuesta': respuesta_dict(respuesta),
|
|
'token': token,
|
|
'enlace': enlace,
|
|
'enviado_por': usuario.name if usuario else '',
|
|
'notificado': notificado,
|
|
}, 201)
|
|
except Exception as e:
|
|
request.env.cr.rollback()
|
|
return _error(str(e), 500)
|
|
|
|
def _encolar_envio(self, respuesta, canal, enlace):
|
|
"""Encola el aviso por el motor de notificaciones (tipo 'formulario').
|
|
|
|
Con `notificaciones.activo=False` (o modo prueba) no se envía nada real;
|
|
el canal 'enlace' no depende del motor y siempre devuelve el enlace.
|
|
"""
|
|
if canal not in ('whatsapp', 'email'):
|
|
return False
|
|
try:
|
|
service = request.env['skeen.notificacion.service'].sudo()
|
|
creadas = service.encolar(
|
|
partner=respuesta.partner_id,
|
|
cita=respuesta.cita_id or None,
|
|
tipo='formulario',
|
|
canal=canal,
|
|
clave_unica='formulario:r%s' % respuesta.id,
|
|
contexto_extra={'enlace': enlace, 'formulario': respuesta.plantilla_id.name},
|
|
)
|
|
cfg = get_notificaciones(request.env)
|
|
for notif in creadas:
|
|
service._intentar(notif, cfg)
|
|
return bool(creadas)
|
|
except Exception:
|
|
logging.getLogger(__name__).exception(
|
|
'No se pudo encolar el formulario %s', respuesta.id)
|
|
return False
|
|
|
|
@http.route(PREFIX + '/respuestas/<int:respuesta_id>/llenar', type='http', auth='none',
|
|
methods=['POST', 'OPTIONS'], csrf=False)
|
|
@require_role('recepcion')
|
|
@require_perm('formularios.crear')
|
|
def llenar_respuesta(self, respuesta_id, **kw):
|
|
respuesta = request.env['skeen.formulario.respuesta'].sudo().browse(respuesta_id)
|
|
if not respuesta.exists():
|
|
return _error('Respuesta no encontrada', 404)
|
|
if respuesta.estado == 'llenado':
|
|
return _error('Esta respuesta ya fue llenada', 409)
|
|
data = _body()
|
|
try:
|
|
plantilla = respuesta.plantilla_id
|
|
campos = plantilla.campos_lista()
|
|
fotos_claves = [a.campo_clave for a in respuesta.foto_ids if a.campo_clave]
|
|
firma_b64 = None
|
|
if data.get('firma_imagen'):
|
|
firma_b64, _mime = parse_data_url(
|
|
data['firma_imagen'], MAX_FIRMA_BYTES, ('image/png', 'image/jpeg'))
|
|
tiene_firma = bool(firma_b64) or bool(respuesta.firma_imagen)
|
|
errores = validar_respuestas(
|
|
plantilla, campos, data.get('respuestas'), fotos_claves,
|
|
tiene_firma, plantilla.requiere_firma, plantilla.requiere_foto)
|
|
if errores:
|
|
return _error(' '.join(errores), 400)
|
|
|
|
limpio = limpiar_respuestas(campos, data.get('respuestas'))
|
|
vals = {
|
|
'respuestas': json.dumps(limpio, ensure_ascii=False),
|
|
'estado': 'llenado',
|
|
'llenado_at': fields.Datetime.now(),
|
|
'creado_por_empleado': True,
|
|
}
|
|
if firma_b64:
|
|
vals['firma_imagen'] = firma_b64
|
|
if 'firma_nombre' in data:
|
|
vals['firma_nombre'] = str(data.get('firma_nombre') or '')[:200]
|
|
respuesta.write(vals)
|
|
return _json({'status': 'success', 'respuesta': respuesta_dict(respuesta, detalle=True)})
|
|
except ValueError as e:
|
|
return _error(str(e), 400)
|
|
except Exception as e:
|
|
request.env.cr.rollback()
|
|
return _error(str(e), 500)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Adjuntos (fotos y archivos)
|
|
# ------------------------------------------------------------------
|
|
@http.route(PREFIX + '/adjuntar', type='http', auth='none',
|
|
methods=['POST', 'OPTIONS'], csrf=False)
|
|
@require_role('recepcion')
|
|
@require_perm('formularios.crear')
|
|
def adjuntar(self, **kw):
|
|
data = _body()
|
|
respuesta = request.env['skeen.formulario.respuesta'].sudo().browse(
|
|
int(data.get('respuesta_id') or 0))
|
|
if not respuesta.exists():
|
|
return _error('Respuesta no encontrada', 404)
|
|
try:
|
|
if len(respuesta.foto_ids) >= 10:
|
|
return _error('Ya hay 10 archivos adjuntos en esta respuesta')
|
|
b64, mimetype = parse_data_url(
|
|
data.get('file') or data.get('file_b64'), MAX_FOTO_BYTES, MIMES_FOTO)
|
|
campo_clave = str(data.get('campo_clave') or '')[:60]
|
|
kind = str(data.get('kind') or '').strip() or _kind_de_campo(respuesta.plantilla_id, campo_clave)
|
|
if kind not in ('antes', 'despues', 'otro'):
|
|
kind = 'otro'
|
|
adj = request.env['skeen.formulario.adjunto'].sudo().create({
|
|
'respuesta_id': respuesta.id,
|
|
'campo_clave': campo_clave,
|
|
'kind': kind,
|
|
'name': str(data.get('name') or 'adjunto')[:200],
|
|
'file': b64,
|
|
'mimetype': mimetype or str(data.get('mimetype') or ''),
|
|
'notes': str(data.get('notes') or '')[:200],
|
|
})
|
|
return _json({'status': 'success', 'adjunto': adjunto_dict(adj)}, 201)
|
|
except ValueError as e:
|
|
return _error(str(e), 400)
|
|
except Exception as e:
|
|
request.env.cr.rollback()
|
|
return _error(str(e), 500)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Binarios (aceptan token de sesión por header o query string)
|
|
# ------------------------------------------------------------------
|
|
def _autorizado_binario(self, kw):
|
|
if current_user(request.env):
|
|
return True
|
|
payload = verify_session_token(request.env, kw.get('token') or '')
|
|
if not payload:
|
|
return False
|
|
user = request.env['skeen.frontend.user'].sudo().browse(payload.get('uid'))
|
|
return bool(user.exists() and user.active)
|
|
|
|
@http.route(PREFIX + '/respuestas/<int:respuesta_id>/firma', type='http', auth='none',
|
|
methods=['GET'], csrf=False)
|
|
def get_firma(self, respuesta_id, **kw):
|
|
if not self._autorizado_binario(kw):
|
|
return _error('No autenticado', 401)
|
|
respuesta = request.env['skeen.formulario.respuesta'].sudo().browse(respuesta_id)
|
|
if not respuesta.exists() or not respuesta.firma_imagen:
|
|
return _error('Firma no encontrada', 404)
|
|
return Response(
|
|
base64.b64decode(respuesta.firma_imagen), status=200, mimetype='image/png',
|
|
headers={'Content-Disposition': 'inline; filename="firma.png"',
|
|
'Access-Control-Allow-Origin': '*'},
|
|
)
|
|
|
|
@http.route(PREFIX + '/adjuntos/<int:adjunto_id>/file', type='http', auth='none',
|
|
methods=['GET'], csrf=False)
|
|
def get_adjunto_file(self, adjunto_id, **kw):
|
|
if not self._autorizado_binario(kw):
|
|
return _error('No autenticado', 401)
|
|
adj = request.env['skeen.formulario.adjunto'].sudo().browse(adjunto_id)
|
|
if not adj.exists() or not adj.file:
|
|
return _error('Adjunto no encontrado', 404)
|
|
return Response(
|
|
base64.b64decode(adj.file), status=200,
|
|
mimetype=adj.mimetype or 'application/octet-stream',
|
|
headers={
|
|
'Content-Disposition': 'inline; filename="%s"' % (adj.name or 'adjunto'),
|
|
'Access-Control-Allow-Origin': '*',
|
|
},
|
|
)
|