Especificación Técnica: `cat_encuesta_preguntas`¶

Esquema: clientes
Base de Datos: crm
Tabla Legacy Origen: c061t_respuesta_pregunta_abierta / c062t_respuesta_pregunta_cerrada
Propósito: Catálogo formal de preguntas y reactivos utilizados en encuestas de prospección, satisfacción y auditoría de punto de venta (reemplaza arrays desnormalizados y JSON libres).

1. Justificación y Mejoras de Arquitectura¶

Estandariza catálogos tipados con UUID, multitenancy RLS por empresa y claves únicas de código de negocio, eliminando cadenas sueltas y tablas desnormalizadas.
Multi-tenancy RLS y Auditoría Estándar: Cada tabla incluye empresa_id BIGINT NOT NULL, utilizado para el aislamiento multiempresa mediante políticas RLS y como identificador de la empresa en core.identidad. Además, incorpora trazabilidad mediante created_at, updated_at, deleted_at, created_by, updated_by y version.
Integridad para opciones y respuestas: Dispone de constraint única (empresa_id, id) que permite a cat_encuesta_opciones y trx_encuestas_cliente_detalle vincularse mediante clave foránea compuesta manteniendo aislamiento multi-tenant.

2. Definiciones de Implementación¶

DBMLSQLSQLAlchemyAlembic

Table clientes.cat_encuesta_preguntas {
    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)']
    codigo          varchar(30)   [not null, note: 'Código único mnemotécnico de la pregunta']
    enunciado       text          [not null, note: 'Texto del reactivo o pregunta formulada']
    tipo_respuesta_id uuid          [not null, note: 'Tipo de respuesta (FK cat_tipos_respuesta_encuesta)']
    seccion         varchar(50)   [not null, default: `'GENERAL'`, note: 'Sección temática de la encuesta']
    orden           int           [not null, default: `1`, note: 'Orden de presentación visual']
    is_obligatoria  boolean       [not null, default: `false`, note: 'Indica si la respuesta es obligatoria']

    // 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_cat_encuesta_preguntas_empresa_id']
        (empresa_id, codigo) [unique, name: 'uq_cat_encuesta_preguntas_empresa_codigo']
        (empresa_id, is_activo) [name: 'ix_cat_encuesta_preguntas_empresa_activo']
        (empresa_id, seccion, orden) [name: 'ix_cat_encuesta_preguntas_seccion']
    }
}
Ref: clientes.cat_encuesta_preguntas.(empresa_id, tipo_respuesta_id) > clientes.cat_tipos_respuesta_encuesta.(empresa_id, id)

CREATE TABLE IF NOT EXISTS clientes.cat_encuesta_preguntas (
    id UUID NOT NULL DEFAULT gen_random_uuid(),
    empresa_id BIGINT NOT NULL,
    codigo VARCHAR(30) NOT NULL,
    enunciado TEXT NOT NULL,
    tipo_respuesta_id UUID NOT NULL,
    seccion VARCHAR(50) NOT NULL DEFAULT 'GENERAL',
    orden INT NOT NULL DEFAULT 1,
    is_obligatoria 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_cat_encuesta_preguntas
        PRIMARY KEY (id),

    CONSTRAINT uq_cat_encuesta_preguntas_empresa_id
        UNIQUE (empresa_id, id),

    CONSTRAINT uq_cat_encuesta_preguntas_empresa_codigo
        UNIQUE (empresa_id, codigo)

    CONSTRAINT fk_cat_encuesta_preguntas_tipo_respuesta
        FOREIGN KEY (empresa_id, tipo_respuesta_id)
        REFERENCES clientes.cat_tipos_respuesta_encuesta (empresa_id, id)
);

CREATE INDEX IF NOT EXISTS ix_cat_encuesta_preguntas_empresa_activo
    ON clientes.cat_encuesta_preguntas (empresa_id, is_activo);

