iframe-proxy | Sunbelt Computer Software

History

223 lines (175 loc) · 8.16 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

# Hydrogram - Telegram MTProto API Client Library for Python

# This file is part of Hydrogram.

# Hydrogram is free software: you can redistribute it and/or modify

# it under the terms of the GNU Lesser General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

# Hydrogram is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

# GNU Lesser General Public License for more details.

# You should have received a copy of the GNU Lesser General Public License

# along with Hydrogram. If not, see <http://www.gnu.org/licenses/>.

from __future__ import annotations

from asyncio import iscoroutinefunction

from typing import Callable

import hydrogram

from hydrogram.types import CallbackQuery, Identifier, Listener, ListenerTypes

from hydrogram.utils import PyromodConfig

from .handler import Handler

class CallbackQueryHandler(Handler):

"""The CallbackQuery handler class. Used to handle callback queries coming from inline buttons.

It is intended to be used with :meth:`~hydrogram.Client.add_handler`

For a nicer way to register this handler, have a look at the

:meth:`~hydrogram.Client.on_callback_query` decorator.

Parameters:

callback (``Callable``):

Pass a function that will be called when a new CallbackQuery arrives. It takes *(client, callback_query)*

as positional arguments (look at the section below for a detailed description).

filters (:obj:`Filters`):

Pass one or more filters to allow only a subset of callback queries to be passed

in your callback function.

Other parameters:

client (:obj:`~hydrogram.Client`):

The Client itself, useful when you want to call other API methods inside the message handler.

callback_query (:obj:`~hydrogram.types.CallbackQuery`):

The received callback query.

"""

def __init__(self, callback: Callable, filters=None):

self.original_callback = callback

super().__init__(self.resolve_future_or_callback, filters)

@staticmethod

def compose_data_identifier(query: CallbackQuery) -> Identifier:

"""

Composes an Identifier object from a CallbackQuery object.

Parameters:

query (:obj:`~hydrogram.types.CallbackQuery`):

The CallbackQuery object to compose of.

Returns:

:obj:`~hydrogram.types.Identifier`: An Identifier object.

"""

from_user = query.from_user

from_user_id = from_user.id if from_user else None

from_user_username = from_user.username if from_user else None

chat_id = None

message_id = None

if query.message:

message_id = getattr(query.message, "id", getattr(query.message, "message_id", None))

if query.message.chat:

chat_id = [query.message.chat.id, query.message.chat.username]

return Identifier(

message_id=message_id,

chat_id=chat_id,

from_user_id=[from_user_id, from_user_username],

inline_message_id=query.inline_message_id,

)

async def check_if_has_matching_listener(

self, client: hydrogram.Client, query: CallbackQuery

) -> tuple[bool, Listener | None]:

"""

Checks if the CallbackQuery object has a matching listener.

Parameters:

client (:obj:`~hydrogram.Client`):

The Client object to check with.

query (:obj:`~hydrogram.types.CallbackQuery`):

The CallbackQuery object to check with.

Returns:

A tuple of whether the CallbackQuery object has a matching listener and its filters does match with the

CallbackQuery and the matching listener;

"""

data = self.compose_data_identifier(query)

listener = client.get_listener_matching_with_data(data, ListenerTypes.CALLBACK_QUERY)

listener_does_match = False

if listener:

filters = listener.filters

if callable(filters):

if iscoroutinefunction(filters.__call__):

listener_does_match = await filters(client, query)

else:

listener_does_match = await client.loop.run_in_executor(

None, filters, client, query

)

else:

listener_does_match = True

return listener_does_match, listener

async def check(self, client: hydrogram.Client, query: CallbackQuery) -> bool:

"""

Checks if the CallbackQuery object has a matching listener or handler.

Parameters:

client (:obj:`~hydrogram.Client`):

The Client object to check with.

query (:obj:`~hydrogram.types.CallbackQuery`):

The CallbackQuery object to check with.

Returns:

``bool``: A boolean indicating whether the CallbackQuery object has a matching listener or the handler filter matches.

"""

listener_does_match, listener = await self.check_if_has_matching_listener(client, query)

if callable(self.filters):

if iscoroutinefunction(self.filters.__call__):

handler_does_match = await self.filters(client, query)

else:

handler_does_match = await client.loop.run_in_executor(

None, self.filters, client, query

)

else:

handler_does_match = True

data = self.compose_data_identifier(query)

if PyromodConfig.unallowed_click_alert:

# matches with the current query but from any user

permissive_identifier = Identifier(

chat_id=data.chat_id,

message_id=data.message_id,

inline_message_id=data.inline_message_id,

from_user_id=None,

)

matches = permissive_identifier.matches(data)

if (

listener

and (matches and not listener_does_match)

and listener.unallowed_click_alert

alert = (

listener.unallowed_click_alert

if isinstance(listener.unallowed_click_alert, str)

else PyromodConfig.unallowed_click_alert_text

)

await query.answer(alert)

return False

# let handler get the chance to handle if listener

# exists but its filters doesn't match

return listener_does_match or handler_does_match

async def resolve_future_or_callback(

self, client: hydrogram.Client, query: CallbackQuery, *args

) -> None:

"""

Resolves the future or calls the callback of the listener. Will call the original handler if no listener.

Parameters:

client (:obj:`~hydrogram.Client`):

The Client object to resolve or call with.

query (:obj:`~hydrogram.types.CallbackQuery`):

The CallbackQuery object to resolve or call with.

args:

The arguments to call the callback with.

Returns:

``None``

"""

listener_does_match, listener = await self.check_if_has_matching_listener(client, query)

if listener and listener_does_match:

client.remove_listener(listener)

if listener.future and not listener.future.done():

listener.future.set_result(query)

raise hydrogram.StopPropagation

if listener.callback:

if iscoroutinefunction(listener.callback):

await listener.callback(client, query, *args)

else:

listener.callback(client, query, *args)

raise hydrogram.StopPropagation

raise ValueError("Listener must have either a future or a callback")

await self.original_callback(client, query, *args)

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Uh oh!

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

Uh oh!

FilesExpand file tree

callback_query_handler.py

Latest commit

History

callback_query_handler.py

File metadata and controls