Sansan Data HubのChange Feed APIを利用して、人物情報を取得する #Python

Sansan Data Hubとは

概要

Sansan株式会社が提供する、顧客データの正規化・統合サービスです。
顧客情報を名寄せ・クレンジングしてくれるだけでなく、売上規模や従業員規模、上場/非上場などの会社情報を付与してリッチにしてくれます。
特に、BtoBでマーケティングに力を入れたい企業におすすめのサービスです。
詳しくは sansanのSansan Data Hubサイトをご確認ください。

Sansan Data Hubを利用するメリット

バラバラな顧客データを自動で正規化・クレンジングしてくれます。
Sansan独自DBや帝国データバンクなどのソースをもとに、顧客情報を自動で最新に更新してくれます。
売上規模や従業員規模、資本金、業種など、マーケティング活動に有用な情報を付与してくれます。
Sansan Data Hubからデータを取得する「Change Feed API」、Sansan Data Hubにデータを取り込む「File Import API」の存在によって、CRMとのシームレスな連携を実現できます。

今回の記事内容

今回は、Sansan Data Hubに蓄積・統合された顧客データについて、APIを利用して取得する方法をご紹介します。
Salesforceとの連携であれば比較的楽にできるようなのですが、それ以外のCRM（例えばkintoneやZOHOなど）と連携したい場合には今回のようなコードが必要になるかと思います。

Change Feed API 公式ドキュメント

【Sansan Data Hub】Change Feed API機能概要
 【Sansan Data Hub】パラメーター仕様書
 【Sansan Data Hub】Change Feed API対象項目一覧

実装

事前準備

Sansan Data Hubを利用するためには、通常のsansanの契約のみならず、Sansan Data Hubの契約が必要です（有料）。
契約後、Sansan担当者と連携して、初期設定を完了しておく必要があります。
初期設定時、Sansan担当者から共有される
- clientId
- clientSecret
- 「会社情報」のFeedId
- 「拠点情報」のFeedId
- 「人物情報」のFeedId　★今回のコードで、PERSON_FEEDIDとしています。
- 「名刺情報」のFeedId

をメモしておいてください。
ちなみに今回は、「人物情報」を取得するコードを書きました。

また、この記事では、Python 3.9.2を使っています。

実際のコード（全体）

main.py

import datetime as dt
import json
import requests


# 事前準備で用意しておきます
CLIENT_ID = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx' 
CLIENT_SECRET = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
PERSON_FEEDID = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'


# STEP1：アクセストークンを取得する関数をつくる
def get_access_token():
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
    }
    data = 'client_id=' + CLIENT_ID + '&client_secret=' + CLIENT_SECRET + '&grant_type=client_credentials'
    response = requests.post('https://account.datahub.sansan.com/connect/token', headers=headers, data=data)
    access_token = response.json()['access_token']
    return access_token

# STEP2：データを取得する関数をつくる
def get_data(feed_id, target_period, headers):
    responses = requests.get('https://api.datahub.sansan.com/change-feed/v1/' + feed_id + '/updated?' + target_period, headers=headers)
    response_list = responses.text.splitlines()
    updated_records = []
    for response in response_list:
        record = json.loads(response)
        record_sorted = dict(sorted(record.items()))
        updated_records.append(record_sorted)
    return updated_records

# STEP3：取得対象の期間を定めて、実際にデータを取得する
def run():
    # 期間のパラメーターをつくる
    today = dt.date.today()
    yesterday = today + dt.timedelta(days=-1)
    target_period = 'start=' + str(yesterday) + 'T00%3A00%3A00.0000000Z&end=' + str(today) + 'T00%3A00%3A00.0000000Z'

    # アクセストークンを取得し、headersを定義する
    access_token = get_access_token()
    headers = {
        'Accept': 'application/ldjson',
        'Authorization': 'Bearer ' + access_token,
    }

    # 実際にChange Feed APIをたたく
    try:
        person_updated_records = get_data(PERSON_FEEDID, target_period, headers)
        # logging.infoなどでログを吐き出しておくと、何かと便利です。
        print(person_updated_records[0])
        return person_updated_records
    except:
        # logging.errorなどでログを吐き出しておくと、何かと便利です。
        return


