Saltar a contenido

Especificación Técnica: trx_avisos_cobro_detalle

Esquema: fuerza_ventas
Base de Datos: crm
Tabla Legacy Origen: c213t (arrays paralelos de avisos normalizados)
Propósito: Desglose de facturas y documentos de deuda vinculados a una orden de cobro en campo.

1. Justificación y Mejoras de Arquitectura

  • Normalización de Colecciones Legacy: Transforma los arreglos paralelos y desnormalizados del sistema legacy en un modelo relacional 1:N estructurado y fuertemente tipado.
  • Integración Cross-Database Desacoplada: El campo factura_erp_id UUID NOT NULL mantiene la correlación lógica con la entidad doc_facturas de la base de datos erp (esquema contable) sin forzar llaves foráneas físicas inter-base.
  • Aislamiento Multi-Tenant (RLS): Integra empresa_id BIGINT NOT NULL con políticas RLS obligatorias (USING y WITH CHECK).
  • Integridad Referencial Compuesta: Clave única compuesta (empresa_id, id) y clave foránea compuesta (empresa_id, aviso_cobro_id) hacia fuerza_ventas.trx_avisos_cobro(empresa_id, id) con eliminación en cascada (ON DELETE CASCADE).
  • Trazabilidad y Auditoría: Auditoría integral (created_at, updated_at, deleted_at, created_by, updated_by) con auditoría integral.

2. Definiciones de Implementación

Table fuerza_ventas.trx_avisos_cobro_detalle {
    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)']
    aviso_cobro_id            uuid          [not null, note: 'Cabecera de la orden de cobro asociada']
    factura_erp_id            uuid          [not null, note: 'Identificador de la factura en erp.contable (correlación lógica cross-database)']
    numero_factura_texto      varchar(50)   [not null, note: 'Número legible o correlativo de la factura']
    fecha_emision_factura     date          [not null, note: 'Fecha de emisión original del documento']
    fecha_vencimiento_factura date          [not null, note: 'Fecha de vencimiento del documento']
    monto_original_factura    numeric(12,2) [not null, note: 'Monto total original de la factura']
    monto_saldo_pendiente     numeric(12,2) [not null, note: 'Saldo exigible pendiente de cobro']
    monto_cobrado_linea       numeric(12,2) [not null, default: `0.00`, note: 'Monto recaudado para esta factura en la gestión']
    is_saldada                boolean       [not null, default: `false`, note: 'Indica si el saldo quedó totalmente cancelado']

    // 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_trx_avisos_cobro_det_empresa_id']
        (empresa_id, is_activo) [name: 'ix_trx_avisos_cobro_det_empresa_activo']
        (empresa_id, aviso_cobro_id) [name: 'ix_trx_avisos_cobro_det_empresa_aviso']
        (empresa_id, factura_erp_id) [name: 'ix_trx_avisos_cobro_det_empresa_factura']
        (empresa_id, is_saldada) [name: 'ix_trx_avisos_cobro_det_empresa_saldada']
    }
}

Ref: fuerza_ventas.trx_avisos_cobro_detalle.(empresa_id, aviso_cobro_id) > fuerza_ventas.trx_avisos_cobro.(empresa_id, id) [delete: cascade]

CREATE TABLE IF NOT EXISTS fuerza_ventas.trx_avisos_cobro_detalle (
    id UUID NOT NULL DEFAULT gen_random_uuid(),
    empresa_id BIGINT NOT NULL,
    aviso_cobro_id UUID NOT NULL,
    factura_erp_id UUID NOT NULL,
    numero_factura_texto VARCHAR(50) NOT NULL,
    fecha_emision_factura DATE NOT NULL,
    fecha_vencimiento_factura DATE NOT NULL,
    monto_original_factura NUMERIC(12,2) NOT NULL,
    monto_saldo_pendiente NUMERIC(12,2) NOT NULL,
    monto_cobrado_linea NUMERIC(12,2) NOT NULL DEFAULT 0.00,
    is_saldada BOOLEAN NOT NULL DEFAULT false,

    -- 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_trx_avisos_cobro_detalle
        PRIMARY KEY (id),

    CONSTRAINT uq_trx_avisos_cobro_det_empresa_id
        UNIQUE (empresa_id, id),

    CONSTRAINT fk_trx_avisos_cobro_det_aviso
        FOREIGN KEY (empresa_id, aviso_cobro_id)
        REFERENCES fuerza_ventas.trx_avisos_cobro (empresa_id, id)
        ON DELETE CASCADE
);

CREATE INDEX IF NOT EXISTS ix_trx_avisos_cobro_det_empresa_activo
    ON fuerza_ventas.trx_avisos_cobro_detalle (empresa_id, is_activo);

