import asyncio from sqlalchemy import text from sqlalchemy.ext.asyncio import create_async_engine from database.models import Base from config import settings from utils.logger import session_logger import urllib.parse async def _ensure_manual_sleep_mode_column(conn) -> None: result = await conn.execute( text( """ SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'device_settings' AND COLUMN_NAME = 'manual_sleep_mode' """ ) ) if int(result.scalar() or 0) > 0: return await conn.execute( text( """ ALTER TABLE device_settings ADD COLUMN manual_sleep_mode TINYINT NOT NULL DEFAULT 0 AFTER sleep_mode """ ) ) async def _ensure_schedule_suppressed_until_column(conn) -> None: result = await conn.execute( text( """ SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'device_settings' AND COLUMN_NAME = 'schedule_suppressed_until' """ ) ) if int(result.scalar() or 0) > 0: return await conn.execute( text( """ ALTER TABLE device_settings ADD COLUMN schedule_suppressed_until DATETIME NULL AFTER manual_sleep_mode """ ) ) async def _ensure_columns(conn, *, table_name: str, columns: list[tuple[str, str]]) -> None: for column_name, definition in columns: result = await conn.execute( text( """ SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :table_name AND COLUMN_NAME = :column_name """ ), {"table_name": table_name, "column_name": column_name}, ) if int(result.scalar() or 0) > 0: continue await conn.execute(text(f"ALTER TABLE {table_name} ADD COLUMN {definition}")) async def _ensure_device_alarm_location_columns(conn) -> None: await _ensure_columns( conn, table_name="device_alarm_events", columns=[ ("coord_type", "coord_type VARCHAR(16) NULL AFTER source_msg_id"), ("lat", "lat DECIMAL(10, 7) NULL AFTER coord_type"), ("lng", "lng DECIMAL(10, 7) NULL AFTER lat"), ("location_updated_at", "location_updated_at DATETIME NULL AFTER lng"), ("address", "address VARCHAR(255) NULL AFTER location_updated_at"), ("address_resolved_at", "address_resolved_at DATETIME NULL AFTER address"), ("address_resolve_status", "address_resolve_status TINYINT NULL AFTER address_resolved_at"), ], ) async def _ensure_child_location_address_columns(conn) -> None: await _ensure_columns( conn, table_name="child_location_current", columns=[ ("address", "address VARCHAR(255) NULL AFTER updated_at"), ("address_resolved_at", "address_resolved_at DATETIME NULL AFTER address"), ("address_resolve_status", "address_resolve_status TINYINT NULL AFTER address_resolved_at"), ], ) async def _ensure_bind_session_card_columns(conn) -> None: await _ensure_columns( conn, table_name="device_bind_sessions", columns=[ ("bind_mode", "bind_mode TINYINT NOT NULL DEFAULT 1 AFTER status"), ("card_uuid", "card_uuid VARCHAR(64) NULL AFTER bind_mode"), ], ) result = await conn.execute( text( """ SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'device_bind_sessions' AND INDEX_NAME = 'idx_device_bind_sessions_card_uuid' """ ) ) if int(result.scalar() or 0) == 0: await conn.execute(text("ALTER TABLE device_bind_sessions ADD INDEX idx_device_bind_sessions_card_uuid (card_uuid)")) async def _ensure_cards_allow_multiple_per_device(conn) -> None: result = await conn.execute( text( """ SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'cards' AND INDEX_NAME = 'idx_cards_device_id' """ ) ) if int(result.scalar() or 0) == 0: await conn.execute(text("ALTER TABLE cards ADD INDEX idx_cards_device_id (device_id)")) result = await conn.execute( text( """ SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'cards' AND INDEX_NAME = 'uq_cards_device_id' """ ) ) if int(result.scalar() or 0) > 0: await conn.execute(text("ALTER TABLE cards DROP INDEX uq_cards_device_id")) async def _ensure_device_family_tables(conn) -> None: await conn.execute( text( """ CREATE TABLE IF NOT EXISTS device_family_members ( id BIGINT NOT NULL AUTO_INCREMENT, device_id VARCHAR(64) NOT NULL, user_id BIGINT NOT NULL, role TINYINT NOT NULL DEFAULT 2, display_name VARCHAR(64) NULL, status TINYINT NOT NULL DEFAULT 1, invited_by_user_id BIGINT NULL, joined_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, removed_at DATETIME NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uq_device_family_member (device_id, user_id), KEY idx_device_family_members_device_status (device_id, status), KEY idx_device_family_members_user_status (user_id, status), KEY idx_device_family_members_invited_by (invited_by_user_id), CONSTRAINT fk_device_family_members_device FOREIGN KEY (device_id) REFERENCES device_auth (device_id), CONSTRAINT fk_device_family_members_user FOREIGN KEY (user_id) REFERENCES parents (user_id), CONSTRAINT fk_device_family_members_invited_by FOREIGN KEY (invited_by_user_id) REFERENCES parents (user_id) ON DELETE SET NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ ) ) await _ensure_columns( conn, table_name="device_family_members", columns=[ ("display_name", "display_name VARCHAR(64) NULL AFTER role"), ], ) await conn.execute( text( """ CREATE TABLE IF NOT EXISTS device_family_invitations ( id BIGINT NOT NULL AUTO_INCREMENT, invite_token CHAR(36) NOT NULL, device_id VARCHAR(64) NOT NULL, owner_user_id BIGINT NOT NULL, status TINYINT NOT NULL DEFAULT 1, expires_at DATETIME NOT NULL, accepted_by_user_id BIGINT NULL, accepted_at DATETIME NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uq_device_family_invite_token (invite_token), KEY idx_device_family_invites_device_status (device_id, status), KEY idx_device_family_invites_owner_status (owner_user_id, status), KEY idx_device_family_invites_expires_at (expires_at), KEY idx_device_family_invites_accepted_by (accepted_by_user_id), CONSTRAINT fk_device_family_invites_device FOREIGN KEY (device_id) REFERENCES device_auth (device_id), CONSTRAINT fk_device_family_invites_owner FOREIGN KEY (owner_user_id) REFERENCES parents (user_id), CONSTRAINT fk_device_family_invites_accepted_by FOREIGN KEY (accepted_by_user_id) REFERENCES parents (user_id) ON DELETE SET NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ ) ) await conn.execute( text( """ INSERT INTO device_family_members ( device_id, user_id, role, status, invited_by_user_id, joined_at, created_at, updated_at ) SELECT db.device_id, db.owner_user_id, 1, 1, NULL, COALESCE(db.bound_at, CURRENT_TIMESTAMP), CURRENT_TIMESTAMP, CURRENT_TIMESTAMP FROM device_bindings AS db LEFT JOIN device_family_members AS dfm ON dfm.device_id = db.device_id AND dfm.user_id = db.owner_user_id WHERE db.status = 1 AND dfm.id IS NULL """ ) ) async def init_db(): """初始化数据库,创建所有表""" try: encoded_password = urllib.parse.quote_plus(settings.db_password) connection_string = f"mysql+aiomysql://{settings.db_user}:{encoded_password}@{settings.db_host}:{settings.db_port}/{settings.db_name}" engine = create_async_engine( connection_string, echo=settings.db_echo ) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) await _ensure_manual_sleep_mode_column(conn) await _ensure_schedule_suppressed_until_column(conn) await _ensure_device_alarm_location_columns(conn) await _ensure_child_location_address_columns(conn) await _ensure_bind_session_card_columns(conn) await _ensure_cards_allow_multiple_per_device(conn) await _ensure_device_family_tables(conn) await engine.dispose() session_logger.info("system", "database", "数据库表已成功创建") return True except Exception as e: session_logger.error("system", "database", f"创建数据库表失败: {str(e)}") return False