كيفية تفويض تقارير بسيطة للروبوت. نكتب روبوتًا بلغة Python وGoogle BigQuery

هل لديك مهام تتكرر يومًا بعد يوم، وأسبوعًا بعد أسبوع؟ على سبيل المثال، كتابة التقارير. يمكنك طلب البيانات وتحليلها وتصورها (إنشاء الرسوم البيانية والمخططات)، ثم إرسالها إلى رئيسك في العمل. ولكن ماذا لو كان كل هذا آليًا؟

سنقوم في هذا البرنامج التعليمي بإنشاء روبوت لـ Telegram سيساعد في أتمتة إعداد التقارير. والأروع من ذلك أن البرنامج بأكمله سيتكون من 50 سطرًا فقط من التعليمات البرمجية! إذا كنت تقوم بإنشاء روبوت لـ Telegram لأول مرة، فيجب عليك أيضًا قراءة هذا الروبوت بعد.

يوصي Skillbox بما يلي: دورة عملية مطور بايثون من الصفر.

نذكر: لجميع قراء "Habr" - خصم 10 روبل عند التسجيل في أي دورة Skillbox باستخدام رمز "Habr" الترويجي.

يهبط

تثبيت المكتبات

سوف نستخدم جوجل السحابية BigQuery للحصول على البيانات من Google BigQuery. matplotlib, نمباي и الباندا سوف تساعدك على تصور البيانات الخاصة بك. بيثون برقية بوت سيتم إرسال البيانات النهائية إلى Telegram.

pip3 تثبيت google-cloud-bigquery matplotlib numpy pandas python-telegram-bot

ربط Google BigQuery API

إذا أردنا استخدام الخدمة، نحتاج إلى توصيل Google BigQuery API. للقيام بذلك نذهب إلى Google Developers Console وإنشاء مشروع جديد (أو تحديد مشروع موجود).

في لوحة التحكم، حدد ENABLE APIS AND SERVICES وابحث عن BigQuery API.

حدد تمكين لتوصيل API.

إنشاء مفتاح الحساب

دعنا نذهب مرة أخرى Google Developers Console، حدد علامة التبويب بيانات الاعتماد، ثم قم بإنشاء بيانات الاعتماد ومفتاح حساب الخدمة.

ثم - حساب خدمة جديد، وأدخل الاسم في حقل اسم حساب الخدمة.

من القائمة المنسدلة الدور، حدد المشروع > المالك، ثم إنشاء.

الملف الذي سيتم تنزيله تلقائيًا يسمى creds.json.

قم بتعيين GOOGLE_APPLICATION_CREDENTIALS، مع تحديد المسار إلى creds.json في الوحدة الطرفية.

تصدير GOOGLE_APPLICATION_CREDENTIALS='[PATH_TO_CREDS.JSON]'

إذا سارت الأمور على ما يرام، فقد حان الوقت لبدء كتابة البرنامج.

قم بإنشاء تطبيق

بالنسبة للبرنامج التعليمي، سنستخدم البيانات من bigquery-public-data.stackoverflow، وبالنسبة لتقريرنا، سنختار عدد المنشورات اليومية.

كل شيء بسيط للغاية.

الاستعلام عن الجدول -> تصور البيانات -> حفظ التصور -> إرسال الصورة

لنقم بإنشاء وظيفة واحدة لتحديد كل موضوع.

الاستعلام إلى BigQuery

أولاً نقوم باستيراد المكتبة.

من google.cloud قم باستيراد BigQuery

نقوم بإنشاء دالة تسمى query_to_bigquery، حيث تكون المعلمة query.

def query_to_bigquery(query):
    client = bigquery.Client()
    query_job = client.query(query)
    result = query_job.result()
    dataframe = result.to_dataframe()
    return dataframe

ستعيد هذه الوظيفة الطلب كإطار بيانات.

تصور البيانات

لحل هذه المشكلة، اختر matplotlib.

استيراد matplotlib.pyplot كـ PLT

نحتاج إلى خمس معلمات، حيث x هي بيانات المحور x، وx_label هو عنوان المحور، وy هو بيانات المحور y، وy_label هو عنوان المحور، والعنوان هو عنوان التمثيل المرئي بأكمله.

def visualize_bar_chart(x, x_label, y, y_label, title):
    plt.title(title)
    plt.xlabel(x_label)
    plt.ylabel(y_label)
    index = np.arange(len(x))
    plt.xticks(index, x, fontsize=5, rotation=30)
    plt.bar(index, y)
    return plt

احفظ الصورة

الآن دعونا نستخدم وظيفتين لإنشاء تصور وحفظه.

وسوف نقوم بإرسال عدد المشاركات المنشورة يوميا. أولا نكتب الطلب.

query = """
        SELECT DATE(creation_date) date, COUNT(*) total_posts
        FROM `bigquery-public-data.stackoverflow.post_history`
        GROUP BY 1
        HAVING date > DATE_SUB('2018-12-02', INTERVAL 14 DAY)
        ORDER BY 1
        """

يساعد الاستعلام في جمع البيانات لمدة أسبوعين بدءًا من 2 ديسمبر 2018.

نستخدم هذا التاريخ لأن 2018-12-02 هو أحدث البيانات المسجلة في bigquery-public-data.stackoverflow.post_history، وفي حالات أخرى يمكنك استخدام CURRENT_DATE() للحصول على أحدث البيانات.

