هل لديك مهام تتكرر يوما بعد يوم ، من أسبوع إلى أسبوع؟ على سبيل المثال ، كتابة التقارير. يمكنك طلب البيانات وإجراء التحليل والتصور (إنشاء رسومات بيانية ومخططات) ثم إرسالها إلى الرئيس. ولكن ماذا لو كان كل هذا آليا؟
في هذا البرنامج التعليمي ، سننشئ روبوتًا لـ Telegram يساعدك في أتمتة التقارير. وأروع شيء هو أن البرنامج بأكمله سيتألف من 50 سطرًا فقط من الكود! إذا كنت تنشئ روبوتًا لـ Telegram للمرة الأولى ، فعليك قراءة هذا
المنشور هنا .
توصي Skillbox بما يلي: مطور Python بالطبع العملي من البداية .
نذكرك: لجميع قراء "Habr" - خصم بقيمة 10،000 روبل عند التسجيل في أي دورة تدريبية في Skillbox باستخدام الرمز "Habr" الترويجي.
النزول
تثبيت المكتباتسنستخدم
google-cloud-bigquery لاسترداد البيانات من Google BigQuery.
سوف matplotlib ،
numpy والباندا تساعد على تصور البيانات. سيرسل
python-telegram-bot البيانات النهائية إلى 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.
حدد تمكين لتوصيل واجهة برمجة التطبيقات.
إنشاء مفتاح حسابننتقل مرة أخرى إلى
Google Developers Console ، وحدد علامة التبويب مفتاح بيانات الاعتماد ، وإنشاء بيانات الاعتماد ، وحساب الخدمة.
ثم - حساب خدمة جديد ، وفي حقل اسم حساب الخدمة ، أدخل الاسم.
من القائمة المنسدلة "الأدوار" ، حدد Project> Owner ، ثم إنشاء.
يُطلق على الملف الذي سيتم تحميله تلقائيًا اسم creds.json.
نحن نكشف GOOGLE_APPLICATION_CREDENTIALS ، مع الإشارة إلى المسار إلى النقاط. json في الجهاز.
تصدير GOOGLE_APPLICATION_CREDENTIALS = '[PATH_TO_CREDS.JSON]'إذا سارت الأمور على ما يرام ، فقد حان الوقت لبدء كتابة البرنامج.
إنشاء التطبيق
بالنسبة للبرنامج التعليمي ، سوف نستخدم البيانات من bigquery-public-data.stackoverflow ، لتقريرنا ، سنختار عدد المنشورات اليومية.
كل شيء بسيط جدا.
الاستعلام عن الجدول -> تخيل البيانات -> حفظ التصور -> إرسال الصورة
دعونا إنشاء وظيفة واحدة لتحديد كل موضوع.
الاستعلام عن BigQueryأولاً ، قم باستيراد المكتبة.
من google.cloud استيراد طلب استفسارإنشاء دالة تسمى query_to_bigquery ، حيث المعلمة هي الاستعلام.
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 للحصول على البيانات.
dataframe = query_to_bigquery (استعلام)ثم استخدم عمود التاريخ للمحور x ، وعمود total_posts للمحور y.
x = dataframe ['date']. tolist ()
y = dataframe ['total_posts']. tolist ()نحن نتصور استخدام دالة visualize_bar_chart وحفظها كصورة.
plt = visualize_bar_chart (x = x، x_label = 'Date'، 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.
نحن نستخدم
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.
نطلق التطبيق عن طريق إدخال الأمر في المحطة:
python3 main.pyكل شيء جاهز. الآن لدينا روبوت يتكون من 50 سطرًا من التعليمات البرمجية التي تنشئ تقارير بدون تدخلنا.
دعونا التحقق من الروبوت
من هنا عن طريق تحديد الأمر / send.
يمكنك الحصول على الكود النهائي في
GitHub الخاص بي .
توصي Skillbox بما يلي: