"""adopt the legacy video discovery tables into Alembic

Revision ID: 20260731_08a
Revises: 20260730_07
Create Date: 2026-07-31

These tables originally came from ``sql/video_discovery_tables.sql``.  Creating
their pre-gate shape here makes a fresh ``alembic upgrade head`` self-contained;
``checkfirst`` behavior keeps existing installations safe.
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import mysql

revision: str = "20260731_08a"
down_revision: str | None = "20260730_07"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def _tables() -> set[str]:
    return set(sa.inspect(op.get_bind()).get_table_names())


def _create_run() -> None:
    op.create_table(
        "video_discovery_run",
        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("run_id", sa.String(64), nullable=False),
        sa.Column("biz_dt", sa.String(32), nullable=True),
        sa.Column("demand_grade_id", sa.BigInteger(), nullable=True),
        sa.Column("demand_word", sa.String(256), nullable=False),
        sa.Column("seed_video_id", sa.String(64), nullable=True),
        sa.Column("seed_video_title", sa.String(512), nullable=True),
        sa.Column("relevant_points_json", sa.Text(), nullable=False),
        sa.Column("intent_summary", sa.Text(), nullable=True),
        sa.Column("status", sa.String(24), server_default="running", nullable=False),
        sa.Column("search_count", sa.Integer(), server_default="0", nullable=False),
        sa.Column("primary_count", sa.Integer(), server_default="0", nullable=False),
        sa.Column("stop_reason", sa.Text(), nullable=True),
        sa.Column("create_time", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
        sa.Column("update_time", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("run_id", name="uk_video_discovery_run_id"),
        sa.UniqueConstraint("biz_dt", "demand_grade_id", name="uk_video_discovery_run_biz_grade"),
        mysql_charset="utf8mb4",
        mysql_collate="utf8mb4_unicode_ci",
    )
    op.create_index("idx_video_discovery_run_demand", "video_discovery_run", ["demand_word"])
    op.create_index("idx_video_discovery_run_grade", "video_discovery_run", ["demand_grade_id"])
    op.create_index("idx_video_discovery_run_biz_dt", "video_discovery_run", ["biz_dt"])
    op.create_index("idx_video_discovery_run_status", "video_discovery_run", ["status"])
    op.execute("ALTER TABLE video_discovery_run MODIFY update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP")


def _create_search() -> None:
    op.create_table(
        "video_discovery_search",
        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("run_id", sa.String(64), nullable=False),
        sa.Column("search_key", sa.String(64), nullable=False),
        sa.Column("keyword", sa.String(256), nullable=False),
        sa.Column("query_reason", sa.Text(), nullable=False),
        sa.Column("source_type", sa.String(32), nullable=False),
        sa.Column("source_value", sa.Text(), nullable=True),
        sa.Column("parent_search_id", sa.BigInteger(), nullable=True),
        sa.Column("provider", sa.String(32), server_default="internal_keyword", nullable=False),
        sa.Column("provider_state_json", sa.Text(), nullable=True),
        sa.Column("content_type", sa.String(16), server_default="视频", nullable=False),
        sa.Column("sort_type", sa.String(32), server_default="综合排序", nullable=False),
        sa.Column("publish_time", sa.String(32), server_default="不限", nullable=False),
        sa.Column("cursor", sa.String(128), server_default="0", nullable=False),
        sa.Column("page_no", sa.Integer(), server_default="1", nullable=False),
        sa.Column("results_count", sa.Integer(), server_default="0", nullable=False),
        sa.Column("new_candidate_count", sa.Integer(), server_default="0", nullable=False),
        sa.Column("has_more", mysql.TINYINT(), server_default="0", nullable=False),
        sa.Column("next_cursor", sa.String(128), nullable=True),
        sa.Column("result_ids_json", sa.Text(), nullable=True),
        sa.Column("status", sa.String(16), server_default="success", nullable=False),
        sa.Column("error_message", sa.Text(), nullable=True),
        sa.Column("create_time", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
        sa.Column("update_time", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
        sa.PrimaryKeyConstraint("id"),
        mysql_charset="utf8mb4",
        mysql_collate="utf8mb4_unicode_ci",
    )
    op.create_index("idx_video_discovery_search_run", "video_discovery_search", ["run_id", "id"])
    op.create_index("idx_video_discovery_search_parent", "video_discovery_search", ["run_id", "parent_search_id"])
    op.create_index("idx_video_discovery_search_keyword", "video_discovery_search", ["keyword"])
    op.execute("ALTER TABLE video_discovery_search MODIFY update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP")


def _create_candidate() -> None:
    op.create_table(
        "video_discovery_candidate",
        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("run_id", sa.String(64), nullable=False),
        sa.Column("search_id", sa.BigInteger(), nullable=True),
        sa.Column("aweme_id", sa.String(64), nullable=False),
        sa.Column("title", sa.String(512), nullable=True),
        sa.Column("content_link", sa.String(1024), nullable=True),
        sa.Column("author_name", sa.String(256), nullable=True),
        sa.Column("author_sec_uid", sa.String(256), nullable=True),
        sa.Column("source_keywords_json", sa.Text(), nullable=True),
        sa.Column("source_search_ids_json", sa.Text(), nullable=True),
        sa.Column("tags_json", sa.Text(), nullable=True),
        sa.Column("play_count", sa.BigInteger(), nullable=True),
        sa.Column("like_count", sa.BigInteger(), nullable=True),
        sa.Column("comment_count", sa.BigInteger(), nullable=True),
        sa.Column("collect_count", sa.BigInteger(), nullable=True),
        sa.Column("share_count", sa.BigInteger(), nullable=True),
        sa.Column("content_age_evidence_json", sa.Text(), nullable=True),
        sa.Column("account_age_evidence_json", sa.Text(), nullable=True),
        sa.Column("age_normalization_json", sa.Text(), nullable=True),
        sa.Column("relevance_score", sa.Numeric(8, 6), nullable=True),
        sa.Column("elder_score", sa.Numeric(8, 6), nullable=True),
        sa.Column("share_score", sa.Numeric(8, 6), nullable=True),
        sa.Column("value_score", sa.Numeric(8, 2), nullable=True),
        sa.Column("decision_reason", sa.Text(), nullable=True),
        sa.Column("decision_bucket", sa.String(24), server_default="unreviewed", nullable=False),
        sa.Column("aigc_crawler_plan_id", sa.String(64), nullable=True),
        sa.Column("aigc_produce_plan_id", sa.String(64), nullable=True),
        sa.Column("aigc_publish_plan_id", sa.String(64), nullable=True),
        sa.Column("aigc_plan_label", sa.String(64), nullable=True),
        sa.Column("create_time", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
        sa.Column("update_time", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
        sa.PrimaryKeyConstraint("id"),
        mysql_charset="utf8mb4",
        mysql_collate="utf8mb4_unicode_ci",
    )
    op.create_index("idx_video_discovery_candidate_search", "video_discovery_candidate", ["search_id", "id"])
    op.create_index("idx_video_discovery_candidate_bucket", "video_discovery_candidate", ["run_id", "decision_bucket"])
    op.create_index("idx_video_discovery_candidate_author", "video_discovery_candidate", ["author_sec_uid"])
    op.execute("ALTER TABLE video_discovery_candidate MODIFY update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP")


def upgrade() -> None:
    tables = _tables()
    if "video_discovery_run" not in tables:
        _create_run()
    if "video_discovery_search" not in tables:
        _create_search()
    if "video_discovery_candidate" not in tables:
        _create_candidate()


def downgrade() -> None:
    # Adoption is forward-only: existing installations may predate Alembic.
    pass
