plantillas-proyectos/backend/alembic/env.py

import importlib.util
import logging
import os
import sys
from logging.config import fileConfig
from urllib.parse import quote_plus

from alembic import context
from alembic.operations import ops
from core.config import settings
from core.database import Base
from sqlalchemy import engine_from_config, pool
from sqlalchemy.engine.url import make_url

logger = logging.getLogger(__name__)

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config


def _strip_env_url(raw: str) -> str:
    """Quita espacios/comillas típicos de secretos CI (.env, Gitea)."""
    url = raw.strip().strip('"').strip("'")
    return url


def _normalize_alembic_sqlalchemy_url(url: str) -> str:
    """Alembic usa el driver síncrono psycopg2; normaliza DSN típicos de app/tests."""
    url = _strip_env_url(url)
    if url.startswith("postgresql+asyncpg://"):
        return url.replace("postgresql+asyncpg://", "postgresql+psycopg2://", 1)
    if url.startswith("postgresql+psycopg2://"):
        return url
    if url.startswith("postgresql://"):
        return url.replace("postgresql://", "postgresql+psycopg2://", 1)
    if url.startswith("postgres://"):
        return url.replace("postgres://", "postgresql+psycopg2://", 1)
    return url


def _validate_sqlalchemy_url(url: str, env_key: str) -> None:
    """Misma validación que create_engine; evita urlparse (falla con esquemas tipo postgresql+psycopg2)."""
    try:
        make_url(url)
    except Exception as e:
        raise RuntimeError(
            f"{env_key} no es una URL de SQLAlchemy válida. "
            "Ejemplo: postgresql://usuario:clave@127.0.0.1:5432/nombre_bd"
        ) from e


def _reject_documentation_placeholder_host(url: str, source: str) -> None:
    """
    Evita el error críptico de DNS: muchos ejemplos usan @host:5432 como texto literal.
    """
    try:
        parsed = make_url(url)
    except Exception:
        return
    h = (parsed.host or "").strip().lower()
    if h == "host":
        raise RuntimeError(
            f"{source}: el hostname \"host\" es un placeholder de documentación, no un servidor real. "
            "Usa el host alcanzable desde el runner (IP, nombre DNS, servicio en docker-compose, "
            "o host.docker.internal si act corre en contenedor y Postgres en tu máquina)."
        )


def get_database_url():
    """Obtiene la URL de la base de datos (PostgreSQL) desde variables de entorno o alembic.ini."""
    # CI / pytest: misma URL que los tests (secret TEST_DATABASE_URL) o DATABASE_URL explícita.
    # CRÍTICO: debe ser tupla con coma final si un solo elemento: ("X",) — si no, ("X") es str y el for
    # itera caracteres; env_key "_" + os.environ["_"] (común en shells) rompe con URL inválida.
    for env_key in ("TEST_DATABASE_URL", "DATABASE_URL"):
        raw = os.environ.get(env_key)
        if raw and raw.strip():
            normalized = _normalize_alembic_sqlalchemy_url(raw)
            _validate_sqlalchemy_url(normalized, env_key)
            _reject_documentation_placeholder_host(normalized, env_key)
            return normalized

    # Construcción desde settings (CORE_DB_* en .env / entorno)
    host = settings.CORE_DB_HOST
    db = settings.CORE_DB_NAME
    user = settings.CORE_DB_USER
    password = settings.CORE_DB_PASSWORD
    port = settings.CORE_DB_PORT

    if host and db and user and password:
        try:
            encoded_user = quote_plus(user)
            encoded_password = quote_plus(password)
            encoded_db = quote_plus(db)
            built = f"postgresql+psycopg2://{encoded_user}:{encoded_password}@{host}:{port}/{encoded_db}"
            _reject_documentation_placeholder_host(built, "CORE_DB_HOST")
            return built
        except Exception as e:
            logger.error(f"Error al construir URL: {e}")

    # Fallback al archivo de configuración
    url = config.get_main_option("sqlalchemy.url")
    if not url or "${" in url or "%(" in url:
        raise RuntimeError(
            "No se ha configurado la cadena de conexión a PostgreSQL. "
            "Define TEST_DATABASE_URL o DATABASE_URL, o variables CORE_DB_*; "
            "sqlalchemy.url en alembic.ini con placeholders ${...} no está soportado."
        )

    _reject_documentation_placeholder_host(url, "alembic.ini sqlalchemy.url")
    return url


# Configurar la URL de la base de datos
database_url = get_database_url()

# Debug: mostrar la URL (sin la contraseña)
if os.environ.get("ALEMBIC_DEBUG"):
    # Ocultar la contraseña para el debug en la URL
    try:
        before, after = database_url.split("@", 1)
        if ":" in before:
            before = before.split(":", 1)[0] + ":***"
        debug_url = before + "@" + after
    except Exception:
        debug_url = "postgresql://***:***@***"
        logger.error("Error al ocultar la contraseña en la URL para debug.")

config.set_main_option("sqlalchemy.url", database_url)

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# Ajusta la ruta para que puedas importar core y módulos
BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, BASE_DIR)

# Configuración de Alembic
config = context.config
fileConfig(config.config_file_name)
target_metadata = Base.metadata


def include_object(object_, name, type_, reflected, compare_to):
    """
    Keep all objects in autogenerate.
    FK noise is cleaned in process_revision_directives.
    """
    return True


def _fk_drop_signature(op_):
    if not isinstance(op_, ops.DropConstraintOp):
        return None
    if getattr(op_, "constraint_type", None) != "foreignkey":
        return None
    return (
        getattr(op_, "schema", None),
        getattr(op_, "table_name", None),
        getattr(op_, "constraint_name", None),
    )


