# -*- coding: utf-8 -*- """CRUD de webhooks salientes y log de entregas (Configuración -> Webhooks).""" import json import secrets from odoo import http from odoo.http import request, Response from odoo.addons.skeen_whatsapp.controllers.auth import require_role, require_perm from odoo.addons.skeen_plataforma.models.webhook import EVENTOS, ESTADOS BASE = '/skeen/frontend/v1' def _json(data, status=200): return Response( json.dumps(data, default=str), status=status, mimetype='application/json', headers={ 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, OPTIONS', 'Access-Control-Allow-Headers': 'Content-Type, Authorization', }, ) def _body(): try: raw = request.httprequest.get_data(as_text=True) return json.loads(raw) if raw else {} except Exception: return {} def _login(): return getattr(request.frontend_user, 'login', '') if hasattr(request, 'frontend_user') else '' def _eventos_list(data): if isinstance(data.get('eventos'), list): return [str(e).strip() for e in data['eventos'] if str(e).strip()] if isinstance(data.get('eventos'), str): return [e.strip() for e in data['eventos'].split(',') if e.strip()] return [] class SkeenWebhooksController(http.Controller): @http.route(BASE + '/webhooks', type='http', auth='none', methods=['GET', 'OPTIONS'], csrf=False) @require_role('gerencia') @require_perm('webhooks.editar') def list_webhooks(self, **kw): try: service = request.env['skeen.webhook.service'].sudo() hooks = request.env['skeen.webhook'].sudo().search([]) catalogo = [{'clave': c, 'label': l, 'ejemplo': service.payload_ejemplo(c)} for c, l in EVENTOS] return _json({ 'status': 'success', 'webhooks': [w.to_public_dict() for w in hooks], 'eventos': catalogo, 'estados': [{'clave': c, 'label': l} for c, l in ESTADOS], 'formato_firma': 'X-Skeen-Signature: sha256=', }) except Exception as e: return _json({'status': 'error', 'message': str(e)}, 500) @http.route(BASE + '/webhooks', type='http', auth='none', methods=['POST', 'OPTIONS'], csrf=False) @require_role('gerencia') @require_perm('webhooks.editar') def create_webhook(self, **kw): try: data = _body() url = (data.get('url') or '').strip() if not (url.startswith('http://') or url.startswith('https://')): return _json({'status': 'error', 'message': 'La URL debe empezar por http:// o https://'}, 400) secreto = (data.get('secreto') or '').strip() or secrets.token_urlsafe(32) headers = data.get('headers') if isinstance(headers, dict): headers = json.dumps(headers, ensure_ascii=False) else: headers = (headers or '').strip() webhook = request.env['skeen.webhook'].sudo().create({ 'name': (data.get('name') or '').strip() or 'Webhook', 'url': url, 'eventos': ','.join(_eventos_list(data)), 'activo': bool(data.get('activo', True)), 'secreto': secreto, 'headers': headers, 'notas': data.get('notas') or '', 'creado_por_login': _login(), }) return _json({'status': 'success', 'webhook': webhook.to_public_dict(), 'secreto': secreto, 'message': 'Webhook creado.'}) except Exception as e: request.env.cr.rollback() return _json({'status': 'error', 'message': str(e)}, 500) @http.route(BASE + '/webhooks/', type='http', auth='none', methods=['PUT', 'OPTIONS'], csrf=False) @require_role('gerencia') @require_perm('webhooks.editar') def update_webhook(self, webhook_id, **kw): try: webhook = request.env['skeen.webhook'].sudo().browse(webhook_id) if not webhook.exists(): return _json({'status': 'error', 'message': 'Webhook no encontrado'}, 404) data = _body() vals = {} if 'name' in data and (data.get('name') or '').strip(): vals['name'] = data['name'].strip() if 'url' in data and (data.get('url') or '').strip(): url = data['url'].strip() if not (url.startswith('http://') or url.startswith('https://')): return _json({'status': 'error', 'message': 'La URL debe empezar por http:// o https://'}, 400) vals['url'] = url if 'eventos' in data: vals['eventos'] = ','.join(_eventos_list(data)) if 'activo' in data: vals['activo'] = bool(data['activo']) if vals['activo']: vals['fallos_consecutivos'] = 0 if 'secreto' in data and (data.get('secreto') or '').strip(): vals['secreto'] = data['secreto'].strip() if 'headers' in data: headers = data['headers'] vals['headers'] = json.dumps(headers, ensure_ascii=False) \ if isinstance(headers, dict) else (headers or '') if 'notas' in data: vals['notas'] = data.get('notas') or '' if vals: webhook.write(vals) return _json({'status': 'success', 'webhook': webhook.to_public_dict()}) except Exception as e: request.env.cr.rollback() return _json({'status': 'error', 'message': str(e)}, 500) @http.route(BASE + '/webhooks/', type='http', auth='none', methods=['DELETE'], csrf=False) @require_role('gerencia') @require_perm('webhooks.editar') def delete_webhook(self, webhook_id, **kw): try: webhook = request.env['skeen.webhook'].sudo().browse(webhook_id) if not webhook.exists(): return _json({'status': 'error', 'message': 'Webhook no encontrado'}, 404) webhook.unlink() return _json({'status': 'success', 'message': 'Webhook eliminado.'}) except Exception as e: request.env.cr.rollback() return _json({'status': 'error', 'message': str(e)}, 500) @http.route(BASE + '/webhooks//eventos', type='http', auth='none', methods=['GET', 'OPTIONS'], csrf=False) @require_role('gerencia') @require_perm('webhooks.editar') def list_webhook_eventos(self, webhook_id, **kw): try: webhook = request.env['skeen.webhook'].sudo().browse(webhook_id) if not webhook.exists(): return _json({'status': 'error', 'message': 'Webhook no encontrado'}, 404) domain = [('webhook_id', '=', webhook.id)] if kw.get('estado'): domain.append(('estado', '=', kw['estado'])) if kw.get('evento'): domain.append(('evento', '=', kw['evento'])) try: limite = min(max(int(kw.get('limit') or 100), 1), 500) except (TypeError, ValueError): limite = 100 Evento = request.env['skeen.webhook.evento'].sudo() registros = Evento.search(domain, order='create_date desc, id desc', limit=limite) resumen = {c: 0 for c, _l in ESTADOS} for grupo in Evento.read_group(domain, ['estado'], ['estado'], lazy=False): if grupo.get('estado'): resumen[grupo['estado']] = grupo.get('__count', 0) return _json({'status': 'success', 'eventos': [e.to_public_dict() for e in registros], 'total': Evento.search_count(domain), 'resumen': resumen}) except Exception as e: return _json({'status': 'error', 'message': str(e)}, 500) @http.route(BASE + '/webhooks//probar', type='http', auth='none', methods=['POST', 'OPTIONS'], csrf=False) @require_role('gerencia') @require_perm('webhooks.editar') def probar_webhook(self, webhook_id, **kw): try: webhook = request.env['skeen.webhook'].sudo().browse(webhook_id) if not webhook.exists(): return _json({'status': 'error', 'message': 'Webhook no encontrado'}, 404) ok, mensaje, evento = request.env['skeen.webhook.service'].sudo().probar( webhook, login=_login()) return _json({'status': 'success' if ok else 'error', 'ok': ok, 'message': mensaje, 'evento': evento.to_public_dict() if evento else None}, 200 if ok else 502) except Exception as e: request.env.cr.rollback() return _json({'status': 'error', 'message': str(e)}, 500) @http.route(BASE + '/webhook_eventos//reintentar', type='http', auth='none', methods=['POST', 'OPTIONS'], csrf=False) @require_role('gerencia') @require_perm('webhooks.editar') def reintentar_evento(self, evento_id, **kw): try: evento = request.env['skeen.webhook.evento'].sudo().browse(evento_id) if not evento.exists(): return _json({'status': 'error', 'message': 'Evento no encontrado'}, 404) estado = request.env['skeen.webhook.service'].sudo().reintentar(evento) return _json({'status': 'success', 'estado': estado, 'evento': evento.to_public_dict()}) except Exception as e: request.env.cr.rollback() return _json({'status': 'error', 'message': str(e)}, 500)