Create a Telegram Bot to get info from PyPi Server: Part I

Link to the source code at the end of the article.

In this Part I we will simple Telegram Bot sync service to work with API and obtain PyPi stats and in next Parts I will improve it.

$ poetry new pypi_observer_bot
$ cd pypi_observer_bot
$ poetry add pyTelegramBotAPI
create new Telegram Bot & got the auth token
import telebot

bot = telebot.TeleBot("mytoken", parse_mode=None)


@bot.message_handler(commands=['start', 'help'])
def send_welcome(message):
bot.reply_to(message, "Howdy, how are you doing?")


@bot.message_handler(func=lambda m: True)
def echo_all(message):
bot.reply_to(message, message.text)


bot.polling()
$ python bot.py
check Bot works & answered well
@bot.message_handler(commands=['start', 'help'])
def send_welcome(message):
print(message.chat.id)
print(message.chat.__dict__)
print(message.__dict__)
bot.reply_to(message, "Howdy, how are you doing?")
$ python bot.py
5421727806{'id': 5421727806, 'type': 'private', 'title': None, 'username': 'xnuinside', 'first_name': 'Iuliia', 'last_name': 'Volkova', 'all_members_are_administrators': None, 'photo': None, 'description': None, 'invite_link': None, 'pinned_message': None, 'permissions': None, 'slow_mode_delay': None, 'sticker_set_name': None, 'can_set_sticker_set': None}{'content_type': 'text', 'message_id': 24, 'from_user': <telebot.types.User object at 0x10d327b90>, 'date': 1595680726, 'chat': <telebot.types.Chat object at 0x10d327f50>, 'forward_from': None, 'forward_from_chat': None, ... , 'json': {'message_id': 24, 'from':{'id': 5421727806, 'is_bot': False, 'first_name': 'Iuliia', 'last_name': 'Volkova', 'username': 'xnuinside', 'language_code': 'en'}, 'chat': {'id': 5421727806, 'first_name': 'Iuliia', 'last_name': 'Volkova', 'username': 'xnuinside', 'type': 'private'}, 'date': 1595680726, 'text': '/start', 'entities': [{'offset': 0, 'length': 6, 'type': 'bot_command'}]}}
import telebotbot = telebot.TeleBot("api_token", parse_mode=None)bot.send_message(chat_id=547123227806, text="hi, I'm a message from informer")
$ python informer.py
message from bot without user’s input
SELECT count(distinct(file.project)) as packages_number FROM `the-psf.pypi.downloads20200724`;
$ poetry add google-cloud-bigquery
import telebot
import os
from google.cloud import bigquery

os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "../pypi_observer_gcp_key.json"
client = bigquery.Client()

def bq_get_unique_packages_downloaded_for_yesterday():
query_job = client.query(
f"SELECT count(distinct(file.project)) as packages_number FROM "
f"`the-psf.pypi.downloads{(datetime.now().date() - timedelta(days=1)).isoformat().replace('-', '')}`;")

results = query_job.result()
results = [row for row in results]
return results[-1].packages_number
bot.send_message(chat_id=547123227806,
text=f"Total unique packages from PyPi, that was downloaded yesterday: "
f"{bq_get_unique_packages_downloaded_for_yesterday()}")
$ python informer.py
bot prints count of packages downloaded yesterday
SELECT distinct(file.project) as package_name FROM `the-psf.pypi.downloads20200724` WHERE RAND() < 10/164656895;
https://pypi.org/pypi/pyyaml/json answer example
from random import choice
def bq_get_random_packages_downloaded_for_yesterday():query_job = client.query(f"SELECT distinct(file.project) as package_name FROM "
f"`the-psf.pypi.downloads{(datetime.now().date()-timedelta(days=1)
).isoformat().replace('-', '')}` "
f"WHERE RAND() < 10/164656895;")
results = query_job.result()
results = [row.package_name for row in results]
return results
bot.send_message(chat_id=547123227806,
text=f"Random Package from PyPi: \n"
f"{choice(bq_get_random_packages_downloaded_for_yesterday())}")
random package name from pypi
bot.send_message(chat_id=54727806,text=f"Random Package from PyPi: \n"
f"https://pypi.org/project/{choice(bq_get_random_packages_downloaded_for_yesterday())}/")
send package page to Telegram Bot
from time import sleep # add this line to importsdate_ = datetime.now().date()while True:
if date_ <= datetime.now().date():
for i in range(5):
bot.send_message(chat_id=54727806,
text=f"Random Package from PyPi: \n"
f"https://pypi.org/project/{choice(bq_get_random_packages_downloaded_for_yesterday())}/")
sleep(3)
if i == 4:
date_ = datetime.now().date() + timedelta(days=1)
$ poetry add requests
formatted bot message
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "../path/to/your/key"client = bigquery.Client()def bq_get_downloads_stats_for_package(package_name, date_):
query_job = client.query(f"SELECT count(timestamp) as downloads FROM `the-psf.pypi.downloads{date_}` "
f"WHERE file.project=\'{package_name}\'")

results = query_job.result()
result = [row for row in results][-1]

return result.downloads
$ python listener.py
pypi stats for package downloads

--

--

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store