CREATE INDEX IF NOT EXISTS ix_cat_encuesta_preguntas_empresa_tipo_respuesta
    ON clientes.cat_encuesta_preguntas (empresa_id, tipo_respuesta_id);

CREATE INDEX IF NOT EXISTS ix_cat_encuesta_preguntas_seccion
    ON clientes.cat_encuesta_preguntas (empresa_id, seccion, orden);

COMMENT ON TABLE clientes.cat_encuesta_preguntas IS
    'Preguntas de Encuestas Comerciales: Catálogo formal de preguntas y reactivos utilizados en encuestas de prospección, satisfacción y auditoría de punto de venta (reemplaza arrays desnormalizados y JSON libres).';

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.id IS
    'Identificador único UUID';

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

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.codigo IS
    'Código único mnemotécnico de la pregunta';

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.enunciado IS
    'Texto del reactivo o pregunta formulada';

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.tipo_respuesta_id IS
    'Tipo de respuesta (FK cat_tipos_respuesta_encuesta)';

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.seccion IS
    'Sección temática de la encuesta';

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.orden IS
    'Orden de presentación visual';

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.is_obligatoria IS
    'Indica si la respuesta es obligatoria';

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.is_activo IS
    'Estado lógico del registro';

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.created_at IS
    'Fecha de creación';

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.updated_at IS
    'Fecha de última actualización';

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.deleted_at IS
    'Fecha de eliminación lógica';

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.created_by IS
    'Usuario creador';

COMMENT ON COLUMN clientes.cat_encuesta_preguntas.updated_by IS
    'Usuario modificador';

ALTER TABLE clientes.cat_encuesta_preguntas
    ENABLE ROW LEVEL SECURITY;

CREATE POLICY rls_cat_encuesta_preguntas_empresa
    ON clientes.cat_encuesta_preguntas
    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 datetime
from typing import Optional

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

from app.db.base import Base


class CatEncuestaPreguntas(Base):
    """
    Preguntas de Encuestas Comerciales.

    Catálogo formal de preguntas y reactivos utilizados en encuestas
    de prospección, satisfacción y auditoría de punto de venta.

    Legacy:
        c061t_respuesta_pregunta_abierta / c062t_respuesta_pregunta_cerrada
    """

    __tablename__ = "cat_encuesta_preguntas"

    __table_args__ = (
        UniqueConstraint(
            "empresa_id",
            "id",
            name="uq_cat_encuesta_preguntas_empresa_id",
        ),
        UniqueConstraint(
            "empresa_id",
            "codigo",
            name="uq_cat_encuesta_preguntas_empresa_codigo",
        ),
        Index(
            "ix_cat_encuesta_preguntas_empresa_activo",
            "empresa_id",
            "is_activo",
        ),
        Index(
            "ix_cat_encuesta_preguntas_seccion",
            "empresa_id",
            "seccion",
            "orden",
        ),
        Index(
            "ix_cat_encuesta_preguntas_empresa_tipo_respuesta",
            "empresa_id",
            "tipo_respuesta_id",
        ),
        ForeignKeyConstraint(
            ["empresa_id", "tipo_respuesta_id"],
            ["clientes.cat_tipos_respuesta_encuesta.empresa_id", "clientes.cat_tipos_respuesta_encuesta.id"],
            name="fk_cat_encuesta_preguntas_tipo_respuesta",
        ),
        {
            "schema": "clientes",
            "comment": "Preguntas de Encuestas Comerciales",
        },
    )

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

    codigo: Mapped[str] = mapped_column(
        String(30),
        nullable=False,
        comment="Código único mnemotécnico de la pregunta",
    )

    enunciado: Mapped[str] = mapped_column(
        Text,
        nullable=False,
        comment="Texto del reactivo o pregunta formulada",
    )

    tipo_respuesta_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        nullable=False,
        comment="Tipo de respuesta (FK cat_tipos_respuesta_encuesta)",
    )

    seccion: Mapped[str] = mapped_column(
        String(50),
        nullable=False,
        server_default=text("'GENERAL'"),
        comment="Sección temática de la encuesta",
    )

    orden: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
        server_default=text("1"),
        comment="Orden de presentación visual",
    )

    is_obligatoria: Mapped[bool] = mapped_column(
        Boolean,
        nullable=False,
        server_default=text("false"),
        comment="Indica si la respuesta es obligatoria",
    )

    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 cat_encuesta_preguntas

