feat: introduce core configuration management and MySQL database integration

config/setting.toml

@@ -3,6 +3,13 @@ api_key = "han1234"
 admin_username = "admin"
 admin_password = "admin"
 
+[database]
+host = "xxxxxxx"
+port = 3306
+user = "xxxxxxx"
+password = "xxxxxxx"
+database = "xxxxxxx"
+
 [sora]
 base_url = "https://sora.chatgpt.com/backend"
 timeout = 120

requirements.txt

@@ -12,4 +12,6 @@ tomli==2.2.1
 toml
 faker==24.0.0
 python-dateutil==2.8.2
-httpx==0.28.1
+httpx==0.28.1
+aiomysql
+cryptography

src/core/config.py

@@ -7,6 +7,8 @@ class Config:
     """Application configuration"""
 
     def __init__(self):
+        from dotenv import load_dotenv
+        load_dotenv()
         self._config = self._load_config()
         self._admin_username: Optional[str] = None
         self._admin_password: Optional[str] = None
@@ -41,6 +43,31 @@ class Config:
         """Set admin username from database"""
         self._admin_username = username
 
+    @property
+    def db_host(self) -> str:
+        import os
+        return os.getenv("DB_HOST", self._config.get("database", {}).get("host", "localhost"))
+
+    @property
+    def db_port(self) -> int:
+        import os
+        return int(os.getenv("DB_PORT", self._config.get("database", {}).get("port", 3306)))
+
+    @property
+    def db_user(self) -> str:
+        import os
+        return os.getenv("DB_USER", self._config.get("database", {}).get("user", ""))
+
+    @property
+    def db_password(self) -> str:
+        import os
+        return os.getenv("DB_PASSWORD", self._config.get("database", {}).get("password", ""))
+
+    @property
+    def db_name(self) -> str:
+        import os
+        return os.getenv("DB_NAME", self._config.get("database", {}).get("database", ""))
+
     @property
     def sora_base_url(self) -> str:
         return self._config["sora"]["base_url"]

src/core/database.py

@@ -1,55 +1,149 @@
 """Database storage layer"""
-import aiosqlite
+import aiomysql
 import json
 from datetime import datetime
-from typing import Optional, List
+from typing import Optional, List, Any
 from pathlib import Path
 from .models import Token, TokenStats, Task, RequestLog, AdminConfig, ProxyConfig, WatermarkFreeConfig, CacheConfig, GenerationConfig, TokenRefreshConfig
+from .config import config
+
+class MySQLCursorWrapper:
+    def __init__(self, cursor):
+        self.cursor = cursor
+
+    def __getattr__(self, name):
+        return getattr(self.cursor, name)
+
+    async def fetchone(self):
+        return await self.cursor.fetchone()
+
+    async def fetchall(self):
+        return await self.cursor.fetchall()
+    
+    @property
+    def lastrowid(self):
+        return self.cursor.lastrowid
+
+class MySQLConnectionWrapper:
+    def __init__(self, pool):
+        self.pool = pool
+        self.conn = None
+        self.cursor = None
+
+    async def __aenter__(self):
+        self.conn = await self.pool.acquire()
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        if self.conn:
+            self.pool.release(self.conn)
+
+    async def execute(self, query: str, args: tuple = None) -> MySQLCursorWrapper:
+        # Convert sqlite '?' placeholder to mysql '%s'
+        query = query.replace('?', '%s')
+        
+        # Create cursor (default to DictCursor for compatibility with explicit dict access, 
+        # but for row_factory we might need to handle it differently. 
+        # Existing code uses DictCursor logic mostly via dict(row))
+        self.cursor = await self.conn.cursor(aiomysql.DictCursor)
+        
+        try:
+            await self.cursor.execute(query, args)
+        except Exception as e:
+            # Add context to error
+            print(f"SQL Error: {e}\nQuery: {query}\nArgs: {args}")
+            raise e
+        return MySQLCursorWrapper(self.cursor)
+
+    async def commit(self):
+        await self.conn.commit()
+    
+    @property
+    def row_factory(self):
+        return None
+    
+    @row_factory.setter
+    def row_factory(self, value):
+        # aiomysql DictCursor yields dict-like objects already, close enough to aiosqlite.Row for most uses
+        pass
 
 class Database:
-    """SQLite database manager"""
+    """MySQL database manager (compatible interface with previous SQLite implementation)"""
 
     def __init__(self, db_path: str = None):
