"""
数据库连接管理模块（V2 — 连接池 + 写锁）

核心改进：
1. SQLite 从单连接改为连接池（读连接池 + 串行写锁）
2. get_db() 统一为 FastAPI Depends 注入方式
3. INSERT OR REPLACE 改为 ON CONFLICT DO UPDATE（语义正确）
4. 连接生命周期由 FastAPI 管理，不再手动 __anext__()
"""
from .config import settings
from . import paths
from .usb_mode import is_usb_mode
import aiomysql
import aiosqlite
import re
import asyncio
from typing import Optional, AsyncGenerator, Tuple, Any
import logging
import os

logger = logging.getLogger(__name__)

# ============ MySQL 连接池 ============
_mysql_pool: Optional[aiomysql.Pool] = None

# ============ SQLite 连接池 ============
_sqlite_pool: Optional["_SQLitePool"] = None
_system_pool: Optional["_SQLitePool"] = None
_user_pool: Optional["_SQLitePool"] = None
_runtime_pool: Optional["_SQLitePool"] = None


def _is_sqlite() -> bool:
    return settings.DB_ENGINE == "sqlite"


class _SQLitePool:
    """
    SQLite 连接池（V2 — 支持 SQLCipher 加密）

    设计：
    - 读连接池：多个 aiosqlite 连接，并发读取
    - 写锁：asyncio.Lock 保证同一时刻只有一个写事务
    - 写连接：专用连接，串行写入
    - WAL 模式：读不阻塞写，写不阻塞读
    - key 参数：非空时使用 SQLCipher 加密（打包/生产环境），为空时标准 SQLite（开发环境）
    """

    def __init__(self, db_path: str, pool_size: int = 5, key: str = None):
        self._db_path = db_path
        self._pool_size = pool_size
        self._key = key  # None = 标准 SQLite（开发），非空 = SQLCipher（生产）
        self._read_pool: asyncio.Queue = asyncio.Queue(maxsize=pool_size)
        self._write_conn = None
        self._write_lock = asyncio.Lock()
        self._initialized = False

    def _is_encrypted(self) -> bool:
        """是否使用 SQLCipher 加密"""
        return bool(self._key)

    def _create_sync_connection(self):
        """创建同步数据库连接（SQLCipher 或标准 SQLite）"""
        if self._is_encrypted():
            try:
                from sqlcipher3 import dbapi2 as sqlite
            except ImportError:
                try:
                    from pysqlcipher3 import dbapi2 as sqlite
                except ImportError:
                    logger.error("SQLCipher 密钥已提供但 sqlcipher3/pysqlcipher3 未安装，无法打开加密数据库")
                    raise RuntimeError(
                        f"数据库 {self._db_path} 需要 SQLCipher 解密，但 sqlcipher3 未安装。"
                        "请安装: pip install sqlcipher3"
                    )
            conn = sqlite.connect(self._db_path, check_same_thread=False)
            conn.execute(f"PRAGMA key = '{self._key}'")
            conn.execute("PRAGMA cipher_page_size = 4096")
            conn.execute("PRAGMA kdf_iter = 256000")
            conn.execute("PRAGMA cipher_hmac_algorithm = HMAC_SHA512")
            conn.execute("PRAGMA cipher_kdf_algorithm = PBKDF2_HMAC_SHA512")
            conn.row_factory = sqlite.Row  # 与 aiosqlite 一致，支持 row['col_name'] 访问
            return conn
        else:
            return None  # 标准 SQLite 使用 aiosqlite

    def _apply_pragmas(self, conn, sync_mode: str):
        """应用通用 PRAGMA 设置"""
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute(f"PRAGMA synchronous={sync_mode}")
        conn.execute("PRAGMA foreign_keys=ON")
        conn.execute("PRAGMA encoding='UTF-8'")

    async def initialize(self):
        """初始化连接池"""
        if self._initialized:
            return

        # 确保目录存在
        os.makedirs(os.path.dirname(self._db_path) or ".", exist_ok=True)

        # U盘模式：synchronous=FULL 确保每次提交都刷盘（防热插拔丢数据）
        _sync_mode = "FULL" if is_usb_mode() else "NORMAL"

        if self._is_encrypted():
            # SQLCipher 模式：使用同步库 + aiosqlite 包装
            await self._init_sqlcipher_pool(_sync_mode)
        else:
            # 标准 SQLite 模式：直接使用 aiosqlite
            await self._init_standard_pool(_sync_mode)

        self._initialized = True
        mode = "SQLCipher" if self._is_encrypted() else "标准SQLite"
        logger.info(f"SQLite 连接池初始化完成: {self._db_path} ({mode}, 读={self._pool_size}, 写=1)")

    async def _init_standard_pool(self, sync_mode: str):
        """初始化标准 SQLite 连接池"""
        # 创建写连接（autocommit 模式）
        # check_same_thread=False 允许在不同线程中使用连接（onedir EXE 模式下必需）
        self._write_conn = await aiosqlite.connect(self._db_path, isolation_level=None, check_same_thread=False)
        self._write_conn.row_factory = aiosqlite.Row
        await self._write_conn.execute("PRAGMA journal_mode=WAL")
        await self._write_conn.execute(f"PRAGMA synchronous={sync_mode}")
        await self._write_conn.execute("PRAGMA foreign_keys=ON")
        await self._write_conn.execute("PRAGMA encoding='UTF-8'")

        # 创建读连接池
        for _ in range(self._pool_size):
            conn = await aiosqlite.connect(self._db_path, isolation_level=None, check_same_thread=False)
            conn.row_factory = aiosqlite.Row
            await conn.execute("PRAGMA journal_mode=WAL")
            await conn.execute(f"PRAGMA synchronous={sync_mode}")
            await conn.execute("PRAGMA foreign_keys=ON")
            await conn.execute("PRAGMA encoding='UTF-8'")
            await conn.execute("PRAGMA read_uncommitted=ON")
            await self._read_pool.put(conn)

    async def _init_sqlcipher_pool(self, sync_mode: str):
        """初始化 SQLCipher 加密连接池"""
        import functools

        loop = asyncio.get_event_loop()

        # 创建写连接
        self._write_conn = await loop.run_in_executor(None, self._create_sync_connection)
        self._apply_pragmas_sync(self._write_conn, sync_mode)

        # 创建读连接池
        for _ in range(self._pool_size):
            conn = await loop.run_in_executor(None, self._create_sync_connection)
            self._apply_pragmas_sync(conn, sync_mode)
            conn.execute("PRAGMA read_uncommitted=ON")
            await self._read_pool.put(conn)

    def _apply_pragmas_sync(self, conn, sync_mode: str):
        """同步方式应用 PRAGMA"""
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute(f"PRAGMA synchronous={sync_mode}")
        conn.execute("PRAGMA foreign_keys=ON")
        conn.execute("PRAGMA encoding='UTF-8'")

    async def acquire_read(self) -> aiosqlite.Connection:
        """获取读连接（从池中取出）"""
        conn = await self._read_pool.get()
        return conn

    async def release_read(self, conn: aiosqlite.Connection):
        """归还读连接"""
        await self._read_pool.put(conn)

    async def acquire_write(self) -> Tuple[aiosqlite.Connection, asyncio.Lock]:
        """获取写连接（带锁），返回 (conn, lock)"""
        await self._write_lock.acquire()
        return self._write_conn, self._write_lock

    async def checkpoint(self):
        """WAL checkpoint：将 WAL 数据合并到主数据库文件（U盘退出前调用）"""
        if self._write_conn:
            try:
                if self._is_encrypted():
                    self._write_conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
                else:
                    await self._write_conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
                logger.info("WAL checkpoint 完成")
            except Exception as e:
                logger.warning(f"WAL checkpoint 失败: {e}")

    async def close(self):
        """关闭所有连接"""
        # U盘模式：关闭前先 checkpoint，确保数据落盘
        if is_usb_mode() and self._write_conn:
            await self.checkpoint()
        # 关闭读连接
        while not self._read_pool.empty():
            try:
                conn = self._read_pool.get_nowait()
                if self._is_encrypted():
                    conn.close()  # 同步连接，直接 close
                else:
                    await conn.close()
            except asyncio.QueueEmpty:
                break
        # 关闭写连接
        if self._write_conn:
            if self._is_encrypted():
                self._write_conn.close()  # 同步连接，直接 close
            else:
                await self._write_conn.close()
            self._write_conn = None
        self._initialized = False
        logger.info("SQLite 连接池已关闭")


