Hubspot のカスタムコードを使って「営業日」の計算をする

この記事を読んで分かること

Hubspot のプロパティを元にN営業日後（前）を計算する方法がわかる。Hubspot のカスタムコードからはライブラリを使うことができないため、営業日を自分で計算する必要がある。計算にはこの3つのプロパティを使う

• 基準日
• N営業日
• 基準日 + N営業日を計算した結果を入れるプロパティ

前準備

APIキーを取得する

1. 設定 > 連携 > 非公開アプリ のページから「非公開アプリを作成」する
2. スコープを指定して権限を付与し、「アプリを作成」する
3. トークンをコピーする

プロパティを作る

今回は仮の名前で、3つのプロパティを作成する

ワークフローからカスタムコードを作成し、プロパティを読み込む

1. ワークフローを「ゼロから作成」する
2. コンタクトベースの空白のワークフローを作成する
3. ワークフローのトリガーを設定する
4. APIKEY という名前でシークレットを作成する
5. 言語を Python に変更する
6. プロパティとして以下の3つを取得して、Python から使える状態にする
   • base_date : 基準日
   • n_days : N営業日
   • hs_object_id : Record ID

import os
from hubspot import HubSpot

def main(event):
  # Use inputs to get data from any action in your workflow and use it in your code instead of having to use the HubSpot API.
  hubspot = HubSpot(access_token=os.getenv('APIKEY'))
  base_date = event["inputFields"]["base_date"]
  n_days = event["inputFields"]["n_days"]
  hs_object_id = event["inputFields"]["hs_object_id"]
  # Return the output data that can be used in later actions in your workflow.
  return {
    "outputFields": {
    }
  }

土日を考慮して営業日を計算する

土日を計算するには date.weekday() を使う。土曜日が 5 , 日曜日が 6 なので、以下のような処理を追加する。ここでは基準日から始めて、N営業日分を1日ずつ、それが土日ではないことを確認する。

from datetime import datetime, timedelta

# n_days が営業日カウント
# business_date に計算中のN営業日後の日付を保存する
def calc_buisiness_date(base_date, n_days):
  base_date = datetime.fromtimestamp(int(base_date) / 1000) # Hubspot では日付をミリ秒単位の Unixtime で保存しているため、1000で割り算した上で通常の日付に変換する
  business_date = base_date
  number = 1 # N営業日「前」を計算するときは1日ずつ前の日付を確認する
  if n_days < 0:
    number = -1
  is_holiday = False
  while n_days != 0 or is_holiday:
    if not is_holiday:
      n_days = n_days - number
    business_date = business_date + timedelta(days = number)
    is_holiday = business_date.weekday() in (5, 6)
  return business_date.strftime('%Y-%m-%d')

def main(event):
  ...
  print(base_date, n_days, "計算結果: ", calc_buisiness_date(base_date, int(n_days)))

動作確認

カスタムコード下の「アクションをテスト」からテスト用のコンタクトを指定して、「テスト」する。ここでは基準日が2023年3月15日で、その10営業日後を計算している。まだ祝日を考慮していないため、ここでは春分の日を無視して29日になる。

祝日を取得する

ここまでの処理で土日を考慮することができたため、次は「祝日」を考慮して営業日を計算できるようにする。営業日を計算するには、まず 政府が公開している祝日情報 を取得する。

from requests import get
from csv import reader

def get_holiday():
  response = get("https://www8.cao.go.jp/chosei/shukujitsu/syukujitsu.csv").content.decode('shift-jis')
  holiday = []
  for row in list(csv.reader(response.splitlines(), delimiter=','))[1:]:
    holiday.append(datetime.strptime(row[0], '%Y/%m/%d'))
  return holiday

def calc_buisiness_date(base_date, n_days):
  holiday = get_holiday()
  ...

動作確認

CSV に含まれる祝日は非常に多いため、結果はすべては表示されない。WARNING になるが問題ない

祝日を考慮して営業日を計算する

土日を考慮して営業日を計算する処理の中で、if 文の中身を変更する。上の処理で祝日一覧を取得できているため、「土日か祝日じゃなければ営業日カウントを減らす」ようにする

def calc_buisiness_date(base_date, n_days):
  ...
  while n_days != 0 or is_holiday:
    if not is_holiday:
      n_days = n_days - number
    business_date = business_date + timedelta(days = number)
    is_holiday = business_date.weekday() in (5, 6) or base_date in holiday # ここに祝日を考慮する処理を追加する
  ...

動作確認

最後に動作確認をする。祝日を考慮していないと29日だったのが、30日になったことが分かる

結果をコンタクトに保存する

最後に、計算結果をプロパティに保存する。ここでは Hubspot の API を呼び出してコンタクトを Update する

from hubspot.crm.objects import SimplePublicObjectInput, ApiException

def update(hubspot, object_id, business_date):
  simple_public_object_input = SimplePublicObjectInput(properties={
    "business_date": business_date
  })
  api_response = hubspot.crm.objects.basic_api.update(object_type="contacts", object_id=object_id, simple_public_object_input=simple_public_object_input)
  return api_response

def main(event):
  ...
  business_date = calc_buisiness_date(base_date, int(n_days))
  # Return the output data that can be used in later actions in your workflow.
  try:
    update(hubspot, hs_object_id, business_date)
  except Exception as e:
    print(e)
    raise
  ...

動作確認

テストを実行して、テストに使ったコンタクトを表示すると、このように 基準日 + N営業日 プロパティが更新されていることがわかる

完成形

import os
from hubspot import HubSpot
from datetime import datetime, timedelta
from requests import get
from csv import reader
from hubspot.crm.objects import SimplePublicObjectInput, ApiException

def get_holiday():
  response = get("https://www8.cao.go.jp/chosei/shukujitsu/syukujitsu.csv").content.decode('shift-jis')
  holiday = []
  for row in list(reader(response.splitlines(), delimiter=','))[1:]:
    holiday.append(datetime.strptime(row[0], '%Y/%m/%d'))
  return holiday

def calc_buisiness_date(base_date, n_days):
  holiday = get_holiday()
  base_date = datetime.fromtimestamp(int(base_date) / 1000)
  business_date = base_date
  number = 1
  if n_days < 0:
    number = -1
  is_holiday = False
  while n_days != 0 or is_holiday:
    if not is_holiday:
      n_days = n_days - number
    business_date = business_date + timedelta(days = number)
    is_holiday = business_date.weekday() in (5, 6) or base_date in holiday
  return business_date.strftime('%Y-%m-%d')

def update(hubspot, object_id, business_date):
  simple_public_object_input = SimplePublicObjectInput(properties={
    "business_date": business_date
  })
  api_response = hubspot.crm.objects.basic_api.update(object_type="contacts", object_id=object_id, simple_public_object_input=simple_public_object_input)
  return api_response

def main(event):
  # Use inputs to get data from any action in your workflow and use it in your code instead of having to use the HubSpot API.
  hubspot = HubSpot(access_token=os.getenv('APIKEY'))
  base_date = event["inputFields"]["base_date"]
  n_days = event["inputFields"]["n_days"]
  hs_object_id = event["inputFields"]["hs_object_id"]
  business_date = calc_buisiness_date(base_date, int(n_days))
  # Return the output data that can be used in later actions in your workflow.
  try:
    update(hubspot, hs_object_id, business_date)
  except Exception as e:
    print(e)
    raise
  return {
    "outputFields": {
    }
  }