Google क्लाउड फ़ंक्शंस का उपयोग करके शेड्यूल पर बिगमैक्स से शेड्यूल कॉल रिपोर्ट्स को कोइएमक्यू से आयात करें

किस लिए

विज्ञापन अभियानों की जटिल संरचना और बड़ी संख्या में कॉल के साथ, इनकमिंग कॉल के बारे में जानकारी के भंडारण, प्रसंस्करण और विश्लेषण के लिए अतिरिक्त उपकरण आवश्यक हो जाते हैं। अक्सर आपको लंबे समय तक डेटा तक त्वरित पहुंच की आवश्यकता होती है। कभी-कभी आपको जटिल डेटा प्रोसेसिंग की आवश्यकता होती है, किसी विशिष्ट चैनल या अभियान के लिए सहसंबंधी कॉल।

काम को गति देने के लिए विकल्पों में से एक, जो अतिरिक्त लाभ भी प्रदान करता है, CoMagic से Google BigQuery को कॉल आयात कर रहा है। बहुत अधिक BigQuery के लाभों के बारे में लिखा गया है, तो चलिए निर्माण के लिए आगे बढ़ते हैं।

एक स्वचालित आयात बनाने के लिए आपको आवश्यकता होगी:

1. Google खाता (यदि यह पहले से नहीं है) निर्मित परियोजना के साथ
2. अजगर ज्ञान
3. पेश है गूगल क्लाउड डॉक्यूमेंटेशन

कैसे एक परियोजना बनाने के लिए यहाँ वर्णित है । प्रोजेक्ट बनने के बाद, आपको BigQuery में एक डेटासेट बनाने की आवश्यकता है। बीक्यू प्रलेखन और डेटासेट बनाने के लिए निर्देश ।

CoMagic से डेटा पुनर्प्राप्त करना

CoMagic प्रलेखन की ओर मुड़ते हुए। कॉल या कॉल की सूची प्राप्त करने के लिए, हमें रिपोर्ट अनुभाग की आवश्यकता है।

हम CoMagic एपीआई के साथ काम करने के लिए एक सरल वर्ग बनाते हैं। GitHub के लिंक में अंत में सभी आवश्यक आवश्यकताओं को इंगित किया जाएगा।

import json import requests import random import pandas as pd class ComagicClient: """     CoMagic""" def __init__(self, login, password): """     CoMagic""" self.login = login self.password = password self.base_url = 'https://dataapi.comagic.ru/v2.0' self.payload_ = {"jsonrpc":"2.0", "id":1, "method":None, "params": None} self.token = self.get_token(self.login, self.password) def base_request(self, method, params): """     CoMagic.      API   .    JSON-like . : https://www.comagic.ru/support/api/data-api/""" id_ = random.randrange(10**7) #  payload = self.payload_.copy() payload["method"] = method payload["params"] = params payload["id"] = id_ self.r = requests.post(self.base_url, data=json.dumps(payload)) self.last_response = json.loads(self.r.text) return self.last_response def get_token(self, login, password): """   .       .  .""" method = "login.user" params = {"login":self.login, "password":self.password} response = self.base_request(method, params) token = response['result']['data']['access_token'] return token def get_report_per_page(self, method, params): """  .      10000 .    .     110000 .     JSON-like .""" response = self.base_request(method, params) print(f"""  c {params["date_from"]}  {params["date_till"]}.  = {params["offset"]}""") result = response['result']['data'] if len(result) < 10000: return result else: params['offset'] += 10000 add_result = self.get_report_per_page(method, params) return result + add_result def get_basic_report(self, method, fields, date_from, date_till, filter=None, offset=0): """   .       method  fields.       .       ,   ,       . method -- <string>   date_from -- <string>  .  "YYYY-MM-DD hh:mm:ss" date_till -- <string>  .  "YYYY-MM-DD hh:mm:ss" fields -- <list>,    filter [] - <dict>  offset [] -- <int>  return -- <list>  """ params = {"access_token":self.token, "limit":10000, "date_from":date_from, "date_till":date_till, "fields": fields, "offset": offset} if filter: params['filter'] = filter report = self.get_report_per_page(method, params) return report 

अब आपको यह निर्धारित करने की आवश्यकता है कि किस प्रकार के डेटा की आवश्यकता है। डेटा को संसाधित करने और दृश्यमान बनाने की आवश्यकता है ताकि इसे BigQuery में लोड किया जा सके।

