feature/seed-digitalizacion

backend/alembic/versions/d2e3f4a5b6c7_seed_document_types_digitization.py

@@ -0,0 +1,57 @@
+"""seed document_types_digitization
+
+Revision ID: d2e3f4a5b6c7
+Revises: f1a2b3c4d5e6
+Create Date: 2026-05-25
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from api.v1.modules.a76.doc_types_dig.seed import seed as doc_types_dig_seed
+
+revision: str = "d2e3f4a5b6c7"
+down_revision: Union[str, None] = "f1a2b3c4d5e6"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Quitar columna active (no se usa en ningún flujo)
+    op.drop_column('document_types_digitization', 'active', schema='a76')
+
+    def format_value(val):
+        if val is None or str(val).strip() == "" or str(val).upper() == "NONE":
+            return "NULL"
+        return f"'{str(val).replace(chr(39), chr(39)*2)}'"
+
+    bind = op.get_bind()
+    session = Session(bind=bind)
+    companies = session.execute(
+        text("SELECT tenant_id, id FROM a76.company WHERE deleted_at IS NULL")
+    ).fetchall()
+
+    for tenant_id, company_id in companies:
+        values = ", ".join([
+            f"({format_value(code)}, {format_value(description)}, {tenant_id}, {company_id})"
+            for code, description in doc_types_dig_seed
+        ])
+        if values:
+            session.execute(text(f"""
+                INSERT INTO a76.document_types_digitization (code, description, tenant_id, company_id)
+                VALUES {values}
+                ON CONFLICT (tenant_id, company_id, code) DO NOTHING;
+            """))
+    session.commit()
+
+
+def downgrade() -> None:
+    op.add_column(
+        'document_types_digitization',
+        sa.Column('active', sa.Boolean(), nullable=False, server_default=sa.text('true')),
+        schema='a76',
+    )