feat: 添加配置文件迁移阶段

pkg/config/migration.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import abc
+import typing
+
+from ..core import app
+
+
+preregistered_migrations: list[typing.Type[Migration]] = []
+
+def migration_class(name: str, number: int):
+    """注册一个迁移
+    """
+    def decorator(cls: typing.Type[Migration]) -> typing.Type[Migration]:
+        cls.name = name
+        cls.number = number
+        preregistered_migrations.append(cls)
+        return cls
+    
+    return decorator
+
+
+class Migration(abc.ABC):
+    """一个版本的迁移
+    """
+
+    name: str
+
+    number: int
+
+    ap: app.Application
+
+    def __init__(self, ap: app.Application):
+        self.ap = ap
+    
+    @abc.abstractmethod
+    async def need_migrate(self) -> bool:
+        """判断当前环境是否需要运行此迁移
+        """
+        pass
+
+    @abc.abstractmethod
+    async def run(self):
+        """执行迁移
+        """
+        pass

pkg/config/migrations/m1_sensitive_word_migration.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import os
+import sys
+
+from .. import migration
+
+
+@migration.migration_class("sensitive-word-migration", 1)
+class SensitiveWordMigration(migration.Migration):
+    """敏感词迁移
+    """
+
+    async def need_migrate(self) -> bool:
+        """判断当前环境是否需要运行此迁移
+        """
+        return os.path.exists("data/config/sensitive-words.json")
+
+    async def run(self):
+        """执行迁移
+        """
+        # 移动文件
+        os.rename("data/config/sensitive-words.json", "data/metadata/sensitive-words.json")
+
+        # 重新加载配置
+        await self.ap.sensitive_meta.load_config()

pkg/core/boot.py

@@ -5,11 +5,12 @@ from ..audit import identifier
 from . import stage
 
 # 引入启动阶段实现以便注册
-from .stages import load_config, setup_logger, build_app
+from .stages import load_config, setup_logger, build_app, migrate
 
 
 stage_order = [
     "LoadConfigStage",
+    "MigrationStage",
     "SetupLoggerStage",
     "BuildAppStage"
 ]

pkg/core/stages/migrate.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import importlib
+
+from .. import stage, app
+from ...config import migration
+from ...config.migrations import m1_sensitive_word_migration
+
+
+@stage.stage_class("MigrationStage")
+class MigrationStage(stage.BootingStage):
+    """迁移阶段
+    """
+
+    async def run(self, ap: app.Application):
+        """启动
+        """
+
+        migrations = migration.preregistered_migrations
+
+        # 按照迁移号排序
+        migrations.sort(key=lambda x: x.number)
+
+        for migration_cls in migrations:
+            migration_instance = migration_cls(ap)
+
+            if await migration_instance.need_migrate():
+                await migration_instance.run()