Especificación Técnica: `log_ubicaciones_promotores`¶

Esquema: fuerza_ventas
Base de Datos: crm
Servicio: svc-fuerza-ventas
Tabla Legacy Origen: c550t (14,2 M filas — particionada por fecha)
Propósito: telemetría GPS en ruta

1. Justificación y Mejoras de Arquitectura¶

Normalización Relacional Rigurosa: Aplica diseño normalizado (1FN a BCNF), eliminando grupos repetidos de booleanos, desnormalizaciones innecesarias y redundancias del sistema legacy.
Aislamiento Multi-Tenant (RLS): Integra empresa_id BIGINT NOT NULL indexado, gobernado por políticas PostgreSQL Row-Level Security (USING y WITH CHECK) para garantizar la total separación de datos por tenant.
Integridad Referencial Compuesta: Todas las claves foráneas internas hacia tablas con tenencia son compuestas (empresa_id, <fk_id>), asegurando la integridad física en PostgreSQL.
Trazabilidad y Auditoría: Auditoría estándar (created_at, updated_at, deleted_at, created_by, updated_by) con auditoría integral.

2. Definiciones de Implementación¶

DBMLSQLSQLAlchemyAlembic

Table fuerza_ventas.log_ubicaciones_promotores {
    id                      uuid          [pk, default: `gen_random_uuid()`, note: 'Identificador único UUID']
    empresa_id                  bigint        [not null, note: 'Identificador BIGINT de la empresa en core.identidad (RLS)']
    promotor_id             uuid          [not null, note: 'Promotor o asesor comercial']
    latitud                 double        [not null, note: 'Campo latitud']
    longitud                double        [not null, note: 'Campo longitud']
    precision_gps_metros    numeric(6,2)  [, note: 'Campo precision_gps_metros']
    bateria_porcentaje      int           [, note: 'Campo bateria_porcentaje']
    is_en_movimiento        boolean       [, note: 'Campo is_en_movimiento']
    fecha_hora_dispositivo  timestamptz   [not null, note: 'Campo fecha_hora_dispositivo']

    // Auditoría General
    is_activo               boolean       [not null, default: `true`, note: 'Estado lógico del registro']
    created_at              timestamptz   [not null, default: `now()`, note: 'Fecha de creación']
    updated_at              timestamptz   [not null, default: `now()`, note: 'Fecha de última actualización']
    deleted_at              timestamptz   [note: 'Fecha de eliminación lógica']
    created_by              uuid          [note: 'Usuario creador']
    updated_by              uuid          [note: 'Usuario modificador']

    Indexes {
        (empresa_id, id) [unique, name: 'uq_log_ubicaciones_promotores_empresa_id']
        (empresa_id, promotor_id) [name: 'ix_log_ubicaciones_promotores_empresa_promotor']
        (empresa_id, is_activo) [name: 'ix_log_ubicaciones_promotores_empresa_activo']
    }
}

Ref: fuerza_ventas.log_ubicaciones_promotores.(empresa_id, promotor_id) > fuerza_ventas.mae_promotores.(empresa_id, id) [delete: restrict]

CREATE TABLE IF NOT EXISTS fuerza_ventas.log_ubicaciones_promotores (
    id UUID NOT NULL DEFAULT gen_random_uuid(),
    empresa_id BIGINT NOT NULL,
    promotor_id UUID NOT NULL,
    latitud NUMERIC(10, 7) NOT NULL,
    longitud NUMERIC(10, 7) NOT NULL,
    precision_gps_metros NUMERIC(6,2),
    bateria_porcentaje INT,
    is_en_movimiento BOOLEAN,
    fecha_hora_dispositivo TIMESTAMPTZ NOT NULL,

    -- Auditoría General
    is_activo BOOLEAN NOT NULL DEFAULT true,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    deleted_at TIMESTAMPTZ,
    created_by UUID,
    updated_by UUID,

    CONSTRAINT pk_log_ubicaciones_promotores PRIMARY KEY (id),
    CONSTRAINT uq_log_ubicaciones_promotores_empresa_id UNIQUE (empresa_id, id),
    CONSTRAINT fk_log_ubicaciones_promotores_promotor_id FOREIGN KEY (empresa_id, promotor_id) REFERENCES fuerza_ventas.mae_promotores (empresa_id, id) ON DELETE RESTRICT
);

CREATE INDEX IF NOT EXISTS ix_log_ubicaciones_promotores_empresa_promotor ON fuerza_ventas.log_ubicaciones_promotores (empresa_id, promotor_id);
CREATE INDEX IF NOT EXISTS ix_log_ubicaciones_promotores_empresa_activo ON fuerza_ventas.log_ubicaciones_promotores (empresa_id, is_activo);

ALTER TABLE fuerza_ventas.log_ubicaciones_promotores ENABLE ROW LEVEL SECURITY;
CREATE POLICY rls_log_ubicaciones_promotores_tenant_isolation ON fuerza_ventas.log_ubicaciones_promotores
    FOR ALL
    USING (empresa_id = current_setting('app.current_empresa_id', true)::BIGINT)
    WITH CHECK (empresa_id = current_setting('app.current_empresa_id', true)::BIGINT);