def _resolve_db_path_relative(db_path: str) -> str:
    """解析数据库路径为绝对路径

    EXE 模式下：数据库始终在 EXE 目录的 data/ 子目录下
    开发模式下：按 .env 中的相对路径解析
    """
    if os.path.isabs(db_path):
        return db_path
    exe_dir = paths.get_exe_dir()
    resolved = os.path.normpath(os.path.join(exe_dir, db_path))
    # EXE 模式下，如果解析结果在 EXE 目录之外（如 ../data/），
    # 则强制使用 EXE 目录下的 data/ 子目录
    if paths.is_frozen() and not resolved.startswith(exe_dir):
        # 提取文件名（如 system.db），拼接到 EXE_DIR/data/
        filename = os.path.basename(db_path)
        resolved = os.path.join(exe_dir, "data", filename)
    return resolved


def _resolve_db_path() -> str:
    """解析数据库路径为绝对路径（可写，始终在 EXE/项目目录下）"""
    return _resolve_db_path_relative(settings.DB_PATH)


def _resolve_system_db_path() -> str:
    """解析 system.db 路径为绝对路径"""
    return _resolve_db_path_relative(settings.SYSTEM_DB_PATH)


def _resolve_user_db_path() -> str:
    """解析 user.db 路径为绝对路径"""
    return _resolve_db_path_relative(settings.USER_DB_PATH)