CREATE INDEX IF NOT EXISTS ix_trx_avisos_cobro_det_empresa_aviso
    ON fuerza_ventas.trx_avisos_cobro_detalle (empresa_id, aviso_cobro_id);

CREATE INDEX IF NOT EXISTS ix_trx_avisos_cobro_det_empresa_factura
    ON fuerza_ventas.trx_avisos_cobro_detalle (empresa_id, factura_erp_id);

CREATE INDEX IF NOT EXISTS ix_trx_avisos_cobro_det_empresa_saldada
    ON fuerza_ventas.trx_avisos_cobro_detalle (empresa_id, is_saldada);

COMMENT ON TABLE fuerza_ventas.trx_avisos_cobro_detalle IS
    'Detalle de Documentos en Avisos de Cobro: Desglose de facturas y documentos de deuda vinculados a una orden de cobro en campo.';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.id IS
    'Identificador único UUID';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.empresa_id IS
    'Identificador UUID de la empresa en core.identidad (RLS)';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.aviso_cobro_id IS
    'Cabecera de la orden de cobro asociada';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.factura_erp_id IS
    'Identificador de la factura en erp.contable (correlación lógica cross-database)';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.numero_factura_texto IS
    'Número legible o correlativo de la factura';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.fecha_emision_factura IS
    'Fecha de emisión original del documento';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.fecha_vencimiento_factura IS
    'Fecha de vencimiento del documento';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.monto_original_factura IS
    'Monto total original de la factura';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.monto_saldo_pendiente IS
    'Saldo exigible pendiente de cobro';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.monto_cobrado_linea IS
    'Monto recaudado para esta factura en la gestión';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.is_saldada IS
    'Indica si el saldo quedó totalmente cancelado';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.is_activo IS
    'Estado lógico del registro';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.created_at IS
    'Fecha de creación';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.updated_at IS
    'Fecha de última actualización';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.deleted_at IS
    'Fecha de eliminación lógica';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.created_by IS
    'Usuario creador';

COMMENT ON COLUMN fuerza_ventas.trx_avisos_cobro_detalle.updated_by IS
    'Usuario modificador';

ALTER TABLE fuerza_ventas.trx_avisos_cobro_detalle
    ENABLE ROW LEVEL SECURITY;

CREATE POLICY rls_trx_avisos_cobro_det_empresa
    ON fuerza_ventas.trx_avisos_cobro_detalle
    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
    );

import uuid
from datetime import date, datetime
from decimal import Decimal
from typing import Optional

from sqlalchemy import (
    BigInteger,
    Boolean,
    Date,
    ForeignKeyConstraint,
    Index,
    Numeric,
    String,
    UniqueConstraint,
    text,
)
from sqlalchemy.dialects.postgresql import TIMESTAMPTZ, UUID
from sqlalchemy.orm import Mapped, mapped_column

from app.db.base import Base