from datetime import datetime
import uuid
from typing import Optional
from sqlalchemy import String, Boolean, DateTime, BigInteger, ForeignKey, ForeignKeyConstraint, UniqueConstraint, Index, Numeric, Integer, Text
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase

class Base(DeclarativeBase):
    pass

class UbicacionesPromotores(Base):
    __tablename__ = 'log_ubicaciones_promotores'
    __table_args__ = (
        UniqueConstraint('empresa_id', 'id', name='uq_log_ubicaciones_promotores_empresa_id'),
        Index('ix_log_ubicaciones_promotores_empresa_promotor', 'empresa_id', 'promotor_id'),
        ForeignKeyConstraint(['empresa_id', 'promotor_id'], ['fuerza_ventas.mae_promotores.empresa_id', 'fuerza_ventas.mae_promotores.id'], name='fk_log_ubicaciones_promotores_promotor_id'),
        Index('ix_log_ubicaciones_promotores_empresa_activo', 'empresa_id', 'is_activo'),
        {'schema': 'fuerza_ventas'}
    )

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    empresa_id: Mapped[int] = mapped_column(
        BigInteger, nullable=False)
    promotor_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False)
    latitud: Mapped[str] = mapped_column(String(200), nullable=False)
    longitud: Mapped[str] = mapped_column(String(200), nullable=False)
    precision_gps_metros: Mapped[Optional[float]] = mapped_column(Numeric(18, 2), nullable=True)
    bateria_porcentaje: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
    is_en_movimiento: Mapped[Optional[bool]] = mapped_column(Boolean, default=True, nullable=True)
    fecha_hora_dispositivo: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

    # Auditoría
    is_activo: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
    deleted_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
    created_by: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)
    updated_by: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)

"""revision: crm_fv_0007
create table fuerza_ventas.log_ubicaciones_promotores
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

revision = 'crm_fv_0007'
down_revision = 'crm_fv_0006'
branch_labels = None
depends_on = None

def upgrade() -> None:
    op.create_table(
        'log_ubicaciones_promotores',
        sa.Column('id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
        sa.Column(
            "empresa_id",
            sa.BigInteger(), nullable=False),
        sa.Column('promotor_id', postgresql.UUID(as_uuid=True), nullable=False),
        sa.Column('latitud', sa.String(length=200), nullable=False),
        sa.Column('longitud', sa.String(length=200), nullable=False),
        sa.Column('precision_gps_metros', sa.Numeric(precision=18, scale=2), nullable=True),
        sa.Column('bateria_porcentaje', sa.Integer(), nullable=True),
        sa.Column('is_en_movimiento', sa.Boolean(), nullable=True),
        sa.Column('fecha_hora_dispositivo', sa.DateTime(timezone=True), nullable=False),
        sa.Column('is_activo', sa.Boolean(), server_default=sa.text('true'), nullable=False),
        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
        sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
        sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
        sa.Column('created_by', postgresql.UUID(as_uuid=True), nullable=True),
        sa.Column('updated_by', postgresql.UUID(as_uuid=True), nullable=True),
        sa.PrimaryKeyConstraint('id', name='pk_log_ubicaciones_promotores'),
        sa.UniqueConstraint('empresa_id', 'id', name='uq_log_ubicaciones_promotores_empresa_id'),
        sa.ForeignKeyConstraint(['empresa_id', 'promotor_id'], ['fuerza_ventas.mae_promotores.empresa_id', 'fuerza_ventas.mae_promotores.id'], name='fk_log_ubicaciones_promotores_promotor_id'),
        schema='fuerza_ventas'
    )
    op.create_index('ix_log_ubicaciones_promotores_empresa_promotor', 'log_ubicaciones_promotores', ['empresa_id', 'promotor_id'], unique=False, schema='fuerza_ventas')
    op.create_index('ix_log_ubicaciones_promotores_empresa_activo', 'log_ubicaciones_promotores', ['empresa_id', 'is_activo'], unique=False, schema='fuerza_ventas')
    op.execute('ALTER TABLE fuerza_ventas.log_ubicaciones_promotores ENABLE ROW LEVEL SECURITY')
    op.execute('''CREATE POLICY rls_log_ubicaciones_promotores_tenant_isolation ON fuerza_ventas.log_ubicaciones_promotores FOR ALL USING (empresa_id = current_setting(\'app.current_empresa_id\', true)::BIGINT) WITH CHECK (empresa_id = current_setting(\'app.current_empresa_id\', true)::BIGINT)''')

def downgrade() -> None:
    op.execute('DROP POLICY IF EXISTS rls_log_ubicaciones_promotores_tenant_isolation ON fuerza_ventas.log_ubicaciones_promotores')
    op.drop_index('ix_log_ubicaciones_promotores_empresa_promotor', table_name='log_ubicaciones_promotores', schema='fuerza_ventas')
    op.drop_index('ix_log_ubicaciones_promotores_empresa_activo', table_name='log_ubicaciones_promotores', schema='fuerza_ventas')
    op.drop_table('log_ubicaciones_promotores', schema='fuerza_ventas')

Especificación Técnica: log_ubicaciones_promotores¶

1. Justificación y Mejoras de Arquitectura¶

2. Definiciones de Implementación¶

Especificación Técnica: `log_ubicaciones_promotores`¶