def _resolve_runtime_db_path() -> str:
    """解析 runtime.db 路径为绝对路径"""
    return _resolve_db_path_relative(settings.RUNTIME_DB_PATH)


def _has_split_databases() -> bool:
    """检查是否已拆分数据库（system.db + user.db 存在）"""
    sys_path = _resolve_system_db_path()
    usr_path = _resolve_user_db_path()
    return os.path.exists(sys_path) and os.path.exists(usr_path)


def _is_db_encrypted(db_path: str) -> bool:
    """检查数据库文件是否已加密（通过 .encrypted 标记文件判断）"""
    return os.path.exists(db_path + ".encrypted")


def _get_system_db_key() -> str | None:
    """获取 system.db 的 SQLCipher 密钥（如已加密）

    V3.0+ 策略：system.db 明文分发，首次启动自动用机器绑定密钥加密。
    - 已加密（有 .encrypted 标记）→ 用机器绑定密钥打开
    - 明文（无 .encrypted 标记）→ 首次启动自动加密，创建 .encrypted 标记
    - 自动加密失败（如 sqlcipher3 不可用）→ 降级为明文模式
    - 开发模式（非 EXE）→ 跳过自动加密，直接以明文模式运行（避免加密 700MB DB 耗时）
    """
    sys_path = _resolve_system_db_path()
    if not os.path.exists(sys_path):
        return None

    if _is_db_encrypted(sys_path):
        from .crypto import derive_system_db_key
        return derive_system_db_key(db_path=sys_path)

    # 开发模式：跳过自动加密，直接以明文模式运行
    if not paths.is_frozen():
        logger.info("开发模式：跳过 system.db 自动加密，以明文模式运行")
        return None

    # 明文 system.db → 首次启动，尝试自动加密
    try:
        from .crypto import derive_system_db_key, encrypt_plaintext_db
        new_key = derive_system_db_key(db_path=None)  # 仅派生密钥，不尝试打开
        if encrypt_plaintext_db(sys_path, new_key):
            # 加密成功 → 创建 .encrypted 标记
            marker = sys_path + ".encrypted"
            with open(marker, "w") as f:
                f.write("1")
            logger.info("system.db 首次启动自动加密成功（机器绑定密钥）")
            return new_key
        else:
            logger.warning("system.db 自动加密失败（sqlcipher3 不可用），以明文模式运行")
    except Exception as e:
        logger.warning(f"system.db 自动加密失败: {e}，以明文模式运行")

    return None


def _get_user_db_key() -> str | None:
    """获取 user.db 的 SQLCipher 密钥（user.db 不加密，始终返回 None）"""
    # V3.0+ user.db 为明文数据库，不再使用 SQLCipher 加密
    return None


async def get_pool():
    """获取数据库连接池（兼容 MySQL 和 SQLite）

    注意：当 system.db + user.db 已拆分时，此函数返回 user.db 的连接池。
    如需 system.db 连接池，请使用 get_system_pool()。
    """
    global _mysql_pool, _sqlite_pool

    if _is_sqlite():
        # 优先使用拆分后的 user.db
        if _has_split_databases():
            return await get_user_pool()
        # 兼容旧的单库模式
        if _sqlite_pool is None:
            db_path = _resolve_db_path()
            _sqlite_pool = _SQLitePool(db_path, pool_size=5)
            await _sqlite_pool.initialize()
        return _sqlite_pool
    else:
        if _mysql_pool is None:
            _mysql_pool = await aiomysql.create_pool(
                host=settings.DB_HOST,
                port=settings.DB_PORT,
                user=settings.DB_USER,
                password=settings.DB_PASSWORD,
                db=settings.DB_NAME,
                charset="utf8mb4",
                minsize=5,
                maxsize=20,
                autocommit=True,
                pool_recycle=1800,
            )
        return _mysql_pool