Revision ID: 0014
Revises: 0013
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 = "0014"
down_revision: Union[str, Sequence[str], None] = "0013"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:

    op.create_table(
        "cat_encuesta_preguntas",

        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(
            "codigo",
            sa.String(length=30),
            nullable=False,
            comment="Código único mnemotécnico de la pregunta",
        ),

        sa.Column(
            "enunciado",
            sa.Text(),
            nullable=False,
            comment="Texto del reactivo o pregunta formulada",
        ),

        sa.Column(
            "tipo_respuesta_id",
            postgresql.UUID(as_uuid=True),
            nullable=False,
            comment="Tipo de respuesta (FK cat_tipos_respuesta_encuesta)",
        ),

        sa.Column(
            "seccion",
            sa.String(length=50),
            nullable=False,
            server_default=sa.text("'GENERAL'"),
            comment="Sección temática de la encuesta",
        ),

        sa.Column(
            "orden",
            sa.Integer(),
            nullable=False,
            server_default=sa.text("1"),
            comment="Orden de presentación visual",
        ),

        sa.Column(
            "is_obligatoria",
            sa.Boolean(),
            nullable=False,
            server_default=sa.text("false"),
            comment="Indica si la respuesta es obligatoria",
        ),

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

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

        sa.UniqueConstraint(
            "empresa_id",
            "codigo",
            name="uq_cat_encuesta_preguntas_empresa_codigo",
        ),
        sa.ForeignKeyConstraint(
            ["empresa_id", "tipo_respuesta_id"],
            ["clientes.cat_tipos_respuesta_encuesta.empresa_id", "clientes.cat_tipos_respuesta_encuesta.id"],
            name="fk_cat_encuesta_preguntas_tipo_respuesta",
        ),

        comment="Preguntas de Encuestas Comerciales",
        schema="clientes",
    )

    op.create_index(
        "ix_cat_encuesta_preguntas_empresa_activo",
        "cat_encuesta_preguntas",
        ["empresa_id", "is_activo"],
        unique=False,
        schema="clientes",
    )

    op.create_index(
        "ix_cat_encuesta_preguntas_seccion",
        "cat_encuesta_preguntas",
        ["empresa_id", "seccion", "orden"],
        unique=False,
        schema="clientes",
    )

    op.execute(
        """
        ALTER TABLE clientes.cat_encuesta_preguntas
        ENABLE ROW LEVEL SECURITY;
        """
    )

    op.execute(
        """
        CREATE POLICY rls_cat_encuesta_preguntas_empresa
        ON clientes.cat_encuesta_preguntas
        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_cat_encuesta_preguntas_empresa
        ON clientes.cat_encuesta_preguntas;
        """
    )

    op.execute(
        """
        ALTER TABLE clientes.cat_encuesta_preguntas
        DISABLE ROW LEVEL SECURITY;
        """
    )

    op.drop_index(
        "ix_cat_encuesta_preguntas_seccion",
        table_name="cat_encuesta_preguntas",
        schema="clientes",
    )

    op.drop_index(
        "ix_cat_encuesta_preguntas_empresa_activo",
        table_name="cat_encuesta_preguntas",
        schema="clientes",
    )

    op.drop_table(
        "cat_encuesta_preguntas",
        schema="clientes",
    )

Especificación Técnica: cat_encuesta_preguntas¶

1. Justificación y Mejoras de Arquitectura¶

2. Definiciones de Implementación¶

Especificación Técnica: `cat_encuesta_preguntas`¶