Cfrtjryk

My program automates the process of creating and publishing 152 blog posts quarterly. The blogging platform HubSpot has a series of APIs that allow developers to access account data from a CLI in Terminal. First step is creating an excel file (must save as CSV) and importing the file for use as a new HubDB table in HubSpot (basically a google sheet built into HS). The HubDB contains all of the content for a single post in a single row.

Not all of the functions defined are used, however I expect the scope to expand and to use this program to automate content creation for other blogs on HubSpot. Therefore, my goal is to make the process agnostic so that it may be applied broadly.

import requests

import json

import time

from urllib.request import urlopen

from pprint import pprint





 """

CRUD (Create, Retrieve, Update, And Delete) HTTP methods for the

ManpowerGroup Employment Outlook Survey hosted in HubSpot

"""



#  Set the API endpoint



APIKEY_VALUE = "XXX-X-X-XXX"

APIKEY = "?hapikey=" + APIKEY_VALUE

HS_API_URL = "http://api.hubapi.com"

PORTAL_ID = 2942250



# Define content group IDs as global variables

TEMPLATE_ID = 5548244265  #  HubSpot Templates

MANPOWERGROUP_ID = 5548244265  #  ManpowerGroup

MEOS_ID = 5664168304  #  ManpowerGroup Employment Outlook Survey

WEF_ID = 5505585256  # World Economic Forum



# List all HubDB table IDs as global variables

MEOS_Q1_2019 = 1029596 #  table ID

MEOS_Q2_2018 = 679969

MEOS_Q3_2018 = 700276

MEOS_Q4_2018 = 850049



# Define JSON Actions

PUBLISH = {"action": "schedule-publish"}

CANCEL = {"action": "cancel-publish"}



# Define publication details

PUBLISH_NOW = str(time.time())

DRAFT = 'draft'

LIVE = 'publish'

LIMIT = 152 # Number of rows in table to publish as posts



# CREATE BLOG POSTS

def create_batch_posts(table_id):

    """Get table by row

    GET /hubdb/api/v2/tables/:tableId/rows

    """

    xurl = "/hubdb/api/v2/tables/" + str(table_id) + "/rows?portalId=" + str(PORTAL_ID)

    url = HS_API_URL + xurl

    response = urlopen(url).read()

    table_data = json.loads(response)



    for row in table_data['objects']:

        name = row["values"]["1"]

        embed = row["values"]["2"]

        title = row["values"]["3"]

        state = row["values"]["4"]

        date = row["values"]["5"]

        year = row["values"]["6"]

        quarter = row["values"]["7"]

        country = row["values"]["9"]

        forecast = row["values"]["10"]

        blog_author_id = row["values"]["11"]

        campaign_id = row["values"]["12"]

        campaign_name = row["values"]["13"]

        content_group_id = row["values"]["14"]

        featured_image = row["values"]["15"]

        year_id = row["values"]["19"]

        quarter_id = row["values"]["20"]

        market_id = row["values"]["21"]

        state_id = row["values"]["22"]

        epoch_date = row["values"]["23"]

        embed_pdf = "<embed src='https://drive.google.com/viewerng/viewer?embedded=true&amp;url=https://go.manpowergroup.com/hubfs/MEOS/2019_Q1/" + str(embed) + ".pdf' width='500' height='675'>"



        POST = {

            "blog_author_id": blog_author_id,

            "campaign": campaign_id,

            "campaign_name": campaign_name,

            "content_group_id": content_group_id,

            "featured_image": featured_image,

            "topic_ids": [year_id, quarter_id, market_id, state_id],

            "meta_description": str(quarter) + " ~ " + str(year) + " ~ " + str(date) + " | " + str(forecast) + " " + str(title) + " " +  str(name) + " | " + str(state) + ", " + str(country),

            "name": str(date) + " | " + str(forecast) + " " + str(title) + " " +  str(name),

            "post_body": embed_pdf,

            "publish_date": str(epoch_date),

            "publish_immediately": False,

            "slug": str(year) + "_" + str(quarter) + "/" + str(embed),

            "use_featured_image": True

        }



        xurl = "/content/api/v2/blog-posts"

        url = HS_API_URL + xurl + APIKEY

        headers = { "content-type" : "application/json" }

        response = requests.post(url, headers=headers, data=json.dumps(POST))

        pprint(response.status_code, response.content, response)





# RETRIEVE CONTENT FROM HubDB

def get_single_row(table_id):

    """Get table by row

    GET /hubdb/api/v2/tables/:tableId/rows

    """

    xurl = "/hubdb/api/v2/tables/" + str(table_id) + "/rows?portalId=" + str(PORTAL_ID)

    url = HS_API_URL + xurl

    response = urlopen(url).read()

    table_data = json.loads(response)

    return(table_data)





def get_all_rows(table_id):

    """Get table by row

    GET /hubdb/api/v2/tables/:tableId/rows

    """

    xurl = "/hubdb/api/v2/tables/" + str(table_id) + "/rows?portalId=" + str(PORTAL_ID)

    url = HS_API_URL + xurl

    response = urlopen(url).read()

    table_data = json.loads(response)

    return(table_data)





def get_all_tables():

    """Get the tables

    GET /hubdb/api/v2/tables

    """

    xurl = "/hubdb/api/v2/tables"

    url = HS_API_URL + xurl + APIKEY

    response = urlopen(url).read()

    all_tables = json.loads(response)

    return all_tables





