"""keywords

Revision ID: 333652ce4b5e
Revises: 88be4993b9b1
Create Date: 2026-08-31 17:10:01.198811

"""

from collections.abc import Sequence

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

# revision identifiers, used by Alembic.
revision: str = "333652ce4b5e"
down_revision: str | Sequence[str] | None = "88be4993b9b1"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


# --- consolidation : fonction + vue `keyword_metrics_latest` (spec §3.3) ---------------
#
# **Une seule implémentation SQL de la règle** : le service API appelle la fonction avec
# `org_settings.preferred_source`, la vue l'appelle avec NULL pour la Matrice et les
# requêtes ad hoc. Jamais de seconde logique, ni côté Python ni côté web.
#
# Règle, par mot-clé et **par champ** (décision n°3.d) :
#   1. si `preferred_source` a une valeur non nulle pour ce champ, elle gagne — **même
#      plus ancienne** que celle d'une autre source ;
#   2. sinon, repli : la valeur non nulle la **plus récente** ;
#   3. tie-break : priorité de source **propre à la métrique** (CASE ci-dessous), puis le
#      nom de la source et l'id, pour que le résultat soit totalement déterministe.
# Une source trouée ne masque donc jamais la valeur d'une autre : chaque champ est choisi
# indépendamment, et la provenance (source + fetched_at) est rendue avec la valeur.
#
# Forme : une seule instruction SELECT, sans CTE ni DISTINCT au niveau supérieur, et des
# LEFT JOIN LATERAL — c'est ce qui permet au planificateur d'inliner la fonction et de
# pousser un filtre sur `keyword_id` jusque dans les sous-requêtes (l'écran réservoir
# n'affiche jamais qu'une page de mots-clés). Chaque latéral se sert du btree de
# `uq_keyword_metrics_keyword_source_fetched`, dont `keyword_id` est la colonne de tête.
#
# Priorités v1 par métrique — défauts d'affichage, pas une vérité (cas résiduel §10.2) :
#   volume, cpc → DataForSEO > Semrush > autres ;  kd → Semrush > DataForSEO > autres ;
#   intent, serp_features, trend → Semrush > SE Ranking > autres ;  generic_csv en dernier.
_PRIORITY_VOLUME_CPC = """CASE m.source
                WHEN 'dataforseo' THEN 1
                WHEN 'semrush_positions' THEN 2
                WHEN 'semrush_magic' THEN 2
                WHEN 'generic_csv' THEN 99
                ELSE 50 END"""
_PRIORITY_KD = """CASE m.source
                WHEN 'semrush_magic' THEN 1
                WHEN 'semrush_positions' THEN 1
                WHEN 'dataforseo' THEN 2
                WHEN 'generic_csv' THEN 99
                ELSE 50 END"""
_PRIORITY_QUALITATIVE = """CASE m.source
                WHEN 'semrush_magic' THEN 1
                WHEN 'semrush_positions' THEN 1
                WHEN 'se_ranking' THEN 2
                WHEN 'generic_csv' THEN 99
                ELSE 50 END"""


def _pick(column: str, priority: str) -> str:
    """Le latéral d'un champ : la valeur retenue, sa source et sa fraîcheur.

    L'alias est suffixé (`volume_pick`, jamais `volume`) : un alias de relation homonyme
    d'un paramètre de sortie de la fonction rendrait la référence ambiguë.
    """
    return f"""
    LEFT JOIN LATERAL (
        SELECT m.{column} AS value, m.source, m.fetched_at
        FROM keyword_metrics m
        WHERE m.keyword_id = b.keyword_id AND m.{column} IS NOT NULL
        ORDER BY (m.source IS NOT DISTINCT FROM preferred_source) DESC,
                 m.fetched_at DESC,
                 {priority},
                 m.source,
                 m.id DESC
        LIMIT 1
    ) {column}_pick ON true"""


