PTV/v6d3vote
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

reformat

Browse Source
This commit is contained in:
AF 2023-10-02 16:57:36 +00:00
parent 21411d5f9a
commit 93818a3dc1
2 changed files with 68 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
v6d3vote/config.py
View File

@ -2,8 +2,8 @@ import os
from v6d0auth.config import root from v6d0auth.config import root
__all__ = ('prefix', 'myroot') __all__ = ("prefix", "myroot")
prefix = os.getenv('v6prefix', '??') prefix = os.getenv("v6prefix", "??")
myroot = root / 'v6d3vote' myroot = root / "v6d3vote"
myroot.mkdir(exist_ok=True) myroot.mkdir(exist_ok=True)

135
v6d3vote/run-bot.py
View File

@ -17,7 +17,7 @@ from v6d2ctx.serve import *
loop = asyncio.new_event_loop() loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
token = loop.run_until_complete(request_token('vote', 'token')) token = loop.run_until_complete(request_token("vote", "token"))
client = discord.Client( client = discord.Client(
intents=discord.Intents( intents=discord.Intents(
members=True, members=True,
@ -30,30 +30,26 @@ client = discord.Client(
message_content=True, message_content=True,
) )
) )
vote_db = Db(myroot / 'vote.db', kvfactory=KVJson()) vote_db = Db(myroot / "vote.db", kvfactory=KVJson())
@client.event @client.event
async def on_ready(): async def on_ready():
print("ready") print("ready")
await client.change_presence( await client.change_presence(activity=discord.Game(name="феноменально"))
activity=discord.Game(
name='феноменально'
)
)
at_of: AtOf[str, command_type] = AtOf() at_of: AtOf[str, command_type] = AtOf()
at, of = at_of() at, of = at_of()
@at('help') @at("help")
async def help_(ctx: Context, args: list[str]) -> None: async def help_(ctx: Context, args: list[str]) -> None:
match args: match args:
case []: case []:
await ctx.reply('poll bot') await ctx.reply("poll bot")
case [name]: case [name]:
await ctx.reply(f'help for {name}: `{name} help`') await ctx.reply(f"help for {name}: `{name} help`")
class SavedPoll(TypedDict): class SavedPoll(TypedDict):
@ -72,7 +68,7 @@ class Poll:
votes: dict[discord.Member, str], votes: dict[discord.Member, str],
emojis: dict[str, str], emojis: dict[str, str],
options: list[str], options: list[str],
title: str title: str,
): ):
self.message = message self.message = message
self.votes = votes self.votes = votes
@ -83,50 +79,40 @@ class Poll:
def saved(self): def saved(self):
return { return {
'votes': { "votes": {str(member.id): option for member, option in self.votes.items()},
str(member.id): option for member, option in self.votes.items() "emojis": self.emojis,
}, "options": self.options,
'emojis': self.emojis, "title": self.title,
'options': self.options,
'title': self.title
} }
def content(self): def content(self):
count: dict[str, int] = {} count: dict[str, int] = {}
for _, option in self.votes.items(): for _, option in self.votes.items():
count[option] = count.get(option, 0) + 1 count[option] = count.get(option, 0) + 1
return ( return f"{self.title}\n" + "\n".join(
f'{self.title}\n' f"{self.emojis[option]} `{count.get(option, 0)}` {option}" for option in self.options
+
'\n'.join(f'{self.emojis[option]} `{count.get(option, 0)}` {option}' for option in self.options)
) )
async def save(self): async def save(self):
await vote_db.set( await vote_db.set(self.message.id, self.saved())
self.message.id, await self.message.edit(content=self.content())
self.saved()
)
await self.message.edit(
content=self.content()
)
@classmethod @classmethod
async def create( async def create(
cls, ctx: Context, options: list[tuple[str, discord.Emoji | str]], title: str, cls, ctx: Context, options: list[tuple[str, discord.Emoji | str]], title: str, votes: dict[discord.Member, str]
votes: dict[discord.Member, str]
): ):
message: discord.Message = await ctx.channel.send('creating poll...') message: discord.Message = await ctx.channel.send("creating poll...")
async with lock_for(message, 'failed to create poll'): async with lock_for(message, "failed to create poll"):
if len(set(emoji for option, emoji in options)) != len(options): if len(set(emoji for option, emoji in options)) != len(options):
raise Explicit('duplicate emojis') raise Explicit("duplicate emojis")
if len(set(option for option, emoji in options)) != len(options): if len(set(option for option, emoji in options)) != len(options):
raise Explicit('duplicate options') raise Explicit("duplicate options")
poll = cls( poll = cls(
message, message,
votes, votes,
{option: str(emoji) for option, emoji in options}, {option: str(emoji) for option, emoji in options},
[option for option, _ in options], [option for option, _ in options],
title title,
) )
for _, emoji in options: for _, emoji in options:
await message.add_reaction(emoji) await message.add_reaction(emoji)
@ -152,31 +138,26 @@ class Poll:
guild: discord.Guild = message.guild guild: discord.Guild = message.guild
return cls( return cls(
message, message,
await cls.load_votes(guild, saved['votes']), await cls.load_votes(guild, saved["votes"]),
saved['emojis'], saved["emojis"],
saved['options'], saved["options"],
saved.get('title', 'unnamed') saved.get("title", "unnamed"),
) )
@classmethod @classmethod
async def _scheduled_save(cls, message: discord.Message, /) -> None: async def _scheduled_save(cls, message: discord.Message, /) -> None:
async with lock_for(message, 'no message'): async with lock_for(message, "no message"):
del cls.tasks[message.id] del cls.tasks[message.id]
print('saving') print("saving")
poll = await cls.load(message) poll = await cls.load(message)
if poll is None: if poll is None:
return return
await vote_db.commit() await vote_db.commit()
await poll.message.edit( await poll.message.edit(content=poll.content())
content=poll.content() print("saved")
)
print('saved')
def schedule_save(self) -> asyncio.Future[None]: def schedule_save(self) -> asyncio.Future[None]:
vote_db.set_nowait( vote_db.set_nowait(self.message.id, self.saved())
self.message.id,
self.saved()
)
if self.message.id not in self.tasks: if self.message.id not in self.tasks:
self.tasks[self.message.id] = asyncio.create_task(self._scheduled_save(self.message)) self.tasks[self.message.id] = asyncio.create_task(self._scheduled_save(self.message))
return self.tasks[self.message.id] return self.tasks[self.message.id]
@ -192,19 +173,19 @@ class Poll:
_channel = guild.get_channel(rrae.channel_id) _channel = guild.get_channel(rrae.channel_id)
assert isinstance(_channel, discord.TextChannel) assert isinstance(_channel, discord.TextChannel)
channel: discord.TextChannel = _channel channel: discord.TextChannel = _channel
print('process? ', rrae.emoji, rrae.event_type) print("process? ", rrae.emoji, rrae.event_type)
async with lock_for(rrae.message_id, 'no message id'): async with lock_for(rrae.message_id, "no message id"):
message: discord.Message = await channel.fetch_message(rrae.message_id) message: discord.Message = await channel.fetch_message(rrae.message_id)
if message.author != client.user: if message.author != client.user:
return return
async with lock_for(message, 'no message'): async with lock_for(message, "no message"):
poll = await cls.load(message) poll = await cls.load(message)
if poll is None: if poll is None:
return return
member: discord.Member = guild.get_member(rrae.user_id) or await guild.fetch_member(rrae.user_id) member: discord.Member = guild.get_member(rrae.user_id) or await guild.fetch_member(rrae.user_id)
print('processing', rrae.emoji, rrae.event_type) print("processing", rrae.emoji, rrae.event_type)
await poll.vote(member, rrae.emoji, rrae.event_type == 'REACTION_REMOVE') await poll.vote(member, rrae.emoji, rrae.event_type == "REACTION_REMOVE")
print('processed ', rrae.emoji, rrae.event_type) print("processed ", rrae.emoji, rrae.event_type)
future = poll.schedule_save() future = poll.schedule_save()
await future await future
@ -226,32 +207,35 @@ async def poll_options(args: list[str]) -> list[tuple[str, discord.Emoji | str]]
while args: while args:
match args: match args:
case [emoji, option, *args]: case [emoji, option, *args]:
if '<' in emoji and '>' in emoji: if "<" in emoji and ">" in emoji:
try: try:
emoji = client.get_emoji( emoji = (
int(''.join(c for c in emoji.rsplit(':', 1)[-1] if c.isdecimal())) client.get_emoji(int("".join(c for c in emoji.rsplit(":", 1)[-1] if c.isdecimal())))
) or emoji or emoji
)
except (discord.NotFound, ValueError): except (discord.NotFound, ValueError):
pass pass
options.append((option, emoji)) options.append((option, emoji))
case _: case _:
raise Explicit('option not specified') raise Explicit("option not specified")
return options return options
@at('poll') @at("poll")
async def create_poll(ctx: Context, args: list[str]) -> None: async def create_poll(ctx: Context, args: list[str]) -> None:
match args: match args:
case ['help']: case ["help"]:
await ctx.reply('`poll title emoji option [emoji option ...]`') await ctx.reply("`poll title emoji option [emoji option ...]`")
await ctx.reply('`poll emoji option [emoji option ...]` (reply fork)') await ctx.reply("`poll emoji option [emoji option ...]` (reply fork)")
case []: case []:
raise Explicit('no options') raise Explicit("no options")
case [*args] if ctx.message.reference is not None and isinstance(ctx.message.reference.resolved, discord.Message): case [*args] if ctx.message.reference is not None and isinstance(
ctx.message.reference.resolved, discord.Message
):
refd: discord.Message = ctx.message.reference.resolved refd: discord.Message = ctx.message.reference.resolved
poll = await Poll.load(refd) poll = await Poll.load(refd)
if poll is None: if poll is None:
raise Explicit('referenced message is not a poll') raise Explicit("referenced message is not a poll")
await Poll.create(ctx, await poll_options(args), poll.title, poll.votes) await Poll.create(ctx, await poll_options(args), poll.title, poll.votes)
case [title, *args]: case [title, *args]:
await Poll.create(ctx, await poll_options(args), title, {}) await Poll.create(ctx, await poll_options(args), title, {})
@ -276,20 +260,21 @@ async def main():
async with vote_db: async with vote_db:
await client.login(token) await client.login(token)
await client.connect() await client.connect()
print('exited') print("exited")
if __name__ == '__main__': if __name__ == "__main__":
from contextlib import ExitStack from contextlib import ExitStack
with ExitStack() as es: with ExitStack() as es:
ALog(client, 'connect').enter(es) ALog(client, "connect").enter(es)
ALog(client, 'close').enter(es) ALog(client, "close").enter(es)
ALog(Db, '__aenter__').enter(es) ALog(Db, "__aenter__").enter(es)
ALog(Db, '__aexit__').enter(es) ALog(Db, "__aexit__").enter(es)
ALog(Db, 'aclose').enter(es) ALog(Db, "aclose").enter(es)
# SLog(Db, '_build_file_sync').enter(es) # SLog(Db, '_build_file_sync').enter(es)
# SLog(Db, '_finish_recovery_sync').enter(es) # SLog(Db, '_finish_recovery_sync').enter(es)
# SLog(Db, '_copy_sync').enter(es) # SLog(Db, '_copy_sync').enter(es)
# ALog(loop, 'run_in_executor').enter(es) # ALog(loop, 'run_in_executor').enter(es)
serve(main(), client, loop) serve(main(), client, loop)
print('after serve') print("after serve")