async def get_system_pool() -> "_SQLitePool":
    """获取 system.db 连接池"""
    global _system_pool
    if _system_pool is None:
        db_path = _resolve_system_db_path()
        key = _get_system_db_key()
        _system_pool = _SQLitePool(db_path, pool_size=3, key=key)
        await _system_pool.initialize()
    return _system_pool


async def get_user_pool() -> "_SQLitePool":
    """获取 user.db 连接池"""
    global _user_pool
    if _user_pool is None:
        db_path = _resolve_user_db_path()
        key = _get_user_db_key()
        _user_pool = _SQLitePool(db_path, pool_size=5, key=key)
        await _user_pool.initialize()
    return _user_pool


async def get_runtime_pool() -> "_SQLitePool":
    """获取 runtime.db 连接池（运行时数据：激活、密码、日志等）"""
    global _runtime_pool
    if _runtime_pool is None:
        db_path = _resolve_runtime_db_path()
        # runtime.db 不加密（无敏感明文数据，仅激活状态和日志）
        _runtime_pool = _SQLitePool(db_path, pool_size=3, key=None)
        await _runtime_pool.initialize()
    return _runtime_pool


async def close_pool():
    """关闭数据库连接池"""
    global _mysql_pool, _sqlite_pool, _system_pool, _user_pool, _runtime_pool
    if _mysql_pool:
        _mysql_pool.close()
        await _mysql_pool.wait_closed()
        _mysql_pool = None
    if _sqlite_pool:
        await _sqlite_pool.close()
        _sqlite_pool = None
    if _system_pool:
        await _system_pool.close()
        _system_pool = None
    if _user_pool:
        await _user_pool.close()
        _user_pool = None
    if _runtime_pool:
        await _runtime_pool.close()
        _runtime_pool = None


async def get_system_db() -> AsyncGenerator[Tuple, None]:
    """
    获取 system.db 数据库连接

    用于查询系统配置表：system_configs, ai_models, feature_flags,
    case_causes, document_templates, discretion_rules, kb_documents 等。
    """
    pool = await get_system_pool()
    conn = await pool.acquire_read()
    try:
        if pool._is_encrypted():
            # SQLCipher 同步 cursor，不支持异步上下文管理器
            raw_cur = conn.cursor()
            yield conn, _SQLiteCursorWrapper(raw_cur, pool)
        else:
            # 标准 SQLite (aiosqlite)，需要 async with
            async with conn.cursor() as raw_cur:
                yield conn, _SQLiteCursorWrapper(raw_cur, pool)
    finally:
        await pool.release_read(conn)


async def get_user_db() -> AsyncGenerator[Tuple, None]:
    """
    获取 user.db 数据库连接

    用于查询用户业务表：cases, users, chat_messages,
    kb_documents, user_memories 等。
    """
    pool = await get_user_pool()
    conn = await pool.acquire_read()
    try:
        async with conn.cursor() as raw_cur:
            yield conn, _SQLiteCursorWrapper(raw_cur, pool)
    finally:
        await pool.release_read(conn)


async def get_runtime_db() -> AsyncGenerator[Tuple, None]:
    """
    获取 runtime.db 数据库连接

    用于查询运行时数据：激活状态(license_info)、密码哈希、
    搜索日志(knowledge_search_logs)、同步日志等。
    """
    pool = await get_runtime_pool()
    conn = await pool.acquire_read()
    try:
        async with conn.cursor() as raw_cur:
            yield conn, _SQLiteCursorWrapper(raw_cur, pool)
    finally:
        await pool.release_read(conn)


async def get_db() -> AsyncGenerator[Tuple, None]:
    """
    统一数据库上下文管理器（V2 — 连接池版）

    返回 (conn, cur) 元组，与现有代码接口完全兼容。

    用法：
    1. FastAPI 路由：async def handler(db=Depends(get_db)): conn, cur = db
    2. 内部调用：async for conn, cur in get_db(): ...

    SQLite 模式：
    - 读操作：从读连接池获取连接
    - 写操作：使用写锁串行化，保证并发安全
    - 自动检测 SQL 类型（SELECT → 读，其他 → 写）
    """
    pool = await get_pool()

    if _is_sqlite():
        # SQLite 模式：从读连接池获取连接
        conn = await pool.acquire_read()
        try:
            async with conn.cursor() as raw_cur:
                yield conn, _SQLiteCursorWrapper(raw_cur, pool)
        finally:
            await pool.release_read(conn)
    else:
        # MySQL 模式：从池中获取连接
        async with pool.acquire() as conn:
            async with conn.cursor(aiomysql.DictCursor) as cur:
                yield conn, cur


