✨ Raise SystemExit exception after core initialization error to shut down the app

core/base/aiobrowser.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from playwright.async_api import Browser, Playwright, async_playwright
+from playwright.async_api import Browser, Playwright, async_playwright, Error
 
 from core.service import Service
 from utils.log import logger
@@ -22,8 +22,15 @@ class AioBrowser(Service):
             try:
                 self.browser = await self._playwright.chromium.launch(timeout=5000)
                 logger.success("[blue]Browser[/] 启动成功", extra={"markup": True})
-            except TimeoutError as err:
+            except Error as err:
-                logger.warning("[blue]Browser[/] 启动失败", extra={"markup": True})
+                if "playwright install" in str(err):
+                    logger.error(
+                        "检查到 [blue]playwright[/] 刚刚安装或者未升级\n"
+                        "请运行以下命令下载新浏览器\n"
+                        "[blue bold]playwright install chromium[/]",
+                        extra={"markup": True},
+                    )
+                    raise RuntimeError("检查到 playwright 刚刚安装或者未升级\n请运行以下命令下载新浏览器\nplaywright install chromium")
                 raise err
 
         return self.browser

core/base/mysql.py

@@ -18,9 +18,8 @@ class MySQL(Service):
         self.user = username
         self.port = port
         self.host = host
-        self.engine = create_async_engine(
+        self.url = f"mysql+asyncmy://{self.user}:{self.password}@{self.host}:{self.port}/{self.database}"
-            f"mysql+asyncmy://{self.user}:{self.password}@{self.host}:{self.port}/{self.database}"
+        self.engine = create_async_engine(self.url)
-        )
         self.Session = sessionmaker(bind=self.engine, class_=AsyncSession)
 
     async def get_session(self):

core/base/redisdb.py

@@ -2,6 +2,7 @@ import asyncio
 
 import fakeredis.aioredis
 from redis import asyncio as aioredis
+from redis.exceptions import ConnectionError as RedisConnectionError, TimeoutError as RedisTimeoutError
 from typing_extensions import Self
 
 from core.config import BotConfig
@@ -27,18 +28,22 @@ class RedisDB(Service):
             logger.info("连接 [red]Redis[/] 失败", extra={"markup": True})
             raise RuntimeError("连接 Redis 失败")
 
+    async def start_fake_redis(self):
+        self.client = fakeredis.aioredis.FakeRedis()
+        await self.ping()
+
     async def start(self):  # pylint: disable=W0221
         if self._loop is None:
             self._loop = asyncio.get_running_loop()
         logger.info("正在尝试建立与 [red]Redis[/] 连接", extra={"markup": True})
         try:
             await self.ping()
-        except (KeyboardInterrupt, SystemExit):
+        except (RedisTimeoutError, RedisConnectionError) as exc:
-            pass
+            if isinstance(exc, RedisTimeoutError):
-        except Exception as exc:
+                logger.warning("连接 [red]Redis[/] 超时，使用 [red]fakeredis[/] 模拟", extra={"markup": True})
-            logger.exception("尝试连接 [red]Redis[/] 失败，使用 [red]fakeredis[/] 模拟", exc_info=exc, extra={"markup": True})
+            if isinstance(exc, RedisConnectionError):
-            self.client = fakeredis.aioredis.FakeRedis()
+                logger.warning("连接 [red]Redis[/] 失败，使用 [red]fakeredis[/] 模拟", extra={"markup": True})
-            await self.ping()
+            await self.start_fake_redis()
 
     async def stop(self):  # pylint: disable=W0221
         await self.client.close()

core/bot.py

@@ -182,9 +182,9 @@ class Bot:
                 await instance.start()
                 logger.success(f'服务 "{base_service_cls.__name__}" 初始化成功')
                 self._services.update({base_service_cls: instance})
-            except Exception as e:  # pylint: disable=W0703
+            except Exception as e:
                 logger.exception(f'服务 "{base_service_cls.__name__}" 初始化失败: {e}')
-                continue
+                raise SystemExit from e
 
     async def start_services(self):
         """启动服务"""