Integraciones

El Sistema A3 se integra con varios servicios externos para extender su funcionalidad. Esta página documenta todas las integraciones implementadas.

Integraciones Principales

1. SAP HANA API

Tecnología: ASP.NET Web API 2 (C#) URL Base: http://atercerosb1.ddns.net Protocolo: HTTP/REST Formato: JSON

Propósito

La API de SAP HANA es el sistema legado que almacena datos maestros de clientes, viviendas, anexos y provisiones. El Sistema A3 se sincroniza bidireccionalmentecon SAP para mantener consistencia de datos.

Endpoints Utilizados

Clientes

POST /api/clientes/agregar
GET /api/clientes/{id}
PUT /api/clientes/actualizar

Viviendas

GET /api/viviendas
GET /api/viviendas/{clave}

Anexos

POST /api/anexos/subir
GET /api/anexos/{id}
DELETE /api/anexos/{id}

Apartados

POST /api/apartados/agregar
GET /api/apartados/{id}
PUT /api/apartados/actualizar

Implementación

# services/sap_service.py
import requests
from django.conf import settings

class SAPService:
    """Servicio para interactuar con SAP HANA API"""

    BASE_URL = settings.SAP_API_URL
    AUTH = (settings.SAP_API_USERNAME, settings.SAP_API_PASSWORD)

    @classmethod
    def crear_cliente(cls, cliente_data):
        """Crea un cliente en SAP"""
        url = f"{cls.BASE_URL}/api/clientes/agregar"

        payload = {
            "RFC": cliente_data['rfc'],
            "Nombre": cliente_data['nombre'],
            "ApellidoPaterno": cliente_data['apellido_paterno'],
            "ApellidoMaterno": cliente_data['apellido_materno'],
            "TipoPersona": cliente_data['tipo_persona'],
            "Plaza": cliente_data['plaza'],
            # ... más campos
        }

        response = requests.post(
            url,
            json=payload,
            auth=cls.AUTH,
            timeout=30
        )

        response.raise_for_status()
        return response.json()

    @classmethod
    def subir_anexo(cls, archivo, tipo, id_referencia):
        """Sube un archivo a SAP"""
        url = f"{cls.BASE_URL}/api/anexos/subir"

        files = {'file': archivo}
        data = {
            'Tipo': tipo,
            'IdReferencia': id_referencia
        }

        response = requests.post(
            url,
            files=files,
            data=data,
            auth=cls.AUTH,
            timeout=60
        )

        response.raise_for_status()
        return response.json()

Manejo de Errores

from requests.exceptions import RequestException, Timeout

try:
    result = SAPService.crear_cliente(data)
except Timeout:
    # Log y notificar al usuario
    logger.error("Timeout al conectar con SAP")
    raise ValueError("El servicio SAP no está disponible")
except RequestException as e:
    # Log del error
    logger.error(f"Error al conectar con SAP: {e}")
    raise ValueError("Error al sincronizar con SAP")

2. AWS S3

Servicio: Amazon Simple Storage Service SDK: boto3 Bucket: sistema-a3-media Región: us-east-1

Propósito

AWS S3 almacena todos los archivos media del sistema: - Anexos de casas - Documentos de clientes - Comprobaciones de gastos - Reportes generados - Imágenes de perfil

Configuración

# settings.py
if USE_S3:
    AWS_ACCESS_KEY_ID = env('AWS_ACCESS_KEY_ID')
    AWS_SECRET_ACCESS_KEY = env('AWS_SECRET_ACCESS_KEY')
    AWS_STORAGE_BUCKET_NAME = env('AWS_STORAGE_BUCKET_NAME')
    AWS_S3_REGION_NAME = 'us-east-1'
    AWS_S3_CUSTOM_DOMAIN = f'{AWS_STORAGE_BUCKET_NAME}.s3.amazonaws.com'
    AWS_S3_OBJECT_PARAMETERS = {
        'CacheControl': 'max-age=86400',
    }
    AWS_DEFAULT_ACL = 'private'
    DEFAULT_FILE_STORAGE = 'storages.backends.s3boto3.S3Boto3Storage'

Uso en Modelos

from django.db import models

class AnexoCasa(models.Model):
    casa = models.ForeignKey('Casa', on_delete=models.CASCADE)
    archivo = models.FileField(upload_to='anexos/casas/%Y/%m/')
    tipo_anexo = models.CharField(max_length=50)

    def get_url(self):
        return self.archivo.url  # URL firmada de S3

Upload Directo

import boto3
from django.conf import settings

def subir_a_s3(archivo, ruta):
    """Sube un archivo directamente a S3"""
    s3_client = boto3.client(
        's3',
        aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
        aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
        region_name=settings.AWS_S3_REGION_NAME
    )

    s3_client.upload_fileobj(
        archivo,
        settings.AWS_STORAGE_BUCKET_NAME,
        ruta,
        ExtraArgs={
            'ContentType': archivo.content_type,
            'ACL': 'private'
        }
    )

    return f"https://{settings.AWS_S3_CUSTOM_DOMAIN}/{ruta}"

3. Web Push Notifications

Protocolo: Web Push Protocol (RFC 8030) Autenticación: VAPID (RFC 8292) Librería: pywebpush

Propósito

Enviar notificaciones push a navegadores y dispositivos móviles (iOS 16.4+) sin necesidad de app nativa.

Configuración

# settings.py
WEBPUSH_SETTINGS = {
    "VAPID_PUBLIC_KEY": env('VAPID_PUBLIC_KEY'),
    "VAPID_PRIVATE_KEY": env('VAPID_PRIVATE_KEY'),
    "VAPID_ADMIN_EMAIL": env('VAPID_ADMIN_EMAIL')
}

Suscripción (Frontend)

// static/js/push.js
async function subscribeToPush() {
    const registration = await navigator.serviceWorker.ready;

    const subscription = await registration.pushManager.subscribe({
        userVisibleOnly: true,
        applicationServerKey: urlBase64ToUint8Array(VAPID_PUBLIC_KEY)
    });

    // Enviar subscription al backend
    await fetch('/api/push/subscribe/', {
        method: 'POST',
        headers: {
            'Content-Type': 'application/json',
            'X-CSRFToken': getCookie('csrftoken')
        },
        body: JSON.stringify(subscription)
    });
}

Envío de Notificaciones (Backend)

# push/utils.py
from pywebpush import webpush, WebPushException
from django.conf import settings
import json

def enviar_push_notification(subscription_info, mensaje):
    """Envía una notificación push"""
    try:
        webpush(
            subscription_info=subscription_info,
            data=json.dumps({
                "title": mensaje['titulo'],
                "body": mensaje['cuerpo'],
                "icon": mensaje.get('icon', '/static/img/logo.png'),
                "url": mensaje.get('url', '/')
            }),
            vapid_private_key=settings.WEBPUSH_SETTINGS['VAPID_PRIVATE_KEY'],
            vapid_claims={
                "sub": f"mailto:{settings.WEBPUSH_SETTINGS['VAPID_ADMIN_EMAIL']}"
            }
        )
        return True
    except WebPushException as ex:
        # Manejar errores (ej: subscription expirada)
        if ex.response.status_code == 410:
            # Subscription ya no es válida, eliminarla
            pass
        return False

4. Django Notifications

Librería: django-notifications-hq Propósito: Notificaciones in-app

Modelos

# El modelo Notification viene de django-notifications-hq
from notifications.models import Notification

# Campos principales:
# - actor: Usuario que genera la notificación
# - recipient: Usuario que recibe
# - verb: Acción realizada
# - action_object: Objeto relacionado
# - target: Objetivo de la acción
# - description: Descripción adicional
# - unread: Boolean

Enviar Notificación

from notifications.signals import notify

# Notificar asignación de ticket
notify.send(
    sender=request.user,
    recipient=ticket.asignado,
    verb='te asignó un ticket',
    action_object=ticket,
    description=ticket.titulo,
    url=f'/tickets/{ticket.id}/'
)

Consultar Notificaciones

# Todas las notificaciones del usuario
notificaciones = request.user.notifications.all()

# Solo no leídas
no_leidas = request.user.notifications.unread()

# Marcar como leída
notificacion.mark_as_read()

# Marcar todas como leídas
request.user.notifications.mark_all_as_read()

Integraciones Futuras

Email (SMTP)

Configuración lista pero no implementada completamente:

# settings.py
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = env('EMAIL_HOST', default='smtp.gmail.com')
EMAIL_PORT = env('EMAIL_PORT', default=587)
EMAIL_USE_TLS = True
EMAIL_HOST_USER = env('EMAIL_HOST_USER')
EMAIL_HOST_PASSWORD = env('EMAIL_HOST_PASSWORD')

Webhooks

Sistema para recibir eventos de servicios externos:

# api/webhooks.py
from rest_framework.decorators import api_view
from rest_framework.response import Response

@api_view(['POST'])
def webhook_sap(request):
    """Recibe eventos de SAP"""
    evento = request.data.get('evento')
    datos = request.data.get('datos')

    if evento == 'cliente_actualizado':
        # Sincronizar cliente
        actualizar_cliente_desde_sap(datos)

    return Response({'status': 'ok'})

APIs de Terceros

Potenciales integraciones:

• Google Maps API: Geocoding y mapas (ya usado parcialmente)
• WhatsApp Business API: Notificaciones por WhatsApp
• Twilio: SMS y llamadas
• SendGrid: Email transaccional
• Stripe/PayPal: Pagos en línea

Monitoreo de Integraciones

Health Checks

# api/health.py
from rest_framework.decorators import api_view
from rest_framework.response import Response
from services.sap_service import SAPService
import boto3

@api_view(['GET'])
def health_check(request):
    """Verifica estado de integraciones"""
    status = {}

    # SAP API
    try:
        SAPService.ping()
        status['sap'] = 'up'
    except:
        status['sap'] = 'down'

    # AWS S3
    try:
        s3 = boto3.client('s3')
        s3.head_bucket(Bucket=settings.AWS_STORAGE_BUCKET_NAME)
        status['s3'] = 'up'
    except:
        status['s3'] = 'down'

    return Response(status)

Logging

import logging

logger = logging.getLogger('integraciones')

# Al hacer llamadas externas
logger.info(f"Llamando a SAP API: {url}")
try:
    response = requests.post(url, data=data)
    logger.info(f"SAP Response: {response.status_code}")
except Exception as e:
    logger.error(f"Error en SAP: {e}")

Seguridad en Integraciones

Credenciales

✅ Nunca en código: Todas en variables de entorno ✅ Rotación: Cambiar periódicamente ✅ Mínimos privilegios: Solo permisos necesarios

Timeouts

# Siempre definir timeouts
requests.post(url, data=data, timeout=30)  # 30 segundos

Retry Logic

from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

def requests_retry_session(
    retries=3,
    backoff_factor=0.3,
    status_forcelist=(500, 502, 504),
):
    session = requests.Session()
    retry = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=backoff_factor,
        status_forcelist=status_forcelist,
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    return session

Rate Limiting

import time
from functools import wraps

def rate_limit(max_per_minute=60):
    """Decorator para limitar rate de requests"""
    min_interval = 60.0 / max_per_minute

    def decorator(func):
        last_called = [0.0]

        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left_to_wait = min_interval - elapsed

            if left_to_wait > 0:
                time.sleep(left_to_wait)

            ret = func(*args, **kwargs)
            last_called[0] = time.time()
            return ret

        return wrapper
    return decorator

@rate_limit(max_per_minute=30)
def llamar_sap_api(data):
    return SAPService.crear_cliente(data)

Documentación de APIs Externas

• SAP API: Ver api.md para documentación completa
• AWS S3: Documentación oficial
• Web Push: MDN Web Docs

---

Las integraciones son cruciales para el funcionamiento del Sistema A3, conectándolo con el ecosistema tecnológico de A Terceros.