utils/client.py

from http.client import HTTPSConnection
from base64 import b64encode
from json import loads
from json import dumps
from keys import data_for_seo_email, data_for_seo_password

class RestClient:
    domain = "api.dataforseo.com"

    def __init__(self, username, password):
        self.username = username
        self.password = password

    def request(self, path, method, data=None):
        connection = HTTPSConnection(self.domain)
        try:
            base64_bytes = b64encode(
                ("%s:%s" % (self.username, self.password)).encode("ascii")
                ).decode("ascii")
            headers = {'Authorization' : 'Basic %s' %  base64_bytes, 'Content-Encoding' : 'gzip'}
            connection.request(method, path, headers=headers, body=data)
            response = connection.getresponse()
            return loads(response.read().decode())
        finally:
            connection.close()

    def get(self, path):
        return self.request(path, 'GET')

    def post(self, path, data):
        if isinstance(data, str):
            data_str = data
        else:
            data_str = dumps(data)
        return self.request(path, 'POST', data_str)



client = RestClient(data_for_seo_email, data_for_seo_password)
# client = RestClient("[email protected]", "cb1661e9ec7c1fba")



def generate_keyword_intent_list(list_of_keywords: list):
    post_data = dict()
    # simple way to set a task
    post_data[len(post_data)] = dict(
        keywords= list_of_keywords,
        language_name="English"
    )
    # POST /v3/dataforseo_labs/google/search_intent/live
    response = client.post("/v3/dataforseo_labs/google/search_intent/live", post_data)
    # you can find the full list of the response codes here https://docs.dataforseo.com/v3/appendix/errors
    if response["status_code"] == 20000:
        # print(response)
        return response["tasks"][0]["result"][0]["items"]
        # do something with result
    else:
        print("error. Code: %d Message: %s" % (response["status_code"], response["status_message"]))

    



def generate_seo_metatitle(keyword, num_query_results=10):
    post_data = dict()
    # You can set only one task at a time
    post_data[len(post_data)] = dict(
        language_code="en",
        location_code=2840,
        keyword=keyword
    )
    
    response = client.post("/v3/serp/google/organic/live/regular", post_data)



    if response["status_code"] == 20000:
        # print(response)
        d= response['tasks'][0]
        # print(d)
        result_dict= d['result'][0]['items']
        summary= ''
        for i in result_dict[:num_query_results]:
            # summary+= i['title']+' '
            x= i['description']
            if x:
                summary+= i['title']+' '+ i['domain']+' '+ i['url']+' '+x+' '
        summary= summary.replace('\n', ' ') 
        summary= summary.replace(',', ' ') 
        summary= summary.replace(';', ' ') 
        print(summary)
        return summary
        
        
        # do something with result
    else:
        print("error. Code: %d Message: %s" % (response["status_code"], response["status_message"] , f" for keyword {keyword}"))
        

def generate_seo_metatitle_train(keyword, num_query_results=10):
    post_data = dict()
    # You can set only one task at a time
    post_data[len(post_data)] = dict(
        language_code="en",
        location_code=2840,
        keyword=keyword
    )
    
    response = client.post("/v3/serp/google/organic/live/regular", post_data)



    if response["status_code"] == 20000:
        # print(response)
        d= response['tasks'][0]
        # print(d)
        result_dict= d['result'][0]['items']
        summary= ''
        for i in result_dict[:num_query_results]:
            # summary+= i['title']+' '
            x= i['description']
            if x:
                summary+= i['title']+' '+ i['domain']+' '+ i['url']+' '+x+' '
        # print(summary)
        return summary.replace('\n', ' ') 
        
        # do something with result
    else:
        print("error. Code: %d Message: %s" % (response["status_code"], response["status_message"] , f" for keyword {keyword}"))
        
        

def generate_top_urls(keyword, num_query_results=10)->list:
    # print(keyword)
    post_data = dict()
    # You can set only one task at a time
    post_data[len(post_data)] = dict(
        language_code="en",
        location_code=2840,
        keyword=keyword
    )
    
    response = client.post("/v3/serp/google/organic/live/regular", post_data)



    if response["status_code"] == 20000:
        # print(response)
        d= response['tasks'][0]
        # print(d)
        result_dict= d['result'][0]['items']
        urls= []
        for i in result_dict[:num_query_results]:
            # summary+= i['title']+' '
            urls.append(i['url'])
            # x= i['description']
            # if x:
            #     summary+= i['title']+' '+ i['domain']+' '+ i['url']+' '+x+' '
        # print(summary)
        # print(urls)
        return (urls)
        
        # do something with result
    else:
        print("error. Code: %d Message: %s" % (response["status_code"], response["status_message"] , f" for keyword {keyword}"))
        
        





def generate_seo_metatitle_train_for_kg(keyword, num_query_results=10):
    post_data = dict()
    # You can set only one task at a time
    post_data[len(post_data)] = dict(
        language_code="en",
        location_code=2840,
        keyword=keyword
    )
    
    response = client.post("/v3/serp/google/organic/live/regular", post_data)



    if response["status_code"] == 20000:
        # print(response)
        d= response['tasks'][0]
        # print(d)
        result_dict= d['result'][0]['items']
        summary= ''
        for i in result_dict[:num_query_results]:
            # summary+= i['title']+' '
            x= i['description']
            if x:
                # summary+= i['title']+' '+ i['domain']+' '+ i['url']+' '+x+' '
                summary+= i['title']+' '+x+' '
        # print(summary)
        return summary.replace('\n', ' ') 
        
        # do something with result
    else:
        print("error. Code: %d Message: %s" % (response["status_code"], response["status_message"] , f" for keyword {keyword}"))
        
        
        
def generate_intent_using_dataforseo(keyword_list):
    post_data = dict()
    # simple way to set a task
    post_data[len(post_data)] = dict(
        keywords= keyword_list,
        language_name="English"
    )
    
    response = client.post("/v3/dataforseo_labs/google/search_intent/live", post_data)
    # you can find the full list of the response codes here https://docs.dataforseo.com/v3/appendix/errors
    if response["status_code"] == 20000:
        # print(response)
        # do something with result
        d= response['tasks'][0]
        # print(d)
        result_dict= d['result'][0]['items']
        intent_dict= dict()
        for i in result_dict:
            key= i['keyword']
            value= i['keyword_intent']['label']
            intent_dict[key]= value
        return intent_dict
    else:
        print("error. Code: %d Message: %s" % (response["status_code"], response["status_message"]))