-        if db_path is None:
-            # Store database in data directory
-            data_dir = Path(__file__).parent.parent.parent / "data"
-            data_dir.mkdir(exist_ok=True)
-            db_path = str(data_dir / "hancat.db")
-        self.db_path = db_path
-
-    def db_exists(self) -> bool:
-        """Check if database file exists"""
-        return Path(self.db_path).exists()
+        self._pool = None
+        # db_path is ignored for MySQL, we use config
+
+    async def get_pool(self):
+        if self._pool is None:
+            self._pool = await aiomysql.create_pool(
+                host=config.db_host,
+                port=config.db_port,
+                user=config.db_user,
+                password=config.db_password,
+                db=config.db_name,
+                autocommit=False,  # We manually commit
+                cursorclass=aiomysql.DictCursor
+            )
+        return self._pool
+
+    def connect(self):
+        """Return a context manager that mimics aiosqlite.connect"""
+        # We can't await in __init__ or synchronous property, so we return an async context manager helper
+        class ConnectContext:
+            def __init__(self, db_instance):
+                self.db = db_instance
+            
+            async def __aenter__(self):
+                pool = await self.db.get_pool()
+                self.wrapper = MySQLConnectionWrapper(pool)
+                return await self.wrapper.__aenter__()
+
+            async def __aexit__(self, exc_type, exc_val, exc_tb):
+                await self.wrapper.__aexit__(exc_type, exc_val, exc_tb)
+
+        return ConnectContext(self)
+
+    async def db_exists(self) -> bool:
+        """Check if database exists (always true for MySQL if we can connect)"""
+        try:
+            pool = await self.get_pool()
+            async with pool.acquire() as conn:
+                async with conn.cursor() as cur:
+                    await cur.execute("SELECT 1")
+            return True
+        except Exception as e:
+            print(f"Database connection failed: {e}")
+            return False
 
     async def _table_exists(self, db, table_name: str) -> bool:
         """Check if a table exists in the database"""
+        # Note: db here is our MySQLConnectionWrapper
         cursor = await db.execute(
-            "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
-            (table_name,)
+            "SELECT count(*) as count FROM information_schema.tables WHERE table_schema = %s AND table_name = %s",
+            (config.db_name, table_name)
         )
         result = await cursor.fetchone()
-        return result is not None
+        return result['count'] > 0
 
     async def _column_exists(self, db, table_name: str, column_name: str) -> bool:
         """Check if a column exists in a table"""
         try:
-            cursor = await db.execute(f"PRAGMA table_info({table_name})")
-            columns = await cursor.fetchall()
-            return any(col[1] == column_name for col in columns)
+            cursor = await db.execute(
+                "SELECT count(*) as count FROM information_schema.columns WHERE table_schema = %s AND table_name = %s AND column_name = %s",
+                (config.db_name, table_name, column_name)
+            )
+            result = await cursor.fetchone()
+            return result['count'] > 0
         except:
             return False
 
     async def _ensure_config_rows(self, db, config_dict: dict = None):
-        """Ensure all config tables have their default rows
-
-        Args:
-            db: Database connection
-            config_dict: Configuration dictionary from setting.toml (optional)
-        """
+        """Ensure all config tables have their default rows"""
         # Ensure admin_config has a row
-        cursor = await db.execute("SELECT COUNT(*) FROM admin_config")
+        cursor = await db.execute("SELECT COUNT(*) as count FROM admin_config")
         count = await cursor.fetchone()
-        if count[0] == 0:
+        if count['count'] == 0:
             # Get admin credentials from config_dict if provided, otherwise use defaults
             admin_username = "admin"
             admin_password = "admin"
@@ -69,9 +163,9 @@ class Database:
             """, (admin_username, admin_password, error_ban_threshold))
 
         # Ensure proxy_config has a row
-        cursor = await db.execute("SELECT COUNT(*) FROM proxy_config")
+        cursor = await db.execute("SELECT COUNT(*) as count FROM proxy_config")
         count = await cursor.fetchone()
-        if count[0] == 0:
+        if count['count'] == 0:
             # Get proxy config from config_dict if provided, otherwise use defaults
             proxy_enabled = False
             proxy_url = None
@@ -89,9 +183,9 @@ class Database:
             """, (proxy_enabled, proxy_url))
 
         # Ensure watermark_free_config has a row
-        cursor = await db.execute("SELECT COUNT(*) FROM watermark_free_config")
+        cursor = await db.execute("SELECT COUNT(*) as count FROM watermark_free_config")
         count = await cursor.fetchone()
-        if count[0] == 0:
+        if count['count'] == 0:
             # Get watermark-free config from config_dict if provided, otherwise use defaults
             watermark_free_enabled = False
             parse_method = "third_party"
@@ -115,9 +209,9 @@ class Database:
             """, (watermark_free_enabled, parse_method, custom_parse_url, custom_parse_token))
 
         # Ensure cache_config has a row
-        cursor = await db.execute("SELECT COUNT(*) FROM cache_config")
+        cursor = await db.execute("SELECT COUNT(*) as count FROM cache_config")
         count = await cursor.fetchone()
-        if count[0] == 0:
+        if count['count'] == 0:
             # Get cache config from config_dict if provided, otherwise use defaults
             cache_enabled = False
             cache_timeout = 600
@@ -137,9 +231,9 @@ class Database:
             """, (cache_enabled, cache_timeout, cache_base_url))
 
         # Ensure generation_config has a row
