1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
# This file is part of python-cinema-club-bot
# contributed in 2024 by Mikhail Kirillov (~w96k) <w96k@runbox.com>
# To the extent possible under law, the author(s) have dedicated all copyright
# and related and neighboring rights to this software to the public domain
# worldwide. This software is distributed without any warranty.
# You should have received a copy of the CC0 Public Domain Dedication along
# with this software. If not, see:
# <http://creativecommons.org/publicdomain/zero/1.0/>
from telegram import Update, error
from telegram.ext import ContextTypes
from collections import deque
from strings import USER_NOT_PROVIDED, USERS_ADDED, USERS_REMOVED, \
EXPECTED_ONE_USER, USER_SET, USER_ADD, USER_REMOVE, ADD_MORE_USERS, \
NEXT_MOVIE_USER, USER_NOT_FOUND, USER_CHOOSE
from utils import context_init, create_users_string
async def set_users(
update: Update,
context: ContextTypes.DEFAULT_TYPE
) -> None:
context_init(context)
if context.args == []:
raise error.TelegramError(USER_NOT_PROVIDED)
context.chat_data["users"] = context.args
await update.message.reply_text(USER_SET)
async def add_users(
update: Update,
context: ContextTypes.DEFAULT_TYPE
) -> None:
context_init(context)
if context.args == []:
raise error.TelegramError(USER_NOT_PROVIDED)
for user in context.args:
context.chat_data["users"].append(user)
await update.message.reply_text(USER_ADD.format(user=user))
await update.message.reply_text(USERS_ADDED)
async def list_users(
update: Update,
context: ContextTypes.DEFAULT_TYPE
) -> None:
context_init(context)
users = context.chat_data["users"]
await update.message.reply_markdown(create_users_string(users))
async def who_is_next(
update: Update,
context: ContextTypes.DEFAULT_TYPE
) -> None:
context_init(context)
users = context.chat_data["users"]
if len(users) > 0:
await update.message.reply_text(NEXT_MOVIE_USER.format(user=users[0]))
else:
await update.message.reply_text(ADD_MORE_USERS)
async def remove_users(
update: Update,
context: ContextTypes.DEFAULT_TYPE
) -> None:
context_init(context)
if context.args == []:
raise error.TelegramError(USER_NOT_PROVIDED)
for user in context.args:
if user == "*":
context.chat_data["users"] = []
break
context.chat_data["users"].remove(user)
await update.message.reply_text(USER_REMOVE.format(user=user))
await update.message.reply_text(USERS_REMOVED)
async def chooser_user(
update: Update,
context: ContextTypes.DEFAULT_TYPE
) -> None:
context_init(context)
if context.args == []:
raise error.TelegramError(USER_NOT_PROVIDED)
if len(context.args) > 1:
raise error.TelegramError(EXPECTED_ONE_USER)
chooser = context.args[0]
users = deque(context.chat_data["users"])
try:
chooser_index = users.index(chooser)
except ValueError:
raise error.TelegramError(USER_NOT_FOUND.format(user=chooser))
users.rotate(-chooser_index)
users = list(users)
context.chat_data["users"] = users
await update.message.reply_text(USER_CHOOSE.format(user=users[0]))
await update.message.reply_markdown(create_users_string(users))
|
