Especificación Técnica: `trx_cobros_campo_imputaciones`¶

Esquema: contable
Base de Datos: erp
Servicio: svc-contable
Tabla Legacy Origen: c213t (descompone arrays paralelos co_aviso / nu_abono_aviso) + c247t
Propósito: Desglose de amortizaciones e imputaciones aplicadas por cada cobro en campo hacia facturas pendientes de cobro u órdenes de cobranza vinculadas.

1. Justificación y Mejoras de Arquitectura¶

Normalización de Colecciones Legacy: Transforma los arrays paralelos desnormalizados del sistema legacy en un modelo relacional 1:N estrictamente tipado con auditoría de saldos previo y posterior a la imputación (monto_saldo_anterior, monto_saldo_posterior).
Integridad Referencial Compuesta Multi-Tenant: Clave única compuesta (empresa_id, id) y clave foránea compuesta (empresa_id, cobro_campo_id) hacia contable.trx_cobros_campo(empresa_id, id) con eliminación en cascada (ON DELETE CASCADE).
Integración y Correlación Desacoplada: El campo factura_erp_id correlaciona con la entidad de facturas dentro del módulo contable del ERP, y aviso_cobro_id correlaciona lógicamente con crm.fuerza_ventas.trx_avisos_cobro sin claves foráneas físicas cross-database.
Aislamiento Multi-Tenant (RLS): Integra empresa_id BIGINT NOT NULL con políticas RLS obligatorias (USING y WITH CHECK).
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¶

DBMLSQLSQLAlchemyAlembic

Table contable.trx_cobros_campo_imputaciones {
    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)']
    cobro_campo_id        uuid          [not null, note: 'Cabecera del cobro en campo asociado']
    factura_erp_id        uuid          [not null, note: 'Identificador de la factura en contable.doc_facturas']
    aviso_cobro_id        uuid          [note: 'Correlación lógica con crm.fuerza_ventas.trx_avisos_cobro']
    monto_imputado        numeric(12,2) [not null, note: 'Monto monetario amortizado a la factura']
    monto_saldo_anterior  numeric(12,2) [not null, note: 'Saldo exigible de la factura previo a la imputación']
    monto_saldo_posterior numeric(12,2) [not null, note: 'Saldo exigible resultante de la factura tras la imputación']

    // 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_cobros_campo_imp_empresa_id']
        (empresa_id, cobro_campo_id) [name: 'ix_trx_cobros_imp_empresa_cobro']
        (empresa_id, factura_erp_id) [name: 'ix_trx_cobros_imp_empresa_factura']
        (empresa_id, aviso_cobro_id) [name: 'ix_trx_cobros_imp_empresa_aviso']
        (empresa_id, is_activo) [name: 'ix_trx_cobros_imp_empresa_activo']
    }
}

Ref: contable.trx_cobros_campo_imputaciones.(empresa_id, cobro_campo_id) > contable.trx_cobros_campo.(empresa_id, id) [delete: cascade]

CREATE TABLE IF NOT EXISTS contable.trx_cobros_campo_imputaciones (
    id UUID NOT NULL DEFAULT gen_random_uuid(),
    empresa_id BIGINT NOT NULL,
    cobro_campo_id UUID NOT NULL,
    factura_erp_id UUID NOT NULL,
    aviso_cobro_id UUID,
    monto_imputado NUMERIC(12,2) NOT NULL,
    monto_saldo_anterior NUMERIC(12,2) NOT NULL,
    monto_saldo_posterior NUMERIC(12,2) 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_trx_cobros_campo_imputaciones PRIMARY KEY (id),
    CONSTRAINT uq_trx_cobros_campo_imp_empresa_id UNIQUE (empresa_id, id),
    CONSTRAINT fk_trx_cobros_campo_imp_cobro FOREIGN KEY (empresa_id, cobro_campo_id)
        REFERENCES contable.trx_cobros_campo (empresa_id, id)
        ON DELETE CASCADE
);

-- Índices
CREATE INDEX IF NOT EXISTS ix_trx_cobros_imp_empresa_cobro
    ON contable.trx_cobros_campo_imputaciones (empresa_id, cobro_campo_id);

