insert2-cogs/threadsummary.py

import nextcord
from nextcord.ext import commands, application_checks
from nextcord import TextInputStyle
import traceback
import random
import aiosqlite as sqlite3
import aiohttp
from urllib.parse import urlparse
import os
import sys

class threadsummary(commands.Cog):

    def __init__(self, bot: commands.Bot):
        self.bot = bot

    @nextcord.slash_command(
        name="threadsummary",
        description="Generate a list of created threads with the owner",
        default_member_permissions=nextcord.Permissions(manage_messages=True),
    )
    async def threadsummary(self, interaction: nextcord.Interaction):
        if interaction.channel.type != nextcord.ChannelType.text:
            await interaction.response.send_message("This is not a text channel", ephemeral=True)
            return
        message = f"Active threads in {interaction.channel.jump_url}\n"
        for t in interaction.channel.threads:
            if t.archived:
                continue
            message = message + f"<@{t.owner_id}>: {t.jump_url}\n"
        await interaction.response.send_message(message, allowed_mentions=nextcord.AllowedMentions.none())

    @commands.Cog.listener('on_thread_update')
    async def threadupdatehandler(self, before: nextcord.Thread, after: nextcord.Thread):
        pins = await after.parent.pins()
        for t in pins:
            if t.author.id == self.bot.application_id and t.content.split("\n")[0] == f"Active threads in {after.parent.jump_url}":
                message = f"Active threads in {after.parent.jump_url}\n"
                for p in after.parent.threads:
                    if p.archived:
                        continue
                    message = message + f"<@{p.owner_id}>: {p.jump_url}\n"
                await t.edit(message)
                return

    @commands.Cog.listener('on_thread_create')
    async def threacreatehandler(self, thread: nextcord.Thread):
        pins = await thread.parent.pins()
        print(pins, file=sys.stderr)
        for t in pins:
            if t.author.id == self.bot.application_id and t.content.split("\n")[0] == f"Active threads in {thread.parent.jump_url}":
                message = f"Active threads in {thread.parent.jump_url}\n"
                for p in thread.parent.threads:
                    if p.archived:
                        continue
                    message = message + f"<@{p.owner_id}>: {p.jump_url}\n"
                await t.edit(message)
                return

    @commands.Cog.listener('on_thread_delete')
    async def threaddelhandler(self, thread: nextcord.Thread):
        pins = await thread.parent.pins()
        for t in pins:
            if t.author.id == self.bot.application_id and t.content.split("\n")[0] == f"Active threads in {thread.parent.jump_url}":
                message = f"Active threads in {thread.parent.jump_url}\n"
                for p in thread.parent.threads:
                    if p.archived:
                        continue
                    message = message + f"<@{p.owner_id}>: {p.jump_url}\n"
                await t.edit(message)
                return

def setup(bot: commands.Bot):
    bot.add_cog(threadsummary(bot))