-        cursor = await db.execute("SELECT COUNT(*) FROM generation_config")
+        cursor = await db.execute("SELECT COUNT(*) as count FROM generation_config")
         count = await cursor.fetchone()
-        if count[0] == 0:
+        if count['count'] == 0:
             # Get generation config from config_dict if provided, otherwise use defaults
             image_timeout = 300
             video_timeout = 1500
@@ -155,9 +249,9 @@ class Database:
             """, (image_timeout, video_timeout))
 
         # Ensure token_refresh_config has a row
-        cursor = await db.execute("SELECT COUNT(*) FROM token_refresh_config")
+        cursor = await db.execute("SELECT COUNT(*) as count FROM token_refresh_config")
         count = await cursor.fetchone()
-        if count[0] == 0:
+        if count['count'] == 0:
             # Get token refresh config from config_dict if provided, otherwise use defaults
             at_auto_refresh_enabled = False
 
@@ -170,15 +264,9 @@ class Database:
                 VALUES (1, ?)
             """, (at_auto_refresh_enabled,))
 
-
     async def check_and_migrate_db(self, config_dict: dict = None):
-        """Check database integrity and perform migrations if needed
-
-        Args:
-            config_dict: Configuration dictionary from setting.toml (optional)
-                        Used to initialize new tables with values from setting.toml
-        """
-        async with aiosqlite.connect(self.db_path) as db:
+        """Check database integrity and perform migrations if needed"""
+        async with self.connect() as db:
             print("Checking database integrity and performing migrations...")
 
             # Check and add missing columns to tokens table
@@ -186,10 +274,10 @@ class Database:
                 columns_to_add = [
                     ("sora2_supported", "BOOLEAN"),
                     ("sora2_invite_code", "TEXT"),
-                    ("sora2_redeemed_count", "INTEGER DEFAULT 0"),
-                    ("sora2_total_count", "INTEGER DEFAULT 0"),
-                    ("sora2_remaining_count", "INTEGER DEFAULT 0"),
-                    ("sora2_cooldown_until", "TIMESTAMP"),
+                    ("sora2_redeemed_count", "INT DEFAULT 0"),
+                    ("sora2_total_count", "INT DEFAULT 0"),
+                    ("sora2_remaining_count", "INT DEFAULT 0"),
+                    ("sora2_cooldown_until", "DATETIME"),
                     ("image_enabled", "BOOLEAN DEFAULT 1"),
                     ("video_enabled", "BOOLEAN DEFAULT 1"),
                 ]
@@ -205,8 +293,8 @@ class Database:
             # Check and add missing columns to admin_config table
             if await self._table_exists(db, "admin_config"):
                 columns_to_add = [
-                    ("admin_username", "TEXT DEFAULT 'admin'"),
-                    ("admin_password", "TEXT DEFAULT 'admin'"),
+                    ("admin_username", "TEXT"),
+                    ("admin_password", "TEXT"),
                 ]
 
                 for col_name, col_type in columns_to_add:
@@ -220,7 +308,7 @@ class Database:
             # Check and add missing columns to watermark_free_config table
             if await self._table_exists(db, "watermark_free_config"):
                 columns_to_add = [
-                    ("parse_method", "TEXT DEFAULT 'third_party'"),
+                    ("parse_method", "TEXT"),
                     ("custom_parse_url", "TEXT"),
                     ("custom_parse_token", "TEXT"),
                 ]
@@ -234,7 +322,6 @@ class Database:
                             print(f"  ✗ Failed to add column '{col_name}': {e}")
 
             # Ensure all config tables have their default rows
-            # Pass config_dict if available to initialize from setting.toml
             await self._ensure_config_rows(db, config_dict)
 
             await db.commit()
@@ -242,33 +329,33 @@ class Database:
 
     async def init_db(self):
         """Initialize database tables - creates all tables and ensures data integrity"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             # Tokens table
             await db.execute("""
                 CREATE TABLE IF NOT EXISTS tokens (
-                    id INTEGER PRIMARY KEY AUTOINCREMENT,
-                    token TEXT UNIQUE NOT NULL,
-                    email TEXT NOT NULL,
-                    username TEXT NOT NULL,
-                    name TEXT NOT NULL,
+                    id INT PRIMARY KEY AUTO_INCREMENT,
+                    token VARCHAR(255) UNIQUE NOT NULL,
+                    email VARCHAR(255) NOT NULL,
+                    username VARCHAR(255) NOT NULL,
+                    name VARCHAR(255) NOT NULL,
                     st TEXT,
                     rt TEXT,
                     remark TEXT,
-                    expiry_time TIMESTAMP,
+                    expiry_time DATETIME,
                     is_active BOOLEAN DEFAULT 1,
-                    cooled_until TIMESTAMP,
+                    cooled_until DATETIME,
                     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                    last_used_at TIMESTAMP,
-                    use_count INTEGER DEFAULT 0,
-                    plan_type TEXT,
-                    plan_title TEXT,
-                    subscription_end TIMESTAMP,
+                    last_used_at DATETIME,
+                    use_count INT DEFAULT 0,
+                    plan_type VARCHAR(50),
+                    plan_title VARCHAR(100),
+                    subscription_end DATETIME,
                     sora2_supported BOOLEAN,
-                    sora2_invite_code TEXT,
-                    sora2_redeemed_count INTEGER DEFAULT 0,
-                    sora2_total_count INTEGER DEFAULT 0,
-                    sora2_remaining_count INTEGER DEFAULT 0,
-                    sora2_cooldown_until TIMESTAMP,
+                    sora2_invite_code VARCHAR(255),
+                    sora2_redeemed_count INT DEFAULT 0,
+                    sora2_total_count INT DEFAULT 0,
+                    sora2_remaining_count INT DEFAULT 0,
+                    sora2_cooldown_until DATETIME,
                     image_enabled BOOLEAN DEFAULT 1,
                     video_enabled BOOLEAN DEFAULT 1
                 )