def get_table_by_id(table_id):

    """Get the tables

    GET /hubdb/api/v2/tables/:tableId

    """

    xurl = "/hubdb/api/v2/tables/" + str(table_id) + "?portalId=" + str(PORTAL_ID)

    url = HS_API_URL + xurl

    response = urlopen(url).read()

    table = json.loads(response)

    return table





# RETRIEVE BLOG POST ID

def list_all_posts():

    """List blog posts

    Hit this URL with a HTTP method of

    GET /content/api/v2/blog-posts

    """

    xurl = "/content/api/v2/blog-posts/"

    url = HS_API_URL + xurl + APIKEY

    blog_list = urlopen(url).read()

    # print(blog_list)

    return blog_list





def list_blog_posts(blog_id):

    """List blog posts

    Hit this URL with a HTTP method of

    GET /content/api/v2/blog-posts

    """

    xurl = "/content/api/v2/blog-posts/" + str(blog_id)

    url = HS_API_URL + xurl + APIKEY

    blog_list = urlopen(url).read()

    # print(blog_list)

    return blog_list





# PUBLISH BLOG POSTS

def publish_blog_post_by_id(blog_post_id):

    """

    Publish, schedule or unpublish a blog post

    POST /content/api/v2/blog-posts/:blog_post_id/publish-action

    """

    xurl = "/content/api/v2/blog-posts/" + str(blog_post_id) + "/publish-action"

    url = HS_API_URL + xurl + APIKEY

    response = requests.post(url)

    headers = { "content-type" : "application/json" }

    response = requests.post(url, headers=headers, data=json.dumps(PUBLISH))

    # pprint(response.status_code, response.content, response)

    return response





def publish_post():

    """Get the blog post by ID

    GET /content/api/v2/blog-posts/

    """

    xurl = "/content/api/v2/blog-posts/"

    url = HS_API_URL + xurl + APIKEY

    response = urlopen(url).read()

    blog_data = json.loads(response)

    for blog_post in blog_data['objects']:

        draft_id = blog_post['id']

        # print(draft_id)

        publish_blog_post_by_id(draft_id) # req. publish_post function





def publish_draft_posts(content_group_id, limit, draft):

    """Get the blog post by ID

    GET /content/api/v2/blog-posts/

    """

    xurl = "/content/api/v2/blog-posts/"

    url = HS_API_URL + xurl + APIKEY

    + '&content_group_id=' + str(content_group_id) # Blog ID

    + '&state=' + str(DRAFT) # DRAFT vs. LIVE

    + '&limit=' + str(LIMIT) # LIMIT TOTAL RESPONSES to ROW NUM

    response = urlopen(url).read()

    blog_data = json.loads(response)

    for blog_post in blog_data['objects']:

        print(blog_post['id'])





def get_blog_post_by_id(post_id):

    """Get the blog by ID

    GET /content/api/v2/blogs/:blog_id

    """

    xurl = "/content/api/v2/blogs/" + str(post_id)

    url = HS_API_URL + xurl + APIKEY

    response = urlopen(url).read()

    blog_info = json.loads(response)

    return blog_info



def cancel_post(blog_post_id):

    """

    Publish, schedule or unpublish a blog post

    POST /content/api/v2/blog-posts/:blog_post_id/publish-action

    """

    xurl = "/content/api/v2/blog-posts/" + str(blog_post_id) + "/publish-action"

    url = HS_API_URL + xurl + APIKEY

    # bin = "http://requestbin.fullcontact.com/1iiiiyo1"

    headers = { "content-type" : "application/json" }

    response = requests.post(url, headers=headers, data=json.dumps(CANCEL))

    # pprint(response.status_code, response.content, response)

    return response





# AUTOMATION FOR MANPOWERGROUP EMPLOYMENT OUTLOOK SURVEY (HUBSPOT)

def main():

    # Create a draft version of the blog posts

    create_batch_posts(1029596)

    # Publish the drafts

    publish_draft_posts(TEMPLATE_ID, 2, DRAFT)

    print("Published")





main()

Please check out my gist for easier reading....

At the top, add a shebang:

#!/usr/bin/env python3

For this line:

APIKEY = "?hapikey=" + APIKEY_VALUE

You're doing string concatenation when the nicer thing to do is pass a dict to the params kwarg of requests. This also applies to the following lines:

xurl = "/hubdb/api/v2/tables/" + str(table_id) + "/rows?portalId=" + str(PORTAL_ID)

url = HS_API_URL + xurl

The alternative is:

url = f'{HS_API_URL}/hubdb/api/v2/tables/{table_id}/rows'

response = requests.get(url, params={'portalId': PORTAL_ID})

And for this line:

urlopen(url).read()

Why, if you have requests? Use requests instead.

table_data = json.loads(response)

For this, use requests; then you can write

table_data = response.json()

For this line:

"meta_description": str(quarter) + " ~ " + str(year) + " ~ " + str(date) + " | " + str(forecast) + " " + str(title) + " " +  str(name) + " | " + str(state) + ", " + str(country),

Stop doing so much string concatenation. Use f-string interpolation like I did above. The same applies to slug and others.

headers = { "content-type" : "application/json" }

You don't need to do that if you use requests and pass the json kwarg with your dict.

For this line:

main()

If someone else imports your file, you should give them the option of not running main. The standard way to do this is if __name__ == '__main__':