एक सहायक वर्ग बनाएं और CoMagic से प्राप्त डेटा को परिभाषित करें।

 class ComagicHandler(ComagicClient): """    ,   CoMagic""" time_partition_field = 'PARTITION_DATE' def __init__(self, login, password, first_call_date): self.day_before_first_call = pd.to_datetime(first_call_date) - pd.Timedelta(days=1) super().__init__(login, password) def get_calls_report(self, date_from, date_till): """        .           .    Pandas DataFrame.      .      Connector    .    .    .  Pnadas.DataFrame""" method = "get.calls_report" fields = ['id', 'visitor_id', 'person_id', 'start_time', 'finish_reason', 'is_lost', 'tags', 'campaign_name','communication_number', 'contact_phone_number', 'talk_duration', 'clean_talk_duration', 'virtual_phone_number', 'ua_client_id', 'ym_client_id', 'entrance_page', 'gclid', 'yclid', 'visitor_type', 'visits_count', 'visitor_first_campaign_name', 'visitor_device', 'site_domain_name','utm_source', 'utm_medium', 'utm_campaign', 'utm_content', 'eq_utm_source', 'eq_utm_medium', 'eq_utm_campaign', 'attributes'] #   CoMagic calls_data = self.get_basic_report(method, fields, date_from, date_till) # DataFrame df = pd.DataFrame(calls_data) #    .    . df[self.time_partition_field] = pd.to_datetime(df.start_time).apply(lambda x: x.date()) #  tags,   BigQuery       ,  # CoMagic.    . df['tags'] = df.tags.apply(lambda x: x if x == None else [i['tag_name'] for i in x]) return df 

BigQuery को डेटा भेजना

CoMagic से डेटा प्राप्त और परिवर्तित होने के बाद, आपको इसे BigQuery को भेजने की आवश्यकता है।

 from google.cloud import bigquery from google.cloud.exceptions import NotFound import pandas as pd class BQTableHanler: """     BigQuery""" time_partition_field = 'PARTITION_DATE' def __init__(self, full_table_id, service_account_file_key_path = None): """       `myproject.mydataset.mytable`.  ,   Application Default Credentials,           .""" self.full_table_id = full_table_id project_id, dataset_id, table_id = full_table_id.split(".") self.project_id = project_id self.dataset_id = dataset_id self.table_id = table_id if service_account_file_key_path: #      from google.oauth2 import service_account self.credentials = service_account.Credentials.from_service_account_file( service_account_file_key_path, scopes=["https://www.googleapis.com/auth/cloud-platform"],) self.bq_client = bigquery.Client(credentials = self.credentials, project = self.project_id) else: self.bq_client = bigquery.Client() self.dataset = self.bq_client.get_dataset(self.dataset_id) self.location = self.dataset.location self.table_ref = self.dataset.table(self.table_id) def get_last_update(self): """        Pandas datetime.      False.""" try: self.bq_client.get_table(self.full_table_id) except NotFound as error: return False query = f"""SELECT MAX({self.time_partition_field}) as last_call FROM `{self.full_table_id}`""" result = self.bq_client.query(query,location=self.location).to_dataframe() date = pd.to_datetime(result.iloc[0,0]).date() return date def insert_dataframe(self, dataframe): """      BigQuery.     Pandas DataFrame.    ,       .""" job_config = bigquery.LoadJobConfig() #     job_config._properties['load']['timePartitioning'] = {'type': 'DAY', 'field': self.time_partition_field} result = self.bq_client.load_table_from_dataframe(dataframe, self.table_ref, job_config=job_config).result() return result 

डेटा अपडेट करने के लिए तर्क का निर्धारण करें

चूंकि CoMagic से प्राप्त डेटा पंक्तियों की संख्या पर एक सीमा है, इसलिए अनुरोधित डेटा की संख्या को सीमित करना आवश्यक है। हम अनुरोध अवधि को सीमित करेंगे। ऐसा करने के लिए, आपको एक सहायक फ़ंक्शन की आवश्यकता होती है जो बड़ी अवधि को एक निर्दिष्ट लंबाई के खंडों में विभाजित करेगा।

 def interval_split(array, interval): """      .   ,      2,    -   ,     -    . : get_intervals([1,2,3,4,5,6,7], 3) => [[1,3], [4,6], [7]] get_intervals([1,2,3], 4) => [[1,3]]""" intervals = [] iw, i = 0, 0 l = len(array) for v in array: if i==0 or (i)%interval==0: intervals.append([v]) if (i+1)%interval == 0 or (i+1) == l: intervals[iw].append(v) iw+=1 i+=1 return intervals 

पहली बार डेटा लोड करते समय यह आवश्यक है, जब आपको लंबे समय तक डेटा डाउनलोड करने की आवश्यकता होती है। अवधि को कई छोटे अवधियों में विभाजित किया जाता है। वैसे, क्लाउड फंक्शन का उपयोग किए बिना ऐसा करना बेहतर है, क्योंकि उनके पास समय सीमा है। खैर, या, एक विकल्प के रूप में, आप कई बार फ़ंक्शन चला सकते हैं।