def _fk_create_signature(op_):
    if not isinstance(op_, ops.CreateForeignKeyOp):
        return None
    local_cols = tuple(getattr(op_, "local_cols", ()) or ())
    remote_cols = tuple(getattr(op_, "remote_cols", ()) or ())
    return (
        getattr(op_, "source_schema", None),
        getattr(op_, "source_table", None),
        getattr(op_, "referent_schema", None),
        getattr(op_, "referent_table", None),
        local_cols,
        remote_cols,
    )


def _drop_to_create_match(drop_op, create_op):
    if not isinstance(drop_op, ops.DropConstraintOp):
        return False
    if not isinstance(create_op, ops.CreateForeignKeyOp):
        return False
    if getattr(drop_op, "constraint_type", None) != "foreignkey":
        return False

    def _normalize_schema(value):
        # PostgreSQL reports default schema inconsistently as None/public.
        return "public" if value in (None, "") else value

    # Prefer structural comparison using Alembic's reverse op when available.
    reverse_create = getattr(drop_op, "_reverse", None)
    if isinstance(reverse_create, ops.CreateForeignKeyOp):
        return (
            _normalize_schema(getattr(reverse_create, "source_schema", None))
            == _normalize_schema(getattr(create_op, "source_schema", None))
            and getattr(reverse_create, "source_table", None) == getattr(create_op, "source_table", None)
            and _normalize_schema(getattr(reverse_create, "referent_schema", None))
            == _normalize_schema(getattr(create_op, "referent_schema", None))
            and getattr(reverse_create, "referent_table", None) == getattr(create_op, "referent_table", None)
            and tuple(getattr(reverse_create, "local_cols", ()) or ())
            == tuple(getattr(create_op, "local_cols", ()) or ())
            and tuple(getattr(reverse_create, "remote_cols", ()) or ())
            == tuple(getattr(create_op, "remote_cols", ()) or ())
        )

    # Fallback for older op payloads: compare source table/schema and name.
    return (
        _normalize_schema(getattr(drop_op, "schema", None)) == _normalize_schema(getattr(create_op, "source_schema", None))
        and getattr(drop_op, "table_name", None) == getattr(create_op, "source_table", None)
        and getattr(drop_op, "constraint_name", None) == getattr(create_op, "constraint_name", None)
    )


def _prune_fk_churn(container):
    if not hasattr(container, "ops"):
        return

    # First recurse into nested containers.
    for op_ in list(container.ops):
        _prune_fk_churn(op_)

    table_ops = container.ops
    kept_ops = []
    consumed_indexes = set()

    for i, op_i in enumerate(table_ops):
        if i in consumed_indexes:
            continue

        if isinstance(op_i, ops.DropConstraintOp) and getattr(op_i, "constraint_type", None) == "foreignkey":
            matched_j = None
            for j in range(i + 1, len(table_ops)):
                if j in consumed_indexes:
                    continue
                op_j = table_ops[j]
                if _drop_to_create_match(op_i, op_j):
                    matched_j = j
                    break
            if matched_j is not None:
                # Drop + recreate same FK detected; remove both.
                consumed_indexes.add(i)
                consumed_indexes.add(matched_j)
                continue

        kept_ops.append(op_i)

    container.ops = kept_ops


def process_revision_directives(context_, revision, directives):
    """
    Remove autogenerate noise where Alembic emits drop/create for equivalent FKs.
    Real FK changes are preserved.
    """
    if not directives:
        return
    script = directives[0]
    _prune_fk_churn(script.upgrade_ops)
    _prune_fk_churn(script.downgrade_ops)


def import_models_from_dir(dir_path: str):
    """Importa recursivamente cualquier archivo models.py desde dir_path y archivos en directorios models/"""
    import sys

    def _load_module(module_path: str):
        rel_path = os.path.relpath(module_path, BASE_DIR)
        module_name = rel_path.replace(os.sep, ".").replace(".py", "")
        # Skip if already loaded to avoid duplicate SQLAlchemy table registrations
        if module_name in sys.modules:
            return
        spec = importlib.util.spec_from_file_location(module_name, module_path)
        mod = importlib.util.module_from_spec(spec)
        # Register in sys.modules before exec so transitive imports resolve correctly
        sys.modules[module_name] = mod
        spec.loader.exec_module(mod)

    for root, dirs, files in os.walk(dir_path):
        # Importar archivos models.py directos
        if "models.py" in files:
            try:
                _load_module(os.path.join(root, "models.py"))
            except Exception as e:
                logger.warning(f"No se pudo importar {os.path.join(root, 'models.py')}: {e}")

        # Importar todos los archivos .py en directorios llamados "models"
        if os.path.basename(root) == "models":
            for file in files:
                if file.endswith(".py") and not file.startswith("__"):
                    try:
                        _load_module(os.path.join(root, file))
                    except Exception as e:
                        logger.warning(f"No se pudo importar {os.path.join(root, file)}: {e}")


# Importar todos los models dentro de api/v1/modules y api/v1/modules/uploads
modules_dir = os.path.join(BASE_DIR, "api", "v1", "modules")
import_models_from_dir(modules_dir)

# Tablas declaradas fuera de models.py / carpeta models/ (autogenerate)
# Agrega aquí imports de models que no estén en archivos models.py estándar.


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.
    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        compare_type=True,
        include_schemas=True,
        include_object=include_object,
        process_revision_directives=process_revision_directives,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.
    """
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
            include_schemas=True,
            include_object=include_object,
            process_revision_directives=process_revision_directives,
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()