discord.py如何通过命令停止Cog中运行的指定循环实例

首先调整任务存储结构，从无差别的列表改为按服务器、城市、通知频道维度分类的嵌套字典，实现精准定位指定任务的需求，完整修改后代码如下：

# 导入需要的额外模块
from discord.ext import commands, tasks
import datetime
import aiohttp # 推荐替换同步requests为异步aiohttp，避免阻塞机器人事件循环

class TaskHandler(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        # 存储结构：{服务器ID: {(城市名, 频道ID): 任务实例}}
        self._tasks = {}

    async def static_loop(self, guild_id, arg1, arg2):
        # arg1为城市名，arg2为频道ID（已提前处理为纯数字）
        channel = self.bot.get_channel(int(arg2))
        if not channel:
            # 频道不存在时自动停止任务
            self._tasks[guild_id].pop((arg1, arg2), None)
            return
        
        # 替换同步requests为异步请求
        async with aiohttp.ClientSession() as session:
            url = f"https://api.pray.zone/v2/times/today.json?city={arg1}"
            async with session.get(url) as resp:
                data = await resp.json()

        local_offset = data.get('results', {}).get('location', {}).get('local_offset', 0)
        now = datetime.datetime.now()
        nowhour = now.hour + local_offset
        # 处理时区转换
        if nowhour >= 24:
            nowhour -= 24
        elif nowhour < 0:
            nowhour += 24
        now_str = f"{nowhour:02d}:{now.minute:02d}"

        pray_times = data.get('results', {}).get('datetime', [{}])[0].get('times', {})
        pray_names = {
            "Fajr": "晨礼",
            "Dhuhr": "晌礼", 
            "Asr": "晡礼",
            "Maghrib": "昏礼",
            "Isha": "宵礼"
        }
        for pray_key, pray_name in pray_names.items():
            if now_str == str(pray_times.get(pray_key, '')):
                await channel.send(f'现在是{arg1}的{pray_name}时间！||@everyone||')
                break

    def task_launcher(self, guild_id, arg1, arg2, **interval): 
        """创建新的定时循环实例"""
        # 如果该服务器下已有相同城市+频道的任务，先停止旧任务避免重复
        if guild_id in self._tasks and (arg1, arg2) in self._tasks[guild_id]:
            old_task = self._tasks[guild_id][(arg1, arg2)]
            if old_task.is_running():
                old_task.stop()
        
        new_task = tasks.loop(**interval)(self.static_loop)
        new_task.start(guild_id, arg1, arg2)
        # 存入字典
        if guild_id not in self._tasks:
            self._tasks[guild_id] = {}
        self._tasks[guild_id][(arg1, arg2)] = new_task
  
    @commands.command()
    async def setup(self, ctx, arg1, arg2):
        # 提前处理频道参数，提取纯数字ID
        try:
            channel_id = int(arg2.replace('<', '').replace('>', '').replace('#', ''))
        except ValueError:
            await ctx.send("频道参数无效，请正确@要通知的频道")
            return
        # 校验城市有效性
        try:
            async with aiohttp.ClientSession() as session:
                url = f"https://api.pray.zone/v2/times/today.json?city={arg1}"
                async with session.get(url) as resp:
                    data = await resp.json()
            if not data.get('results'):
                raise ValueError
        except:
            await ctx.send("城市名称无效")
            return
        
        self.task_launcher(ctx.guild.id, arg1, channel_id, seconds=60)
        await ctx.send("定时提醒已设置完成")

    @commands.command()
    async def stop(self, ctx, city: str, channel_arg: str):
        """停止当前服务器下指定城市、指定频道的提醒任务
        使用示例：!stop 北京 #通知频道
        """
        guild_id = ctx.guild.id
        # 处理频道参数
        try:
            channel_id = int(channel_arg.replace('<', '').replace('>', '').replace('#', ''))
        except ValueError:
            await ctx.send("频道参数无效，请正确@要停止任务的频道")
            return
        
        # 检查任务是否存在
        if guild_id not in self._tasks or (city, channel_id) not in self._tasks[guild_id]:
            await ctx.send("未找到对应任务，请检查城市名和频道是否正确")
            return
        
        # 停止并删除任务
        task = self._tasks[guild_id].pop((city, channel_id))
        if task.is_running():
            task.stop()
        # 如果当前服务器没有剩余任务，清空对应键节省空间
        if not self._tasks[guild_id]:
            del self._tasks[guild_id]
        await ctx.send(f"已停止{city}对应频道的提醒任务")

    @commands.command()
    @commands.has_permissions(administrator=True)
    async def stop_all(self, ctx):
        """停止当前服务器下所有的提醒任务，需要管理员权限"""
        guild_id = ctx.guild.id
        if guild_id not in self._tasks:
            await ctx.send("当前服务器没有运行中的提醒任务")
            return
        
        # 停止所有任务
        for task in self._tasks[guild_id].values():
            if task.is_running():
                task.stop()
        del self._tasks[guild_id]
        await ctx.send("已停止当前服务器下所有的提醒任务")

• 调整了任务存储结构，支持按服务器ID、城市名、频道ID三个维度精准定位任务，不会出现跨服务器误停的问题
• 新增了stop命令，支持停止当前服务器下指定城市和频道的任务
• 新增了stop_all命令，支持管理员一键停止当前服务器所有任务
• 修复了原有代码中client未定义的问题，替换为cog绑定的self.bot实例
• 将同步requests替换为异步aiohttp，避免请求阻塞机器人的事件循环
• 优化了时间计算逻辑，简化了代码冗余
• setup命令新增重复任务校验，同一服务器下相同城市和频道的任务不会重复创建

内容的提问来源于stack exchange，提问作者Ramy0204