KEYWORD_METRICS_LATEST_FUNCTION = f"""
CREATE FUNCTION keyword_metrics_latest(preferred_source text DEFAULT NULL)
RETURNS TABLE (
    keyword_id uuid,
    org_id text,
    volume integer,
    volume_source text,
    volume_fetched_at timestamptz,
    kd numeric,
    kd_source text,
    kd_fetched_at timestamptz,
    cpc numeric,
    cpc_source text,
    cpc_fetched_at timestamptz,
    intent text,
    intent_source text,
    intent_fetched_at timestamptz,
    serp_features jsonb,
    serp_features_source text,
    serp_features_fetched_at timestamptz,
    trend smallint[],
    trend_source text,
    trend_fetched_at timestamptz,
    last_updated timestamptz
)
LANGUAGE sql
STABLE
AS $$
    SELECT b.keyword_id,
           b.org_id,
           volume_pick.value, volume_pick.source, volume_pick.fetched_at,
           kd_pick.value, kd_pick.source, kd_pick.fetched_at,
           cpc_pick.value, cpc_pick.source, cpc_pick.fetched_at,
           intent_pick.value, intent_pick.source, intent_pick.fetched_at,
           serp_features_pick.value, serp_features_pick.source, serp_features_pick.fetched_at,
           trend_pick.value, trend_pick.source, trend_pick.fetched_at,
           b.last_updated
    FROM (
        SELECT m.keyword_id, min(m.org_id) AS org_id, max(m.fetched_at) AS last_updated
        FROM keyword_metrics m
        GROUP BY m.keyword_id
    ) b{_pick("volume", _PRIORITY_VOLUME_CPC)}{_pick("kd", _PRIORITY_KD)}{
    _pick("cpc", _PRIORITY_VOLUME_CPC)
}{_pick("intent", _PRIORITY_QUALITATIVE)}{_pick("serp_features", _PRIORITY_QUALITATIVE)}{
    _pick("trend", _PRIORITY_QUALITATIVE)
}
$$
"""

# Confort de lecture (Matrice, requêtes ad hoc) : la consolidation par défaut, sans source
# préférée. Une vue et une fonction peuvent porter le même nom — catalogues distincts.
KEYWORD_METRICS_LATEST_VIEW = """
CREATE VIEW keyword_metrics_latest AS SELECT * FROM keyword_metrics_latest(NULL::text)
"""