CREATE INDEX IF NOT EXISTS ix_trx_cobros_imp_empresa_factura
    ON contable.trx_cobros_campo_imputaciones (empresa_id, factura_erp_id);

CREATE INDEX IF NOT EXISTS ix_trx_cobros_imp_empresa_aviso
    ON contable.trx_cobros_campo_imputaciones (empresa_id, aviso_cobro_id);

CREATE INDEX IF NOT EXISTS ix_trx_cobros_imp_empresa_activo
    ON contable.trx_cobros_campo_imputaciones (empresa_id, is_activo);

-- Row-Level Security
ALTER TABLE contable.trx_cobros_campo_imputaciones ENABLE ROW LEVEL SECURITY;

CREATE POLICY rls_trx_cobros_campo_imp_empresa
    ON contable.trx_cobros_campo_imputaciones
    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
    );

-- Comentarios
COMMENT ON TABLE contable.trx_cobros_campo_imputaciones IS 'Desglose de imputaciones de cobros en campo a facturas contables';
COMMENT ON COLUMN contable.trx_cobros_campo_imputaciones.cobro_campo_id IS 'Identificador del cobro en campo cabecera';
COMMENT ON COLUMN contable.trx_cobros_campo_imputaciones.factura_erp_id IS 'Identificador de la factura en contable.doc_facturas (correlación interna ERP)';
COMMENT ON COLUMN contable.trx_cobros_campo_imputaciones.aviso_cobro_id IS 'Correlación lógica con crm.fuerza_ventas.trx_avisos_cobro (sin FK cross-DB)';

from datetime import datetime
from typing import Optional
import uuid

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


class Base(DeclarativeBase):
    pass


class TrxCobroCampoImputacion(Base):
    __tablename__ = "trx_cobros_campo_imputaciones"
    __table_args__ = (
        UniqueConstraint(
            "empresa_id",
            "id",
            name="uq_trx_cobros_campo_imp_empresa_id",
        ),
        ForeignKeyConstraint(
            ["empresa_id", "cobro_campo_id"],
            ["contable.trx_cobros_campo.empresa_id", "contable.trx_cobros_campo.id"],
            ondelete="CASCADE",
            name="fk_trx_cobros_campo_imp_cobro",
        ),
        Index(
            "ix_trx_cobros_imp_empresa_cobro",
            "empresa_id",
            "cobro_campo_id",
        ),
        Index(
            "ix_trx_cobros_imp_empresa_factura",
            "empresa_id",
            "factura_erp_id",
        ),
        Index(
            "ix_trx_cobros_imp_empresa_aviso",
            "empresa_id",
            "aviso_cobro_id",
        ),
        Index(
            "ix_trx_cobros_imp_empresa_activo",
            "empresa_id",
            "is_activo",
        ),
        {"schema": "contable"},
    )

    # Identificadores y Claves
    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)",
    )

    cobro_campo_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        nullable=False,
        comment="Cabecera del cobro en campo asociado",
    )

    factura_erp_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        nullable=False,
        comment="Identificador de la factura en contable.doc_facturas",
    )

    aviso_cobro_id: Mapped[Optional[uuid.UUID]] = mapped_column(
        UUID(as_uuid=True),
        nullable=True,
        comment="Correlación lógica con crm.fuerza_ventas.trx_avisos_cobro",
    )

    # Importes Financieros
    monto_imputado: Mapped[float] = mapped_column(
        Numeric(12, 2),
        nullable=False,
        comment="Monto monetario amortizado a la factura",
    )

    monto_saldo_anterior: Mapped[float] = mapped_column(
        Numeric(12, 2),
        nullable=False,
        comment="Saldo exigible de la factura previo a la imputación",
    )

    monto_saldo_posterior: Mapped[float] = mapped_column(
        Numeric(12, 2),
        nullable=False,
        comment="Saldo exigible resultante de la factura tras la imputación",
    )

    # Auditoría General
    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()"),
        onupdate=datetime.utcnow,
        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",
    )

