2021年10月に日本語化されたことを機にNotionというアプリ使い始めました。さまざまな利用用途の中で、タスク管理を目的使い始めた際に、繰り返しタスクを登録する機能がNotionにはなく、非常に不便だと感じました。今回は、Notionをより上手に活用するために、繰り返しタスクを登録するプログラムを作成したので、その紹介をしたいと思います。
記事のポイント
- Notionでタスク管理を始めたが、繰り返しタスクを毎回作るのが大変
- NotionAPIを使って繰り返しタスクを自動作成
- Notionに不満を感じた繰り返しタスク登録機能がない点
- 繰り返しタスク自動登録のプログラムの説明
- 処理概要
- 事前準備① - 繰り返しタスクページとタスク管理ページの準備
- 事前準備② - インテグレーショントークンの発行とページの共有
- 事前準備③ - Pythonのパッケージ追加
- 事前準備④ - データベースIDの確認
- プログラム紹介
- プログラム全体
- プログラム詳細-1：インテグレーショントークンやデータベースIDの設定とリクエスト情報
- プログラム詳細-2：各ページのデータベース項目・プロパティ値をリスト化
- プログラム詳細-3：繰り返しタスクページから情報を取得
- プログラム詳細-4：取得情報から各レコードのIDを取得
- プログラム詳細-5：取得情報をpandasデータフレーム構造にする
- プログラム詳細-6：レコード毎に処理を実施
- プログラム詳細-7：プログラム実行日が次回処理日以降かどうか判定
- プログラム詳細-8：更新頻度と実施日の値でタスク予定開始日を設定します
- プログラム詳細-9：タスクの予定開始・終了日の値を設定
- プログラム詳細-10：タスク予定開始日が有効期間の終了日以前かを判定
- プログラム詳細-11：タスク管理ページへタスクを登録
- プログラム詳細-12：タスク管理ページへタスクを登録
- プログラム詳細-13：タスク管理ページへタスクを登録
- まとめ
Notionに不満を感じた繰り返しタスク登録機能がない点
プログラムの紹介を始める間に、私が不満を感じた背景や、プログラム作成前にしたことをまずは紹介いたします。
タスク管理の目的は効率的にタスクを消化すること
お仕事でも、日常でもタスク管理をされている方は多いと思うのですが、タスクには不定期に都度発生する通常タスクと 定期的な周期で行う、繰り返しのタスクがあります。
Notionでタスク管理をはじめると、まずこの繰り返しタスクの管理が非常に手間だと感じます。 その理由は通常タスクであっても、 繰り返しタスクであってもタスクの登録方法は同じであることです。 同じ周期で行う繰り返しタスクは繰り返す回数分、タスクの登録作業をしないといけません。 タスクの数が多いので管理を行うという人にとって、この回数分同じ作業を繰り返すというのは手間になります。 タスク管理はタスクを効率的に消化していくことが大きな目的で、管理作業自体に時間を費やしてしまうのは非常に時間がもったいないです。 Notionでタスク管理を始めて、私はまずこの手間が嫌になりました。 そこで、繰り返しタスクの登録を自動化できないかとまずは考えました。
繰り返しタスクの自動登録ができる外部サービス
繰り返しタスクを自動登録するにあたって、まずは他のサービスを利用できないかを調べました。 そこで私が試してみたのが以下3つのサービスでした。どれもNotionと連携できるサービスです。
- Zapier
- Automate.io
- IFTTT
| サービス名 | 作成可能ジョブ数 | 複雑な処理 | 実行回数（月） | データチェック間隔 |
|---|---|---|---|---|
| Zapier | 5 | Single処理のみ | 100 | 15分 |
| Automate.io | 5 | Single処理のみ | 300 | 5分 |
| IFTTT | 5 | Single処理のみ | 無制限 | 1時間以内 |
それぞれ、GUI操作で設定ができ、繰り返しタスクを作成することは可能なのですが『無料利用枠』という中で できる範囲には限界があり、残念ながら外部サービスを使って実現することを諦めました。 どのサービスに共通して言える点で、私が諦めた理由（『無料利用枠』の限界）になったポイントは以下になります。
- 登録できるジョブの数があまり多くない
- 複雑な処理パターンには対応できない
他にも3つのサービスを組み合わせて使うことも考えましたが、管理の煩雑さが気になり組み合わせることもやめました。
繰り返しタスク自動登録のプログラムの説明
外部サービス利用も厳しいため、今回は自分でNotionAPIを使って繰り返しタスクの登録をするプログラムを作成することにしました。
処理概要
以下の画面キャプチャのように繰り返しタスクというページに設定した繰り返しタスクを、
Pythonを使ってタスク管理ページに都度反映をするというしくみになります。詳細は後述します。
事前準備① - 繰り返しタスクページとタスク管理ページの準備
処理概要で記載したとおり、Notion側で画面を2つ用意します。
繰り返しタスクページの画面項目
繰り返しタスクのページでは以下表の項目名のものを用意しています。 Activateや有効期間を見て、タスク管理ページに反映するかどうかを確認します。 次回処理日でpython実行日が処理対象日かどうかを判定します。 その後、タスク名や実施日など他の項目を参照しタスク管理ページにデータを反映します。
| 項目名 | プロパティ | 用途 |
|---|---|---|
| Activate | チェックボックス | 繰り返しタスクの有効・無効の制御（チェック有が有効） |
| 有効期間 | 日付 | 繰り返しタスクの有効期間 |
| 更新頻度 | セレクト | タスクの頻度 |
| 次回処理日 | 日付 | 次回繰り返しタスクを登録する日 |
| タスクプロジェクト | セレクト | タスクが所属するプロジェクト |
| タスク名 | タイトル | タスク |
| タスクカテゴリ | セレクト | タスクのカテゴリ |
| 実施日 | セレクト | 曜日で記載 |
| タスク開始予定時間 | セレクト | タスクの開始時間 |
| タスク終了予定時間 | セレクト | タスクの終了時間 |
| タスク予定工数 | テキスト | タスクの予定工数 |
| 最終更新日時 | 最終更新日時 | データベース要素の最終更新日時 |
タスク管理ページの画面項目
タスク管理画面では以下表の項目名のものを用意しています。 今回のプログラムでは反映欄で★☆になっている項目を登録対象としています。 その中で、★は繰り返しタスクページの内容を登録します。 ☆はプログラムの中で設定した固定の値を登録しています。
| 反映 | 項目名 | プロパティ | 用途 |
|---|---|---|---|
| ★ | プロジェクト | リレーション | タスクが所属するプロジェクト |
| ★ | タスク名 | タイトル | タスク名 |
| ★ | カテゴリ | セレクト | タスクのカテゴリ |
| ☆ | 優先度 | セレクト | 優先度 |
| ★ | 予定開始-終了日 | 日付 | タスクの予定開始日と予定終了日 |
| ★ | 予定工数(H) | 数値 | タスクに費やす予定の時間 |
| - | 実績開始日 | 日付 | タスクの実施開始日 |
| - | 実績終了日 | 日付 | タスクの実施完了日 |
| - | 実績工数(H) | 数値 | タスクに費やした実績の時間 |
| ☆ | 進捗 | 数値 | 0～100%までの間でのタスク状況を管理 |
| ☆ | ステータス | セレクト | 未着手なのか、完了なのか、保留なのかステータスを管理 |
事前準備② - インテグレーショントークンの発行とページの共有
NotionAPIを使うためにはインテグレーショントークンを発行する必要があります。 そして作成したインテグレーショントークンを各画面に対して共有することでAPI操作が可能になります。
インテグレーショントークンの発行手順
https://www.notion.so/my-integrations 上記のURLからインテグレーショントークン管理画面を表示します。
『新しいインテグレーション』をクリックしてインテグレーション登録画面に移ります。
基本情報画面で各種情報を入力をし、『送信』をクリックします。
作成されたインテグレーショントークンの画面で登録内容が確認できます。
この画面で確認していただきたいのはインテグレーションの種類です。
内部かパブリックか選択が可能ですが、特別な要件がなければ内部にしておく方が良いでしょう。
また、トークンは*******となっていますが、表示をクリックすることで内容を確認できます。 「secret_」で始まっているものがインテグレーショントークンになります。
ページの共有方法
続いて発行したインテグレーショントークンに対してページ共有を行います。
画面右上の『共有』をクリック、続いて『招待』をクリックします。
共有するインテグレーションを選択し、招待をクリックします。
画面のように招待したインテグレーションが表示されれば、共有完了です。
今回私の方で共有したページは以下3つのページです。
- 繰り返しタスクページ
- タスク管理ページ
- プロジェクト管理ページ(※)
タスクに紐付けるプロジェクトが今回はあるため、以下のようなページを用意して共有しています。
事前準備③ - Pythonのパッケージ追加
今回はpythonというプログラムを使ってAPIを操作します。pythonのインストール方法や プログラムの実行方法については紹介はしませんので、ご了承ください。 ただ、プログラムを動かすにあたって私の環境や事前にインストールしたパッケージを紹介いたします。
今回私の環境下のpythonは以下のバージョンです。
- Python 3.9.0
また、追加でpandasをインストールしています。
pip install pandas
事前準備④ - データベースIDの確認
プログラムを紹介する前に操作するページのデータベースIDを確認します。 データベースIDはブラウザのURLから確認ができます。 『https://www.notion.so/XXXXXXXXXXXXXXXXXXXXXXXX?v=yyyyyyyyyyyyyyyyyyyy』 画面共有したページをブラウザで表示すると上記のようなURLになっているかと思います。 データベースIDは上記URLの『XXXXXXXXXXXXXXXXXXXXXXXX』の部分にあたります。
プログラム紹介
いよいよ、プログラムの紹介です。 実行前にプログラムの以下の部分はご自身の環境に合わせて修正をしてください。
NOTION_ACCESS_TOKEN = 'secret_xxxxxxxxxxxxxxxxxxxxx' REGULAR_TASK_DB_ID = 'yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy' TASK_MANAGEMENT_DB_ID = 'zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz'
自身の環境に合わせて修正
- 『NOTION_ACCESS_TOKEN』・・・インテグレーショントークンを設定
- 『REGULAR_TASK_DB_ID』・・・繰り返しタスクページのデータベースIDを設定
- 『TASK_MANAGEMENT_DB_ID』・・・タスク管理ページのデータベースIDを設定
上記3点をご自身の環境に合わせて修正いただき、同じレイアウトのページを作っていただければプログラムは動くと思います。
プログラム全体
プログラムの全体は以下になります。
import requests import pandas as pd import datetime import re #プロパティ値の取得 def get_property_value(d): """ プロパティの値を取得する関数 """ property_type = d.get('type') if property_type in ['title']: return d[property_type][0]['plain_text'] if d[property_type] else pd.NA elif property_type in ['number', 'checkbox', 'url', 'email', 'phone_number', 'created_time', 'last_edited_time']: return d[property_type] elif property_type in ['select']: return d[property_type]['name'] elif property_type == 'multi_select': return [i.get('name') for i in d[property_type]] elif property_type == 'date': return {'start': d[property_type]['start'], 'end': d[property_type]['end']} elif property_type == 'people': return [i['id'] for i in d[property_type]] elif property_type == 'files': return [i['name'] for i in d[property_type]] elif property_type in ['created_by', 'last_edited_by']: return d[property_type]['id'] elif property_type == 'formula': return get_property_value(d[property_type]) elif property_type == 'relation': return d[property_type][0]['id'] if d[property_type] else pd.NA elif property_type == 'rollup': return get_property_value(d[property_type]['array'][0]) elif property_type in ['rich_text']: return d[property_type][0]['plain_text'] if d[property_type] else pd.NA else: return pd.NA #項目の存在チェック def validColumExists(data,list): statusMessage = "" if not list["frequency"] in data.columns : statusMessage = statusMessage + list["frequency"] + "が存在しません" + '\n' if not list["procday"] in data.columns : statusMessage = statusMessage + list["procday"] + "が存在しません" + '\n' if not list["planestimate"] in data.columns : statusMessage = statusMessage + list["planestimate"] + "が存在しません" + '\n' if not statusMessage == "": statusMessage = "タスク名:"+data[list["task"]]+"\n" + statusMessage return statusMessage #入力チェック&入力データの整合性チェック def validDataCheck(data,list,frequency): statusMessage = "" if pd.isnull(data[list["task"]]): statusMessage = "タスク名を入力してください\n" else: if pd.isnull(data[list["frequency"]]): statusMessage = statusMessage + list["frequency"]+"の値を入力してください\n" else: if (data[list["frequency"]] == frequency["week"]) or (data[list["frequency"]] == frequency["2week"]): if pd.isnull(data[list["procday"]]): statusMessage = statusMessage + list["frequency"]+"が"+data[list["frequency"]]+"の場合は"+list["procday"]+"の値を入力してください\n" if pd.isnull(data[list["planestimate"]]): statusMessage = statusMessage + list["planestimate"]+"の値を入力してください\n" m = None if not pd.isnull(data[list["plansdatetime"]]): m = re.match(r'^([01]?[0-9]|2[0-3]):([0-5][0-9])$', data[list["plansdatetime"]]) if m == None: statusMessage = statusMessage + list["plansdatetime"] +"("+data[list["plansdatetime"]]+")を24時間表記(HH:MM)にしてください\n" if not pd.isnull(data[list["planedatetime"]]): m = re.match(r'^([01]?[0-9]|2[0-3]):([0-5][0-9])$', data[list["planedatetime"]]) if m == None: statusMessage = statusMessage + list["planedatetime"] +"("+data[list["planedatetime"]]+")を24時間表記(HH:MM)にしてください\n" if not m == None: stime = datetime.datetime.strptime(data[list["plansdatetime"]]+":00","%H:%M:%S") etime = datetime.datetime.strptime(data[list["planedatetime"]]+":00","%H:%M:%S") if stime > etime: statusMessage = statusMessage + list["plansdatetime"] +"("+data[list["plansdatetime"]]+")と"+list["planedatetime"] +"("+data[list["planedatetime"]]+")"+"の前後関係を見直してください\n" if not statusMessage == "": statusMessage = "タスク名:"+data[list["task"]]+"\n" + statusMessage return statusMessage #空文字登録 def enptyData(df,row,item): returndata = None if item in df.columns : if not pd.isnull(row[item]): returndata = row[item] return returndata # 翌日を返す def get_nextday_target_date(date): # dateに加算 next_target_date = date + datetime.timedelta(days = 1) return next_target_date # 翌週の指定した曜日を返す def get_nextWeek_target_date(date, target_week): week = ['月','火','水','木','金','土','日'] # 曜日を数値型で取得 weekday = date.weekday() # dateから指定した曜日までの加算日数を計算 add_days = 7 - weekday + week.index(target_week) # dateに加算 next_target_date = date + datetime.timedelta(days = add_days) return next_target_date # 2週間後の指定した曜日の日付を返す def get_next2Week_target_date(date, target_week): week = ['月','火','水','木','金','土','日'] # 曜日を数値型で取得 weekday = date.weekday() # dateから指定した曜日までの加算日数を計算 add_days = 7 - weekday + week.index(target_week) # dateに加算 next_target_date = date + datetime.timedelta(days = add_days+7) return next_target_date # 登録データの作成(予定開始-終了日：時間有り) def createPageDate(databaseId,task_management_list,update_task_val,nextday,plantime): data = { "parent": { "database_id": databaseId } } properties = { task_management_list["task"]:{"title": [ {"text": {"content": update_task_val["task"] + "(" + nextday.strftime('%m/%d') + ")"}} ] }, task_management_list["priority"]: {"select": {"name":update_task_val["priority"]} }, task_management_list["planestimate"]:{ "number":update_task_val["planestimate"] }, task_management_list["status"]: {"select": {"name":update_task_val["status"]} }, task_management_list["progress"]:{ "number":update_task_val["progress"] }, } if not update_task_val["project"] == None: properties[task_management_list["project"]] = {"relation": [ {"id":update_task_val["project"]} ] } if not update_task_val["category"] == None: properties[task_management_list["category"]] = {"select": {"name":update_task_val["category"]} } if plantime: properties[task_management_list["plandate"]] = { "date":{ "time_zone":"Asia/Tokyo", "start":update_task_val["planStartDate"], "end":update_task_val["planEndDate"] } } else: properties[task_management_list["plandate"]] = { "date":{ "start":update_task_val["planStartDate"], "end":update_task_val["planEndDate"] } } data["properties"] = properties return data # 定期タスクの次回処理日を登録 def updateNextProcDate(regular_task_list,updateDate): data = { "properties": { regular_task_list["nextproc"]:{ "date":{ "start":updateDate.strftime('%Y-%m-%d'), "end": None } } } } return data # 定期タスクのActivateを登録 def updateEffect(regular_task_list,updateDate): data = { "properties": { regular_task_list["activate"]:{ "checkbox": updateDate } } } return data # NOTIONのTOKEN・URL・ヘッダー情報を設定 NOTION_ACCESS_TOKEN = 'secret_xxxxxxxxxxxxxxxxxxxxx' REGULAR_TASK_DB_ID = 'yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy' TASK_MANAGEMENT_DB_ID = 'zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz' r_url = f"https://api.notion.com/v1/databases/{REGULAR_TASK_DB_ID}/query" t_url = f"https://api.notion.com/v1/pages" headers = { 'Authorization': 'Bearer ' + NOTION_ACCESS_TOKEN, 'Notion-Version': '2021-05-13', 'Content-Type': 'application/json', } regular_task_list= { "activate":"Activate" ,"period":"有効期間" ,"frequency":"更新頻度" ,"nextproc":"次回処理日" ,"task":"タスク名" ,"project":"タスクプロジェクト" ,"category":"タスクカテゴリ" ,"procday":"タスク実施日" ,"plansdatetime":"タスク開始予定時間" ,"planedatetime":"タスク終了予定時間" ,"planestimate":"タスク予定工数" } task_management_list= { "task":"タスク名" ,"project":"プロジェクト" ,"category":"カテゴリ" ,"priority":"優先度" ,"plandate":"予定開始-終了日" ,"planestimate":"予定工数(H)" ,"status":"ステータス" ,"progress":"進捗" } update_task_val = { "task":"" ,"project":"" ,"category":"" ,"priority":"★★★" ,"planestimate":"" ,"planStartDate":"" ,"planEndDate":"" ,"status":"未着手" ,"progress":0 } frequency = { "day":"日" ,"week":"週" ,"2week":"2週間" } # 現在の日付を取得する。 today = datetime.date.today() # APIを実行して結果を取得 r = requests.post(r_url, headers=headers) data = r.json().get('results') content = [] # 取得したデータからpropertiesとIDだけを取得 for i in data: idlist = {'No':{'type': 'rich_text','rich_text':[{'plain_text': i['id'],'type':'text'}]}} idlist.update(i['properties']) content.append(idlist) # 取得結果をpandaでDataFrame構造にする df = pd.DataFrame(content) # 欠損値を空の辞書に置換 df = df.mask(df.isnull(), {}) # 全てのフィールドに対して、関数を実行 df = df.applymap(get_property_value) # 取得項目の存在チェック error = validColumExists(df,regular_task_list) if error == "": #１行ずつ処理する for index, row in df.iterrows(): if row[regular_task_list["activate"]]: if regular_task_list["period"] in df.columns : # 有効期間の内容チェック if pd.isnull(row[regular_task_list["period"]]): sdate =today edate =today else : sdatetime = datetime.datetime.strptime(row[regular_task_list["period"]]['start'],'%Y-%m-%d') sdate = datetime.date(sdatetime.year, sdatetime.month, sdatetime.day) if row[regular_task_list["period"]]["end"] is not None : edatetime = datetime.datetime.strptime(row[regular_task_list["period"]]["end"],'%Y-%m-%d') edate = datetime.date(edatetime.year, edatetime.month, edatetime.day) else : edate =today else : sdate =today edate =today # 処理有効期間内かどうかの判定(未入力は有効と判定) if sdate <= today and today <= edate: #次回実行日かどうかの判定 if regular_task_list["nextproc"] in df.columns : if pd.isnull(row[regular_task_list["nextproc"]]): tdate = today else: tdatetime = datetime.datetime.strptime(row[regular_task_list["nextproc"]]["start"],'%Y-%m-%d') tdate = datetime.date(tdatetime.year, tdatetime.month, tdatetime.day) else : tdate = today if tdate <= today: #エラーチェック error=validDataCheck(row,regular_task_list,frequency) if error == "": # 更新頻度が日次の場合 if row[regular_task_list["frequency"]] == frequency["day"]: next_day = get_nextday_target_date(tdate) elif row[regular_task_list["frequency"]] == frequency["week"]: next_day = get_nextWeek_target_date(tdate,row[regular_task_list["procday"]]) elif row[regular_task_list["frequency"]] == frequency["2week"]: next_day = get_next2Week_target_date(tdate,row[regular_task_list["procday"]]) else: next_day = edate + datetime.timedelta(days = 1) #タスク開始予定時間・タスク終了予定時間の存在・入力チェック sdateCheck = True #開始時間有 edateCheck = True #終了時間有 if regular_task_list["plansdatetime"] in df.columns : if pd.isnull(row[regular_task_list["plansdatetime"]]): sdateCheck = False else: sdateCheck = False if regular_task_list["planedatetime"] in df.columns : if pd.isnull(row[regular_task_list["planedatetime"]]): edateCheck = False else: edateCheck = False if sdateCheck: planStartDate = next_day.strftime('%Y-%m-%d')+ 'T'+ row[regular_task_list["plansdatetime"]] + ':00.000+09:00' planEndDate = next_day.strftime('%Y-%m-%d')+ 'T'+ row[regular_task_list["planedatetime"]] + ':00.000+09:00' else: planStartDate = next_day.strftime('%Y-%m-%d') planEndDate = next_day.strftime('%Y-%m-%d') if regular_task_list["period"] in df.columns : if pd.isnull(row[regular_task_list["period"]]): edate = next_day else: edate = next_day if next_day <= edate: project = enptyData(df,row,regular_task_list["project"]) category = enptyData(df,row,regular_task_list["category"]) #タスクの登録データを整形 update_task_val.update( task=row[regular_task_list["task"]] ,project=project ,category=category ,planStartDate=planStartDate ,planEndDate=planEndDate ,planestimate=row[regular_task_list["planestimate"]] ) #登録処理を実施 jsondata = createPageDate(TASK_MANAGEMENT_DB_ID,task_management_list,update_task_val,next_day,sdateCheck) r = requests.post(t_url, headers=headers,json=jsondata) if r.status_code ==200: jsondata = updateNextProcDate(regular_task_list,next_day) r = requests.patch(t_url+"/"+row["No"], headers=headers, json=jsondata) if r.status_code ==200: print("正常終了") else: print("定期タスクの次回処理日の更新失敗") print(str(r.status_code) + ":" + r.text) else: print("タスク管理画面への登録失敗") print(str(r.status_code) + ":" + r.text) else: jsondata = updateEffect(regular_task_list,False) r = requests.patch(t_url+"/"+row["No"], headers=headers, json=jsondata) if r.status_code ==200: print("正常終了") else: print("定期タスクのActivateの更新失敗") print(str(r.status_code) + ":" + r.text) else: print(error) else : print(error)
プログラム詳細-1：インテグレーショントークンやデータベースIDの設定とリクエスト情報
ここからはプログラムの詳細を説明します。
NOTION_ACCESS_TOKEN = 'secret_xxxxxxxxxxxxxxxxxxxxx' REGULAR_TASK_DB_ID = 'yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy' TASK_MANAGEMENT_DB_ID = 'zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz' r_url = f"https://api.notion.com/v1/databases/{REGULAR_TASK_DB_ID}/query" t_url = f"https://api.notion.com/v1/pages" headers = { 'Authorization': 'Bearer ' + NOTION_ACCESS_TOKEN, 'Notion-Version': '2021-05-13', 'Content-Type': 'application/json', }
NOTION_ACCESS_TOKEN・REGULAR_TASK_DB_ID・TASK_MANAGEMENT_DB_IDは上述しているので割愛します。 r_urlは繰り返しタスクからデータを取得するためのリクエストするURLを設定しています。 t_urlはデータ登録・更新する際にリクエストするURLを設定しています。 headersはリクエストする際のヘッダー情報になります。
プログラム詳細-2：各ページのデータベース項目・プロパティ値をリスト化
regular_task_list= {
"activate":"Activate"
,"period":"有効期間"
・
・
・
}
task_management_list= {
"task":"タスク名"
,"project":"プロジェクト"
・
・
・
}
update_task_val = {
"task":""
,"project":""
,"category":""
・
・
・
}
frequency = {
"day":"日"
,"week":"週"
,"2week":"2週間"
}
regular_task_listは繰り返しタスクページのデータベース項目をリスト化しています。 task_management_listはタスク管理ページの更新項目をリスト化しています。 update_task_valはタスク管理ページに登録項目の値を事前にリスト化しています。空文字の項目は後半の処理で上書きをしています。 frequencyは更新頻度のプロパティ値をリスト化しています。
プログラム詳細-3：繰り返しタスクページから情報を取得
r = requests.post(r_url, headers=headers)
data = r.json().get('results')
繰り返しタスクページからデータを取得します。response情報はjson形式で戻ってくるのですが その中のresult項目の値をここではdataという変数に格納しています。
プログラム詳細-4：取得情報から各レコードのIDを取得
for i in data: idlist = {'No':{'type': 'rich_text','rich_text':[{'plain_text': i['id'],'type':'text'}]}} idlist.update(i['properties']) content.append(idlist)
取得したデータの各レコードのIDを取得しています。 このIDは次回処理日やActivateを更新するために必要になります。
プログラム詳細-5：取得情報をpandasデータフレーム構造にする
df = pd.DataFrame(content)
df = df.mask(df.isnull(), {})
df = df.applymap(get_property_value)
error = validColumExists(df,regular_task_list)
if error == "":
ここで取得した情報をpandasのデータフレーム構造に格納し、欠損値対応をpandasのmask関数を使いクレンジングしています。 また、Notionデータベースの各要素によって値の入っている箇所が異なるので、 get_property_value関数でそれぞれの要素のプロパティに応じて値を取得しています。 最後にNotionからデータを取得した際に全てのレコードに値がなかった場合、そのカラム自体がレスポンスとして返ってきません。 そのため、必要なカラムが存在しているかを事前にvalidColumExists関数でチェックしています。
プログラム詳細-6：レコード毎に処理を実施
for index, row in df.iterrows(): if row[regular_task_list["activate"]]: if regular_task_list["period"] in df.columns : if pd.isnull(row[regular_task_list["period"]]): sdate =today edate =today else : sdatetime = datetime.datetime.strptime(row[regular_task_list["period"]]['start'],'%Y-%m-%d') sdate = datetime.date(sdatetime.year, sdatetime.month, sdatetime.day) if row[regular_task_list["period"]]["end"] is not None : edatetime = datetime.datetime.strptime(row[regular_task_list["period"]]["end"],'%Y-%m-%d') edate = datetime.date(edatetime.year, edatetime.month, edatetime.day) else : edate =today else : sdate =today edate =today if sdate <= today and today <= edate:
取得データを1行ずつ処理します。その際、Activateの項目にチェックが有るかどうかをしています。 チェックが入っていない場合、そのレコードは処理対象外として次の処理に移ります。 チェックが入っている場合は、有効期間の値を補正します。 有効期間欄に値がなかった場合、プログラム実行日は有効としています。 値がある場合は、その期間を見てプログラム実行日が有効期間内かどうかはチェックしています。
プログラム詳細-7：プログラム実行日が次回処理日以降かどうか判定
if regular_task_list["nextproc"] in df.columns : if pd.isnull(row[regular_task_list["nextproc"]]): tdate = today else: tdatetime = datetime.datetime.strptime(row[regular_task_list["nextproc"]]["start"],'%Y-%m-%d') tdate = datetime.date(tdatetime.year, tdatetime.month, tdatetime.day) else : tdate = today if tdate <= today:
次回処理日が空の場合はプログラム実行日を次回処理日として処理するようにします。 次回処理日が入力済の場合は、プログラム実行日が次回処理日以降かどうかを判定しています。
プログラム詳細-8：更新頻度と実施日の値でタスク予定開始日を設定します
error=validDataCheck(row,regular_task_list,frequency)
if error == "":
if row[regular_task_list["frequency"]] == frequency["day"]:
next_day = get_nextday_target_date(tdate)
elif row[regular_task_list["frequency"]] == frequency["week"]:
next_day = get_nextWeek_target_date(tdate,row[regular_task_list["procday"]])
elif row[regular_task_list["frequency"]] == frequency["2week"]:
next_day = get_next2Week_target_date(tdate,row[regular_task_list["procday"]])
else:
next_day = edate + datetime.timedelta(days = 1)
更新頻度の値をもとにタスク管理ページに反映する予定開始日を取得しています。 get_next***関数では翌日・もしくは実施日に設定した曜日の日付を取得しています
プログラム詳細-9：タスクの予定開始・終了日の値を設定
sdateCheck = True edateCheck = True if regular_task_list["plansdatetime"] in df.columns : if pd.isnull(row[regular_task_list["plansdatetime"]]): sdateCheck = False else: sdateCheck = False if regular_task_list["planedatetime"] in df.columns : if pd.isnull(row[regular_task_list["planedatetime"]]): edateCheck = False else: edateCheck = False if sdateCheck: planStartDate = next_day.strftime('%Y-%m-%d')+ 'T'+ row[regular_task_list["plansdatetime"]] + ':00Z' planEndDate = next_day.strftime('%Y-%m-%d')+ 'T'+ row[regular_task_list["planedatetime"]] + ':00Z' else: planStartDate = next_day.strftime('%Y-%m-%d') planEndDate = next_day.strftime('%Y-%m-%d')
タスクの予定開始日・終了日の値を設定しています。その際繰り返しタスクページに予定開始時間・終了時間にある場合は その時間をタスク管理ページに反映するように値を設定しています。 予定開始時間・終了時間に値がない場合は、日付までをタスク管理ページに反映するように値を設定しています。
プログラム詳細-10：タスク予定開始日が有効期間の終了日以前かを判定
if regular_task_list["period"] in df.columns : if pd.isnull(row[regular_task_list["period"]]): edate = next_day else: edate = next_day if next_day <= edate:
タスク予定開始日が繰り返しタスクページの有効期間の終了日以前かを判定し、 終了日以前であれば、後続の登録作業を行います。
プログラム詳細-11：タスク管理ページへタスクを登録
project = enptyData(df,row,regular_task_list["project"]) category = enptyData(df,row,regular_task_list["category"]) update_task_val.update( task=row[regular_task_list["task"]] ,project=project ,category=category ,planStartDate=planStartDate ,planEndDate=planEndDate ,planestimate=row[regular_task_list["planestimate"]] ) jsondata = createPageDate(TASK_MANAGEMENT_DB_ID,task_management_list,update_task_val,next_day,sdateCheck) r = requests.post(t_url, headers=headers,json=jsondata)
createPageDateでNotionにリクエストするデータの準備をを行っています。 r = requests.post(t_url, headers=headers,json=jsondata)でNotionにpostしてタスク管理ページにデータを登録しています。
プログラム詳細-12：タスク管理ページへタスクを登録
jsondata = updateNextProcDate(regular_task_list,next_day)
r = requests.patch(t_url+"/"+row["No"], headers=headers, json=jsondata)
繰り返しタスクページの次回処理日を反映しています。
プログラム詳細-13：タスク管理ページへタスクを登録
jsondata = updateEffect(regular_task_list,False) r = requests.patch(t_url+"/"+row["No"], headers=headers, json=jsondata)
有効期間を過ぎたタスクに関して、繰り返しタスクページのActivateを無効化しています。
まとめ
今回は私が作成したNotionAPIを使って定期タスクを登録するプログラムを紹介しました。 私はこのプログラムを自宅のNASを使って定期タスクとして実行をしていて 毎日自動で繰り返しタスクがタスク管理画面に反映しています。 おかげで通常タスクを整理することに時間をかけることができるようになりました。 今回の私の記事がNotion利用者のお役に立ち生産性のアップに繋がってくれていれば幸いです。