class TrxAvisosCobroDetalle(Base):
    """
    Detalle de Documentos en Avisos de Cobro.

    Desglose de facturas y documentos de deuda vinculados
    a una orden de cobro asignada a promotores en campo.

    Legacy:
        c213t (arrays paralelos de avisos normalizados)
    """

    __tablename__ = "trx_avisos_cobro_detalle"

    __table_args__ = (
        UniqueConstraint(
            "empresa_id",
            "id",
            name="uq_trx_avisos_cobro_det_empresa_id",
        ),
        ForeignKeyConstraint(
            ["empresa_id", "aviso_cobro_id"],
            ["fuerza_ventas.trx_avisos_cobro.empresa_id", "fuerza_ventas.trx_avisos_cobro.id"],
            ondelete="CASCADE",
            name="fk_trx_avisos_cobro_det_aviso",
        ),
        Index(
            "ix_trx_avisos_cobro_det_empresa_activo",
            "empresa_id",
            "is_activo",
        ),
        Index(
            "ix_trx_avisos_cobro_det_empresa_aviso",
            "empresa_id",
            "aviso_cobro_id",
        ),
        Index(
            "ix_trx_avisos_cobro_det_empresa_factura",
            "empresa_id",
            "factura_erp_id",
        ),
        Index(
            "ix_trx_avisos_cobro_det_empresa_saldada",
            "empresa_id",
            "is_saldada",
        ),
        {
            "schema": "fuerza_ventas",
            "comment": "Detalle de Documentos en Avisos de Cobro",
        },
    )

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        primary_key=True,
        server_default=text("gen_random_uuid()"),
        comment="Identificador único UUID",
    )

    empresa_id: Mapped[int] = mapped_column(
        BigInteger,
        nullable=False,
        comment="Identificador UUID de la empresa en core.identidad (RLS)",
    )

    aviso_cobro_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        nullable=False,
        comment="Cabecera de la orden de cobro asociada",
    )

    factura_erp_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        nullable=False,
        comment="Identificador de la factura en erp.contable (correlación lógica cross-database)",
    )

    numero_factura_texto: Mapped[str] = mapped_column(
        String(50),
        nullable=False,
        comment="Número legible o correlativo de la factura",
    )

    fecha_emision_factura: Mapped[date] = mapped_column(
        Date,
        nullable=False,
        comment="Fecha de emisión original del documento",
    )

    fecha_vencimiento_factura: Mapped[date] = mapped_column(
        Date,
        nullable=False,
        comment="Fecha de vencimiento del documento",
    )

    monto_original_factura: Mapped[Decimal] = mapped_column(
        Numeric(12, 2),
        nullable=False,
        comment="Monto total original de la factura",
    )

    monto_saldo_pendiente: Mapped[Decimal] = mapped_column(
        Numeric(12, 2),
        nullable=False,
        comment="Saldo exigible pendiente de cobro",
    )

    monto_cobrado_linea: Mapped[Decimal] = mapped_column(
        Numeric(12, 2),
        nullable=False,
        server_default=text("0.00"),
        comment="Monto recaudado para esta factura en la gestión",
    )

    is_saldada: Mapped[bool] = mapped_column(
        Boolean,
        nullable=False,
        server_default=text("false"),
        comment="Indica si el saldo quedó totalmente cancelado",
    )

    is_activo: Mapped[bool] = mapped_column(
        Boolean,
        nullable=False,
        server_default=text("true"),
        comment="Estado lógico del registro",
    )

    created_at: Mapped[datetime] = mapped_column(
        TIMESTAMPTZ,
        nullable=False,
        server_default=text("now()"),
        comment="Fecha de creación",
    )

    updated_at: Mapped[datetime] = mapped_column(
        TIMESTAMPTZ,
        nullable=False,
        server_default=text("now()"),
        comment="Fecha de última actualización",
    )

    deleted_at: Mapped[Optional[datetime]] = mapped_column(
        TIMESTAMPTZ,
        nullable=True,
        comment="Fecha de eliminación lógica",
    )

    created_by: Mapped[Optional[uuid.UUID]] = mapped_column(
        UUID(as_uuid=True),
        nullable=True,
        comment="Usuario creador",
    )

    updated_by: Mapped[Optional[uuid.UUID]] = mapped_column(
        UUID(as_uuid=True),
        nullable=True,
        comment="Usuario modificador",
    )

"""create trx_avisos_cobro_detalle

Revision ID: trx_0007
Revises: trx_0006
Create Date: 2026-09-09
"""

from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql


