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