Skip to content

Instantly share code, notes, and snippets.

@gagarski
Created December 7, 2020 20:39
Show Gist options
  • Save gagarski/e8a99ee746041c2504fd7764a1417dae to your computer and use it in GitHub Desktop.
Save gagarski/e8a99ee746041c2504fd7764a1417dae to your computer and use it in GitHub Desktop.
import requests
BOT_TOKEN = "xxx:yyy"
STICKERS = []
def get_updates(offset=None):
return requests.get(f"https://api.telegram.org/bot{BOT_TOKEN}/getUpdates",
params={"offset": offset, "timeout": 30}).json()
def to_cached_sticker(id, file_id):
return {
"type": "sticker",
"id": id,
"sticker_file_id": file_id
}
def persist_sticker(msg):
file_id = msg.get("sticker", {}).get("file_id")
if not file_id:
return
STICKERS.append(file_id)
print(f"Storage size is: {len(STICKERS)}")
def answer_inline_query(query):
print(">>>>>")
print(f'received offset: {query.get("offset")} query id: {query.get("id")}')
offset_str = query.get("offset", "0") or "0"
try:
offset = int(offset_str) or 0
except ValueError:
return
stickers_slice = list(enumerate(STICKERS[offset:offset + 20]))
next_offset = str(offset + 20) if len(stickers_slice) == 20 else None
print(f'sending answer with next_offset: {next_offset}')
print("<<<<<")
resp = {
"inline_query_id": query.get("id"),
"results": [to_cached_sticker(ix, file_id) for (ix, file_id) in stickers_slice],
"cache_time": 0
}
if next_offset:
resp["next_offset"] = next_offset
return requests.post(f"https://api.telegram.org/bot{BOT_TOKEN}/answerInlineQuery", json=resp)
offset = None
while True:
updates = get_updates(offset)
results = updates.get("result")
if results is None or 0 == len(results):
continue
offset = results[-1].get("update_id") + 1
for u in results:
iq = u.get("inline_query")
msg = u.get("message")
if iq is not None:
answer_inline_query(iq)
elif msg is not None:
persist_sticker(msg)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment