"""create the complete video discovery persistence tables

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

The tables predate Alembic and were previously installed from files under
``sql/``.  ``checkfirst``-style inspection keeps this migration safe for
databases where those legacy DDL files were already applied, while making a
fresh ``alembic upgrade head`` self-contained.
"""
from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

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 _table_names() -> set[str]:
    return set(sa.inspect(op.get_bind()).get_table_names())


def _create_run_table() -> None:
    op.create_table(
        "video_discovery_run",
        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("run_id", sa.String(length=64), nullable=False, comment="Agent 运行标识"),
        sa.Column("biz_dt", sa.String(length=32), nullable=True, comment="业务日 YYYYMMDD"),
        sa.Column("demand_grade_id", sa.BigInteger(), nullable=True),
        sa.Column("demand_word", sa.String(length=256), nullable=False),
        sa.Column("seed_video_id", sa.String(length=64), nullable=True),
        sa.Column("seed_video_title", sa.String(length=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(length=24),
            nullable=False,
            server_default="running",
        ),
        sa.Column("search_count", sa.Integer(), nullable=False, server_default="0"),
        sa.Column("primary_count", sa.Integer(), nullable=False, server_default="0"),
        sa.Column("stop_reason", sa.Text(), nullable=True),
        sa.Column(
            "create_time",
            sa.DateTime(),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.Column(
            "update_time",
            sa.DateTime(),
            nullable=False,
            server_default=sa.func.now(),
        ),
        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",
    )
    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"],
    )


def _create_search_table() -> None:
    op.create_table(
        "video_discovery_search",
        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("run_id", sa.String(length=64), nullable=False),
        sa.Column("search_key", sa.String(length=64), nullable=False),
        sa.Column("keyword", sa.String(length=256), nullable=False),
        sa.Column("query_reason", sa.Text(), nullable=False),
        sa.Column("source_type", sa.String(length=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(length=32),
            nullable=False,
            server_default="internal_keyword",
        ),
        sa.Column("provider_state_json", sa.Text(), nullable=True),
        sa.Column(
            "content_type",
            sa.String(length=16),
            nullable=False,
            server_default="视频",
        ),
        sa.Column(
            "sort_type",
            sa.String(length=32),
            nullable=False,
            server_default="综合排序",
        ),
        sa.Column(
            "publish_time",
            sa.String(length=32),
            nullable=False,
            server_default="不限",
        ),
        sa.Column(
            "cursor",
            sa.String(length=128),
            nullable=False,
            server_default="0",
        ),
        sa.Column("page_no", sa.Integer(), nullable=False, server_default="1"),
        sa.Column("results_count", sa.Integer(), nullable=False, server_default="0"),
        sa.Column(
            "new_candidate_count",
            sa.Integer(),
            nullable=False,
            server_default="0",
        ),
        sa.Column("has_more", sa.Integer(), nullable=False, server_default="0"),
        sa.Column("next_cursor", sa.String(length=128), nullable=True),
        sa.Column("result_ids_json", sa.Text(), nullable=True),
        sa.Column(
            "status",
            sa.String(length=16),
            nullable=False,
            server_default="success",
        ),
        sa.Column("error_message", sa.Text(), nullable=True),
        sa.Column(
            "create_time",
            sa.DateTime(),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.Column(
            "update_time",
            sa.DateTime(),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.PrimaryKeyConstraint("id"),
        mysql_charset="utf8mb4",
    )
    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"],
    )


def _create_candidate_table() -> None:
    op.create_table(
        "video_discovery_candidate",
        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("run_id", sa.String(length=64), nullable=False),
        sa.Column(
            "search_id",
            sa.BigInteger(),
            sa.ForeignKey(
                "video_discovery_search.id",
                name="fk_video_discovery_candidate_search",
                ondelete="RESTRICT",
            ),
            nullable=True,
        ),
        sa.Column("aweme_id", sa.String(length=64), nullable=False),
        sa.Column("title", sa.String(length=512), nullable=True),
        sa.Column("content_link", sa.String(length=1024), nullable=True),
        sa.Column("author_name", sa.String(length=256), nullable=True),
        sa.Column("author_sec_uid", sa.String(length=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(length=24),
            nullable=False,
            server_default="pending_evaluation",
        ),
        sa.Column("aigc_crawler_plan_id", sa.String(length=64), nullable=True),
        sa.Column("aigc_produce_plan_id", sa.String(length=64), nullable=True),
        sa.Column("aigc_publish_plan_id", sa.String(length=64), nullable=True),
        sa.Column("aigc_plan_label", sa.String(length=64), nullable=True),
        sa.Column(
            "create_time",
            sa.DateTime(),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.Column(
            "update_time",
            sa.DateTime(),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.PrimaryKeyConstraint("id"),
        mysql_charset="utf8mb4",
    )
    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"],
    )


def upgrade() -> None:
    tables = _table_names()
    if "video_discovery_run" not in tables:
        _create_run_table()
    if "video_discovery_search" not in tables:
        _create_search_table()
    if "video_discovery_candidate" not in tables:
        _create_candidate_table()


def downgrade() -> None:
    # These tables may have existed before Alembic took ownership of the schema.
    # A destructive automatic downgrade could delete legacy production data, so
    # this adoption migration is intentionally forward-only.
    pass