"""Hardware-asset financial lifecycle — depreciation policy + book-value snapshot + disposal P&L columns + monthly snapshot cron seed. Revision ID: 283 Revises: 282 Create Date: 2026-06-13 Closes the finance-persona parity gap: finance could see software / cloud / AI / procurement cost in depth but had ZERO visibility into hardware asset value. This migration adds the storage for the hardware financial lifecycle (acquisition → depreciation → disposal P&L): 1. ``asset_depreciation_policies`` — per-category depreciation rule (method / useful_life_months / residual_rate / declining_rate). One ``asset_category IS NULL`` row = the platform default fallback. Seeded with a 3-year straight-line / 5% salvage default. 2. ``asset_book_value_snapshots`` — monthly materialised fleet net book value per ``(period, asset_category)``, CNY-normalised, backing the leadership depreciation-trend chart. 3. ``asset_items`` gains ``disposal_proceeds`` / ``disposal_book_value`` / ``disposal_currency`` — the disposal gain/loss inputs, frozen at disposal time and mirrored into the immutable disposal operation. 4. ``scheduled_jobs`` seed — ``asset.book_value_snapshot``, monthly on the 1st at 02:00 Asia/Shanghai (after month-end, before CN work hours). Enabled; idempotent on prior-month aggregates. No person FK on any new column (policy keys on the asset_category enum string; snapshots are aggregates) — TiDB-3780-safe by construction. Idempotent: tables guarded by information_schema; columns guarded by column-existence checks; job insert guarded by NOT EXISTS on handler_key. """ from __future__ import annotations import uuid import sqlalchemy as sa from alembic import op revision = "283" down_revision = "282" branch_labels = None depends_on = None _HANDLER = "asset.book_value_snapshot" _DESC = ( "每月 1 日 02:00(Asia/Shanghai)物化上一月末活跃硬件资产的账面价值" "(按品类、CNY 归一),写入 asset_book_value_snapshots,支撑折旧趋势图。" "幂等:按 (period, asset_category) upsert。" ) def _table_exists(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n LIMIT 1" ), {"n": name}, ).scalar() is not None def _column_exists(bind, table: str, column: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t " " AND COLUMN_NAME = :c LIMIT 1" ), {"t": table, "c": column}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() if not _table_exists(bind, "asset_depreciation_policies"): op.create_table( "asset_depreciation_policies", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("asset_category", sa.String(32), nullable=True), sa.Column( "method", sa.String(32), nullable=False, server_default="straight_line", ), sa.Column( "useful_life_months", sa.Integer, nullable=False, server_default="36", ), sa.Column( "residual_rate", sa.Numeric(5, 4), nullable=False, server_default="0.0500", ), sa.Column("declining_rate", sa.Numeric(6, 4), nullable=True), sa.Column( "is_active", sa.Boolean, nullable=False, server_default=sa.text("1"), ), sa.Column("notes", sa.Text, nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False, ), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), onupdate=sa.func.now(), nullable=False, ), sa.UniqueConstraint( "asset_category", name="uq_asset_depreciation_policies_category", ), ) # Seed the platform default fallback (NULL category). bind.execute( sa.text( "INSERT INTO asset_depreciation_policies " "(id, asset_category, method, useful_life_months, " " residual_rate, declining_rate, is_active, notes, " " created_at, updated_at) " "VALUES (:id, NULL, 'straight_line', 36, 0.0500, NULL, 1, " ":notes, NOW(), NOW())" ), { "id": str(uuid.uuid4()), "notes": "平台默认折旧策略:3 年直线法,5% 残值。", }, ) if not _table_exists(bind, "asset_book_value_snapshots"): op.create_table( "asset_book_value_snapshots", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("period", sa.String(7), nullable=False), sa.Column("asset_category", sa.String(32), nullable=True), sa.Column( "original_cost_cny", sa.Numeric(16, 2), nullable=False, server_default="0", ), sa.Column( "accumulated_depreciation_cny", sa.Numeric(16, 2), nullable=False, server_default="0", ), sa.Column( "net_book_value_cny", sa.Numeric(16, 2), nullable=False, server_default="0", ), sa.Column( "asset_count", sa.Integer, nullable=False, server_default="0", ), sa.Column( "undepreciable_count", sa.Integer, nullable=False, server_default="0", ), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False, ), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), onupdate=sa.func.now(), nullable=False, ), sa.UniqueConstraint( "period", "asset_category", name="uq_asset_bv_snapshots_period_category", ), sa.Index("ix_asset_bv_snapshots_period", "period"), ) for col, ddl in ( ("disposal_proceeds", sa.Column("disposal_proceeds", sa.Numeric(12, 2), nullable=True)), ("disposal_book_value", sa.Column("disposal_book_value", sa.Numeric(12, 2), nullable=True)), ("disposal_currency", sa.Column("disposal_currency", sa.String(8), nullable=True)), ): if not _column_exists(bind, "asset_items", col): op.add_column("asset_items", ddl) if _table_exists(bind, "scheduled_jobs"): bind.execute( sa.text( "INSERT INTO scheduled_jobs " "(id, name, display_name, description, handler_key, " " trigger_type, cron_expr, timezone, enabled, config, " " run_count, failure_count, created_at, updated_at) " "SELECT :id, :name, :display, :desc, :handler, 'cron', " ":cron, 'Asia/Shanghai', 1, '{}', 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " "(SELECT 1 FROM scheduled_jobs WHERE handler_key = :handler)" ), { "id": str(uuid.uuid4()), "name": _HANDLER, "display": "硬件资产账面价值月度物化", "desc": _DESC, "handler": _HANDLER, "cron": "0 2 1 * *", }, ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "scheduled_jobs"): bind.execute( sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :k"), {"k": _HANDLER}, ) for col in ("disposal_currency", "disposal_book_value", "disposal_proceeds"): if _column_exists(bind, "asset_items", col): op.drop_column("asset_items", col) if _table_exists(bind, "asset_book_value_snapshots"): op.drop_table("asset_book_value_snapshots") if _table_exists(bind, "asset_depreciation_policies"): op.drop_table("asset_depreciation_policies")