@@ -277,12 +364,12 @@ class Database:
             # Token stats table
             await db.execute("""
                 CREATE TABLE IF NOT EXISTS token_stats (
-                    id INTEGER PRIMARY KEY AUTOINCREMENT,
-                    token_id INTEGER NOT NULL,
-                    image_count INTEGER DEFAULT 0,
-                    video_count INTEGER DEFAULT 0,
-                    error_count INTEGER DEFAULT 0,
-                    last_error_at TIMESTAMP,
+                    id INT PRIMARY KEY AUTO_INCREMENT,
+                    token_id INT NOT NULL,
+                    image_count INT DEFAULT 0,
+                    video_count INT DEFAULT 0,
+                    error_count INT DEFAULT 0,
+                    last_error_at DATETIME,
                     FOREIGN KEY (token_id) REFERENCES tokens(id)
                 )
             """)
@@ -290,17 +377,17 @@ class Database:
             # Tasks table
             await db.execute("""
                 CREATE TABLE IF NOT EXISTS tasks (
-                    id INTEGER PRIMARY KEY AUTOINCREMENT,
-                    task_id TEXT UNIQUE NOT NULL,
-                    token_id INTEGER NOT NULL,
-                    model TEXT NOT NULL,
+                    id INT PRIMARY KEY AUTO_INCREMENT,
+                    task_id VARCHAR(255) UNIQUE NOT NULL,
+                    token_id INT NOT NULL,
+                    model VARCHAR(255) NOT NULL,
                     prompt TEXT NOT NULL,
-                    status TEXT NOT NULL DEFAULT 'processing',
+                    status VARCHAR(50) NOT NULL DEFAULT 'processing',
                     progress FLOAT DEFAULT 0,
                     result_urls TEXT,
                     error_message TEXT,
                     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                    completed_at TIMESTAMP,
+                    completed_at DATETIME,
                     FOREIGN KEY (token_id) REFERENCES tokens(id)
                 )
             """)