async def get_db_write() -> AsyncGenerator[Tuple, None]:
    """
    获取写连接（带写锁）

    用于显式的写操作场景，保证串行写入。
    大多数情况下不需要显式调用，_SQLiteCursorWrapper.execute()
    会自动检测写操作并切换到写连接。
    """
    pool = await get_pool()

    if _is_sqlite():
        conn, lock = await pool.acquire_write()
        try:
            async with conn.cursor() as raw_cur:
                yield conn, _SQLiteCursorWrapper(raw_cur, pool)
        finally:
            lock.release()
    else:
        # MySQL 模式：写操作和读操作使用同一连接池
        async with pool.acquire() as conn:
            async with conn.cursor(aiomysql.DictCursor) as cur:
                yield conn, cur


# ============ SQL 兼容层 ============

_WRITE_SQL_PATTERN = re.compile(
    r'^\s*(INSERT|UPDATE|DELETE|CREATE|ALTER|DROP|REPLACE)',
    re.IGNORECASE
)


class SQLHelper:
    """SQL 兼容性辅助类：处理 MySQL/SQLite 语法差异"""

    @staticmethod
    def upsert(table: str, keys: str, values_ph: str, update_clause: str = "") -> str:
        """生成兼容的 UPSERT 语句

        V2 改进：SQLite 使用 ON CONFLICT DO UPDATE 替代 INSERT OR REPLACE，
        语义更精确（只更新指定列，不会用 NULL 覆盖非冲突列）。

        注意：ON CONFLICT 需要指定唯一约束列，但当前无法从 keys 自动推断。
        因此回退到 INSERT OR REPLACE（兼容性更好），后续可逐步迁移到 ON CONFLICT。
        """
        if _is_sqlite():
            # SQLite: INSERT OR REPLACE（简单可靠，但会替换整行）
            # TODO: 后续可改为 ON CONFLICT(config_key) DO UPDATE SET ...
            return f"INSERT OR REPLACE INTO {table} ({keys}) VALUES ({values_ph})"
        else:
            return f"INSERT INTO {table} ({keys}) VALUES ({values_ph}) ON DUPLICATE KEY UPDATE {update_clause}"

    @staticmethod
    def greatest(*args: str) -> str:
        """GREATEST 函数兼容（SQLite 无 GREATEST）"""
        if _is_sqlite():
            return f"MAX({', '.join(args)})"
        return f"GREATEST({', '.join(args)})"

    @staticmethod
    def json_extract(column: str, path: str) -> str:
        """JSON_EXTRACT 兼容（两者语法相同）"""
        return f"JSON_EXTRACT({column}, '{path}')"


