4
\$\begingroup\$

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 = XXXXXXX

# 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....

\$\endgroup\$
1
  • \$\begingroup\$ @200_success I have added the entire program to the post. \$\endgroup\$
    – FlySpaceAge
    Commented Dec 11, 2018 at 18:02

1 Answer 1

Reset to default
2
\$\begingroup\$

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__':

\$\endgroup\$
2
  • \$\begingroup\$ Thanks for hitting on these points, in particular using f string interpolation. Is this for both readability and speed? \$\endgroup\$
    – FlySpaceAge
    Commented Dec 17, 2018 at 18:34
  • \$\begingroup\$ Speed is dubious; that would have to be tested. It's more for legibility and ease of maintenance. \$\endgroup\$
    – Reinderien
    Commented Dec 17, 2018 at 19:27

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.