Discord.py 动态命令选项：无需重启实时更新数据库内容

import discord
from discord import app_commands, Interaction
from typing import List

# 假设 LessonRepository 是一个与数据库交互的类
class LessonRepository:
    @staticmethod
    def get_all_lessons():
        # 模拟从数据库获取所有课程
        # 实际上这里会进行数据库查询
        class LessonDTO:
            def __init__(self, title):
                self.title = title
        return [LessonDTO("数学"), LessonDTO("物理"), LessonDTO("化学")]

def lesson_choices() -> List[app_commands.Choice[str]]:
    """从数据库获取课程列表并转换为 Choices"""
    return [
        app_commands.Choice(name=lesson.title, value=lesson.title)
        for lesson in LessonRepository.get_all_lessons()
    ]

class SomeCog(discord.ext.commands.Cog):
    def __init__(self, bot: discord.ext.commands.Bot):
        self.bot = bot

    @app_commands.command(name="create_or_update_mark")
    @app_commands.default_permissions(administrator=True)
    @app_commands.choices(lesson=lesson_choices()) # 这里的 choices 在机器人启动时确定
    async def create_or_update_mark(self, interaction: Interaction,
                                    student: discord.Member,
                                    lesson: app_commands.Choice[str],
                                    logiks: app_commands.Range[int, 0, 8]):
        # ... 处理逻辑 ...
        await interaction.response.send_message(f"已更新 {student.display_name} 的 {lesson.value} 成绩。", ephemeral=True)

# 机器人启动代码
# bot = commands.Bot(command_prefix="!", intents=discord.Intents.all())
# async def setup():
#     await bot.add_cog(SomeCog(bot))
#     await bot.tree.sync()
# bot.run("YOUR_TOKEN")

# ... (LessonRepository 和 LessonDTO 定义同上) ...

async def lesson_autocomplete(interaction: Interaction, current: str) -> List[app_commands.Choice[str]]:
    """为课程参数提供自动补全建议"""
    # 每次用户输入字符时都会调用此函数
    lessons = [lesson_dto.title for lesson_dto in LessonRepository.get_all_lessons()]
    return [
        app_commands.Choice(name=lesson, value=lesson)
        for lesson in lessons if current.lower() in lesson.lower()
    ]

class SomeCog(discord.ext.commands.Cog):
    # ... (init 方法同上) ...

    @app_commands.command(name="create_or_update_mark")
    @app_commands.default_permissions(administrator=True)
    @app_commands.autocomplete(lesson=lesson_autocomplete) # 使用自动补全
    async def create_or_update_mark(self, interaction: Interaction,
                                    student: discord.Member,
                                    lesson: str, # 注意这里 lesson 的类型变为 str
                                    logiks: app_commands.Range[int, 0, 8]):
        # ... 处理逻辑，lesson 现在是用户输入的字符串 ...
        await interaction.response.send_message(f"已更新 {student.display_name} 的 {lesson} 成绩。", ephemeral=True)

from typing import TYPE_CHECKING, Dict, List, Any, Optional, Union
import discord
from discord.ext import commands
from discord import app_commands
import difflib

# 类型别名，增加可读性
GUILD_ID = int
MEMBER_ID = int
LESSON_ID = int

# 假设 Lesson 类代表数据库中的课程数据
class Lesson:
    def __init__(self, id: int, title: str):
        self.id = id
        self.title = title

    def __repr__(self):
        return f"<Lesson id={self.id} title='{self.title}'>"

# 模拟异步数据库操作
class AsyncLessonRepository:
    @staticmethod
    async def get_all_lessons():
        # 模拟异步数据库查询
        await discord.utils.sleep_until(discord.utils.utcnow() + discord.timedelta(milliseconds=50))
        return [
            Lesson(1, "数学"),
            Lesson(2, "物理"),
            Lesson(3, "化学"),
            Lesson(4, "生物"),
            Lesson(5, "历史"),
            Lesson(6, "地理"),
            Lesson(7, "计算机科学"),
        ]

class MyBot(commands.Bot):
    # 类型提示，方便 IDE 识别
    if TYPE_CHECKING:
        some_function_for_loading_the_lessons_cache: Any

    def __init__(self, *args: Any, **kwargs: Any):
        super().__init__(*args, **kwargs)
        # 缓存结构：{guild_id: {student_id: {lesson_id: Lesson_object}}}
        # 这里的缓存设计可以根据实际需求调整，例如只缓存所有课程，不按学生区分
        self.lessons_cache: Dict[GUILD_ID, Dict[MEMBER_ID, Dict[LESSON_ID, Lesson]]] = {}
        self.all_lessons_flat_cache: Dict[LESSON_ID, Lesson] = {} # 用于全局课程查找

    async def setup_hook(self):
        """机器人启动时调用，用于加载缓存和同步命令"""
        print("Bot setup_hook called.")
        await self.tree.sync() # 同步斜杠命令
        await self._load_lessons_cache() # 加载课程缓存
        print("Commands synced and cache loaded.")

    async def _load_lessons_cache(self):
        """从数据库加载所有课程到缓存"""
        print("Loading lessons cache...")
        all_lessons = await AsyncLessonRepository.get_all_lessons()
        self.all_lessons_flat_cache = {lesson.id: lesson for lesson in all_lessons}
        # 模拟为每个学生/公会填充缓存，实际中可能需要更复杂的逻辑
        # 这里仅为演示，假设所有课程对所有学生都可用
        for guild_id in self.guilds: # 假设已经加入了公会
            self.lessons_cache[guild_id.id] = {}
            # 简化处理：假设每个学生都能看到所有课程
            # 真实场景下，可能需要根据学生ID从数据库加载其专属课程
            # for student_id in some_student_ids:
            #     self.lessons_cache[guild_id.id][student_id] = self.all_lessons_flat_cache.copy()
        print(f"Lessons cache loaded: {len(self.all_lessons_flat_cache)} lessons.")

    async def update_lesson_cache(self):
        """在数据库更新后，手动调用此函数以刷新缓存"""
        print("Updating lessons cache...")
        await self._load_lessons_cache()
        print("Lessons cache updated.")


