修改数据库版本更新

alembic/versions/20260727_01_pipeline_control_plane.py

@@ -1,219 +0,0 @@
-"""create durable pipeline control plane
-
-Revision ID: 20260727_01
-Revises:
-Create Date: 2026-07-27
-"""
-from __future__ import annotations
-
-from collections.abc import Sequence
-from datetime import datetime
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "20260727_01"
-down_revision: str | None = None
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "pipeline_run",
-        sa.Column("run_id", sa.String(36), primary_key=True),
-        sa.Column("dedupe_key", sa.String(191), nullable=False),
-        sa.Column("pipeline_key", sa.String(64), nullable=False),
-        sa.Column("biz_dt", sa.String(8), nullable=False),
-        sa.Column("trigger_type", sa.String(24), nullable=False),
-        sa.Column("trigger_source", sa.String(64), nullable=True),
-        sa.Column("triggered_by", sa.String(128), nullable=True),
-        sa.Column("trigger_reason", sa.Text(), nullable=True),
-        sa.Column("parent_run_id", sa.String(36), nullable=True),
-        sa.Column("run_mode", sa.String(24), nullable=False),
-        sa.Column("dry_run", sa.Boolean(), nullable=False),
-        sa.Column("status", sa.String(32), nullable=False),
-        sa.Column("current_step", sa.String(64), nullable=True),
-        sa.Column("scheduled_for", sa.DateTime(), nullable=True),
-        sa.Column("deadline_at", sa.DateTime(), nullable=False),
-        sa.Column("started_at", sa.DateTime(), nullable=True),
-        sa.Column("finished_at", sa.DateTime(), nullable=True),
-        sa.Column("heartbeat_at", sa.DateTime(), nullable=True),
-        sa.Column("lease_owner", sa.String(191), nullable=True),
-        sa.Column("lease_until", sa.DateTime(), nullable=True),
-        sa.Column("code_version", sa.String(128), nullable=True),
-        sa.Column("config_snapshot_json", sa.JSON(), nullable=False),
-        sa.Column("date_snapshot_json", sa.JSON(), nullable=False),
-        sa.Column("summary_json", sa.JSON(), nullable=True),
-        sa.Column("error_code", sa.String(64), nullable=True),
-        sa.Column("error_message", sa.Text(), nullable=True),
-        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
-        sa.Column("updated_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
-        sa.UniqueConstraint("dedupe_key", name="uk_pipeline_run_dedupe_key"),
-        mysql_charset="utf8mb4",
-    )
-    op.create_index(
-        "idx_pipeline_run_biz",
-        "pipeline_run",
-        ["pipeline_key", "biz_dt", "created_at"],
-    )
-    op.create_index(
-        "idx_pipeline_run_status",
-        "pipeline_run",
-        ["status", "lease_until"],
-    )
-    op.create_index(
-        "idx_pipeline_run_deadline",
-        "pipeline_run",
-        ["status", "deadline_at"],
-    )
-
-    op.create_table(
-        "pipeline_step_run",
-        sa.Column("step_run_id", sa.String(36), primary_key=True),
-        sa.Column(
-            "run_id",
-            sa.String(36),
-            sa.ForeignKey("pipeline_run.run_id", ondelete="CASCADE"),
-            nullable=False,
-        ),
-        sa.Column("step_key", sa.String(64), nullable=False),
-        sa.Column("step_order", sa.Integer(), nullable=False),
-        sa.Column("attempt", sa.Integer(), nullable=False),
-        sa.Column("status", sa.String(32), nullable=False),
-        sa.Column("critical", sa.Boolean(), nullable=False),
-        sa.Column("dependency_snapshot_json", sa.JSON(), nullable=False),
-        sa.Column("input_snapshot_json", sa.JSON(), nullable=False),
-        sa.Column("timeout_seconds", sa.Integer(), nullable=False),
-        sa.Column("max_attempts", sa.Integer(), nullable=False),
-        sa.Column("retryable", sa.Boolean(), nullable=False),
-        sa.Column("next_retry_at", sa.DateTime(), nullable=True),
-        sa.Column("started_at", sa.DateTime(), nullable=True),
-        sa.Column("finished_at", sa.DateTime(), nullable=True),
-        sa.Column("heartbeat_at", sa.DateTime(), nullable=True),
-        sa.Column("lease_owner", sa.String(191), nullable=True),
-        sa.Column("lease_until", sa.DateTime(), nullable=True),
-        sa.Column("exit_code", sa.Integer(), nullable=True),
-        sa.Column("result_summary_json", sa.JSON(), nullable=True),
-        sa.Column("metrics_json", sa.JSON(), nullable=True),
-        sa.Column("error_code", sa.String(64), nullable=True),
-        sa.Column("error_message", sa.Text(), nullable=True),
-        sa.Column("log_uri", sa.String(1024), nullable=True),
-        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
-        sa.Column("updated_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
-        sa.UniqueConstraint(
-            "run_id",
-            "step_key",
-            "attempt",
-            name="uk_pipeline_step_run_attempt",
-        ),
-        mysql_charset="utf8mb4",
-    )
-    op.create_index(
-        "idx_pipeline_step_claim",
-        "pipeline_step_run",
-        ["status", "next_retry_at", "step_order"],
-    )
-    op.create_index(
-        "idx_pipeline_step_run",
-        "pipeline_step_run",
-        ["run_id", "step_order", "attempt"],
-    )
-    op.create_index(
-        "idx_pipeline_step_lease",
-        "pipeline_step_run",
-        ["status", "lease_until"],
-    )
-
-    op.create_table(
-        "pipeline_lock",
-        sa.Column("lock_key", sa.String(191), primary_key=True),
-        sa.Column("owner_run_id", sa.String(36), nullable=False),
-        sa.Column("owner_instance", sa.String(191), nullable=False),
-        sa.Column("lease_until", sa.DateTime(), nullable=False),
-        sa.Column("heartbeat_at", sa.DateTime(), nullable=False),
-        sa.Column("version", sa.BigInteger(), nullable=False),
-        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
-        sa.Column("updated_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
-        mysql_charset="utf8mb4",
-    )
-    pipeline_lock = sa.table(
-        "pipeline_lock",
-        sa.column("lock_key", sa.String()),
-        sa.column("owner_run_id", sa.String()),
-        sa.column("owner_instance", sa.String()),
-        sa.column("lease_until", sa.DateTime()),
-        sa.column("heartbeat_at", sa.DateTime()),
-        sa.column("version", sa.BigInteger()),
-    )
-    epoch = datetime(1970, 1, 1)
-    op.bulk_insert(
-        pipeline_lock,
-        [
-            {
-                "lock_key": "pipeline:claim_guard",
-                "owner_run_id": "control-plane",
-                "owner_instance": "claim-coordinator",
-                "lease_until": epoch,
-                "heartbeat_at": epoch,
-                "version": 1,
-            }
-        ],
-    )
-
-    op.create_table(
-        "pipeline_outbox",
-        sa.Column("outbox_id", sa.String(36), primary_key=True),
-        sa.Column(
-            "run_id",
-            sa.String(36),
-            sa.ForeignKey("pipeline_run.run_id", ondelete="CASCADE"),
-            nullable=False,
-        ),
-        sa.Column(
-            "step_run_id",
-            sa.String(36),
-            sa.ForeignKey("pipeline_step_run.step_run_id", ondelete="CASCADE"),
-            nullable=False,
-        ),
-        sa.Column("effect_type", sa.String(64), nullable=False),
-        sa.Column("idempotency_key", sa.String(191), nullable=False),
-        sa.Column("payload_hash", sa.String(64), nullable=False),
-        sa.Column("payload_json", sa.JSON(), nullable=True),
-        sa.Column("payload_uri", sa.String(1024), nullable=True),
-        sa.Column("status", sa.String(32), nullable=False),
-        sa.Column("dry_run", sa.Boolean(), nullable=False),
-        sa.Column("record_error", sa.Text(), nullable=True),
-        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
-        sa.Column("updated_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
-        sa.UniqueConstraint(
-            "idempotency_key",
-            name="uk_pipeline_outbox_idempotency",
-        ),
-        mysql_charset="utf8mb4",
-    )
-    op.create_index(
-        "idx_pipeline_outbox_run",
-        "pipeline_outbox",
-        ["run_id", "step_run_id"],
-    )
-    op.create_index(
-        "idx_pipeline_outbox_status",
-        "pipeline_outbox",
-        ["status", "created_at"],
-    )
-
-
-def downgrade() -> None:
-    op.drop_index("idx_pipeline_outbox_status", table_name="pipeline_outbox")
-    op.drop_index("idx_pipeline_outbox_run", table_name="pipeline_outbox")
-    op.drop_table("pipeline_outbox")
-    op.drop_table("pipeline_lock")
-    op.drop_index("idx_pipeline_step_lease", table_name="pipeline_step_run")
-    op.drop_index("idx_pipeline_step_run", table_name="pipeline_step_run")
-    op.drop_index("idx_pipeline_step_claim", table_name="pipeline_step_run")
-    op.drop_table("pipeline_step_run")
-    op.drop_index("idx_pipeline_run_deadline", table_name="pipeline_run")
-    op.drop_index("idx_pipeline_run_status", table_name="pipeline_run")
-    op.drop_index("idx_pipeline_run_biz", table_name="pipeline_run")
-    op.drop_table("pipeline_run")

alembic/versions/20260727_02_nullable_manual_deadline.py

@@ -1,49 +0,0 @@
-"""allow manual pipeline runs without a deadline
-
-Revision ID: 20260727_02
-Revises: 20260727_01
-Create Date: 2026-07-27
-"""
-from __future__ import annotations
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "20260727_02"
-down_revision: str | None = "20260727_01"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-
-def upgrade() -> None:
-    op.alter_column(
-        "pipeline_run",
-        "deadline_at",
-        existing_type=sa.DateTime(),
-        nullable=True,
-    )
-    op.execute(
-        sa.text(
-            "UPDATE pipeline_run "
-            "SET deadline_at = NULL "
-            "WHERE trigger_type NOT IN ('cron', 'reconcile')"
-        )
-    )
-
-
-def downgrade() -> None:
-    op.execute(
-        sa.text(
-            "UPDATE pipeline_run "
-            "SET deadline_at = UTC_TIMESTAMP() "
-            "WHERE deadline_at IS NULL"
-        )
-    )
-    op.alter_column(
-        "pipeline_run",
-        "deadline_at",
-        existing_type=sa.DateTime(),
-        nullable=False,
-    )

alembic/versions/20260727_03_agent_document_injection.py

@@ -1,75 +0,0 @@
-"""create persisted Agent document injection table
-
-Revision ID: 20260727_03
-Revises: 20260727_02
-Create Date: 2026-07-27
-"""
-from __future__ import annotations
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from alembic import op
-from sqlalchemy.dialects import mysql
-
-revision: str = "20260727_03"
-down_revision: str | None = "20260727_02"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "agent_document_injection",
-        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
-        sa.Column(
-            "agent_name",
-            sa.String(length=128),
-            nullable=False,
-            comment="Agent name",
-        ),
-        sa.Column(
-            "content",
-            sa.Text().with_variant(mysql.LONGTEXT(), "mysql"),
-            nullable=False,
-            comment="Document content injected after the base system prompt",
-        ),
-        sa.Column(
-            "enabled",
-            sa.Integer(),
-            server_default=sa.text("1"),
-            nullable=False,
-            comment="Whether the document injection is active",
-        ),
-        sa.Column(
-            "create_time",
-            sa.DateTime(),
-            server_default=sa.func.now(),
-            nullable=False,
-        ),
-        sa.Column(
-            "update_time",
-            sa.DateTime(),
-            server_default=sa.func.now(),
-            nullable=False,
-        ),
-        sa.PrimaryKeyConstraint("id"),
-        sa.UniqueConstraint(
-            "agent_name",
-            name="uk_agent_document_injection_agent_name",
-        ),
-        mysql_charset="utf8mb4",
-    )
-    op.create_index(
-        "ix_agent_document_injection_agent_name",
-        "agent_document_injection",
-        ["agent_name"],
-    )
-
-
-def downgrade() -> None:
-    op.drop_index(
-        "ix_agent_document_injection_agent_name",
-        table_name="agent_document_injection",
-    )
-    op.drop_table("agent_document_injection")

alembic/versions/20260728_01_drop_unused_scheduler_artifacts.py

@@ -1,75 +0,0 @@
-"""drop unused scheduler leftovers
-
-Revision ID: 20260728_01
-Revises: 20260727_03
-Create Date: 2026-07-28
-"""
-from __future__ import annotations
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from alembic import op
-from sqlalchemy.dialects import mysql
-
-revision: str = "20260728_01"
-down_revision: str | None = "20260727_03"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-
-def upgrade() -> None:
-    op.execute("DROP TABLE IF EXISTS scheduler_job_execution")
-    op.drop_column("pipeline_run", "triggered_by")
-    op.drop_column("pipeline_run", "parent_run_id")
-    op.drop_column("pipeline_run", "scheduled_for")
-    op.drop_column("pipeline_step_run", "metrics_json")
-
-
-def downgrade() -> None:
-    op.add_column(
-        "pipeline_step_run",
-        sa.Column("metrics_json", sa.JSON(), nullable=True),
-    )
-    op.add_column(
-        "pipeline_run",
-        sa.Column("scheduled_for", sa.DateTime(), nullable=True),
-    )
-    op.add_column(
-        "pipeline_run",
-        sa.Column("parent_run_id", sa.String(length=36), nullable=True),
-    )
-    op.add_column(
-        "pipeline_run",
-        sa.Column("triggered_by", sa.String(length=128), nullable=True),
-    )
-    op.create_table(
-        "scheduler_job_execution",
-        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
-        sa.Column("run_id", sa.String(length=64), nullable=False),
-        sa.Column("job_name", sa.String(length=128), nullable=False),
-        sa.Column("job_id", sa.String(length=128), nullable=False),
-        sa.Column("status", sa.String(length=32), nullable=False),
-        sa.Column("event_time", sa.DateTime(), nullable=False),
-        sa.Column("biz_dt", sa.String(length=8), nullable=True),
-        sa.Column("started_at", sa.DateTime(), nullable=True),
-        sa.Column("finished_at", sa.DateTime(), nullable=True),
-        sa.Column("duration_seconds", sa.Float(), nullable=True),
-        sa.Column("error_message", sa.Text(), nullable=True),
-        sa.Column("detail", mysql.LONGTEXT(), 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 ON UPDATE CURRENT_TIMESTAMP"),
-            nullable=False,
-        ),
-        sa.PrimaryKeyConstraint("id"),
-        mysql_charset="utf8mb4",
-        mysql_collate="utf8mb4_unicode_ci",
-    )

alembic/versions/20260729_01_search_candidate_occurrences.py

@@ -1,117 +0,0 @@
-"""store every search and every candidate occurrence
-
-Revision ID: 20260729_01
-Revises: 20260728_01
-Create Date: 2026-07-29
-"""
-from __future__ import annotations
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "20260729_01"
-down_revision: str | None = "20260728_01"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-
-def upgrade() -> None:
-    inspector = sa.inspect(op.get_bind())
-    tables = set(inspector.get_table_names())
-    required_tables = {"video_discovery_search", "video_discovery_candidate"}
-    if not required_tables.issubset(tables):
-        # Fresh databases create these tables later from ORM metadata.
-        return
-
-    search_uniques = {
-        item.get("name")
-        for item in inspector.get_unique_constraints("video_discovery_search")
-    }
-    if "uk_video_discovery_search_key" in search_uniques:
-        op.drop_constraint(
-            "uk_video_discovery_search_key",
-            "video_discovery_search",
-            type_="unique",
-        )
-
-    candidate_uniques = {
-        item.get("name")
-        for item in inspector.get_unique_constraints("video_discovery_candidate")
-    }
-    if "uk_video_discovery_candidate_run_aweme" in candidate_uniques:
-        op.drop_constraint(
-            "uk_video_discovery_candidate_run_aweme",
-            "video_discovery_candidate",
-            type_="unique",
-        )
-
-    candidate_columns = {
-        item["name"]
-        for item in inspector.get_columns("video_discovery_candidate")
-    }
-    if "search_id" not in candidate_columns:
-        op.add_column(
-            "video_discovery_candidate",
-            sa.Column(
-                "search_id",
-                sa.BigInteger(),
-                nullable=True,
-                comment="直接关联 video_discovery_search.id；历史数据允许为空",
-            ),
-        )
-
-    candidate_indexes = {
-        item.get("name")
-        for item in inspector.get_indexes("video_discovery_candidate")
-    }
-    if "idx_video_discovery_candidate_search" not in candidate_indexes:
-        op.create_index(
-            "idx_video_discovery_candidate_search",
-            "video_discovery_candidate",
-            ["search_id", "id"],
-        )
-
-    candidate_foreign_keys = {
-        item.get("name")
-        for item in inspector.get_foreign_keys("video_discovery_candidate")
-    }
-    if "fk_video_discovery_candidate_search" not in candidate_foreign_keys:
-        op.create_foreign_key(
-            "fk_video_discovery_candidate_search",
-            "video_discovery_candidate",
-            "video_discovery_search",
-            ["search_id"],
-            ["id"],
-            ondelete="RESTRICT",
-        )
-
-
-def downgrade() -> None:
-    inspector = sa.inspect(op.get_bind())
-    tables = set(inspector.get_table_names())
-    required_tables = {"video_discovery_search", "video_discovery_candidate"}
-    if not required_tables.issubset(tables):
-        return
-
-    op.drop_constraint(
-        "fk_video_discovery_candidate_search",
-        "video_discovery_candidate",
-        type_="foreignkey",
-    )
-    op.drop_index(
-        "idx_video_discovery_candidate_search",
-        table_name="video_discovery_candidate",
-    )
-    op.drop_column("video_discovery_candidate", "search_id")
-    op.create_unique_constraint(
-        "uk_video_discovery_candidate_run_aweme",
-        "video_discovery_candidate",
-        ["run_id", "aweme_id"],
-    )
-    op.create_unique_constraint(
-        "uk_video_discovery_search_key",
-        "video_discovery_search",
-        ["run_id", "search_key"],
-    )

alembic/versions/20260729_02_remove_candidate_audit_fields.py

@@ -1,95 +0,0 @@
-"""remove unused candidate audit and explanation fields
-
-Revision ID: 20260729_02
-Revises: 20260729_01
-Create Date: 2026-07-29
-"""
-from __future__ import annotations
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "20260729_02"
-down_revision: str | None = "20260729_01"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-_REMOVED_COLUMNS = (
-    "hit_points_json",
-    "publish_timestamp",
-    "detail_verified",
-    "content_portrait_attempted",
-    "account_portrait_attempted",
-    "age_portraits_normalized",
-    "expansion_worthy_tags_json",
-    "confidence",
-    "relevance_reason",
-    "elder_reason",
-    "share_reason",
-    "manual_review_note",
-    "manual_review_status",
-)
-
-
-def _candidate_columns() -> set[str]:
-    inspector = sa.inspect(op.get_bind())
-    if "video_discovery_candidate" not in set(inspector.get_table_names()):
-        return set()
-    return {
-        str(item["name"])
-        for item in inspector.get_columns("video_discovery_candidate")
-    }
-
-
-def upgrade() -> None:
-    columns = _candidate_columns()
-    for column_name in _REMOVED_COLUMNS:
-        if column_name in columns:
-            op.drop_column("video_discovery_candidate", column_name)
-
-
-def downgrade() -> None:
-    columns = _candidate_columns()
-    if not columns:
-        return
-
-    definitions = (
-        sa.Column("hit_points_json", sa.Text(), nullable=True),
-        sa.Column("publish_timestamp", sa.BigInteger(), nullable=True),
-        sa.Column(
-            "detail_verified",
-            sa.Integer(),
-            nullable=False,
-            server_default=sa.text("0"),
-        ),
-        sa.Column(
-            "content_portrait_attempted",
-            sa.Integer(),
-            nullable=False,
-            server_default=sa.text("0"),
-        ),
-        sa.Column(
-            "account_portrait_attempted",
-            sa.Integer(),
-            nullable=False,
-            server_default=sa.text("0"),
-        ),
-        sa.Column(
-            "age_portraits_normalized",
-            sa.Integer(),
-            nullable=False,
-            server_default=sa.text("0"),
-        ),
-        sa.Column("expansion_worthy_tags_json", sa.Text(), nullable=True),
-        sa.Column("confidence", sa.String(length=16), nullable=True),
-        sa.Column("relevance_reason", sa.Text(), nullable=True),
-        sa.Column("elder_reason", sa.Text(), nullable=True),
-        sa.Column("share_reason", sa.Text(), nullable=True),
-        sa.Column("manual_review_note", sa.Text(), nullable=True),
-        sa.Column("manual_review_status", sa.String(length=32), nullable=True),
-    )
-    for column in definitions:
-        if column.name not in columns:
-            op.add_column("video_discovery_candidate", column)

alembic/versions/20260729_03_value_score_range_comment.py

@@ -1,53 +0,0 @@
-"""set video discovery value score comment to 0-1
-
-Revision ID: 20260729_03
-Revises: 20260729_02
-Create Date: 2026-07-29
-"""
-from __future__ import annotations
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "20260729_03"
-down_revision: str | None = "20260729_02"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-
-def _has_value_score_column() -> bool:
-    inspector = sa.inspect(op.get_bind())
-    if "video_discovery_candidate" not in set(inspector.get_table_names()):
-        return False
-    return "value_score" in {
-        str(item["name"])
-        for item in inspector.get_columns("video_discovery_candidate")
-    }
-
-
-def upgrade() -> None:
-    if not _has_value_score_column():
-        return
-    op.alter_column(
-        "video_discovery_candidate",
-        "value_score",
-        existing_type=sa.Numeric(8, 2),
-        existing_nullable=True,
-        comment="联合价值 V，范围 0~1",
-        existing_comment="联合价值 V，范围 0~100",
-    )
-
-
-def downgrade() -> None:
-    if not _has_value_score_column():
-        return
-    op.alter_column(
-        "video_discovery_candidate",
-        "value_score",
-        existing_type=sa.Numeric(8, 2),
-        existing_nullable=True,
-        comment="联合价值 V，范围 0~100",
-        existing_comment="联合价值 V，范围 0~1",
-    )