"""add credit record billing snapshots and frontend user kind

Revision ID: 0a8da2d3c091
Revises: 5e2c124e5484
Create Date: 2026-06-25 13:25:21.246025
"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = "0a8da2d3c091"
down_revision: Union[str, None] = "5e2c124e5484"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # 1) 表结构变更。
    # 注意：autogenerate 生成的 upload_task.other_info 删除语句与本次需求无关，已移除。
    op.add_column("users", sa.Column("frontend_user_kind", sa.String(length=16), server_default="external", nullable=False))

    op.add_column("credit_records", sa.Column("owner_type", sa.String(length=64), nullable=True))
    op.add_column("credit_records", sa.Column("owner_id", sa.String(length=64), nullable=True))
    op.add_column("credit_records", sa.Column("attempt_no", sa.Integer(), nullable=True))
    op.add_column("credit_records", sa.Column("charge_kind", sa.String(length=32), nullable=True))
    op.add_column("credit_records", sa.Column("charge_action", sa.String(length=16), nullable=True))
    op.add_column("credit_records", sa.Column("credit_subject", sa.String(length=32), nullable=True))
    op.add_column("credit_records", sa.Column("media_type", sa.String(length=16), nullable=True))
    op.add_column("credit_records", sa.Column("billing_scene", sa.String(length=64), nullable=True))
    op.add_column("credit_records", sa.Column("source_module", sa.String(length=64), nullable=True))
    op.add_column("credit_records", sa.Column("source_project_id", sa.String(length=64), nullable=True))
    op.add_column("credit_records", sa.Column("source_step_id", sa.String(length=64), nullable=True))
    op.add_column("credit_records", sa.Column("source_step_code", sa.String(length=64), nullable=True))
    op.add_column("credit_records", sa.Column("token_usage_id", sa.String(length=32), nullable=True))
    op.add_column("credit_records", sa.Column("input_tokens", sa.Integer(), nullable=True))
    op.add_column("credit_records", sa.Column("output_tokens", sa.Integer(), nullable=True))
    op.add_column("credit_records", sa.Column("total_tokens", sa.Integer(), nullable=True))
    op.add_column("credit_records", sa.Column("engine_type", sa.String(length=16), nullable=True))
    op.add_column("credit_records", sa.Column("engine_id", sa.String(length=32), nullable=True))
    op.add_column("credit_records", sa.Column("engine_name", sa.String(length=128), nullable=True))
    op.add_column("credit_records", sa.Column("engine_provider", sa.String(length=64), nullable=True))
    op.add_column("credit_records", sa.Column("engine_model_name", sa.String(length=128), nullable=True))
    op.add_column("credit_records", sa.Column("user_type_snapshot", sa.String(length=16), nullable=True))
    op.add_column("credit_records", sa.Column("frontend_user_kind_snapshot", sa.String(length=16), nullable=True))

    op.add_column("module_generation_steps", sa.Column("token_usage_id", sa.String(length=32), nullable=True))
    op.add_column("module_generation_steps", sa.Column("model_config_id", sa.String(length=32), nullable=True))
    op.add_column("module_generation_steps", sa.Column("input_tokens", sa.Integer(), nullable=True))
    op.add_column("module_generation_steps", sa.Column("output_tokens", sa.Integer(), nullable=True))
    op.add_column("module_generation_steps", sa.Column("total_tokens", sa.Integer(), nullable=True))
    op.add_column("module_generation_steps", sa.Column("text_credits_cost", sa.Float(), nullable=True))

    op.add_column("token_usage", sa.Column("owner_type", sa.String(length=64), nullable=True))
    op.add_column("token_usage", sa.Column("owner_id", sa.String(length=64), nullable=True))
    op.add_column("token_usage", sa.Column("biz_key", sa.String(length=160), nullable=True))
    op.add_column("token_usage", sa.Column("source_module", sa.String(length=64), nullable=True))
    op.add_column("token_usage", sa.Column("source_step_code", sa.String(length=64), nullable=True))

    # 2) 历史数据补齐。
    _backfill_legacy_data()

    # 3) 索引。放在历史补齐之后，避免大批量 UPDATE 时维护过多新索引。
    op.create_index(op.f("ix_users_frontend_user_kind"), "users", ["frontend_user_kind"], unique=False)
    op.create_index(op.f("ix_users_user_type"), "users", ["user_type"], unique=False)

    op.create_index(op.f("ix_credit_records_billing_scene"), "credit_records", ["billing_scene"], unique=False)
    op.create_index(op.f("ix_credit_records_charge_kind"), "credit_records", ["charge_kind"], unique=False)
    op.create_index(op.f("ix_credit_records_credit_subject"), "credit_records", ["credit_subject"], unique=False)
    op.create_index(op.f("ix_credit_records_engine_id"), "credit_records", ["engine_id"], unique=False)
    op.create_index(op.f("ix_credit_records_engine_type"), "credit_records", ["engine_type"], unique=False)
    op.create_index(op.f("ix_credit_records_frontend_user_kind_snapshot"), "credit_records", ["frontend_user_kind_snapshot"], unique=False)
    op.create_index(op.f("ix_credit_records_media_type"), "credit_records", ["media_type"], unique=False)
    op.create_index("ix_credit_records_owner", "credit_records", ["owner_type", "owner_id"], unique=False)
    op.create_index(op.f("ix_credit_records_owner_id"), "credit_records", ["owner_id"], unique=False)
    op.create_index(op.f("ix_credit_records_owner_type"), "credit_records", ["owner_type"], unique=False)
    op.create_index(op.f("ix_credit_records_source_module"), "credit_records", ["source_module"], unique=False)
    op.create_index("ix_credit_records_source_module_scene", "credit_records", ["source_module", "billing_scene"], unique=False)
    op.create_index(op.f("ix_credit_records_source_project_id"), "credit_records", ["source_project_id"], unique=False)
    op.create_index(op.f("ix_credit_records_source_step_code"), "credit_records", ["source_step_code"], unique=False)
    op.create_index(op.f("ix_credit_records_source_step_id"), "credit_records", ["source_step_id"], unique=False)
    op.create_index("ix_credit_records_subject_media", "credit_records", ["credit_subject", "media_type"], unique=False)
    op.create_index(op.f("ix_credit_records_token_usage_id"), "credit_records", ["token_usage_id"], unique=False)
    op.create_index(op.f("ix_credit_records_type"), "credit_records", ["type"], unique=False)
    op.create_index("ix_credit_records_user_kind_time", "credit_records", ["user_type_snapshot", "frontend_user_kind_snapshot", "created_at"], unique=False)
    op.create_index(op.f("ix_credit_records_user_type_snapshot"), "credit_records", ["user_type_snapshot"], unique=False)

    op.create_index(op.f("ix_module_generation_steps_model_config_id"), "module_generation_steps", ["model_config_id"], unique=False)
    op.create_index(op.f("ix_module_generation_steps_token_usage_id"), "module_generation_steps", ["token_usage_id"], unique=False)

    op.create_index("ix_token_usage_biz_key", "token_usage", ["biz_key"], unique=False)
    op.create_index("ix_token_usage_owner", "token_usage", ["owner_type", "owner_id"], unique=False)
    op.create_index(op.f("ix_token_usage_owner_id"), "token_usage", ["owner_id"], unique=False)
    op.create_index(op.f("ix_token_usage_owner_type"), "token_usage", ["owner_type"], unique=False)
    op.create_index(op.f("ix_token_usage_source_module"), "token_usage", ["source_module"], unique=False)
    op.create_index(op.f("ix_token_usage_source_step_code"), "token_usage", ["source_step_code"], unique=False)


def _backfill_legacy_data() -> None:
    """修复历史账务流水的归类、模块、步骤、token、模型和引擎快照。

    设计原则：
    - 只补能确定的数据，不用“同用户 + 时间接近 + token 相等”这类不可靠方式强行匹配。
    - ModuleGenerationStep 包含 deleted_at 不为空的软删数据，保证历史流水仍可展示来源。
    - CreditRecord 是账务事实表，历史业务数据软删不影响流水展示。
    - CreditRecord 统一使用 engine_* 承载执行配置快照：model/image/video。
    """

    # users.frontend_user_kind 默认外部用户。
    op.execute(
        """
        UPDATE users
        SET frontend_user_kind = 'external'
        WHERE frontend_user_kind IS NULL OR frontend_user_kind NOT IN ('internal', 'external')
        """
    )

    # 用户类型快照：按迁移时 users 当前值兜底。
    op.execute(
        """
        UPDATE credit_records cr
        SET
            user_type_snapshot = COALESCE(NULLIF(u.user_type, ''), CASE WHEN COALESCE(u.is_admin, false) THEN 'admin' ELSE 'frontend' END),
            frontend_user_kind_snapshot = CASE
                WHEN COALESCE(NULLIF(u.user_type, ''), CASE WHEN COALESCE(u.is_admin, false) THEN 'admin' ELSE 'frontend' END) = 'frontend'
                    THEN COALESCE(NULLIF(u.frontend_user_kind, ''), 'external')
                ELSE COALESCE(NULLIF(u.frontend_user_kind, ''), 'external')
            END
        FROM users u
        WHERE cr.user_id = u.id
        """
    )

    # 解析标准 biz_key：owner_type:owner_id:attempt:{n}:charge_kind:charge/refund。
    op.execute(
        """
        UPDATE credit_records
        SET
            owner_type = split_part(biz_key, ':', 1),
            owner_id = split_part(biz_key, ':', 2),
            attempt_no = split_part(biz_key, ':', 4)::integer,
            charge_kind = split_part(biz_key, ':', 5),
            charge_action = split_part(biz_key, ':', 6)
        WHERE biz_key IS NOT NULL
          AND biz_key ~ '^[^:]+:[^:]+:attempt:[0-9]+:[^:]+:(charge|refund)$'
        """
    )

    # 少数退款记录如果自身 biz_key 不完整，用 refund_for_biz_key 兜底解析业务归属。
    op.execute(
        """
        UPDATE credit_records
        SET
            owner_type = split_part(refund_for_biz_key, ':', 1),
            owner_id = split_part(refund_for_biz_key, ':', 2),
            attempt_no = split_part(refund_for_biz_key, ':', 4)::integer,
            charge_kind = split_part(refund_for_biz_key, ':', 5),
            charge_action = 'refund'
        WHERE type = 'refund'
          AND (owner_type IS NULL OR owner_type = 'unknown' OR owner_id IS NULL)
          AND refund_for_biz_key IS NOT NULL
          AND refund_for_biz_key ~ '^[^:]+:[^:]+:attempt:[0-9]+:[^:]+:(charge|refund)$'
        """
    )

    # 支付充值流水。
    op.execute(
        """
        UPDATE credit_records cr
        SET
            owner_type = COALESCE(NULLIF(cr.owner_type, 'unknown'), 'payment_order'),
            owner_id = COALESCE(cr.owner_id, po.id),
            charge_kind = COALESCE(NULLIF(cr.charge_kind, 'unknown'), 'recharge'),
            charge_action = COALESCE(cr.charge_action, 'charge'),
            credit_subject = COALESCE(NULLIF(cr.credit_subject, 'unknown'), 'recharge'),
            billing_scene = COALESCE(NULLIF(cr.billing_scene, 'unknown'), 'recharge'),
            source_module = COALESCE(NULLIF(cr.source_module, 'unknown'), 'payment')
        FROM payment_orders po
        WHERE cr.related_id = po.id
          AND cr.type = 'recharge'
          AND (cr.description IS NULL OR cr.description NOT LIKE '管理员调整:%')
        """
    )

    # 支付退款扣回积分。历史上它通常是 type=consume，但属于支付退款，不应该混入生成消费。
    op.execute(
        """
        UPDATE credit_records cr
        SET
            owner_type = COALESCE(NULLIF(cr.owner_type, 'unknown'), 'payment_order'),
            owner_id = COALESCE(cr.owner_id, po.id),
            charge_kind = COALESCE(NULLIF(cr.charge_kind, 'unknown'), 'refund'),
            charge_action = COALESCE(cr.charge_action, 'refund'),
            credit_subject = COALESCE(NULLIF(cr.credit_subject, 'unknown'), 'refund'),
            billing_scene = COALESCE(NULLIF(cr.billing_scene, 'unknown'), 'refund'),
            source_module = COALESCE(NULLIF(cr.source_module, 'unknown'), 'payment')
        FROM payment_orders po
        WHERE cr.related_id = po.id
          AND cr.type = 'consume'
          AND (cr.description ILIKE '%退款%' OR po.status = 'refunded')
        """
    )

    # 管理员手动调整，不混入普通充值/消费。
    op.execute(
        """
        UPDATE credit_records cr
        SET
            owner_type = COALESCE(NULLIF(cr.owner_type, 'unknown'), 'admin_adjust'),
            owner_id = COALESCE(cr.owner_id, cr.related_id),
            charge_kind = 'admin_adjust',
            charge_action = CASE WHEN cr.type = 'refund' THEN 'refund' ELSE 'charge' END,
            credit_subject = 'admin_adjust',
            billing_scene = 'admin_adjust',
            source_module = 'admin'
        WHERE cr.description LIKE '管理员调整:%'
        """
    )

    # GenerationRecord 旧版提示词优化裸扣费：generation.py /optimize 历史上未传 related_id / biz_key / record_meta。
    # 只做严格唯一匹配：同用户、同项目名称描述、同扣费金额，并且 credit_record 与 generation_record 均唯一时才回填。
    op.execute(
        """
        WITH candidates AS (
            SELECT
                cr.id AS credit_record_id,
                gr.id AS generation_record_id,
                gr.text_tokens_used,
                gr.text_credits_cost
            FROM credit_records cr
            JOIN generation_records gr
              ON gr.user_id = cr.user_id
            JOIN projects p
              ON p.id = gr.project_id
            WHERE cr.type = 'consume'
              AND cr.biz_key IS NULL
              AND cr.refund_for_biz_key IS NULL
              AND cr.related_id IS NULL
              AND (cr.owner_type IS NULL OR cr.owner_type = 'unknown')
              AND (cr.source_module IS NULL OR cr.source_module = 'unknown')
              AND (cr.billing_scene IS NULL OR cr.billing_scene = 'unknown')
              AND COALESCE(gr.text_credits_cost, 0) > 0
              AND COALESCE(gr.text_tokens_used, 0) > 0
              AND cr.description = '提示词优化 - ' || p.name
              AND ABS(ABS(cr.amount)::numeric - ROUND(COALESCE(gr.text_credits_cost, 0)::numeric, 2)) < 0.01
        ),
        unique_credit AS (
            SELECT credit_record_id
            FROM candidates
            GROUP BY credit_record_id
            HAVING COUNT(*) = 1
        ),
        unique_generation AS (
            SELECT generation_record_id
            FROM candidates
            GROUP BY generation_record_id
            HAVING COUNT(*) = 1
        ),
        matched AS (
            SELECT c.*
            FROM candidates c
            JOIN unique_credit uc ON uc.credit_record_id = c.credit_record_id
            JOIN unique_generation ug ON ug.generation_record_id = c.generation_record_id
        )
        UPDATE credit_records cr
        SET
            related_id = COALESCE(cr.related_id, matched.generation_record_id),
            owner_type = 'generation_record',
            owner_id = matched.generation_record_id,
            attempt_no = COALESCE(cr.attempt_no, 1),
            charge_kind = 'text_prompt',
            charge_action = 'charge',
            credit_subject = 'text',
            billing_scene = 'generation_record_text_prompt_optimize',
            source_module = 'generation_record',
            total_tokens = COALESCE(cr.total_tokens, NULLIF(matched.text_tokens_used, 0)),
            engine_type = COALESCE(NULLIF(cr.engine_type, 'unknown'), 'model')
        FROM matched
        WHERE cr.id = matched.credit_record_id
        """
    )

    # GenerationRecord：项目记录提词/文件解析/图片理解/图片生成/视频生成。
    op.execute(
        """
        UPDATE credit_records cr
        SET
            owner_type = COALESCE(NULLIF(cr.owner_type, 'unknown'), 'generation_record'),
            owner_id = COALESCE(cr.owner_id, gr.id),
            related_id = COALESCE(cr.related_id, gr.id),
            source_module = COALESCE(NULLIF(cr.source_module, 'unknown'), 'generation_record'),
            media_type = CASE
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' AND lower(COALESCE(gr.gen_type, '')) IN ('image', 'video') THEN lower(gr.gen_type)
                ELSE cr.media_type
            END,
            total_tokens = CASE
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'text_prompt' THEN COALESCE(cr.total_tokens, NULLIF(gr.text_tokens_used, 0))
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' AND lower(COALESCE(gr.gen_type, '')) = 'image' THEN COALESCE(cr.total_tokens, NULLIF(gr.image_tokens_used, 0))
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' AND lower(COALESCE(gr.gen_type, '')) = 'video' THEN COALESCE(cr.total_tokens, NULLIF(gr.video_tokens_used, 0))
                ELSE cr.total_tokens
            END,
            credit_subject = CASE
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' THEN 'media'
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') IN ('text_prompt', 'file_parse', 'vision_input') THEN 'text'
                ELSE cr.credit_subject
            END,
            billing_scene = CASE
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'text_prompt' THEN 'generation_record_text_prompt_optimize'
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'file_parse' THEN 'generation_record_file_parse'
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'vision_input' THEN 'generation_record_vision_input'
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' AND lower(COALESCE(gr.gen_type, '')) = 'image' THEN 'generation_record_image_generate'
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' AND lower(COALESCE(gr.gen_type, '')) = 'video' THEN 'generation_record_video_generate'
                ELSE cr.billing_scene
            END,
            engine_type = CASE
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') IN ('text_prompt', 'file_parse', 'vision_input') THEN COALESCE(NULLIF(cr.engine_type, 'unknown'), 'model')
                ELSE cr.engine_type
            END
        FROM generation_records gr
        WHERE (cr.owner_type = 'generation_record' AND cr.owner_id = gr.id)
           OR ((cr.owner_type IS NULL OR cr.owner_type = 'unknown') AND cr.related_id = gr.id)
        """
    )

    # ModuleGenerationStep 提词优化：从 output_json.usage.credit_biz_key 精准匹配 CreditRecord.biz_key。
    # 包含软删步骤，不加 deleted_at 过滤。
    op.execute(
        """
        WITH step_usage AS (
            SELECT
                s.id AS step_id,
                s.project_id,
                s.module,
                s.step_code,
                (s.output_json::jsonb -> 'usage') AS usage,
                COALESCE(
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'token_usage_id', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'tokenUsageId', ''),
                    NULLIF(s.output_json::jsonb ->> 'token_usage_id', ''),
                    NULLIF(s.output_json::jsonb ->> 'tokenUsageId', '')
                ) AS token_usage_id,
                COALESCE(
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'model_config_id', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'modelConfigId', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'model_config' ->> 'id', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'modelConfig' ->> 'id', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'model_config_snapshot' ->> 'id', ''),
                    NULLIF(s.output_json::jsonb ->> 'model_config_id', ''),
                    NULLIF(s.output_json::jsonb ->> 'modelConfigId', ''),
                    NULLIF(s.output_json::jsonb -> 'model_config' ->> 'id', ''),
                    NULLIF(s.output_json::jsonb -> 'modelConfig' ->> 'id', ''),
                    NULLIF(s.output_json::jsonb -> 'model_config_snapshot' ->> 'id', '')
                ) AS model_config_id
            FROM module_generation_steps s
            WHERE s.output_json IS NOT NULL
              AND jsonb_typeof(s.output_json::jsonb) = 'object'
              AND jsonb_typeof(s.output_json::jsonb -> 'usage') = 'object'
        )
        UPDATE module_generation_steps s
        SET
            token_usage_id = COALESCE(s.token_usage_id, step_usage.token_usage_id),
            model_config_id = COALESCE(s.model_config_id, step_usage.model_config_id),
            input_tokens = COALESCE(
                s.input_tokens,
                CASE WHEN COALESCE(step_usage.usage ->> 'input_tokens', '') ~ '^[0-9]+$' THEN (step_usage.usage ->> 'input_tokens')::integer ELSE NULL END
            ),
            output_tokens = COALESCE(
                s.output_tokens,
                CASE WHEN COALESCE(step_usage.usage ->> 'output_tokens', '') ~ '^[0-9]+$' THEN (step_usage.usage ->> 'output_tokens')::integer ELSE NULL END
            ),
            total_tokens = COALESCE(
                s.total_tokens,
                CASE WHEN COALESCE(step_usage.usage ->> 'total_tokens', '') ~ '^[0-9]+$' THEN (step_usage.usage ->> 'total_tokens')::integer ELSE NULL END,
                CASE
                    WHEN COALESCE(step_usage.usage ->> 'input_tokens', '') ~ '^[0-9]+$'
                     AND COALESCE(step_usage.usage ->> 'output_tokens', '') ~ '^[0-9]+$'
                    THEN (step_usage.usage ->> 'input_tokens')::integer + (step_usage.usage ->> 'output_tokens')::integer
                    ELSE NULL
                END
            ),
            text_credits_cost = COALESCE(
                s.text_credits_cost,
                CASE WHEN COALESCE(step_usage.usage ->> 'text_credits_cost', '') ~ '^[0-9]+(\\.[0-9]+)?$' THEN (step_usage.usage ->> 'text_credits_cost')::double precision ELSE NULL END
            )
        FROM step_usage
        WHERE s.id = step_usage.step_id
        """
    )

    op.execute(
        """
        WITH step_usage AS (
            SELECT
                s.id AS step_id,
                s.project_id,
                s.module,
                s.step_code,
                (s.output_json::jsonb -> 'usage') AS usage,
                COALESCE(
                    NULLIF(s.token_usage_id, ''),
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'token_usage_id', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'tokenUsageId', ''),
                    NULLIF(s.output_json::jsonb ->> 'token_usage_id', ''),
                    NULLIF(s.output_json::jsonb ->> 'tokenUsageId', '')
                ) AS step_token_usage_id,
                COALESCE(
                    NULLIF(s.model_config_id, ''),
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'model_config_id', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'modelConfigId', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'model_config' ->> 'id', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'modelConfig' ->> 'id', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'model_config_snapshot' ->> 'id', ''),
                    NULLIF(s.output_json::jsonb ->> 'model_config_id', ''),
                    NULLIF(s.output_json::jsonb ->> 'modelConfigId', ''),
                    NULLIF(s.output_json::jsonb -> 'model_config' ->> 'id', ''),
                    NULLIF(s.output_json::jsonb -> 'modelConfig' ->> 'id', ''),
                    NULLIF(s.output_json::jsonb -> 'model_config_snapshot' ->> 'id', '')
                ) AS step_model_config_id,
                COALESCE(
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'model_config_name', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'modelConfigName', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'model_config' ->> 'name', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'modelConfig' ->> 'name', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'model_config_snapshot' ->> 'name', ''),
                    NULLIF(s.output_json::jsonb ->> 'model_config_name', ''),
                    NULLIF(s.output_json::jsonb ->> 'modelConfigName', ''),
                    NULLIF(s.output_json::jsonb -> 'model_config' ->> 'name', ''),
                    NULLIF(s.output_json::jsonb -> 'modelConfig' ->> 'name', ''),
                    NULLIF(s.output_json::jsonb -> 'model_config_snapshot' ->> 'name', '')
                ) AS step_model_config_name,
                COALESCE(
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'model_provider', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'modelProvider', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'provider', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'model_config' ->> 'provider', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'modelConfig' ->> 'provider', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'model_config_snapshot' ->> 'provider', ''),
                    NULLIF(s.output_json::jsonb ->> 'model_provider', ''),
                    NULLIF(s.output_json::jsonb ->> 'modelProvider', ''),
                    NULLIF(s.output_json::jsonb ->> 'provider', ''),
                    NULLIF(s.output_json::jsonb -> 'model_config' ->> 'provider', ''),
                    NULLIF(s.output_json::jsonb -> 'modelConfig' ->> 'provider', ''),
                    NULLIF(s.output_json::jsonb -> 'model_config_snapshot' ->> 'provider', '')
                ) AS step_model_provider,
                COALESCE(
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'model_name', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' ->> 'modelName', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'model_config' ->> 'model_name', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'modelConfig' ->> 'model_name', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'model_config_snapshot' ->> 'model_name', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'model_config' ->> 'modelName', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'modelConfig' ->> 'modelName', ''),
                    NULLIF(s.output_json::jsonb -> 'usage' -> 'model_config_snapshot' ->> 'modelName', ''),
                    NULLIF(s.output_json::jsonb ->> 'model_name', ''),
                    NULLIF(s.output_json::jsonb ->> 'modelName', ''),
                    NULLIF(s.output_json::jsonb -> 'model_config' ->> 'model_name', ''),
                    NULLIF(s.output_json::jsonb -> 'modelConfig' ->> 'model_name', ''),
                    NULLIF(s.output_json::jsonb -> 'model_config_snapshot' ->> 'model_name', ''),
                    NULLIF(s.output_json::jsonb -> 'model_config' ->> 'modelName', ''),
                    NULLIF(s.output_json::jsonb -> 'modelConfig' ->> 'modelName', ''),
                    NULLIF(s.output_json::jsonb -> 'model_config_snapshot' ->> 'modelName', '')
                ) AS step_model_name
            FROM module_generation_steps s
            WHERE s.output_json IS NOT NULL
              AND jsonb_typeof(s.output_json::jsonb) = 'object'
              AND jsonb_typeof(s.output_json::jsonb -> 'usage') = 'object'
        )
        UPDATE credit_records cr
        SET
            owner_type = COALESCE(NULLIF(cr.owner_type, 'unknown'), 'module_generation_step'),
            owner_id = COALESCE(cr.owner_id, step_usage.step_id),
            charge_kind = COALESCE(NULLIF(cr.charge_kind, 'unknown'), 'text_prompt'),
            charge_action = COALESCE(cr.charge_action, 'charge'),
            credit_subject = COALESCE(NULLIF(cr.credit_subject, 'unknown'), 'text'),
            billing_scene = COALESCE(
                NULLIF(cr.billing_scene, 'unknown'),
                CASE
                    WHEN step_usage.module = 'hot_opening_replicate' AND step_usage.step_code = 'image_prompt_optimize' THEN 'hot_opening_image_prompt_optimize'
                    WHEN step_usage.module = 'hot_opening_replicate' AND step_usage.step_code = 'video_prompt_optimize' THEN 'hot_opening_video_prompt_optimize'
                    WHEN step_usage.module = 'shot_replicate' AND step_usage.step_code = 'image_prompt_optimize' THEN 'shot_image_prompt_optimize'
                    WHEN step_usage.module = 'shot_replicate' AND step_usage.step_code = 'video_prompt_optimize' THEN 'shot_video_prompt_optimize'
                    ELSE 'unknown'
                END
            ),
            source_module = COALESCE(NULLIF(cr.source_module, 'unknown'), step_usage.module),
            source_project_id = COALESCE(cr.source_project_id, step_usage.project_id),
            source_step_id = COALESCE(cr.source_step_id, step_usage.step_id),
            source_step_code = COALESCE(cr.source_step_code, step_usage.step_code),
            token_usage_id = COALESCE(cr.token_usage_id, step_usage.step_token_usage_id),
            input_tokens = COALESCE(
                cr.input_tokens,
                CASE WHEN COALESCE(step_usage.usage ->> 'input_tokens', '') ~ '^[0-9]+$' THEN (step_usage.usage ->> 'input_tokens')::integer ELSE NULL END
            ),
            output_tokens = COALESCE(
                cr.output_tokens,
                CASE WHEN COALESCE(step_usage.usage ->> 'output_tokens', '') ~ '^[0-9]+$' THEN (step_usage.usage ->> 'output_tokens')::integer ELSE NULL END
            ),
            total_tokens = COALESCE(
                cr.total_tokens,
                CASE WHEN COALESCE(step_usage.usage ->> 'total_tokens', '') ~ '^[0-9]+$' THEN (step_usage.usage ->> 'total_tokens')::integer ELSE NULL END,
                CASE
                    WHEN COALESCE(step_usage.usage ->> 'input_tokens', '') ~ '^[0-9]+$'
                     AND COALESCE(step_usage.usage ->> 'output_tokens', '') ~ '^[0-9]+$'
                    THEN (step_usage.usage ->> 'input_tokens')::integer + (step_usage.usage ->> 'output_tokens')::integer
                    ELSE NULL
                END
            ),
            engine_type = COALESCE(NULLIF(cr.engine_type, 'unknown'), 'model'),
            engine_id = COALESCE(cr.engine_id, step_usage.step_model_config_id),
            engine_name = COALESCE(cr.engine_name, step_usage.step_model_config_name),
            engine_provider = COALESCE(cr.engine_provider, step_usage.step_model_provider),
            engine_model_name = COALESCE(cr.engine_model_name, step_usage.step_model_name)
        FROM step_usage
        WHERE cr.biz_key = step_usage.usage ->> 'credit_biz_key'
        """
    )

    # TokenUsage 如果 usage 中已经记录 token_usage_id，则可以安全补 owner；否则不强行按时间/token 猜。
    op.execute(
        """
        WITH step_usage AS (
            SELECT
                s.id AS step_id,
                s.module,
                s.step_code,
                (s.output_json::jsonb -> 'usage') AS usage
            FROM module_generation_steps s
            WHERE s.output_json IS NOT NULL
              AND jsonb_typeof(s.output_json::jsonb) = 'object'
              AND jsonb_typeof(s.output_json::jsonb -> 'usage') = 'object'
              AND NULLIF(s.output_json::jsonb -> 'usage' ->> 'token_usage_id', '') IS NOT NULL
        )
        UPDATE token_usage tu
        SET
            owner_type = COALESCE(NULLIF(tu.owner_type, 'unknown'), 'module_generation_step'),
            owner_id = COALESCE(tu.owner_id, step_usage.step_id),
            biz_key = COALESCE(tu.biz_key, NULLIF(step_usage.usage ->> 'credit_biz_key', '')),
            source_module = COALESCE(NULLIF(tu.source_module, 'unknown'), step_usage.module),
            source_step_code = COALESCE(tu.source_step_code, step_usage.step_code)
        FROM step_usage
        WHERE tu.id = step_usage.usage ->> 'token_usage_id'
        """
    )

    # 拆镜复刻原视频整体分析流水：归属到任务集。
    op.execute(
        """
        UPDATE credit_records cr
        SET
            owner_type = COALESCE(NULLIF(cr.owner_type, 'unknown'), 'shot_replicate_task_set'),
            owner_id = COALESCE(cr.owner_id, ts.id),
            attempt_no = COALESCE(cr.attempt_no, 1),
            charge_kind = COALESCE(NULLIF(cr.charge_kind, 'unknown'), 'video_analysis'),
            charge_action = COALESCE(cr.charge_action, CASE WHEN cr.type = 'refund' THEN 'refund' ELSE 'charge' END),
            credit_subject = COALESCE(NULLIF(cr.credit_subject, 'unknown'), 'analysis'),
            media_type = COALESCE(cr.media_type, 'video'),
            billing_scene = COALESCE(NULLIF(cr.billing_scene, 'unknown'), 'shot_original_video_analysis'),
            source_module = COALESCE(NULLIF(cr.source_module, 'unknown'), 'shot_replicate'),
            source_project_id = COALESCE(cr.source_project_id, ts.id),
            source_step_code = COALESCE(cr.source_step_code, 'video_analysis'),
            engine_type = COALESCE(NULLIF(cr.engine_type, 'unknown'), 'model')
        FROM shot_replicate_task_sets ts
        WHERE (
                (cr.owner_type = 'shot_replicate_task_set' AND cr.owner_id = ts.id)
             OR ((cr.owner_type IS NULL OR cr.owner_type = 'unknown') AND cr.related_id = ts.id)
          )
          AND cr.type IN ('consume', 'refund')
          AND (
                cr.charge_kind = 'video_analysis'
             OR cr.billing_scene IN ('shot_video_analysis', 'shot_original_video_analysis')
             OR cr.description ILIKE '%视频分析%'
             OR cr.description ILIKE '%原视频分析%'
          )
        """
    )

    # 拆镜复刻手动片段分析流水：归属到片段。
    op.execute(
        """
        UPDATE credit_records cr
        SET
            owner_type = COALESCE(NULLIF(cr.owner_type, 'unknown'), 'shot_replicate_segment'),
            owner_id = COALESCE(cr.owner_id, sg.id),
            attempt_no = COALESCE(cr.attempt_no, 1),
            charge_kind = COALESCE(NULLIF(cr.charge_kind, 'unknown'), 'video_analysis'),
            charge_action = COALESCE(cr.charge_action, CASE WHEN cr.type = 'refund' THEN 'refund' ELSE 'charge' END),
            credit_subject = COALESCE(NULLIF(cr.credit_subject, 'unknown'), 'analysis'),
            media_type = COALESCE(cr.media_type, 'video'),
            billing_scene = COALESCE(NULLIF(cr.billing_scene, 'unknown'), 'shot_segment_video_analysis'),
            source_module = COALESCE(NULLIF(cr.source_module, 'unknown'), 'shot_replicate'),
            source_project_id = COALESCE(cr.source_project_id, sg.task_set_id),
            source_step_id = COALESCE(cr.source_step_id, sg.id),
            source_step_code = COALESCE(cr.source_step_code, 'video_analysis'),
            engine_type = COALESCE(NULLIF(cr.engine_type, 'unknown'), 'model')
        FROM shot_replicate_segments sg
        WHERE (
                (cr.owner_type = 'shot_replicate_segment' AND cr.owner_id = sg.id)
             OR ((cr.owner_type IS NULL OR cr.owner_type = 'unknown') AND cr.related_id = sg.id)
          )
          AND cr.type IN ('consume', 'refund')
          AND (
                cr.charge_kind = 'video_analysis'
             OR cr.billing_scene IN ('shot_video_analysis', 'shot_segment_video_analysis')
             OR cr.description ILIKE '%片段分析%'
             OR cr.description ILIKE '%手动%分析%'
          )
        """
    )

    # Module media generation：ChatGenerationTask 被模块步骤引用时，优先归到具体模块/步骤，而不是 AI 创作。
    op.execute(
        """
        WITH step_task AS (
            SELECT DISTINCT ON (s.chat_task_id)
                s.chat_task_id,
                s.id AS step_id,
                s.project_id,
                s.module,
                s.step_code
            FROM module_generation_steps s
            WHERE s.chat_task_id IS NOT NULL
            ORDER BY s.chat_task_id, s.is_current DESC, s.updated_at DESC
        )
        UPDATE credit_records cr
        SET
            source_module = COALESCE(NULLIF(cr.source_module, 'unknown'), step_task.module),
            source_project_id = COALESCE(cr.source_project_id, step_task.project_id),
            source_step_id = COALESCE(cr.source_step_id, step_task.step_id),
            source_step_code = COALESCE(cr.source_step_code, step_task.step_code),
            billing_scene = COALESCE(
                NULLIF(cr.billing_scene, 'unknown'),
                CASE
                    WHEN step_task.module = 'hot_opening_replicate' AND step_task.step_code = 'image_generate' THEN 'hot_opening_image_generate'
                    WHEN step_task.module = 'hot_opening_replicate' AND step_task.step_code = 'video_generate' THEN 'hot_opening_video_generate'
                    WHEN step_task.module = 'shot_replicate' AND step_task.step_code = 'image_generate' THEN 'shot_image_generate'
                    WHEN step_task.module = 'shot_replicate' AND step_task.step_code = 'video_generate' THEN 'shot_video_generate'
                    ELSE cr.billing_scene
                END
            )
        FROM step_task
        WHERE cr.owner_type = 'chat_generation_task'
          AND cr.owner_id = step_task.chat_task_id
        """
    )

    # ChatGenerationTask：AI 创作图片/视频生成，以及媒体 token/引擎快照。
    op.execute(
        """
        UPDATE credit_records cr
        SET
            owner_type = COALESCE(NULLIF(cr.owner_type, 'unknown'), 'chat_generation_task'),
            owner_id = COALESCE(cr.owner_id, t.id),
            source_module = COALESCE(NULLIF(cr.source_module, 'unknown'), 'ai_creation'),
            credit_subject = CASE WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' THEN 'media' ELSE cr.credit_subject END,
            media_type = CASE
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' AND lower(COALESCE(t.gen_type, '')) IN ('image', 'video') THEN lower(t.gen_type)
                ELSE cr.media_type
            END,
            billing_scene = COALESCE(
                NULLIF(cr.billing_scene, 'unknown'),
                CASE
                    WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' AND lower(COALESCE(t.gen_type, '')) = 'image' THEN 'ai_creation_image_generate'
                    WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' AND lower(COALESCE(t.gen_type, '')) = 'video' THEN 'ai_creation_video_generate'
                    ELSE cr.billing_scene
                END
            ),
            engine_type = CASE
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' AND lower(COALESCE(t.gen_type, '')) IN ('image', 'video') THEN lower(t.gen_type)
                ELSE cr.engine_type
            END,
            engine_id = COALESCE(cr.engine_id, t.engine_id),
            total_tokens = CASE
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' AND lower(COALESCE(t.gen_type, '')) = 'image' THEN COALESCE(cr.total_tokens, NULLIF(t.image_tokens_used, 0))
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') = 'media' AND lower(COALESCE(t.gen_type, '')) = 'video' THEN COALESCE(cr.total_tokens, NULLIF(t.video_tokens_used, 0))
                ELSE cr.total_tokens
            END
        FROM chat_generation_tasks t
        WHERE (cr.owner_type = 'chat_generation_task' AND cr.owner_id = t.id)
           OR ((cr.owner_type IS NULL OR cr.owner_type = 'unknown') AND cr.related_id = t.id)
        """
    )

    # 从 TokenUsage 补 CreditRecord 的 token 和 ModelConfig 执行快照，前提是已经能安全匹配 token_usage_id。
    op.execute(
        """
        UPDATE credit_records cr
        SET
            input_tokens = COALESCE(cr.input_tokens, NULLIF(tu.input_tokens, 0)),
            output_tokens = COALESCE(cr.output_tokens, NULLIF(tu.output_tokens, 0)),
            total_tokens = COALESCE(cr.total_tokens, NULLIF(tu.total_tokens, 0)),
            engine_type = CASE
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') IN ('text_prompt', 'file_parse', 'vision_input', 'video_analysis')
                  OR COALESCE(NULLIF(cr.credit_subject, 'unknown'), '') IN ('text', 'analysis')
                THEN COALESCE(NULLIF(cr.engine_type, 'unknown'), 'model')
                ELSE cr.engine_type
            END,
            engine_id = CASE
                WHEN COALESCE(NULLIF(cr.charge_kind, 'unknown'), '') IN ('text_prompt', 'file_parse', 'vision_input', 'video_analysis')
                  OR COALESCE(NULLIF(cr.credit_subject, 'unknown'), '') IN ('text', 'analysis')
                THEN COALESCE(cr.engine_id, tu.model_config_id)
                ELSE cr.engine_id
            END
        FROM token_usage tu
        WHERE cr.token_usage_id = tu.id
        """
    )

    # ModelConfig 快照统一冷备到 engine_*，engine_type=model。
    op.execute(
        """
        UPDATE credit_records cr
        SET
            engine_type = COALESCE(NULLIF(cr.engine_type, 'unknown'), 'model'),
            engine_name = COALESCE(cr.engine_name, mc.name),
            engine_provider = COALESCE(cr.engine_provider, mc.provider),
            engine_model_name = COALESCE(cr.engine_model_name, mc.model_name)
        FROM model_configs mc
        WHERE cr.engine_id = mc.id
          AND (
                cr.engine_type = 'model'
             OR cr.charge_kind IN ('text_prompt', 'file_parse', 'vision_input', 'video_analysis')
             OR cr.credit_subject IN ('text', 'analysis')
          )
        """
    )

    # 图片引擎快照。
    op.execute(
        """
        UPDATE credit_records cr
        SET
            engine_name = COALESCE(cr.engine_name, ie.name),
            engine_provider = COALESCE(cr.engine_provider, ie.provider),
            engine_model_name = COALESCE(cr.engine_model_name, ie.model_name),
            engine_type = COALESCE(NULLIF(cr.engine_type, 'unknown'), 'image')
        FROM image_engines ie
        WHERE cr.engine_id = ie.id
          AND (cr.engine_type = 'image' OR cr.media_type = 'image' OR cr.engine_type IS NULL OR cr.engine_type = 'unknown')
        """
    )

    # 视频引擎快照。
    op.execute(
        """
        UPDATE credit_records cr
        SET
            engine_name = COALESCE(cr.engine_name, ve.name),
            engine_provider = COALESCE(cr.engine_provider, ve.provider),
            engine_model_name = COALESCE(cr.engine_model_name, ve.model_name),
            engine_type = COALESCE(NULLIF(cr.engine_type, 'unknown'), 'video')
        FROM video_engines ve
        WHERE cr.engine_id = ve.id
          AND (cr.engine_type = 'video' OR cr.media_type = 'video' OR cr.engine_type IS NULL OR cr.engine_type = 'unknown')
        """
    )

    # 退款流水复制原消费流水快照，但 billing_scene 固定为 refund，方便筛选回退流水。
    op.execute(
        """
        UPDATE credit_records r
        SET
            owner_type = COALESCE(NULLIF(r.owner_type, 'unknown'), c.owner_type),
            owner_id = COALESCE(r.owner_id, c.owner_id),
            attempt_no = COALESCE(r.attempt_no, c.attempt_no),
            charge_kind = COALESCE(NULLIF(r.charge_kind, 'unknown'), c.charge_kind),
            charge_action = 'refund',
            credit_subject = COALESCE(NULLIF(r.credit_subject, 'unknown'), c.credit_subject, 'refund'),
            media_type = COALESCE(r.media_type, c.media_type),
            billing_scene = 'refund',
            source_module = COALESCE(NULLIF(r.source_module, 'unknown'), c.source_module),
            source_project_id = COALESCE(r.source_project_id, c.source_project_id),
            source_step_id = COALESCE(r.source_step_id, c.source_step_id),
            source_step_code = COALESCE(r.source_step_code, c.source_step_code),
            token_usage_id = COALESCE(r.token_usage_id, c.token_usage_id),
            input_tokens = COALESCE(r.input_tokens, c.input_tokens),
            output_tokens = COALESCE(r.output_tokens, c.output_tokens),
            total_tokens = COALESCE(r.total_tokens, c.total_tokens),
            engine_type = COALESCE(NULLIF(r.engine_type, 'unknown'), c.engine_type),
            engine_id = COALESCE(r.engine_id, c.engine_id),
            engine_name = COALESCE(r.engine_name, c.engine_name),
            engine_provider = COALESCE(r.engine_provider, c.engine_provider),
            engine_model_name = COALESCE(r.engine_model_name, c.engine_model_name),
            user_type_snapshot = COALESCE(r.user_type_snapshot, c.user_type_snapshot),
            frontend_user_kind_snapshot = COALESCE(r.frontend_user_kind_snapshot, c.frontend_user_kind_snapshot)
        FROM credit_records c
        WHERE r.type = 'refund'
          AND r.refund_for_biz_key = c.biz_key
        """
    )

    # 基础分类兜底。
    op.execute(
        """
        UPDATE credit_records
        SET
            charge_action = COALESCE(charge_action, CASE WHEN type = 'refund' THEN 'refund' ELSE 'charge' END),
            charge_kind = COALESCE(
                NULLIF(charge_kind, 'unknown'),
                CASE
                    WHEN type = 'recharge' THEN 'recharge'
                    WHEN type = 'refund' THEN 'refund'
                    ELSE 'unknown'
                END
            ),
            credit_subject = COALESCE(
                NULLIF(credit_subject, 'unknown'),
                CASE
                    WHEN type = 'recharge' THEN 'recharge'
                    WHEN type = 'refund' THEN 'refund'
                    WHEN charge_kind = 'media' THEN 'media'
                    WHEN charge_kind IN ('text_prompt', 'file_parse', 'vision_input') THEN 'text'
                    WHEN charge_kind = 'module_create' THEN 'module'
                    WHEN charge_kind = 'video_analysis' THEN 'analysis'
                    WHEN charge_kind = 'video_split' THEN 'split'
                    WHEN charge_kind = 'admin_adjust' THEN 'admin_adjust'
                    ELSE 'unknown'
                END
            ),
            source_module = COALESCE(NULLIF(source_module, 'unknown'), CASE WHEN type = 'recharge' THEN 'payment' ELSE 'unknown' END),
            billing_scene = COALESCE(
                NULLIF(billing_scene, 'unknown'),
                CASE
                    WHEN type = 'recharge' THEN 'recharge'
                    WHEN type = 'refund' THEN 'refund'
                    ELSE 'unknown'
                END
            ),
            owner_type = COALESCE(NULLIF(owner_type, 'unknown'), 'unknown'),
            engine_type = CASE
                WHEN charge_kind IN ('text_prompt', 'file_parse', 'vision_input', 'video_analysis')
                  OR credit_subject IN ('text', 'analysis')
                THEN COALESCE(NULLIF(engine_type, 'unknown'), 'model')
                ELSE engine_type
            END
        WHERE charge_kind IS NULL
           OR charge_kind = 'unknown'
           OR charge_action IS NULL
           OR credit_subject IS NULL
           OR credit_subject = 'unknown'
           OR source_module IS NULL
           OR source_module = 'unknown'
           OR billing_scene IS NULL
           OR billing_scene = 'unknown'
           OR owner_type IS NULL
           OR owner_type = 'unknown'
        """
    )

    # 用户快照兜底，避免后台筛选 NULL。
    op.execute(
        """
        UPDATE credit_records
        SET
            user_type_snapshot = COALESCE(user_type_snapshot, 'frontend'),
            frontend_user_kind_snapshot = COALESCE(frontend_user_kind_snapshot, 'external')
        WHERE user_type_snapshot IS NULL OR frontend_user_kind_snapshot IS NULL
        """
    )


def downgrade() -> None:
    op.drop_index(op.f("ix_users_user_type"), table_name="users")
    op.drop_index(op.f("ix_users_frontend_user_kind"), table_name="users")

    op.drop_index(op.f("ix_token_usage_source_step_code"), table_name="token_usage")
    op.drop_index(op.f("ix_token_usage_source_module"), table_name="token_usage")
    op.drop_index(op.f("ix_token_usage_owner_type"), table_name="token_usage")
    op.drop_index(op.f("ix_token_usage_owner_id"), table_name="token_usage")
    op.drop_index("ix_token_usage_owner", table_name="token_usage")
    op.drop_index("ix_token_usage_biz_key", table_name="token_usage")

    op.drop_index(op.f("ix_module_generation_steps_token_usage_id"), table_name="module_generation_steps")
    op.drop_index(op.f("ix_module_generation_steps_model_config_id"), table_name="module_generation_steps")

    op.drop_index(op.f("ix_credit_records_user_type_snapshot"), table_name="credit_records")
    op.drop_index("ix_credit_records_user_kind_time", table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_type"), table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_token_usage_id"), table_name="credit_records")
    op.drop_index("ix_credit_records_subject_media", table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_source_step_id"), table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_source_step_code"), table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_source_project_id"), table_name="credit_records")
    op.drop_index("ix_credit_records_source_module_scene", table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_source_module"), table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_owner_type"), table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_owner_id"), table_name="credit_records")
    op.drop_index("ix_credit_records_owner", table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_media_type"), table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_frontend_user_kind_snapshot"), table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_engine_type"), table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_engine_id"), table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_credit_subject"), table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_charge_kind"), table_name="credit_records")
    op.drop_index(op.f("ix_credit_records_billing_scene"), table_name="credit_records")

    op.drop_column("token_usage", "source_step_code")
    op.drop_column("token_usage", "source_module")
    op.drop_column("token_usage", "biz_key")
    op.drop_column("token_usage", "owner_id")
    op.drop_column("token_usage", "owner_type")

    op.drop_column("module_generation_steps", "text_credits_cost")
    op.drop_column("module_generation_steps", "total_tokens")
    op.drop_column("module_generation_steps", "output_tokens")
    op.drop_column("module_generation_steps", "input_tokens")
    op.drop_column("module_generation_steps", "model_config_id")
    op.drop_column("module_generation_steps", "token_usage_id")

    op.drop_column("credit_records", "frontend_user_kind_snapshot")
    op.drop_column("credit_records", "user_type_snapshot")
    op.drop_column("credit_records", "engine_model_name")
    op.drop_column("credit_records", "engine_provider")
    op.drop_column("credit_records", "engine_name")
    op.drop_column("credit_records", "engine_id")
    op.drop_column("credit_records", "engine_type")
    op.drop_column("credit_records", "total_tokens")
    op.drop_column("credit_records", "output_tokens")
    op.drop_column("credit_records", "input_tokens")
    op.drop_column("credit_records", "token_usage_id")
    op.drop_column("credit_records", "source_step_code")
    op.drop_column("credit_records", "source_step_id")
    op.drop_column("credit_records", "source_project_id")
    op.drop_column("credit_records", "source_module")
    op.drop_column("credit_records", "billing_scene")
    op.drop_column("credit_records", "media_type")
    op.drop_column("credit_records", "credit_subject")
    op.drop_column("credit_records", "charge_action")
    op.drop_column("credit_records", "charge_kind")
    op.drop_column("credit_records", "attempt_no")
    op.drop_column("credit_records", "owner_id")
    op.drop_column("credit_records", "owner_type")

    op.drop_column("users", "frontend_user_kind")
