Update app.py

app.py

@@ -7,6 +7,7 @@ import os
 import secrets
 import threading
 import traceback
+import time
 from collections import deque
 from datetime import datetime
 from pathlib import Path
@@ -237,14 +238,21 @@ def _ensure_data_layout():
     global db_initialized
     if db_initialized:
         return
+
+    logger.info("DB layout ensure begin.")
     with db_init_lock:
         if db_initialized:
+            logger.info("DB layout already initialized by another worker.")
             return
+        logger.info("DB schema initialization begin.")
         _init_db_schema()
         db_initialized = True
+        logger.info("DB schema initialization complete.")
 
     try:
+        logger.info("Legacy migration stage begin.")
         _migrate_legacy_file_data_if_needed()
+        logger.info("Legacy migration stage complete.")
     except Exception as exc:
         logger.warning("Legacy data migration skipped due to error: %s", exc)
 
@@ -363,7 +371,9 @@ def _build_mysql_conn_kwargs():
         "database": db_name,
         "charset": "utf8mb4",
         "autocommit": True,
-        "connect_timeout": int(os.getenv("MYSQL_CONNECT_TIMEOUT", "10")),
+        "connect_timeout": int(os.getenv("MYSQL_CONNECT_TIMEOUT", "4")),
+        "read_timeout": int(os.getenv("MYSQL_READ_TIMEOUT", "8")),
+        "write_timeout": int(os.getenv("MYSQL_WRITE_TIMEOUT", "8")),
         "cursorclass": pymysql.cursors.DictCursor,
     }
 
@@ -377,11 +387,43 @@ def _build_mysql_conn_kwargs():
 
 
 def _db_connect():
+    kwargs = _build_mysql_conn_kwargs()
+    host = kwargs.get("host")
+    port = kwargs.get("port")
+    database = kwargs.get("database")
+    connect_timeout = kwargs.get("connect_timeout")
+    read_timeout = kwargs.get("read_timeout")
+    write_timeout = kwargs.get("write_timeout")
+    has_ssl = bool(kwargs.get("ssl"))
+    started_at = time.perf_counter()
+
+    logger.info(
+        "MySQL connect begin. host=%s port=%s db=%s connect_timeout=%ss read_timeout=%ss write_timeout=%ss ssl=%s",
+        host,
+        port,
+        database,
+        connect_timeout,
+        read_timeout,
+        write_timeout,
+        has_ssl,
+    )
     try:
-        conn = pymysql.connect(**_build_mysql_conn_kwargs())
+        conn = pymysql.connect(**kwargs)
     except Exception as exc:
+        elapsed = time.perf_counter() - started_at
+        logger.warning(
+            "MySQL connect failed after %.2fs. host=%s port=%s db=%s error=%s",
+            elapsed,
+            host,
+            port,
+            database,
+            exc,
+        )
         _update_db_status(False, exc)
         raise
+
+    elapsed = time.perf_counter() - started_at
+    logger.info("MySQL connect success. host=%s port=%s db=%s elapsed=%.2fs", host, port, database, elapsed)
     _update_db_status(True)
     return conn
 
@@ -417,7 +459,7 @@ def _db_execute(query: str, params=()):
 
 
 def _init_db_schema():
-    _ensure_data_layout()
+    logger.info("DB schema creation SQL begin.")
     _db_execute(
         f"""
         CREATE TABLE IF NOT EXISTS `{USERS_TABLE}` (
@@ -434,6 +476,7 @@ def _init_db_schema():
         ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
         """
     )
+    logger.info("DB schema creation SQL complete.")
 
 
 def _legacy_load_json(path: Path, default):
@@ -444,11 +487,15 @@ def _legacy_load_json(path: Path, default):
 
 
 def _migrate_legacy_file_data_if_needed():
+    logger.info("Legacy migration check. path=%s", LEGACY_USERS_META_PATH)
     if not LEGACY_USERS_META_PATH.exists():
+        logger.info("Legacy migration skipped: users.json not found.")
         return
 
     row = _db_query_one(f"SELECT COUNT(*) AS cnt FROM `{USERS_TABLE}`")