قم باستدعاء الدالة query_to_bigquery للحصول على البيانات.

إطار البيانات = query_to_bigquery(استعلام)

ثم نستخدم عمود بيانات التاريخ للمحور السيني، وعمود Total_posts للمحور الصادي.

x = dataframe['date'].tolist()
y = dataframe['total_posts'].tolist()

نحن نتصورها باستخدام وظيفة visualize_bar_chart ونحفظها كصورة.

plt = visualize_bar_chart(x=x, x_label='التاريخ', y=y, y_label='إجمالي المشاركات', title='المشاركات اليومية')
plt.savefig('viz.png')

نقوم بلف هذا الكود في دالة تسمى get_and_save_image.

def get_and_save_image():
    query = """
            SELECT DATE(creation_date) date, COUNT(*) total_posts
            FROM `bigquery-public-data.stackoverflow.post_history`
            GROUP BY 1
            HAVING date > DATE_SUB('2018-12-02', INTERVAL 14 DAY)
            ORDER BY 1
            """
    dataframe = query_to_bigquery(query)  
    x = dataframe['date'].tolist()
    y = dataframe['total_posts'].tolist()
    plt = visualize_bar_chart(x=x, x_label='Date', y=y, y_label='Total Posts', title='Daily Posts')
    plt.savefig('viz.png')

أرسل صورة

لإرسال تقرير إلى المستلم، تحتاج إلى معرفة معلمة chat_id.

نحن نستخدم com.userinfobot واكتب /ابدأ. يستجيب الروبوت بالمعلومات الضرورية، ويتم تضمين chat_id في حقل المعرف.

لنقم الآن بإنشاء وظيفة send_image. سيستخدم وظيفة get_and_save_image لاسترداد الصورة وحفظها. وبعد ذلك نرسل كل شيء إلى جهة الاتصال الصحيحة.

def send_image(bot, update):
    get_and_save_image()
    chat_id = 'CHAT_ID_RECEIVER'
    bot.send_photo(chat_id=chat_id, photo=open('viz.png','rb'))

البرنامج الرئيسي

أخيرًا، نقوم بإنشاء وظيفة أخرى، رئيسية، لتشغيل التطبيق. لا تنس تغيير YOUR_TOKEN للروبوت.

تذكر: سيقوم هذا البرنامج بإرسال الصورة تلقائيًا في الوقت الذي تحدده. على سبيل المثال، سنرسل تقريرًا في الساعة التاسعة صباحًا كل يوم.

def main():
    updater = Updater('YOUR_TOKEN')
    updater.job_queue.run_daily(send_image, time=datetime.datetime.strptime('9:00AM', '%I:%M%p').time(), days=(0,1,2,3,4,5,6))
    updater.start_polling()
    updater.idle()
 
if __name__ == '__main__':
    main()

ونتيجة لذلك، سيبدو طلبنا كما يلي:

from google.cloud import bigquery
from telegram.ext import Updater
 
import matplotlib.pyplot as plt
import numpy as np
import datetime
 
def query_to_bigquery(query):
    client = bigquery.Client()
    query_job = client.query(query)
    result = query_job.result()
    dataframe = result.to_dataframe()
    return dataframe
 
def visualize_bar_chart(x, x_label, y, y_label, title):
    plt.title(title)
    plt.xlabel(x_label)
    plt.ylabel(y_label)
    index = np.arange(len(x))
    plt.xticks(index, x, fontsize=5, rotation=30)
    plt.bar(index, y)
    return plt
 
def get_and_save_image():
    query = """
            SELECT DATE(creation_date) date, COUNT(*) total_posts
            FROM `bigquery-public-data.stackoverflow.post_history`
            GROUP BY 1
            HAVING date > DATE_SUB('2018-12-02', INTERVAL 14 DAY)
            ORDER BY 1
            """
    dataframe = query_to_bigquery(query)  
    x = dataframe['date'].tolist()
    y = dataframe['total_posts'].tolist()
    plt = visualize_bar_chart(x=x, x_label='Date', y=y, y_label='Total Posts', title='Daily Posts')
    plt.savefig('viz.png')
 
def send_image(bot, update):
    get_and_save_image()
    chat_id = 'CHAT_ID_RECEIVER'
    bot.send_photo(chat_id=chat_id, photo=open('viz.png', 'rb'))
 
def main():
    updater = Updater('YOUR_TOKEN')
    updater.job_queue.run_daily(send_image, time=datetime.datetime.strptime('9:00AM', '%I:%M%p').time(), days=(0,1,2,3,4,5,6))
    updater.start_polling()
    updater.idle()
 
if __name__ == '__main__':
main()

احفظ الملف وأطلق عليه اسم main.py.

نقوم بتشغيل التطبيق عن طريق إدخال الأمر في المحطة:

بيثون3 main.py

كل شيء جاهز. الآن لدينا روبوت يتكون من 50 سطرًا من التعليمات البرمجية التي تولد التقارير دون تدخلنا.

دعونا نتحقق من الروبوت من هناعن طريق اختيار الأمر /إرسال.

يمكنك الحصول على الكود النهائي في بلدي جيثب.

يوصي Skillbox بما يلي:

• دورة عملية لمدة عامين "أنا مطور ويب للمحترفين".
• دورة على شبكة الإنترنت "C # developer من 0".
• دورة سنة عملية "مطور PHP من 0 إلى PRO".

المصدر: www.habr.com