342 lines
13 KiB
Python
342 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Utilidades compartidas por los controladores de formularios.
|
|
|
|
Aquí vive lo que no debe duplicarse entre la API autenticada y el portal
|
|
público: firma/verificación del token HMAC, validación de las respuestas contra
|
|
la plantilla, lectura acotada de imágenes en data URL y los serializadores.
|
|
"""
|
|
|
|
import base64
|
|
import binascii
|
|
import hashlib
|
|
import hmac
|
|
import io
|
|
import json
|
|
import time
|
|
|
|
from PIL import Image
|
|
|
|
from odoo.addons.skeen_whatsapp.controllers.auth import _get_secret
|
|
|
|
# El token de un formulario viaja en un enlace que el paciente puede abrir
|
|
# días después: mismo TTL que el portal de reservas (180 días).
|
|
FORM_TOKEN_TTL = 60 * 60 * 24 * 180
|
|
FORM_TOKEN_KIND = 'formulario'
|
|
|
|
# Límites de subida. Las fotos de celular pesan; se acotan por foto y en número.
|
|
MAX_FOTO_BYTES = 5 * 1024 * 1024
|
|
MAX_FIRMA_BYTES = 3 * 1024 * 1024
|
|
MAX_FOTOS = 10
|
|
# Solo formatos que el navegador produce de forma habitual y que PIL sabe leer
|
|
# (así una imagen dañada se rechaza con 400 y no revienta el guardado).
|
|
MIMES_FOTO = ('image/jpeg', 'image/png', 'image/webp')
|
|
|
|
MAX_TEXTO_RESP = 5000
|
|
|
|
|
|
def _b64url_encode(raw):
|
|
return base64.urlsafe_b64encode(raw).rstrip(b'=').decode('ascii')
|
|
|
|
|
|
def _b64url_decode(s):
|
|
return base64.urlsafe_b64decode(s + '=' * (-len(s) % 4))
|
|
|
|
|
|
def firmar_token(env, respuesta):
|
|
"""Token HMAC de una respuesta (mismo formato que el portal de reservas)."""
|
|
payload = {'rid': respuesta.id, 'kind': FORM_TOKEN_KIND,
|
|
'exp': int(time.time()) + FORM_TOKEN_TTL}
|
|
body = _b64url_encode(json.dumps(payload, separators=(',', ':')).encode('utf-8'))
|
|
sig = hmac.new(_get_secret(env).encode('utf-8'),
|
|
body.encode('ascii'), hashlib.sha256).hexdigest()
|
|
return '%s.%s' % (body, sig)
|
|
|
|
|
|
def verificar_token(env, token):
|
|
"""Payload del token o None si es inválido, de otro tipo o expiró."""
|
|
if not token or '.' not in str(token):
|
|
return None
|
|
try:
|
|
body, sig = str(token).rsplit('.', 1)
|
|
expected = hmac.new(_get_secret(env).encode('utf-8'),
|
|
body.encode('ascii'), hashlib.sha256).hexdigest()
|
|
if not hmac.compare_digest(sig, expected):
|
|
return None
|
|
payload = json.loads(_b64url_decode(body).decode('utf-8'))
|
|
except Exception:
|
|
return None
|
|
if payload.get('kind') != FORM_TOKEN_KIND:
|
|
return None
|
|
if int(payload.get('exp', 0)) < int(time.time()):
|
|
return None
|
|
return payload
|
|
|
|
|
|
def respuesta_de_token(env, token):
|
|
"""Respuesta vinculada al token, o recordset vacío."""
|
|
payload = verificar_token(env, token)
|
|
if not payload:
|
|
return env['skeen.formulario.respuesta'].sudo().browse()
|
|
respuesta = env['skeen.formulario.respuesta'].sudo().browse(int(payload.get('rid') or 0))
|
|
return respuesta if respuesta.exists() else env['skeen.formulario.respuesta'].sudo().browse()
|
|
|
|
|
|
def parse_data_url(valor, max_bytes, mimes_ok):
|
|
"""Extrae (base64_limpio, mimetype) de un data URL. Lanza ValueError.
|
|
|
|
Acepta también base64 plano (sin encabezado) asumiendo el primer mime
|
|
permitido. Valida el tamaño ya decodificado y que la imagen se pueda leer:
|
|
una imagen dañada se rechaza aquí con 400 en vez de reventar el guardado
|
|
(Odoo reprocesa las imágenes al escribir un Binary como adjunto).
|
|
"""
|
|
if not valor or not isinstance(valor, str):
|
|
raise ValueError('Imagen vacía')
|
|
raw = valor.strip()
|
|
mimetype = mimes_ok[0] if mimes_ok else 'application/octet-stream'
|
|
b64 = raw
|
|
if raw.startswith('data:'):
|
|
try:
|
|
cabecera, b64 = raw.split(',', 1)
|
|
except ValueError:
|
|
raise ValueError('Formato de imagen no válido')
|
|
if ';base64' not in cabecera:
|
|
raise ValueError('La imagen debe venir en base64')
|
|
tipo = cabecera[5:].split(';')[0].strip().lower()
|
|
if mimes_ok and tipo not in mimes_ok:
|
|
raise ValueError('Tipo de imagen no permitido: %s' % (tipo or 'desconocido'))
|
|
mimetype = tipo or mimetype
|
|
b64 = ''.join(b64.split())
|
|
try:
|
|
crudo = base64.b64decode(b64, validate=True)
|
|
except (binascii.Error, ValueError):
|
|
raise ValueError('La imagen no es base64 válido')
|
|
if not crudo:
|
|
raise ValueError('La imagen está vacía')
|
|
if len(crudo) > max_bytes:
|
|
raise ValueError('La imagen excede el tamaño máximo permitido')
|
|
_validar_imagen(crudo)
|
|
return b64, mimetype
|
|
|
|
|
|
def _validar_imagen(crudo):
|
|
"""Comprueba que los bytes son una imagen legible. Lanza ValueError si no."""
|
|
try:
|
|
with Image.open(io.BytesIO(crudo)) as img:
|
|
img.verify()
|
|
with Image.open(io.BytesIO(crudo)) as img:
|
|
img.load()
|
|
except Exception:
|
|
raise ValueError('La imagen no se pudo leer (archivo dañado o formato no soportado)')
|
|
|
|
|
|
def _texto(valor, limite=MAX_TEXTO_RESP):
|
|
return str(valor)[:limite]
|
|
|
|
|
|
def limpiar_respuestas(campos, respuestas):
|
|
"""Deja solo las claves conocidas de la plantilla, con su valor normalizado."""
|
|
respuestas = respuestas if isinstance(respuestas, dict) else {}
|
|
out = {}
|
|
for campo in campos:
|
|
clave = campo['clave']
|
|
if clave not in respuestas:
|
|
continue
|
|
valor = respuestas[clave]
|
|
tipo = campo['tipo']
|
|
if tipo in ('texto', 'texto_largo'):
|
|
out[clave] = _texto(valor)
|
|
elif tipo == 'numero':
|
|
try:
|
|
out[clave] = float(valor)
|
|
except (TypeError, ValueError):
|
|
out[clave] = None
|
|
elif tipo == 'fecha':
|
|
out[clave] = _texto(valor, 20)
|
|
elif tipo == 'seleccion':
|
|
out[clave] = _texto(valor, 300)
|
|
elif tipo == 'multi_seleccion':
|
|
out[clave] = [_texto(v, 300) for v in valor] if isinstance(valor, list) else []
|
|
elif tipo in ('checkbox', 'consentimiento'):
|
|
out[clave] = bool(valor)
|
|
elif tipo == 'escala':
|
|
try:
|
|
out[clave] = int(valor)
|
|
except (TypeError, ValueError):
|
|
out[clave] = None
|
|
elif tipo == 'nps':
|
|
try:
|
|
n = int(valor)
|
|
out[clave] = n if 0 <= n <= 10 else None
|
|
except (TypeError, ValueError):
|
|
out[clave] = None
|
|
elif tipo in ('firma', 'foto', 'archivo'):
|
|
out[clave] = bool(valor)
|
|
else:
|
|
out[clave] = _texto(valor)
|
|
return out
|
|
|
|
|
|
def validar_respuestas(plantilla, campos, respuestas, fotos_claves,
|
|
tiene_firma, requiere_firma, requiere_foto):
|
|
"""Errores de validación de un llenado (lista de mensajes claros)."""
|
|
errores = []
|
|
respuestas = respuestas if isinstance(respuestas, dict) else {}
|
|
fotos_claves = set(fotos_claves or [])
|
|
for campo in campos:
|
|
clave = campo['clave']
|
|
etiqueta = campo.get('etiqueta') or clave
|
|
tipo = campo['tipo']
|
|
obligatorio = bool(campo.get('obligatorio'))
|
|
valor = respuestas.get(clave)
|
|
|
|
if tipo == 'consentimiento':
|
|
if obligatorio and not valor:
|
|
errores.append('Debes aceptar: %s' % etiqueta)
|
|
continue
|
|
if tipo == 'checkbox':
|
|
continue
|
|
if tipo == 'firma':
|
|
if obligatorio and not tiene_firma:
|
|
errores.append('Falta la firma: %s' % etiqueta)
|
|
continue
|
|
if tipo in ('foto', 'archivo'):
|
|
if obligatorio and clave not in fotos_claves:
|
|
errores.append('Falta adjuntar: %s' % etiqueta)
|
|
continue
|
|
if tipo == 'escala':
|
|
if valor in (None, ''):
|
|
if obligatorio:
|
|
errores.append('Falta responder: %s' % etiqueta)
|
|
else:
|
|
try:
|
|
n = int(valor)
|
|
if not (1 <= n <= 10):
|
|
errores.append('%s debe estar entre 1 y 10' % etiqueta)
|
|
except (TypeError, ValueError):
|
|
errores.append('%s debe ser un número del 1 al 10' % etiqueta)
|
|
continue
|
|
if tipo == 'nps':
|
|
if valor in (None, ''):
|
|
if obligatorio:
|
|
errores.append('Falta responder: %s' % etiqueta)
|
|
else:
|
|
try:
|
|
n = int(valor)
|
|
if not (0 <= n <= 10):
|
|
errores.append('%s debe estar entre 0 y 10' % etiqueta)
|
|
except (TypeError, ValueError):
|
|
errores.append('%s debe ser un número del 0 al 10' % etiqueta)
|
|
continue
|
|
if tipo == 'numero':
|
|
if valor in (None, ''):
|
|
if obligatorio:
|
|
errores.append('Falta responder: %s' % etiqueta)
|
|
else:
|
|
try:
|
|
float(valor)
|
|
except (TypeError, ValueError):
|
|
errores.append('%s debe ser un número' % etiqueta)
|
|
continue
|
|
if tipo == 'multi_seleccion':
|
|
opciones = campo.get('opciones') or []
|
|
if valor in (None, '', []):
|
|
if obligatorio:
|
|
errores.append('Selecciona al menos una opción en: %s' % etiqueta)
|
|
elif not isinstance(valor, list):
|
|
errores.append('%s debe ser una lista de opciones' % etiqueta)
|
|
else:
|
|
invalidas = [v for v in valor if v not in opciones]
|
|
if invalidas:
|
|
errores.append('%s tiene opciones no válidas' % etiqueta)
|
|
continue
|
|
if tipo == 'seleccion':
|
|
opciones = campo.get('opciones') or []
|
|
if valor in (None, ''):
|
|
if obligatorio:
|
|
errores.append('Falta elegir: %s' % etiqueta)
|
|
elif valor not in opciones:
|
|
errores.append('%s tiene una opción no válida' % etiqueta)
|
|
continue
|
|
# texto, texto_largo, fecha
|
|
if obligatorio and (valor is None or str(valor).strip() == ''):
|
|
errores.append('Falta responder: %s' % etiqueta)
|
|
|
|
if requiere_firma and not tiene_firma:
|
|
errores.append('Este formulario requiere firma')
|
|
if requiere_foto and not fotos_claves:
|
|
errores.append('Este formulario requiere al menos una foto')
|
|
return errores
|
|
|
|
|
|
def plantilla_dict(plantilla, public=False):
|
|
data = {
|
|
'id': plantilla.id,
|
|
'name': plantilla.name,
|
|
'tipo': plantilla.tipo,
|
|
'tipo_label': dict(plantilla._fields['tipo'].selection).get(plantilla.tipo),
|
|
'activo': plantilla.activo,
|
|
'version': plantilla.version,
|
|
'descripcion': plantilla.descripcion or '',
|
|
'campos': plantilla.campos_lista(),
|
|
'texto_legal': plantilla.texto_legal or '',
|
|
'requiere_firma': plantilla.requiere_firma,
|
|
'requiere_foto': plantilla.requiere_foto,
|
|
'duracion_estimada_min': plantilla.duracion_estimada_min,
|
|
'secuencia': plantilla.secuencia,
|
|
'respuestas_count': plantilla.respuestas_count,
|
|
}
|
|
if not public:
|
|
data['sucursales'] = [{'id': s.id, 'name': s.name} for s in plantilla.sucursales]
|
|
data['servicios'] = [{'id': s.id, 'name': s.name} for s in plantilla.servicios]
|
|
return data
|
|
|
|
|
|
def adjunto_dict(adj):
|
|
return {
|
|
'id': adj.id,
|
|
'campo_clave': adj.campo_clave or '',
|
|
'kind': adj.kind or 'otro',
|
|
'kind_label': dict(adj._fields['kind'].selection).get(adj.kind),
|
|
'name': adj.name or '',
|
|
'mimetype': adj.mimetype or '',
|
|
'notes': adj.notes or '',
|
|
'create_date': adj.create_date.strftime('%Y-%m-%d %H:%M') if adj.create_date else None,
|
|
'url': '/skeen/frontend/v1/formularios/adjuntos/%s/file' % adj.id,
|
|
}
|
|
|
|
|
|
def respuesta_dict(respuesta, detalle=False):
|
|
data = {
|
|
'id': respuesta.id,
|
|
'name': respuesta.name,
|
|
'plantilla_id': respuesta.plantilla_id.id,
|
|
'plantilla': respuesta.plantilla_id.name,
|
|
'plantilla_tipo': respuesta.plantilla_id.tipo,
|
|
'plantilla_version': respuesta.plantilla_version,
|
|
'partner_id': respuesta.partner_id.id,
|
|
'patient': respuesta.partner_id.name,
|
|
'cita_id': respuesta.cita_id.id if respuesta.cita_id else None,
|
|
'estado': respuesta.estado,
|
|
'estado_label': dict(respuesta._fields['estado'].selection).get(respuesta.estado),
|
|
'canal_envio': respuesta.canal_envio,
|
|
'creado_por_empleado': respuesta.creado_por_empleado,
|
|
'tiene_firma': bool(respuesta.firma_imagen),
|
|
'firma_nombre': respuesta.firma_nombre or '',
|
|
'enviado_at': respuesta.enviado_at.strftime('%Y-%m-%d %H:%M') if respuesta.enviado_at else None,
|
|
'llenado_at': respuesta.llenado_at.strftime('%Y-%m-%d %H:%M') if respuesta.llenado_at else None,
|
|
'create_date': respuesta.create_date.strftime('%Y-%m-%d %H:%M') if respuesta.create_date else None,
|
|
'fotos_count': len(respuesta.foto_ids),
|
|
}
|
|
if detalle:
|
|
try:
|
|
data['respuestas'] = json.loads(respuesta.respuestas or '{}')
|
|
except (ValueError, TypeError):
|
|
data['respuestas'] = {}
|
|
data['campos'] = respuesta.plantilla_id.campos_lista()
|
|
data['texto_legal'] = respuesta.plantilla_id.texto_legal or ''
|
|
data['notas'] = respuesta.notas or ''
|
|
data['firma_url'] = ('/skeen/frontend/v1/formularios/respuestas/%s/firma' % respuesta.id
|
|
if respuesta.firma_imagen else None)
|
|
data['fotos'] = [adjunto_dict(a) for a in respuesta.foto_ids]
|
|
return data
|