# revision identifiers, used by Alembic.
revision: str = "trx_0007"
down_revision: Union[str, Sequence[str], None] = "trx_0006"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:

    op.create_table(
        "trx_avisos_cobro_detalle",

        sa.Column(
            "id",
            postgresql.UUID(as_uuid=True),
            nullable=False,
            server_default=sa.text("gen_random_uuid()"),
            comment="Identificador único UUID",
        ),

        sa.Column(
            "empresa_id",
            sa.BigInteger(),
            nullable=False,
            comment="Identificador UUID de la empresa en core.identidad (RLS)",
        ),

        sa.Column(
            "aviso_cobro_id",
            postgresql.UUID(as_uuid=True),
            nullable=False,
            comment="Cabecera de la orden de cobro asociada",
        ),

        sa.Column(
            "factura_erp_id",
            postgresql.UUID(as_uuid=True),
            nullable=False,
            comment="Identificador de la factura en erp.contable (correlación lógica cross-database)",
        ),

        sa.Column(
            "numero_factura_texto",
            sa.String(length=50),
            nullable=False,
            comment="Número legible o correlativo de la factura",
        ),

        sa.Column(
            "fecha_emision_factura",
            sa.Date(),
            nullable=False,
            comment="Fecha de emisión original del documento",
        ),

        sa.Column(
            "fecha_vencimiento_factura",
            sa.Date(),
            nullable=False,
            comment="Fecha de vencimiento del documento",
        ),

        sa.Column(
            "monto_original_factura",
            sa.Numeric(precision=12, scale=2),
            nullable=False,
            comment="Monto total original de la factura",
        ),

        sa.Column(
            "monto_saldo_pendiente",
            sa.Numeric(precision=12, scale=2),
            nullable=False,
            comment="Saldo exigible pendiente de cobro",
        ),

        sa.Column(
            "monto_cobrado_linea",
            sa.Numeric(precision=12, scale=2),
            nullable=False,
            server_default=sa.text("0.00"),
            comment="Monto recaudado para esta factura en la gestión",
        ),

        sa.Column(
            "is_saldada",
            sa.Boolean(),
            nullable=False,
            server_default=sa.text("false"),
            comment="Indica si el saldo quedó totalmente cancelado",
        ),

        # Auditoría General
        sa.Column(
            "is_activo",
            sa.Boolean(),
            nullable=False,
            server_default=sa.text("true"),
            comment="Estado lógico del registro",
        ),

        sa.Column(
            "created_at",
            sa.TIMESTAMP(timezone=True),
            nullable=False,
            server_default=sa.text("now()"),
            comment="Fecha de creación",
        ),

        sa.Column(
            "updated_at",
            sa.TIMESTAMP(timezone=True),
            nullable=False,
            server_default=sa.text("now()"),
            comment="Fecha de última actualización",
        ),

        sa.Column(
            "deleted_at",
            sa.TIMESTAMP(timezone=True),
            nullable=True,
            comment="Fecha de eliminación lógica",
        ),

        sa.Column(
            "created_by",
            postgresql.UUID(as_uuid=True),
            nullable=True,
            comment="Usuario creador",
        ),

        sa.Column(
            "updated_by",
            postgresql.UUID(as_uuid=True),
            nullable=True,
            comment="Usuario modificador",
        ),

        sa.PrimaryKeyConstraint(
            "id",
            name="pk_trx_avisos_cobro_detalle",
        ),

        sa.UniqueConstraint(
            "empresa_id",
            "id",
            name="uq_trx_avisos_cobro_det_empresa_id",
        ),

        sa.ForeignKeyConstraint(
            ["empresa_id", "aviso_cobro_id"],
            ["fuerza_ventas.trx_avisos_cobro.empresa_id", "fuerza_ventas.trx_avisos_cobro.id"],
            ondelete="CASCADE",
            name="fk_trx_avisos_cobro_det_aviso",
        ),

        comment="Detalle de Documentos en Avisos de Cobro",
        schema="fuerza_ventas",
    )

    op.create_index(
        "ix_trx_avisos_cobro_det_empresa_activo",
        "trx_avisos_cobro_detalle",
        ["empresa_id", "is_activo"],
        unique=False,
        schema="fuerza_ventas",
    )

    op.create_index(
        "ix_trx_avisos_cobro_det_empresa_aviso",
        "trx_avisos_cobro_detalle",
        ["empresa_id", "aviso_cobro_id"],
        unique=False,
        schema="fuerza_ventas",
    )

    op.create_index(
        "ix_trx_avisos_cobro_det_empresa_factura",
        "trx_avisos_cobro_detalle",
        ["empresa_id", "factura_erp_id"],
        unique=False,
        schema="fuerza_ventas",
    )

    op.create_index(
        "ix_trx_avisos_cobro_det_empresa_saldada",
        "trx_avisos_cobro_detalle",
        ["empresa_id", "is_saldada"],
        unique=False,
        schema="fuerza_ventas",
    )

    op.execute(
        """
        ALTER TABLE fuerza_ventas.trx_avisos_cobro_detalle
        ENABLE ROW LEVEL SECURITY;
        """
    )

    op.execute(
        """
        CREATE POLICY rls_trx_avisos_cobro_det_empresa
        ON fuerza_ventas.trx_avisos_cobro_detalle
        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_trx_avisos_cobro_det_empresa
        ON fuerza_ventas.trx_avisos_cobro_detalle;
        """
    )

    op.execute(
        """
        ALTER TABLE fuerza_ventas.trx_avisos_cobro_detalle
        DISABLE ROW LEVEL SECURITY;
        """
    )

    op.drop_index(
        "ix_trx_avisos_cobro_det_empresa_saldada",
        table_name="trx_avisos_cobro_detalle",
        schema="fuerza_ventas",
    )

    op.drop_index(
        "ix_trx_avisos_cobro_det_empresa_factura",
        table_name="trx_avisos_cobro_detalle",
        schema="fuerza_ventas",
    )

    op.drop_index(
        "ix_trx_avisos_cobro_det_empresa_aviso",
        table_name="trx_avisos_cobro_detalle",
        schema="fuerza_ventas",
    )

    op.drop_index(
        "ix_trx_avisos_cobro_det_empresa_activo",
        table_name="trx_avisos_cobro_detalle",
        schema="fuerza_ventas",
    )

    op.drop_table(
        "trx_avisos_cobro_detalle",
        schema="fuerza_ventas",
    )