def upgrade() -> None:
    """Upgrade schema."""
    # --- opérations manuelles (l'autogénération ne les produit pas, spec §3.13) ---------
    # `pg_trgm` : index GIN de repli flou sur `keywords.keyword`. Extension « trusted »,
    # le propriétaire de la base la crée lui-même — cette ligne est donc auto-suffisante.
    # `vector` : type `halfvec(1024)` de `keywords.embedding`. **Non trusted** : elle doit
    # être installée par un superutilisateur au préalable (`scripts/db-init.sh` la pose
    # dans le schéma `extensions`, hors du `public` que `app.db_reset` recrée). L'appel
    # ci-dessous est alors un no-op ; sur une base non préparée, il échoue ici, tôt et
    # lisiblement, plutôt qu'à la création de la table.
    op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")
    op.execute("CREATE EXTENSION IF NOT EXISTS vector")
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        "keywords",
        sa.Column("id", sa.UUID(), server_default=sa.text("uuidv7()"), nullable=False),
        sa.Column("org_id", sa.Text(), nullable=False),
        sa.Column("locale", sa.Text(), nullable=False),
        sa.Column("keyword", sa.Text(), nullable=False),
        sa.Column(
            "keyword_tsv",
            postgresql.TSVECTOR(),
            sa.Computed("to_tsvector('simple', keyword)", persisted=True),
            nullable=False,
        ),
        sa.Column("embedding", pgvector.sqlalchemy.halfvec.HALFVEC(dim=1024), nullable=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.CheckConstraint("locale ~ '^[a-z]{2}-[A-Z]{2}$'", name="ck_keywords_locale"),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("org_id", "locale", "keyword", name="uq_keywords_org_locale_keyword"),
    )
    op.create_index(
        "ix_keywords_keyword_trgm",
        "keywords",
        ["keyword"],
        unique=False,
        postgresql_using="gin",
        postgresql_ops={"keyword": "gin_trgm_ops"},
    )
    op.create_index(
        "ix_keywords_tsv", "keywords", ["keyword_tsv"], unique=False, postgresql_using="gin"
    )
    op.create_table(
        "org_settings",
        sa.Column("org_id", sa.Text(), nullable=False),
        sa.Column("preferred_source", sa.Text(), nullable=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.PrimaryKeyConstraint("org_id"),
    )
    op.create_table(
        "tags",
        sa.Column("id", sa.UUID(), server_default=sa.text("uuidv7()"), nullable=False),
        sa.Column("org_id", sa.Text(), nullable=False),
        sa.Column("name", sa.Text(), nullable=False),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("org_id", "name", name="uq_tags_org_name"),
    )
    op.create_table(
        "keyword_metrics",
        sa.Column("id", sa.UUID(), server_default=sa.text("uuidv7()"), nullable=False),
        sa.Column("org_id", sa.Text(), nullable=False),
        sa.Column("keyword_id", sa.UUID(), nullable=False),
        sa.Column("source", sa.Text(), nullable=False),
        sa.Column("fetched_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("volume", sa.Integer(), nullable=True),
        sa.Column("kd", sa.Numeric(), nullable=True),
        sa.Column("cpc", sa.Numeric(), nullable=True),
        sa.Column("intent", sa.Text(), nullable=True),
        sa.Column("serp_features", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
        sa.Column("trend", postgresql.ARRAY(sa.SmallInteger()), nullable=True),
        sa.Column("raw", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
        sa.ForeignKeyConstraint(
            ["keyword_id"],
            ["keywords.id"],
            name="fk_keyword_metrics_keyword_id",
            ondelete="CASCADE",
        ),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint(
            "keyword_id", "source", "fetched_at", name="uq_keyword_metrics_keyword_source_fetched"
        ),
    )
    op.create_index(op.f("ix_keyword_metrics_org_id"), "keyword_metrics", ["org_id"], unique=False)
    op.create_table(
        "keyword_tags",
        sa.Column("keyword_id", sa.UUID(), nullable=False),
        sa.Column("tag_id", sa.UUID(), nullable=False),
        sa.Column("org_id", sa.Text(), nullable=False),
        sa.ForeignKeyConstraint(
            ["keyword_id"], ["keywords.id"], name="fk_keyword_tags_keyword_id", ondelete="CASCADE"
        ),
        sa.ForeignKeyConstraint(
            ["tag_id"], ["tags.id"], name="fk_keyword_tags_tag_id", ondelete="CASCADE"
        ),
        sa.PrimaryKeyConstraint("keyword_id", "tag_id"),
    )
    op.create_index(op.f("ix_keyword_tags_org_id"), "keyword_tags", ["org_id"], unique=False)
    op.create_index("ix_keyword_tags_tag_id", "keyword_tags", ["tag_id"], unique=False)
    op.create_table(
        "keyword_lists",
        sa.Column("id", sa.UUID(), server_default=sa.text("uuidv7()"), nullable=False),
        sa.Column("org_id", sa.Text(), nullable=False),
        sa.Column("site_target_id", sa.UUID(), nullable=False),
        sa.Column("name", sa.Text(), nullable=False),
        sa.Column("source", sa.Text(), nullable=True),
        sa.Column("created_from_import_id", sa.UUID(), nullable=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        # `fk_keyword_lists_created_from_import_id` est posée plus bas par un ALTER TABLE :
        # `imports` référence `keyword_lists` et réciproquement (§3.4 / §3.11), la table
        # cible n'existe pas encore ici. Le modèle la déclare avec `use_alter=True`, mais
        # `op.create_table` ne l'émettrait pas — elle serait silencieusement perdue.
        sa.ForeignKeyConstraint(
            ["site_target_id"],
            ["site_targets.id"],
            name="fk_keyword_lists_site_target_id",
            ondelete="CASCADE",
        ),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("site_target_id", "name", name="uq_keyword_lists_target_name"),
    )
    op.create_index(op.f("ix_keyword_lists_org_id"), "keyword_lists", ["org_id"], unique=False)
    op.create_table(
        "keyword_page_map",
        sa.Column("id", sa.UUID(), server_default=sa.text("uuidv7()"), nullable=False),
        sa.Column("org_id", sa.Text(), nullable=False),
        sa.Column("site_target_id", sa.UUID(), nullable=False),
        sa.Column("keyword_id", sa.UUID(), nullable=False),
        sa.Column("url_normalized", sa.Text(), nullable=False),
        sa.Column("page_id", sa.UUID(), nullable=True),
        sa.Column("source", sa.Text(), nullable=False),
        sa.Column("position", sa.Numeric(), nullable=True),
        sa.Column("seen_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.CheckConstraint(
            "source IN ('import_url', 'manual', 'gsc')", name="ck_keyword_page_map_source"
        ),
        sa.ForeignKeyConstraint(
            ["keyword_id"],
            ["keywords.id"],
            name="fk_keyword_page_map_keyword_id",
            ondelete="CASCADE",
        ),
        sa.ForeignKeyConstraint(
            ["site_target_id"],
            ["site_targets.id"],
            name="fk_keyword_page_map_site_target_id",
            ondelete="CASCADE",
        ),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint(
            "site_target_id",
            "keyword_id",
            "url_normalized",
            "source",
            name="uq_keyword_page_map_target_keyword_url_source",
        ),
    )
    op.create_index(
        "ix_keyword_page_map_keyword_id", "keyword_page_map", ["keyword_id"], unique=False
    )
    op.create_index(
        op.f("ix_keyword_page_map_org_id"), "keyword_page_map", ["org_id"], unique=False
    )
    op.create_index(
        "ix_keyword_page_map_pending",
        "keyword_page_map",
        ["site_target_id", "url_normalized"],
        unique=False,
        postgresql_where=sa.text("page_id IS NULL"),
    )
    op.create_index(
        "uq_keyword_page_map_manual",
        "keyword_page_map",
        ["site_target_id", "keyword_id"],
        unique=True,
        postgresql_where=sa.text("source = 'manual'"),
    )
    op.create_table(
        "imports",
        sa.Column("id", sa.UUID(), server_default=sa.text("uuidv7()"), nullable=False),
        sa.Column("org_id", sa.Text(), nullable=False),
        sa.Column("site_target_id", sa.UUID(), nullable=False),
        sa.Column("source", sa.Text(), nullable=False),
        sa.Column("kind", sa.Text(), nullable=False),
        sa.Column("filename", sa.Text(), nullable=True),
        sa.Column("file_sha256", sa.Text(), nullable=True),
        sa.Column("detected_format", sa.Text(), nullable=True),
        sa.Column("detected_country", sa.Text(), nullable=True),
        sa.Column("fetched_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("keyword_list_id", sa.UUID(), nullable=True),
        sa.Column("job_id", sa.UUID(), nullable=False),
        sa.Column("status", sa.Text(), nullable=False),
        sa.Column("error", sa.Text(), nullable=True),
        sa.Column("report", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.CheckConstraint("kind IN ('csv', 'api')", name="ck_imports_kind"),
        sa.CheckConstraint(
            "status IN ('queued', 'running', 'done', 'failed')", name="ck_imports_status"
        ),
        sa.ForeignKeyConstraint(["job_id"], ["jobs.id"], name="fk_imports_job_id"),
        sa.ForeignKeyConstraint(
            ["keyword_list_id"],
            ["keyword_lists.id"],
            name="fk_imports_keyword_list_id",
            ondelete="SET NULL",
        ),
        sa.ForeignKeyConstraint(
            ["site_target_id"],
            ["site_targets.id"],
            name="fk_imports_site_target_id",
            ondelete="CASCADE",
        ),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index(op.f("ix_imports_org_id"), "imports", ["org_id"], unique=False)
    op.create_index(
        "ix_imports_target_sha256", "imports", ["site_target_id", "file_sha256"], unique=False
    )
    op.create_table(
        "keyword_list_items",
        sa.Column("list_id", sa.UUID(), nullable=False),
        sa.Column("keyword_id", sa.UUID(), nullable=False),
        sa.Column("org_id", sa.Text(), nullable=False),
        sa.Column(
            "added_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False
        ),
        sa.ForeignKeyConstraint(
            ["keyword_id"],
            ["keywords.id"],
            name="fk_keyword_list_items_keyword_id",
            ondelete="CASCADE",
        ),
        sa.ForeignKeyConstraint(
            ["list_id"],
            ["keyword_lists.id"],
            name="fk_keyword_list_items_list_id",
            ondelete="CASCADE",
        ),
        sa.PrimaryKeyConstraint("list_id", "keyword_id"),
    )
    op.create_index(
        "ix_keyword_list_items_keyword_id", "keyword_list_items", ["keyword_id"], unique=False
    )
    op.create_index(
        op.f("ix_keyword_list_items_org_id"), "keyword_list_items", ["org_id"], unique=False
    )
    # ### end Alembic commands ###

    # --- opérations manuelles (suite) --------------------------------------------------
    # La moitié « différée » du cycle keyword_lists ↔ imports, maintenant que les deux
    # tables existent. SET NULL : supprimer un import ne supprime jamais la liste qu'il a
    # créée — les mots-clés importés restent dans le réservoir de la cible.
    op.create_foreign_key(
        "fk_keyword_lists_created_from_import_id",
        "keyword_lists",
        "imports",
        ["created_from_import_id"],
        ["id"],
        ondelete="SET NULL",
    )
    op.execute(KEYWORD_METRICS_LATEST_FUNCTION)
    op.execute(KEYWORD_METRICS_LATEST_VIEW)


def downgrade() -> None:
    """Downgrade schema."""
    # --- symétrique des opérations manuelles ------------------------------------------
    # La vue d'abord : elle dépend de la fonction.
    op.execute("DROP VIEW IF EXISTS keyword_metrics_latest")
    op.execute("DROP FUNCTION IF EXISTS keyword_metrics_latest(text)")
    # Le cycle keyword_lists ↔ imports empêcherait le DROP TABLE dans l'ordre inverse.
    op.drop_constraint(
        "fk_keyword_lists_created_from_import_id", "keyword_lists", type_="foreignkey"
    )
    # **Aucun DROP EXTENSION ici** : `pg_trgm` et `vector` sont des objets de base de
    # données, pas de cette migration — d'autres schémas ou d'autres migrations peuvent en
    # dépendre, et un downgrade ne doit jamais casser ce qu'il n'a pas créé.
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index(op.f("ix_keyword_list_items_org_id"), table_name="keyword_list_items")
    op.drop_index("ix_keyword_list_items_keyword_id", table_name="keyword_list_items")
    op.drop_table("keyword_list_items")
    op.drop_index("ix_imports_target_sha256", table_name="imports")
    op.drop_index(op.f("ix_imports_org_id"), table_name="imports")
    op.drop_table("imports")
    op.drop_index(
        "uq_keyword_page_map_manual",
        table_name="keyword_page_map",
        postgresql_where=sa.text("source = 'manual'"),
    )
    op.drop_index(
        "ix_keyword_page_map_pending",
        table_name="keyword_page_map",
        postgresql_where=sa.text("page_id IS NULL"),
    )
    op.drop_index(op.f("ix_keyword_page_map_org_id"), table_name="keyword_page_map")
    op.drop_index("ix_keyword_page_map_keyword_id", table_name="keyword_page_map")
    op.drop_table("keyword_page_map")
    op.drop_index(op.f("ix_keyword_lists_org_id"), table_name="keyword_lists")
    op.drop_table("keyword_lists")
    op.drop_index("ix_keyword_tags_tag_id", table_name="keyword_tags")
    op.drop_index(op.f("ix_keyword_tags_org_id"), table_name="keyword_tags")
    op.drop_table("keyword_tags")
    op.drop_index(op.f("ix_keyword_metrics_org_id"), table_name="keyword_metrics")
    op.drop_table("keyword_metrics")
    op.drop_table("tags")
    op.drop_table("org_settings")
    op.drop_index("ix_keywords_tsv", table_name="keywords", postgresql_using="gin")
    op.drop_index(
        "ix_keywords_keyword_trgm",
        table_name="keywords",
        postgresql_using="gin",
        postgresql_ops={"keyword": "gin_trgm_ops"},
    )
    op.drop_table("keywords")
    # ### end Alembic commands ###