@@ -308,12 +395,12 @@ class Database:
             # Request logs table
             await db.execute("""
                 CREATE TABLE IF NOT EXISTS request_logs (
-                    id INTEGER PRIMARY KEY AUTOINCREMENT,
-                    token_id INTEGER,
-                    operation TEXT NOT NULL,
+                    id INT PRIMARY KEY AUTO_INCREMENT,
+                    token_id INT,
+                    operation VARCHAR(255) NOT NULL,
                     request_body TEXT,
                     response_body TEXT,
-                    status_code INTEGER NOT NULL,
+                    status_code INT NOT NULL,
                     duration FLOAT NOT NULL,
                     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                     FOREIGN KEY (token_id) REFERENCES tokens(id)
@@ -323,88 +410,92 @@ class Database:
             # Admin config table
             await db.execute("""
                 CREATE TABLE IF NOT EXISTS admin_config (
-                    id INTEGER PRIMARY KEY DEFAULT 1,
-                    admin_username TEXT DEFAULT 'admin',
-                    admin_password TEXT DEFAULT 'admin',
-                    error_ban_threshold INTEGER DEFAULT 3,
-                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                    id INT PRIMARY KEY DEFAULT 1,
+                    admin_username VARCHAR(255) DEFAULT 'admin',
+                    admin_password VARCHAR(255) DEFAULT 'admin',
+                    error_ban_threshold INT DEFAULT 3,
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
                 )
             """)
 
             # Proxy config table
             await db.execute("""
                 CREATE TABLE IF NOT EXISTS proxy_config (
-                    id INTEGER PRIMARY KEY DEFAULT 1,
+                    id INT PRIMARY KEY DEFAULT 1,
                     proxy_enabled BOOLEAN DEFAULT 0,
                     proxy_url TEXT,
                     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
                 )
             """)
 
             # Watermark-free config table
             await db.execute("""
                 CREATE TABLE IF NOT EXISTS watermark_free_config (
-                    id INTEGER PRIMARY KEY DEFAULT 1,
+                    id INT PRIMARY KEY DEFAULT 1,
                     watermark_free_enabled BOOLEAN DEFAULT 0,
-                    parse_method TEXT DEFAULT 'third_party',
+                    parse_method VARCHAR(50) DEFAULT 'third_party',
                     custom_parse_url TEXT,
                     custom_parse_token TEXT,
                     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
                 )
             """)
 
             # Cache config table
             await db.execute("""
                 CREATE TABLE IF NOT EXISTS cache_config (
-                    id INTEGER PRIMARY KEY DEFAULT 1,
+                    id INT PRIMARY KEY DEFAULT 1,
                     cache_enabled BOOLEAN DEFAULT 0,
-                    cache_timeout INTEGER DEFAULT 600,
+                    cache_timeout INT DEFAULT 600,
                     cache_base_url TEXT,
                     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
                 )
             """)
 
             # Generation config table
             await db.execute("""
                 CREATE TABLE IF NOT EXISTS generation_config (
-                    id INTEGER PRIMARY KEY DEFAULT 1,
-                    image_timeout INTEGER DEFAULT 300,
-                    video_timeout INTEGER DEFAULT 1500,
+                    id INT PRIMARY KEY DEFAULT 1,
+                    image_timeout INT DEFAULT 300,
+                    video_timeout INT DEFAULT 1500,
                     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
                 )
             """)
 
             # Token refresh config table
             await db.execute("""
                 CREATE TABLE IF NOT EXISTS token_refresh_config (
-                    id INTEGER PRIMARY KEY DEFAULT 1,
+                    id INT PRIMARY KEY DEFAULT 1,
                     at_auto_refresh_enabled BOOLEAN DEFAULT 0,
                     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
                 )
             """)
 
             # Create indexes
-            await db.execute("CREATE INDEX IF NOT EXISTS idx_task_id ON tasks(task_id)")
-            await db.execute("CREATE INDEX IF NOT EXISTS idx_task_status ON tasks(status)")
-            await db.execute("CREATE INDEX IF NOT EXISTS idx_token_active ON tokens(is_active)")
+            # MySQL handles index creation slightly differently/implicitly but this is standard enough
+            # Use separate calls and catch error if index exists (or use CREATE INDEX IF NOT EXISTS which MySQL supports in 8.0+, assuming 5.7+ check needed?)
+            # MySQL 8.0+ supports IF NOT EXISTS. 
+            try:
+                await db.execute("CREATE INDEX idx_task_id ON tasks(task_id)")
+            except: pass
+            
+            try:
+                await db.execute("CREATE INDEX idx_task_status ON tasks(status)")
+            except: pass
+            
+            try:
+                await db.execute("CREATE INDEX idx_token_active ON tokens(is_active)")
+            except: pass
 
             await db.commit()
 
     async def init_config_from_toml(self, config_dict: dict, is_first_startup: bool = True):
-        """
-        Initialize database configuration from setting.toml
-
-        Args:
-            config_dict: Configuration dictionary from setting.toml
-            is_first_startup: If True, only update if row doesn't exist. If False, always update.
-        """
-        async with aiosqlite.connect(self.db_path) as db:
-            # On first startup, ensure all config rows exist with values from setting.toml
+        """Initialize database configuration from setting.toml"""
+        async with self.connect() as db:
             if is_first_startup:
                 await self._ensure_config_rows(db, config_dict)
 
@@ -412,13 +503,11 @@ class Database:
             admin_config = config_dict.get("admin", {})
             error_ban_threshold = admin_config.get("error_ban_threshold", 3)
 
-            # Get admin credentials from global config
             global_config = config_dict.get("global", {})
             admin_username = global_config.get("admin_username", "admin")
             admin_password = global_config.get("admin_password", "admin")
 
             if not is_first_startup:
-                # On upgrade, update the configuration
                 await db.execute("""
                     UPDATE admin_config
                     SET admin_username = ?, admin_password = ?, error_ban_threshold = ?, updated_at = CURRENT_TIMESTAMP
@@ -429,12 +518,11 @@ class Database:
             proxy_config = config_dict.get("proxy", {})
             proxy_enabled = proxy_config.get("proxy_enabled", False)
             proxy_url = proxy_config.get("proxy_url", "")
-            # Convert empty string to None
             proxy_url = proxy_url if proxy_url else None
 
             if is_first_startup:
                 await db.execute("""
-                    INSERT OR IGNORE INTO proxy_config (id, proxy_enabled, proxy_url)
+                    INSERT IGNORE INTO proxy_config (id, proxy_enabled, proxy_url)
                     VALUES (1, ?, ?)
                 """, (proxy_enabled, proxy_url))
             else:
@@ -450,14 +538,13 @@ class Database:
             parse_method = watermark_config.get("parse_method", "third_party")
             custom_parse_url = watermark_config.get("custom_parse_url", "")
             custom_parse_token = watermark_config.get("custom_parse_token", "")
-
-            # Convert empty strings to None
+            
             custom_parse_url = custom_parse_url if custom_parse_url else None
             custom_parse_token = custom_parse_token if custom_parse_token else None
 
             if is_first_startup:
                 await db.execute("""
-                    INSERT OR IGNORE INTO watermark_free_config (id, watermark_free_enabled, parse_method, custom_parse_url, custom_parse_token)
+                    INSERT IGNORE INTO watermark_free_config (id, watermark_free_enabled, parse_method, custom_parse_url, custom_parse_token)
                     VALUES (1, ?, ?, ?, ?)
                 """, (watermark_free_enabled, parse_method, custom_parse_url, custom_parse_token))
             else:
@@ -473,12 +560,11 @@ class Database:
             cache_enabled = cache_config.get("enabled", False)
             cache_timeout = cache_config.get("timeout", 600)
             cache_base_url = cache_config.get("base_url", "")
-            # Convert empty string to None
             cache_base_url = cache_base_url if cache_base_url else None
 
             if is_first_startup:
                 await db.execute("""
-                    INSERT OR IGNORE INTO cache_config (id, cache_enabled, cache_timeout, cache_base_url)
+                    INSERT IGNORE INTO cache_config (id, cache_enabled, cache_timeout, cache_base_url)
                     VALUES (1, ?, ?, ?)
                 """, (cache_enabled, cache_timeout, cache_base_url))
             else:
@@ -495,7 +581,7 @@ class Database:
 
             if is_first_startup:
                 await db.execute("""
-                    INSERT OR IGNORE INTO generation_config (id, image_timeout, video_timeout)
+                    INSERT IGNORE INTO generation_config (id, image_timeout, video_timeout)
                     VALUES (1, ?, ?)
                 """, (image_timeout, video_timeout))
             else:
@@ -511,7 +597,7 @@ class Database:
 
             if is_first_startup:
                 await db.execute("""
-                    INSERT OR IGNORE INTO token_refresh_config (id, at_auto_refresh_enabled)
+                    INSERT IGNORE INTO token_refresh_config (id, at_auto_refresh_enabled)
                     VALUES (1, ?)
                 """, (at_auto_refresh_enabled,))
             else:
@@ -526,7 +612,7 @@ class Database:
     # Token operations
     async def add_token(self, token: Token) -> int:
         """Add a new token"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             cursor = await db.execute("""
                 INSERT INTO tokens (token, email, username, name, st, rt, remark, expiry_time, is_active,
                                    plan_type, plan_title, subscription_end, sora2_supported, sora2_invite_code,
@@ -553,8 +639,7 @@ class Database:
     
     async def get_token(self, token_id: int) -> Optional[Token]:
         """Get token by ID"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("SELECT * FROM tokens WHERE id = ?", (token_id,))
             row = await cursor.fetchone()
             if row:
@@ -563,8 +648,7 @@ class Database:
     
     async def get_token_by_value(self, token: str) -> Optional[Token]:
         """Get token by value"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("SELECT * FROM tokens WHERE token = ?", (token,))
             row = await cursor.fetchone()
             if row:
@@ -573,29 +657,29 @@ class Database:
     
     async def get_active_tokens(self) -> List[Token]:
         """Get all active tokens (enabled, not cooled down, not expired)"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
+            # MySQL handles booleans as TinyInt, so is_active = 1 works.
+            # CURRENT_TIMESTAMP is standard.
             cursor = await db.execute("""
                 SELECT * FROM tokens
                 WHERE is_active = 1
                 AND (cooled_until IS NULL OR cooled_until < CURRENT_TIMESTAMP)
                 AND expiry_time > CURRENT_TIMESTAMP
-                ORDER BY last_used_at ASC NULLS FIRST
+                ORDER BY last_used_at ASC
             """)
             rows = await cursor.fetchall()
             return [Token(**dict(row)) for row in rows]
     
     async def get_all_tokens(self) -> List[Token]:
         """Get all tokens"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("SELECT * FROM tokens ORDER BY created_at DESC")
             rows = await cursor.fetchall()
             return [Token(**dict(row)) for row in rows]
     
     async def update_token_usage(self, token_id: int):
         """Update token usage"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE tokens 
                 SET last_used_at = CURRENT_TIMESTAMP, use_count = use_count + 1
@@ -605,7 +689,7 @@ class Database:
     
     async def update_token_status(self, token_id: int, is_active: bool):
         """Update token status"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE tokens SET is_active = ? WHERE id = ?
             """, (is_active, token_id))
@@ -614,7 +698,7 @@ class Database:
     async def update_token_sora2(self, token_id: int, supported: bool, invite_code: Optional[str] = None,
                                 redeemed_count: int = 0, total_count: int = 0, remaining_count: int = 0):
         """Update token Sora2 support info"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE tokens
                 SET sora2_supported = ?, sora2_invite_code = ?, sora2_redeemed_count = ?, sora2_total_count = ?, sora2_remaining_count = ?
@@ -624,7 +708,7 @@ class Database:
 
     async def update_token_sora2_remaining(self, token_id: int, remaining_count: int):
         """Update token Sora2 remaining count"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE tokens SET sora2_remaining_count = ? WHERE id = ?
             """, (remaining_count, token_id))
@@ -632,7 +716,7 @@ class Database:
 
     async def update_token_sora2_cooldown(self, token_id: int, cooldown_until: Optional[datetime]):
         """Update token Sora2 cooldown time"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE tokens SET sora2_cooldown_until = ? WHERE id = ?
             """, (cooldown_until, token_id))