```python """create trx_cobros_campo_imputaciones

Revision ID: erp_0005 Revises: erp_0004 Create Date: 2026-09-10 """

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 = "erp_0005" down_revision: Union[str, Sequence[str], None] = "erp_0004" branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None

def upgrade() -> None:

op.create_table(
    "trx_cobros_campo_imputaciones",

    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(
        "cobro_campo_id",
        postgresql.UUID(as_uuid=True),
        nullable=False,
        comment="Cabecera del cobro en campo asociado",
    ),

    sa.Column(
        "factura_erp_id",
        postgresql.UUID(as_uuid=True),
        nullable=False,
        comment="Identificador de la factura en contable.doc_facturas",
    ),

    sa.Column(
        "aviso_cobro_id",
        postgresql.UUID(as_uuid=True),
        nullable=True,
        comment="Correlación lógica con crm.fuerza_ventas.trx_avisos_cobro",
    ),

    sa.Column(
        "monto_imputado",
        sa.Numeric(precision=12, scale=2),
        nullable=False,
        comment="Monto monetario amortizado a la factura",
    ),

    sa.Column(
        "monto_saldo_anterior",
        sa.Numeric(precision=12, scale=2),
        nullable=False,
        comment="Saldo exigible de la factura previo a la imputación",
    ),

    sa.Column(
        "monto_saldo_posterior",
        sa.Numeric(precision=12, scale=2),
        nullable=False,
        comment="Saldo exigible resultante de la factura tras la imputación",
    ),

    # 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_cobros_campo_imputaciones",
    ),

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

    sa.ForeignKeyConstraint(
        ["empresa_id", "cobro_campo_id"],
        ["contable.trx_cobros_campo.empresa_id", "contable.trx_cobros_campo.id"],
        ondelete="CASCADE",
        name="fk_trx_cobros_campo_imp_cobro",
    ),

    comment="Imputaciones Contables de Cobros a Facturas / Avisos",
    schema="contable",
)

op.create_index(
    "ix_trx_cobros_imp_empresa_activo",
    "trx_cobros_campo_imputaciones",
    ["empresa_id", "is_activo"],
    unique=False,
    schema="contable",
)

op.create_index(
    "ix_trx_cobros_imp_empresa_cobro",
    "trx_cobros_campo_imputaciones",
    ["empresa_id", "cobro_campo_id"],
    unique=False,
    schema="contable",
)

op.create_index(
    "ix_trx_cobros_imp_empresa_factura",
    "trx_cobros_campo_imputaciones",
    ["empresa_id", "factura_erp_id"],
    unique=False,
    schema="contable",
)

op.create_index(
    "ix_trx_cobros_imp_empresa_aviso",
    "trx_cobros_campo_imputaciones",
    ["empresa_id", "aviso_cobro_id"],
    unique=False,
    schema="contable",
)

op.execute(
    """
    ALTER TABLE contable.trx_cobros_campo_imputaciones
    ENABLE ROW LEVEL SECURITY;
    """
)

op.execute(
    """
    CREATE POLICY rls_trx_cobros_campo_imp_empresa
    ON contable.trx_cobros_campo_imputaciones
    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_cobros_campo_imp_empresa
    ON contable.trx_cobros_campo_imputaciones;
    """
)

op.execute(
    """
    ALTER TABLE contable.trx_cobros_campo_imputaciones
    DISABLE ROW LEVEL SECURITY;
    """
)

op.drop_index(
    "ix_trx_cobros_imp_empresa_aviso",
    table_name="trx_cobros_campo_imputaciones",
    schema="contable",
)

op.drop_index(
    "ix_trx_cobros_imp_empresa_factura",
    table_name="trx_cobros_campo_imputaciones",
    schema="contable",
)

op.drop_index(
    "ix_trx_cobros_imp_empresa_cobro",
    table_name="trx_cobros_campo_imputaciones",
    schema="contable",
)

op.drop_index(
    "ix_trx_cobros_imp_empresa_activo",
    table_name="trx_cobros_campo_imputaciones",
    schema="contable",
)

op.drop_table(
    "trx_cobros_campo_imputaciones",
    schema="contable",
)

Especificación Técnica: trx_cobros_campo_imputaciones¶

1. Justificación y Mejoras de Arquitectura¶

2. Definiciones de Implementación¶

revision identifiers, used by Alembic.¶

Especificación Técnica: `trx_cobros_campo_imputaciones`¶