class LessonTransformer(app_commands.Transformer):
    async def find_similar_lesson_titles(self, lessons: Dict[LESSON_ID, Lesson], title: str) -> Dict[LESSON_ID, Lesson]:
        """使用 difflib 查找相似的课程标题"""
        # map(lambda l: l.title, lessons.values()) 提取所有课程的标题
        all_titles = [lesson.title for lesson in lessons.values()]
        # difflib.get_close_matches 查找与输入标题相似的标题
        similar_titles = difflib.get_close_matches(title, all_titles, n=15, cutoff=0.6) # cutoff 可调整匹配严格度
        return {lesson.id: lesson for lesson in lessons.values() if lesson.title in similar_titles}

    async def autocomplete(self, interaction: discord.Interaction[MyBot], value: str, /) -> List[app_commands.Choice[str]]:
        """为课程参数提供自动补全建议"""
        # 前提：此命令只能在服务器（公会）中调用
        assert interaction.guild is not None

        # 从机器人的全局缓存中获取所有课程
        all_lessons_in_guild = interaction.client.all_lessons_flat_cache

        # 检查用户是否已填写 "student" 参数
        student: Optional[discord.Member] = interaction.namespace.get('student')

        target_lessons: Dict[LESSON_ID, Lesson]
        if student is None:
            # 如果没有指定学生，则显示所有可用课程
            target_lessons = all_lessons_in_guild
        else:
            # 如果指定了学生，可以根据学生ID进一步过滤课程
            # 这里的逻辑需要根据您的实际缓存结构和业务需求调整
            # 示例中，我们假设所有课程对所有学生都可用，所以这里仍然使用全部课程
            # 如果您的缓存是 {guild_id: {student_id: {lesson_id: Lesson}}}
            # 则这里会是 interaction.client.lessons_cache.get(interaction.guild.id, {}).get(student.id, {})
            target_lessons = all_lessons_in_guild

        # 查找与用户输入相似的课程
        similar_lessons = await self.find_similar_lesson_titles(target_lessons, value)

        # 返回自动补全建议，value 存储课程 ID
        return [
            app_commands.Choice(name=lesson.title, value=str(lesson_id))
            for lesson_id, lesson in similar_lessons.items()
        ]

    async def transform(self, interaction: discord.Interaction[MyBot], value: str, /) -> Union[Lesson, LESSON_ID]:
        """将用户最终选择的或输入的字符串转换为 Lesson 对象或其 ID"""
        assert interaction.guild is not None

        # 自动补全建议的 value 是课程 ID (字符串形式)，用户也可能手动输入
        if not value.isdigit():
            # 如果用户输入的不是数字（即不是课程 ID），可以尝试根据标题查找
            # 或者直接抛出错误，取决于您的业务逻辑
            # 这里简单处理为抛出错误，要求用户必须从建议中选择或输入有效ID
            raise app_commands.AppCommandError("无效的课程ID或标题。请从建议中选择。")

        lesson_id = int(value)

        # 再次检查学生参数，以便更精确地验证
        student: Optional[discord.Member] = interaction.namespace.get('student')

        # 从缓存中查找课程
        lesson = interaction.client.all_lessons_flat_cache.get(lesson_id)

        if lesson is None:
            raise app_commands.AppCommandError("未找到匹配的课程。")

        # 如果需要，可以在这里添加更复杂的验证，例如检查该课程是否属于指定学生
        # if student is not None and lesson_id not in interaction.client.lessons_cache.get(interaction.guild.id, {}).get(student.id, {}):
        #     raise app_commands.AppCommandError("该课程不属于指定学生。")

        return lesson # 返回 Lesson 对象，方便命令回调函数直接使用

class SomeCog(commands.Cog):
    def __init__(self, bot: MyBot):
        self.bot = bot

    @app_commands.command(name="create_or_update_mark_dynamic")
    @app_commands.guild_only() # 确保只在公会中运行
    @app_commands.default_permissions(administrator=True)
    async def create_or_update_mark_dynamic(
        self,
        interaction: discord.Interaction[MyBot],
        student: discord.Member,
        lesson: app_commands.Transform[Lesson, LessonTransformer], # 使用 Transformer
        logiks: app_commands.Range[int, 0, 8],
    ):
        """动态更新学生成绩的命令"""
        assert interaction.guild is not None

        # lesson 现在已经是经过 Transformer 转换后的 Lesson 对象
        # 我们可以直接访问其属性，例如 lesson.id, lesson.title

        # 假设这里是与数据库交互的逻辑
        # with MarkRepository(student_id=student.id, lesson_title=lesson.title) as lr:
        #     lr.create_or_update(mark_dto=MarkCreateDTO(logiks=logiks))

        await interaction.response.send_message(
            f"已成功更新 {student.display_name} 的 {lesson.title} 课程成绩为 {logiks}。",
            ephemeral=True
        )

# 机器人启动示例
# intents = discord.Intents.default()
# intents.members = True # 如果需要访问成员信息
# bot = MyBot(command_prefix="!", intents=intents)

# @bot.event
# async def on_ready():
#     print(f'Logged in as {bot.user} (ID: {bot.user.id})')
#     print('------')

# async def main():
#     async with bot:
#         await bot.add_cog(SomeCog(bot))
#         await bot.start("YOUR_BOT_TOKEN")

# if __name__ == "__main__":
#     import asyncio
#     asyncio.run(main())