@@ -640,7 +724,7 @@ class Database:
 
     async def update_token_cooldown(self, token_id: int, cooled_until: datetime):
         """Update token cooldown"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE tokens SET cooled_until = ? WHERE id = ?
             """, (cooled_until, token_id))
@@ -648,7 +732,7 @@ class Database:
     
     async def delete_token(self, token_id: int):
         """Delete token"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("DELETE FROM token_stats WHERE token_id = ?", (token_id,))
             await db.execute("DELETE FROM tokens WHERE id = ?", (token_id,))
             await db.commit()
@@ -665,7 +749,7 @@ class Database:
                           image_enabled: Optional[bool] = None,
                           video_enabled: Optional[bool] = None):
         """Update token (AT, ST, RT, remark, expiry_time, subscription info, image_enabled, video_enabled)"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             # Build dynamic update query
             updates = []
             params = []
@@ -719,8 +803,7 @@ class Database:
     # Token stats operations
     async def get_token_stats(self, token_id: int) -> Optional[TokenStats]:
         """Get token statistics"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("SELECT * FROM token_stats WHERE token_id = ?", (token_id,))
             row = await cursor.fetchone()
             if row:
@@ -729,7 +812,7 @@ class Database:
     
     async def increment_image_count(self, token_id: int):
         """Increment image generation count"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE token_stats SET image_count = image_count + 1 WHERE token_id = ?
             """, (token_id,))
@@ -737,7 +820,7 @@ class Database:
     
     async def increment_video_count(self, token_id: int):
         """Increment video generation count"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE token_stats SET video_count = video_count + 1 WHERE token_id = ?
             """, (token_id,))
@@ -745,7 +828,7 @@ class Database:
     
     async def increment_error_count(self, token_id: int):
         """Increment error count"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE token_stats 
                 SET error_count = error_count + 1, last_error_at = CURRENT_TIMESTAMP