हम BigQuery टेबल को लिंक करने के लिए एक कनेक्टर क्लास बनाते हैं जहां हम डेटा और डेटा को कोमजिक से स्टोर करना चाहते हैं।

 from helpfunctions import interval_split import pandas as pd class Connector: """      """ time_partition_field = 'PARTITION_DATE' #  -.       def __init__ (self, source, dest): """          """ self.source = source self.dest = dest self.source.time_partition_field = self.time_partition_field self.dest.time_partition_field = self.time_partition_field def insert_data_in_dest(self, start_date, end_date): """      .          ,     .""" dates = pd.date_range(start_date, end_date) week_intervals = interval_split(dates, 7) #     7  for week_interval in week_intervals: date_from = week_interval[0].strftime("%Y-%m-%d") + " 00:00:00" date_till = week_interval[1].strftime("%Y-%m-%d") + " 23:59:59" calls_df = self.source.get_calls_report(date_from, date_till) self.dest.insert_dataframe(calls_df) print (f"  {date_from}  {date_till}   ") return True def update_dest_data(self): #     BigQuery last_date = self.dest.get_last_update() if not last_date: #    last_date = self.source.day_before_first_call yesterday = pd.Timestamp.today(tz='Europe/Moscow').date() - pd.Timedelta(days=1) if last_date == yesterday: print("  ") else: last_date = last_date + pd.Timedelta(days=1) self.insert_data_in_dest(last_date, yesterday) return True 

अगला, हम डेटा को अपडेट करने के लिए मुख्य फ़ंक्शन को निर्धारित करते हैं, जो एक शेड्यूल पर लॉन्च किया जाएगा।

 from connector import Connector from bqhandler import BQTableHanler from comagichandler import ComagicHandler from credfile import * def main(event, context): """    event, context  : https://cloud.google.com/functions/docs/writing/background#functions-writing-background-hello-pubsub-python""" #import base64 #pubsub_message = base64.b64decode(event['data']).decode('utf-8') # c      comagic_handler = ComagicHandler(COMAGIC_LOGIN, COMAGIC_PASSWORD, FIRST_CALL_DATE) bq_handelr = BQTableHanler(full_table_id, google_credintials_key_path) #  connector = Connector(comagic_handler, bq_handelr) #     connector.update_dest_data() 

Google क्लाउड प्लेटफ़ॉर्म कॉन्फ़िगर करें

हम एक ज़िप संग्रह में सभी फ़ाइलों को इकट्ठा करते हैं। Credfile.py फ़ाइल में, हम टोकन प्राप्त करने के लिए CoMagic लॉगिन और पासवर्ड दर्ज करते हैं, साथ ही BigQuery में तालिका का पूरा नाम और स्थानीय मशीन से स्क्रिप्ट लॉन्च होने पर सेवा खाता फ़ाइल का पथ।

क्लाउड फंक्शन बनाएं

• कंसोल पर जाएं
• यदि कोई फ़ंक्शन अभी तक नहीं बनाया गया है, तो "फ़ंक्शन बनाएँ" पर क्लिक करें
• ट्रिगर फ़ील्ड में, PUB / SUB चुनें
• PUB / SUB के लिए एक नई थीम बनाएँ। उदाहरण के लिए, 'update_calls'
• स्रोत: ज़िप अपलोड (स्थानीय ज़िप फ़ाइल)
• पर्यावरण: पायथन 3.7
• ज़िप फ़ाइल डाउनलोड करें
• क्लाउड स्टोरेज का एक अस्थायी सेगमेंट चुनना
• क्षेत्र में 'फ़ंक्शन कहा जाता है' हम 'मुख्य' लिखते हैं
• आवंटित स्मृति: वैकल्पिक

शेड्यूलर और PUB / SUB को कॉन्फ़िगर करना

अंतिम चरण में, हमने `update_calls` ट्रिगर बनाया। यह स्वचालित विषय विषयों की सूची में दिखाई दिया है।

अब, क्लाउड शेड्यूलर के साथ आपको ट्रिगर कॉन्फ़िगर करने की आवश्यकता है। जब आग लगेगी और जीसीएफ शुरू हो जाएगा।

• कंसोल पर जाएं
• CRON प्रारूप में आवृत्ति क्षेत्र में, सेट करें जब ट्रिगर में आग लगनी चाहिए और फ़ंक्शन शुरू हो जाए।
• गंतव्य: पब / उप
• विषय: फ़ंक्शन बनाते समय निर्दिष्ट की गई विषयवस्तु को पंजीकृत करें: "update_calls"
• पेलोड्स * (पेलोड्स) - यह वह सूचना है जिसे पब / सब और मुख्य समारोह में स्थानांतरित किया जाएगा

अब स्क्रिप्ट को दैनिक 01:00 बजे लॉन्च किया जाएगा और कॉल डेटा को पिछले दिन के अंत में अपडेट किया जाएगा।

GitHub को स्थानीय कंप्यूटर से चलाने के लिए लिंक करें
GitHub जिप फाइल का लिंक