筆者について

name → Koki Teramoto
age → 18
language → ja_JP, en_US

本編

やりたいことは、題名の通りです。参考までに、データベースのイメージを下に記しておきます。

ということで、順番にやっていきましょう。

変数の設定

main.py

API_TOKEN_QIITA = "YOUR QIITA TOKEN"
API_TOKEN_NOTION ="YOUR NOTION TOKEN"

API_ID_DB = "YOUR DATABASE ID"

API_URL_ITEMS = "https://qiita.com/api/v2/authenticated_user/items"
API_URL_USER = "https://qiita.com/api/v2/users/{USERNAME}"
API_URL_DB_POST = "https://api.notion.com/v1/pages"
API_URL_DB_QUERY = f"https://api.notion.com/v1/databases/{API_ID_DB}/query"

API_HEADERS_QIITA = {
    "Authorization": f"Bearer {API_TOKEN_QIITA}"
}

API_HEADERS_NOTION = {
    "Authorization": f'Bearer {API_TOKEN_NOTION}',
    "Notion-Version": '2022-06-28'
}

上記の変数を設定しています。

Qiitaから記事一覧を取得する

ということで、まずはQiitaから記事の一覧を取得しましょう。

まず、Qiitaの記事情報を全て取得するAPIのリクエスト先は以下の通りです。

GET:https://qiita.com/api/v2/authenticated_user/items

ただし、これには一つ問題があります。それは、一回のリクエストで取得する記事の数に限界があるということです。ということで、記事の総数を取得する必要があります。そのためには、下のURLにリクエストを投げる必要があります。

GET:https://qiita.com/api/v2/users/{USERNAME

また、リクエストの数について、リクエストの数は一回のリクエストで取得する記事の数で記事の総数を割った時の商+1回やる必要があります。これはあまりの分のリクエストがあるからですね。また、古い記事から順番にIDをつけています。

main.py

res_user = requests.get(API_URL_USER)
num_page = 10
num_count = res_user.json()["items_count"]
num_items =  num_count // num_page
list_allitem = []
list_itemid = []

for i in range(num_items + 1):
    res_articles = requests.get(f'{API_URL_ITEMS}?page={i + 1}&per_page={num_page}', headers=API_HEADERS_QIITA)
    for l, item in enumerate(res_articles.json()):
        data = {
            "id": num_count - ((num_page * i) + l),
            "title":item["title"],
            "url":item["url"],
            "date":item["created_at"]
        }
        list_allitem.append(data)

list_allitem = sorted(list_allitem, key=lambda x : x["id"])

ちなみに、一番下のソートの指揮は必要ありませんが、ログを出したいときは見やすいと思いました。

Notionに一覧をポストする

あとはNotionにポストするだけですが、取得したたびにそのデータをアップロードすると更新ではなく新しい記事として投稿されます。ので、すでにデータベースにすでに追加されている記事を除外する必要があります。ので、まずはデータベースから記事のID一覧を取得します。ちなみにリクエスト先は以下の通りです。

GET https://api.notion.com/v1/databases/{API_ID_DB}/query

また取得したのはあくまで辞書の配列なのでここからIDを取得した配列を作ります。

main.py

for l in requests.post(API_URL_DB_QUERY, headers=API_HEADERS_NOTION).json()["results"]:
    list_itemid.append(l['properties']['ID']["number"])

これができたらあとは除外するプログラムを書くだけです。悩んだのですが、インデックスを検索してもしもそこでエラーが出たらポストする例外処理をするプログラムを作りました。ポスト先のURLは以下のとおりです。

POST https://api.notion.com/v1/pages

あとログに出すときにポストが成功したかについて出力しています。確認はres_db.okを使っています。

main.py

for db_new in list_allitem:
    try:
        list_itemid.index(db_new["id"])
        print(f'{db_new["id"]}:Skipped (Already Exists)')
    except:
        data_db = {
          "parent": { "database_id": API_ID_DB },
          "properties": {
                  "ID": {
                      "number":db_new["id"]
                  },
                  "記事タイトル": {
                      "title": [{"type":"text","text": {"content":db_new["title"]}}]
                  },
                  "記事URL": {
                      "url": db_new["url"]
                  },
                  "作成日": {
                      "date": {
                              "start": db_new["date"]
                            }
                  }
              }
        }
        res_db = requests.post(API_URL_DB_POST, json=data_db, headers=API_HEADERS_NOTION)
        if(res_db.ok):
            print(f'{db_new["id"]}:Posted (Succeed))')
        else:
            print(f'{db_new["id"]}:Skipped (Faild)')

最後に

今回は久しぶりにAPIたるものを触りました。いや〜、やっぱ定期的に触らないとダメですね。今後もNotion系のAPIは触っていきたいなと思っています。