@@ -755,7 +838,7 @@ class Database:
     
     async def reset_error_count(self, token_id: int):
         """Reset error count"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE token_stats SET error_count = 0 WHERE token_id = ?
             """, (token_id,))
@@ -764,7 +847,7 @@ class Database:
     # Task operations
     async def create_task(self, task: Task) -> int:
         """Create a new task"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             cursor = await db.execute("""
                 INSERT INTO tasks (task_id, token_id, model, prompt, status, progress)
                 VALUES (?, ?, ?, ?, ?, ?)
@@ -775,7 +858,7 @@ class Database:
     async def update_task(self, task_id: str, status: str, progress: float, 
                          result_urls: Optional[str] = None, error_message: Optional[str] = None):
         """Update task status"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             completed_at = datetime.now() if status in ["completed", "failed"] else None
             await db.execute("""
                 UPDATE tasks 
@@ -786,8 +869,7 @@ class Database:
     
     async def get_task(self, task_id: str) -> Optional[Task]:
         """Get task by ID"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("SELECT * FROM tasks WHERE task_id = ?", (task_id,))
             row = await cursor.fetchone()
             if row:
@@ -797,7 +879,7 @@ class Database:
     # Request log operations
     async def log_request(self, log: RequestLog):
         """Log a request"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 INSERT INTO request_logs (token_id, operation, request_body, response_body, status_code, duration)
                 VALUES (?, ?, ?, ?, ?, ?)
@@ -807,8 +889,7 @@ class Database:
     
     async def get_recent_logs(self, limit: int = 100) -> List[dict]:
         """Get recent logs with token email"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("""
                 SELECT
                     rl.id,
@@ -831,19 +912,16 @@ class Database:
     # Admin config operations
     async def get_admin_config(self) -> AdminConfig:
         """Get admin configuration"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("SELECT * FROM admin_config WHERE id = 1")
             row = await cursor.fetchone()
             if row:
                 return AdminConfig(**dict(row))
-            # If no row exists, return a default config with placeholder values
-            # This should not happen in normal operation as _ensure_config_rows should create it
             return AdminConfig(admin_username="admin", admin_password="admin")
     
     async def update_admin_config(self, config: AdminConfig):
         """Update admin configuration"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE admin_config
                 SET admin_username = ?, admin_password = ?, error_ban_threshold = ?, updated_at = CURRENT_TIMESTAMP