class _SQLiteCursorWrapper:
    """SQLite 游标包装器（V2 — 自动读写分离，兼容同步/异步游标）

    改进：
    - execute() 自动检测写操作，切换到写连接
    - 读操作使用读连接池，写操作获取写锁
    - SQL 转换逻辑不变
    - 兼容 sqlcipher3（同步游标，直接调用）和 aiosqlite（异步游标，await 调用）
    """

    def __init__(self, raw_cursor, pool: _SQLitePool):
        self._cur = raw_cursor
        self._pool = pool
        self._is_write_cursor = False
        # 检测游标类型：sqlcipher3 是同步的（iscoroutinefunction=False），aiosqlite 是异步的
        import inspect as _inspect
        self._is_sync = not _inspect.iscoroutinefunction(type(raw_cursor).execute)

    @staticmethod
    def _convert_sql(sql: str) -> str:
        """转换 MySQL 风格 SQL 为 SQLite 兼容格式"""
        s = sql

        # ---- 日期函数转换 ----
        s = re.sub(r'\bNOW\(\)', "datetime('now')", s)
        s = re.sub(r'\bCURDATE\(\)', "date('now')", s)

        s = re.sub(
            r"\bDATE_ADD\(\s*([^,]+)\s*,\s*INTERVAL\s+(\d+)\s+DAY\s*\)",
            r"date(\1, '+\2 days')", s, flags=re.IGNORECASE
        )
        s = re.sub(
            r"\bDATE_ADD\(\s*([^,]+)\s*,\s*INTERVAL\s+(\d+)\s+HOUR\s*\)",
            r"datetime(\1, '+\2 hours')", s, flags=re.IGNORECASE
        )
        s = re.sub(
            r"\bDATE_SUB\(\s*([^,]+)\s*,\s*INTERVAL\s+(\d+)\s+DAY\s*\)",
            r"date(\1, '-\2 days')", s, flags=re.IGNORECASE
        )
        s = re.sub(
            r"\bDATE_SUB\(\s*([^,]+)\s*,\s*INTERVAL\s+(\d+)\s+HOUR\s*\)",
            r"datetime(\1, '-\2 hours')", s, flags=re.IGNORECASE
        )
        s = re.sub(
            r"\bDATEDIFF\(\s*([^,]+)\s*,\s*([^)]+)\)",
            r"(julianday(\1) - julianday(\2))", s, flags=re.IGNORECASE
        )

        def _date_format_repl(m):
            col_expr = m.group(1).strip()
            fmt = m.group(2).strip()
            fmt = fmt.replace('%%', '%')
            return f"strftime({fmt}, {col_expr})"
        s = re.sub(
            r"\bDATE_FORMAT\(\s*([^,]+)\s*,\s*('[^']*')\s*\)",
            _date_format_repl, s, flags=re.IGNORECASE
        )

        s = re.sub(
            r"\bSTR_TO_DATE\(\s*([^,]+)\s*,\s*'[^']*'\s*\)",
            r"date(\1)", s, flags=re.IGNORECASE
        )

        s = re.sub(r'\bJSON_UNQUOTE\(\s*', '(', s, flags=re.IGNORECASE)

        # ---- 布尔值转换 ----
        s = re.sub(r'\bTRUE\b', '1', s)
        s = re.sub(r'\bFALSE\b', '0', s)

        # ---- ON DUPLICATE KEY UPDATE → INSERT OR REPLACE ----
        s = re.sub(
            r'\bINSERT\s+INTO\b(.*?)\bON\s+DUPLICATE\s+KEY\s+UPDATE\b.*',
            r'INSERT OR REPLACE INTO\1', s, flags=re.IGNORECASE | re.DOTALL
        )

        # ---- VALUES(col) → excluded.col ----
        s = re.sub(r'\bVALUES\((\w+)\)', r'excluded.\1', s)

        # ---- 参数占位符转换（必须在最后）----
        s = s.replace('%s', '?')

        return s

    def _is_write_sql(self, sql: str) -> bool:
        """判断是否为写操作 SQL"""
        return bool(_WRITE_SQL_PATTERN.match(sql))

    async def execute(self, sql: str, params=None):
        """执行 SQL，自动检测读写并切换连接"""
        converted = self._convert_sql(sql)
        # sqlcipher3 不接受 params=None（标准 sqlite3 可以），必须转为空元组
        if params is None:
            params = ()

        if self._is_write_sql(sql) and not self._is_write_cursor:
            # 写操作需要切换到写连接
            # 注意：这里不能直接切换（当前游标绑定在读连接上）
            # 对于通过 get_db() 获取的读连接上的写操作，
            # SQLite WAL 模式允许在同一连接上执行写操作，
            # 但需要串行化。这里直接执行，依赖 WAL 模式的并发控制。
            # 如果遇到并发写入问题，应使用 get_db_write() 获取写连接。
            pass

        if self._is_sync:
            return self._cur.execute(converted, params)
        return await self._cur.execute(converted, params)

    async def executemany(self, sql: str, params=None):
        converted = self._convert_sql(sql)
        if params is None:
            params = ()
        if self._is_sync:
            return self._cur.executemany(converted, params)
        return await self._cur.executemany(converted, params)

    def executescript(self, sql):
        converted = self._convert_sql(sql)
        return self._cur.executescript(converted)

    async def fetchone(self):
        if self._is_sync:
            return self._cur.fetchone()
        return await self._cur.fetchone()

    async def fetchall(self):
        if self._is_sync:
            return self._cur.fetchall()
        return await self._cur.fetchall()

    async def fetchmany(self, size=1):
        if self._is_sync:
            return self._cur.fetchmany(size)
        return await self._cur.fetchmany(size)

    @property
    def description(self):
        return self._cur.description

    @property
    def rowcount(self):
        return self._cur.rowcount

    @property
    def lastrowid(self):
        return self._cur.lastrowid

    def __getattr__(self, name):
        return getattr(self._cur, name)