-    if row and int(row.get("cnt", 0)) > 0:
+    existing_count = int(row.get("cnt", 0)) if row else 0
+    if existing_count > 0:
+        logger.info("Legacy migration skipped: database already has %s users.", existing_count)
         return
 
     try:
@@ -459,8 +506,10 @@ def _migrate_legacy_file_data_if_needed():
 
     users = raw.get("users", []) if isinstance(raw, dict) else []
     if not users:
+        logger.info("Legacy migration skipped: legacy users list is empty.")
         return
 
+    logger.info("Legacy migration loaded %s legacy users.", len(users))
     migrated = 0
     for item in users:
         username = str(item.get("username", "")).strip()
@@ -506,11 +555,13 @@ def _migrate_legacy_file_data_if_needed():
         except Exception as exc:
             logger.warning("迁移用户失败。user=%s error=%s", username, exc)
 
+    logger.info("Legacy migration finished. migrated=%s total=%s", migrated, len(users))
     if migrated > 0:
         logger.info("已完成旧版文件数据迁移，共迁移 %s 个用户。", migrated)
 
 
 def _load_users_meta():
+    logger.info("Load users meta begin.")
     _ensure_data_layout()
     rows = _db_query_all(
         f"""
@@ -519,6 +570,7 @@ def _load_users_meta():
         ORDER BY username ASC
         """
     )
+    logger.info("Load users meta complete. count=%s", len(rows))
     return {str(row["username"]): row for row in rows}
 
 
@@ -785,15 +837,20 @@ async def _run_user_tasks(username: str):
 def _sync_user_jobs_from_meta(users_map: dict[str, Any], run_startup_tasks: bool = False):
     global scheduler_bootstrapped
 
+    logger.info("Sync user jobs begin. count=%s run_startup_tasks=%s", len(users_map), run_startup_tasks)
     for username in users_map.keys():
+        logger.info("Sync user job. username=%s", username)
         _schedule_user_job(username)
         if run_startup_tasks:
             cfg = _load_user_config(username)
             run_on_startup = bool(cfg.get("scheduler", {}).get("runOnStartup", False))
+            logger.info("Startup run flag loaded. username=%s run_on_startup=%s", username, run_on_startup)
             if run_on_startup:
+                logger.info("Trigger startup run. username=%s", username)
                 _start_background_run(username, "startup")
 
     scheduler_bootstrapped = True
+    logger.info("Sync user jobs complete. count=%s", len(users_map))
 
 
 def _start_scheduler_bootstrap(run_startup_tasks: bool):
@@ -801,6 +858,7 @@ def _start_scheduler_bootstrap(run_startup_tasks: bool):
 
     with scheduler_bootstrap_lock:
         if scheduler_bootstrap_running:
+            logger.info("Scheduler bootstrap already running; skip duplicate start.")
             return False
         scheduler_bootstrap_running = True
 
@@ -808,9 +866,19 @@ def _start_scheduler_bootstrap(run_startup_tasks: bool):
         global scheduler_bootstrapped, scheduler_bootstrap_running
         try:
             logger.info("Scheduler bootstrap started. run_startup_tasks=%s", run_startup_tasks)
+
+            logger.info("Bootstrap stage begin: ensure_data_layout")
             _ensure_data_layout()
+            logger.info("Bootstrap stage complete: ensure_data_layout")
+
+            logger.info("Bootstrap stage begin: load_users_meta")
             users_map = _load_users_meta()
+            logger.info("Bootstrap stage complete: load_users_meta count=%s", len(users_map))
+
+            logger.info("Bootstrap stage begin: sync_user_jobs")
             _sync_user_jobs_from_meta(users_map, run_startup_tasks=run_startup_tasks)
+            logger.info("Bootstrap stage complete: sync_user_jobs count=%s", len(users_map))
+
             logger.info("Scheduler bootstrap completed. users=%s", len(users_map))
         except Exception as exc:
             scheduler_bootstrapped = False