概要
作ってみた話 系の投稿をしたかったのと、サーバーレスを触ってみたかったので、
毎回録画している好きな番組岩合光昭の世界ネコ歩きの番組表を取得し、カレンダー(timetree)に自動登録するシステムを作ってみました。
使用したもの
- NHK番組表API -> 番組表の取得
- Google Cloud Function
- cron-job.org -> 無料定期実行サービス
- timetree API
NHK番組表APIというものがあるので、それを1日1回取得し、データをフィルターしてカレンダーのAPIで登録するようにします。
定期実行にはcron-job.orgという無料サービスがあり、指定した時間に指定したURLにリクエストを送れるものがあるので、利用しました。
今回は、なんとなく1日で完成させたかったので、一番慣れているPythonを書けるGoogle Cloud Functionを利用しました。
作成
大したコードではないものが続きます。すみません。
最初はエラー検知でメール送信とか考えてたんですが、Google Cloud Functionについて調べていたら気がついたら夜・・・
心のザッカーバーグがあの言葉を言ってきたのでとりあえず作成しました。
番組表取得
最初はクラス分けてやろうとしてた痕跡が残ってます。
# default
import os
from typing import List
from datetime import datetime, timedelta
import json
# from pip
import requests
apikey = os.environ["NHK_API_KEY"]
class NHKApi():
area = "130"
genre = "1001"
base_url = "http://api.nhk.or.jp/v2/pg/genre"
# NHK総合 g1
# BS プレミアム s3
@classmethod
def url(cls, service: str, date_str: str) -> str:
url = f"{cls.base_url}/{cls.area}/{service}/{cls.genre}/{date_str}.json"
return url
def get_g1_data(date_str: str):
url = NHKApi.url("g1", date_str)
response = requests.get(url, params={"key": apikey})
if response.status_code == 200:
return response.json()
else:
return {}
def get_s3_data(date_str: str):
url = NHKApi.url("s3", date_str)
response = requests.get(url, params={"key": apikey})
if response.status_code == 200:
return response.json()
else:
return {}
def check_is_nekoaruki(service: str, program: dict) -> bool:
"""番組表データにネコ歩きが入っているかどうかを判定"""
is_nekoaruki = False
try:
title = program["title"]
if "ネコ歩き" in title:
is_nekoaruki = True
except KeyError:
print("data type is invalided")
return is_nekoaruki
def filter_nekoaruki(service: str, data: dict) -> List[dict]:
filtered_programs: list = []
if data and data.get("list"):
try:
programs = data["list"][service]
filtered_programs = [i for i in programs if check_is_nekoaruki(service, i)]
except KeyError:
print("data type is invalided")
return filtered_programs
def get_days() -> List[str]:
days_ls = []
dt_format = "%Y-%m-%d"
search_day = 6 # 6日分取得
current = datetime.now()
for i in range(search_day):
days_ls.append((current + timedelta(days=i)).strftime(dt_format))
return days_ls
def get_nekoaruki() -> List[dict]:
days = get_days()
programs: list = []
for day in days:
g1_data = filter_nekoaruki("g1", get_g1_data(day))
s3_data = filter_nekoaruki("s3", get_s3_data(day))
one_day_data = g1_data + s3_data
if one_day_data:
for data in one_day_data:
programs.append(data)
return programs
カレンダー追加
ここもカレンダーID取得 -> 登録済データと比較 -> 追加しているだけです。
妥協の嵐。
class TimeTreeAPI():
url = "https://timetreeapis.com"
api_key = os.environ["TIMETREE_API_KEY"]
headers = {'Authorization': f'Bearer {api_key}',
"Accept": "application/vnd.timetree.v1+json",
"Content-Type": "application/json"}
def get_calendar() -> str:
response = requests.get(TimeTreeAPI.url + "/calendars", headers=TimeTreeAPI.headers)
if response.status_code == 200:
data = response.json()
calendars = data["data"]
for calendar in calendars:
# 1つしかカレンダー使ってないので最初のカレンダーでいいや
if calendar["attributes"]["order"] == 0:
return calendar
else:
pass
else:
return response.text
def check_upcoming_events(calendar_id: str):
"""7日分の登録済イベントを取得"""
response = requests.get(TimeTreeAPI.url + f"/calendars/{calendar_id}/upcoming_events",
headers=TimeTreeAPI.headers,
params={"days": 7})
if response.status_code == 200:
data = response.json()
return data
else:
return None
def convert_to_timetree_style(data: dict, calendar_id: str):
timetree_dict = {
"data": {
"attributes": {
"title": data["title"],
"category": "schedule",
"all_day": False,
"start_at": data["start_time"],
"end_at": data["end_time"],
"description": data["title"] + "\n" + data["content"],
"location": data["service"]["name"],
"url": "https://www4.nhk.or.jp/nekoaruki/"
},
"relationships": {
"label": {
"data": {
"id": f"{calendar_id},1",
"type": "label"
}
}
}
}
}
return timetree_dict
def add_event(data: dict, calendar_id: str):
"""イベントをAPIに送信し追加する"""
json_data = json.dumps(data)
response = requests.post(TimeTreeAPI.url + f"/calendars/{calendar_id}/events",
headers=TimeTreeAPI.headers, data=json_data)
if response.status_code == 201:
return True
else:
return False
def convert_all(programs: dict, cal_id: str):
events: list = []
for program in programs:
events.append(convert_to_timetree_style(program, cal_id))
return events
def post_events(data_ls: List[dict], calendar_id: str, registered: List[dict]):
"""登録済イベントと取得したデータを比較して追加"""
add_events: list = []
title_ls = [i["title"] for i in registered]
for data in data_ls:
# タイトルが登録済ならスキップ
# 登録してから放送時間変わったの感知出来ないけどいいや
if data["data"]["attributes"]["title"] in title_ls:
pass
else:
add_events.append(data)
if add_events:
for event in add_events:
add_event(event, calendar_id)
def extract_registered_data(data_ls: List[dict]):
"""登録済のデータからネコ歩きイベントのみ抽出"""
filtered_registered_events = filter(lambda x: "ネコ歩き" in x["attributes"]["title"], data_ls)
extracted: list = []
# 最初は開始時間が変わったら更新するつもりだった
for program in filtered_registered_events:
extracted.append({"title": program["attributes"]["title"],
"start": program["attributes"]["start_at"]})
return extracted
def main(request):
if request.get_json()["hoge"] == "hoge":
# get programs
nekoaruki_programs = get_nekoaruki()
# get cal_id
cal_id = get_calendar()["id"]
# get upcoming events
registered_events = check_upcoming_events(cal_id)["data"]
# filter upcoming events
extracted = extract_registered_data(registered_events)
data_ls = convert_all(nekoaruki_programs, cal_id)
post_events(data_ls, cal_id, extracted)
return "success!"
else:
return "failed..."
Functionの設定
プロジェクトを作成 -> Function作成 -> コード貼り付け -> 環境変数等設定します。
環境変数にはAPI Keyの他にTZ=Asia/Tokyoを入れ、タイムゾーンを変更するのを忘れずに。
requirements.txtもrequests==2.22.0を追加。
定期実行の設定
cron-job.orgでアカウントを作成し、Cronjobs-> Create cronjobからjobを設定するだけです。
Postするデータも設定出来ます。
まとめ
ただ関数を並べただけのコードを投稿するのは恥ずかしいですね・・・
しかし、動けばいいで作れるのが趣味プロダクトのいいところでもあると思います。
すべて無料でできるのもいいですね。
サーバーレス環境も簡単に作成できることがわかったのも良い収穫でした。
これでネコ歩きの放送時間を忘れることはありません。
岩合光昭の世界ネコ歩き 見ましょう。