@@ -854,19 +932,16 @@ class Database:
     # Proxy config operations
     async def get_proxy_config(self) -> ProxyConfig:
         """Get proxy configuration"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("SELECT * FROM proxy_config WHERE id = 1")
             row = await cursor.fetchone()
             if row:
                 return ProxyConfig(**dict(row))
-            # If no row exists, return a default config
-            # This should not happen in normal operation as _ensure_config_rows should create it
             return ProxyConfig(proxy_enabled=False)
     
     async def update_proxy_config(self, enabled: bool, proxy_url: Optional[str]):
         """Update proxy configuration"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE proxy_config
                 SET proxy_enabled = ?, proxy_url = ?, updated_at = CURRENT_TIMESTAMP
@@ -877,20 +952,17 @@ class Database:
     # Watermark-free config operations
     async def get_watermark_free_config(self) -> WatermarkFreeConfig:
         """Get watermark-free configuration"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("SELECT * FROM watermark_free_config WHERE id = 1")
             row = await cursor.fetchone()
             if row:
                 return WatermarkFreeConfig(**dict(row))
-            # If no row exists, return a default config
-            # This should not happen in normal operation as _ensure_config_rows should create it
             return WatermarkFreeConfig(watermark_free_enabled=False, parse_method="third_party")
 
     async def update_watermark_free_config(self, enabled: bool, parse_method: str = None,
                                           custom_parse_url: str = None, custom_parse_token: str = None):
         """Update watermark-free configuration"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             if parse_method is None and custom_parse_url is None and custom_parse_token is None:
                 # Only update enabled status
                 await db.execute("""
@@ -911,27 +983,22 @@ class Database:
     # Cache config operations
     async def get_cache_config(self) -> CacheConfig:
         """Get cache configuration"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("SELECT * FROM cache_config WHERE id = 1")
             row = await cursor.fetchone()
             if row:
                 return CacheConfig(**dict(row))
-            # If no row exists, return a default config
-            # This should not happen in normal operation as _ensure_config_rows should create it
             return CacheConfig(cache_enabled=False, cache_timeout=600)
 
     async def update_cache_config(self, enabled: bool = None, timeout: int = None, base_url: Optional[str] = None):
         """Update cache configuration"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             # Get current config first
-            db.row_factory = aiosqlite.Row
             cursor = await db.execute("SELECT * FROM cache_config WHERE id = 1")
             row = await cursor.fetchone()
 
             if row:
                 current = dict(row)
-                # Update only provided fields
                 new_enabled = enabled if enabled is not None else current.get("cache_enabled", False)
                 new_timeout = timeout if timeout is not None else current.get("cache_timeout", 600)
                 new_base_url = base_url if base_url is not None else current.get("cache_base_url")
@@ -953,27 +1020,21 @@ class Database:
     # Generation config operations
     async def get_generation_config(self) -> GenerationConfig:
         """Get generation configuration"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("SELECT * FROM generation_config WHERE id = 1")
             row = await cursor.fetchone()
             if row:
                 return GenerationConfig(**dict(row))
-            # If no row exists, return a default config
-            # This should not happen in normal operation as _ensure_config_rows should create it
             return GenerationConfig(image_timeout=300, video_timeout=1500)
 
     async def update_generation_config(self, image_timeout: int = None, video_timeout: int = None):
         """Update generation configuration"""
-        async with aiosqlite.connect(self.db_path) as db:
-            # Get current config first
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("SELECT * FROM generation_config WHERE id = 1")
             row = await cursor.fetchone()
 
             if row:
                 current = dict(row)
-                # Update only provided fields
                 new_image_timeout = image_timeout if image_timeout is not None else current.get("image_timeout", 300)
                 new_video_timeout = video_timeout if video_timeout is not None else current.get("video_timeout", 1500)
             else:
@@ -990,23 +1051,19 @@ class Database:
     # Token refresh config operations
     async def get_token_refresh_config(self) -> TokenRefreshConfig:
         """Get token refresh configuration"""
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
+        async with self.connect() as db:
             cursor = await db.execute("SELECT * FROM token_refresh_config WHERE id = 1")
             row = await cursor.fetchone()
             if row:
                 return TokenRefreshConfig(**dict(row))
-            # If no row exists, return a default config
-            # This should not happen in normal operation as _ensure_config_rows should create it
             return TokenRefreshConfig(at_auto_refresh_enabled=False)
 
     async def update_token_refresh_config(self, at_auto_refresh_enabled: bool):
         """Update token refresh configuration"""
-        async with aiosqlite.connect(self.db_path) as db:
+        async with self.connect() as db:
             await db.execute("""
                 UPDATE token_refresh_config
                 SET at_auto_refresh_enabled = ?, updated_at = CURRENT_TIMESTAMP
                 WHERE id = 1
             """, (at_auto_refresh_enabled,))
             await db.commit()
-