if __name__ == '__main__':
     run()

返ってくるデータのイメージ（一つのレコード）

{
  '_id': 'xxxxxxxxxxx',
  '_ts': 000000000,
  'businessLocation.address': '東京都〇〇区〇〇1-1-1',
  'businessLocation.address_building': '',
  'businessLocation.address_city': '〇〇区',
  'businessLocation.address_prefecture': '東京都',
  'businessLocation.address_street': '〇〇区〇〇1-1-1',
  'nta.addressInside': '東京都〇〇区〇〇1-1-1',
  'nta.addressInside_postalCode': '1500031',
  'nta.address_city': '〇〇区',
  'nta.address_prefecture': '東京都',
  'nta.address_street': '〇〇1-1-1',
  'nta.corporateName': '株式会社サンプル商事',
  'organization.address_city': '〇〇区',
  'organization.address_prefecture': '東京都',
  'organization.address_street': '〇〇1-1-1',
  'organization.fax': '03-xxxx-xxxx',
  'organization.name': '株式会社サンプル商事',
  'organization.phone': '03-xxxx-xxxx',
  'person.address_city': '〇〇区',
  'person.address_prefecture': '東京都',
  'person.address_street': '〇〇町1-1-1',
  'person.email': 'test_taro@testshoji.co.jp',
  'person.firstName': '太郎',
  'person.lastName': 'テスト',
  'person.mobilePhone': '080-xxxx-xxxx',
  'person.phone': '03-xxxx-xxxx',
  'person.positionRank': 6,
  'personId': 'xxxxxxxxxxxxxxxxxxxxxx',
  'slc': 'xxxxxxxxxxxxx',
  'soc': 'xxxxxxxxxxxxx',
  'tdb.address': '東京都〇〇区〇〇1-1-1',
  'tdb.address_city': '〇〇区',
  'tdb.address_prefecture': '東京都',
  'tdb.address_street': '〇〇町1-1-1',
  'tdb.employeeNumberRange_ge': 10,
  'tdb.employeeNumberRange_lt': 30,
  'tdb.establishedIn': '2023/2',
  'tdb.foundedIn': '2023/1',
  'tdb.latestSalesAccountingTermSalesRange_ge': 0,
  'tdb.latestSalesAccountingTermSalesRange_lt': 50,
  'tdb.legalCapitalRange_ge': 10000,
  'tdb.legalCapitalRange_lt': 50000,
  'tdb.mainIndustrialClassName': '〇〇業',
  'tdb.phone': '03-xxxx-xxxx',
  'tdb.representativeName': 'サンプル花子',
  'tdb.representativeTitle': '代表取締役',
  'tdb.subIndustrialClassName': '〇〇業',
  ・・・
}

※nta：ソース元が国税庁の情報
※tdb：ソース元が帝国データバンクの情報

ポイント

1. 期間は開始日時と終了日時の両方を指定する

リクエストURLのパラメーターに

/updated?start={開始日時}&end={終了日時}

で指定できます。
Sansan Data Hubの顧客情報はリッチ化される分、データ量としては結構なものになります。
適切な期間を設定しないと、データ量が大きすぎてタイムエラーなどが起きる可能性があります。

2. 返却されたデータが改行されているので対処する

Change Feed APIをたたいて返ってきたデータは、レコードごとに改行されています。
改行文字で分割してリスト化してくれる splitlines() で対処します。

response_list = responses.text.splitlines()

また、JSON形式の文字列をpythonで辞書型として扱うために、json.loads() を使っています。

record = json.loads(response)

3. 返ってきたデータをkeyでソートしておく

sorted()を使ってkeyでソートしておくと、後々見やすくなります。
sorted()のみだとタプルになるので、dict()で辞書型に戻しています。

record_sorted = dict(sorted(record.items()))

感想・補足

公式ドキュメントが充実しているのでありがたいです。
logを蓄積しておくと、後々検証するのに便利です。
awsならLambdaに載せるなどすれば、毎日自動で更新してくれます。