[PYTHON] I tried to make a system to automatically acquire the program guide → register it in the calendar in one day
Overview
I wanted to post a story about what I made, and I wanted to touch serverless, so I tried to create a system that acquires the program guide of my favorite program ** Mitsuaki Iwago's World Cat Walk ** that I record every time and automatically registers it in the calendar (timetree).
What was used
-NHK Program Guide API-> Get Program Guide
- Google Cloud Function -cron-job.org-> Free subscription service
- timetree API
There is an NHK program guide API, so get it once a day, filter the data, and register it with the calendar API. There is a free service called cron-job.org for regular execution, and I used it because there is one that can send a request to the specified URL at the specified time.
This time, I wanted to complete it in one day, so I used Google Cloud Function, which allows me to write the most familiar Python.
Create
Followed by something that isn't a lot of code. I'm sorry.
At first, I was thinking about sending an email by detecting an error, but when I was researching Google Cloud Function, I noticed it at night ... Zuckerberg in my heart said that word, so I created it for the time being. did.
Program guide acquisition
At first, there are traces of trying to classify.
# 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 General g1
#BS premium 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:
"""Determine if the program guide data includes cat walking"""
is_nekoaruki = False
try:
title = program["title"]
if "Cat walking" 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 #Get for 6 days
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
Add calendar
Here too, I just added the calendar ID acquisition-> compare with registered data->. A storm of compromise.
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:
#I only use one calendar, so the first calendar is fine
if calendar["attributes"]["order"] == 0:
return calendar
else:
pass
else:
return response.text
def check_upcoming_events(calendar_id: str):
"""Get 7 days worth of registered events"""
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):
"""Send event to API and add"""
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 by comparing registered events with acquired data"""
add_events: list = []
title_ls = [i["title"] for i in registered]
for data in data_ls:
#Skip if title is registered
#I can't detect that the broadcast time has changed since I registered, but no
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]):
"""Extract only cat walking events from registered data"""
filtered_registered_events = filter(lambda x: "Cat walking" in x["attributes"]["title"], data_ls)
extracted: list = []
#At first I was going to update when the start time changed
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 settings
Create project-> Create Function-> Paste code-> Set environment variables.
Don't forget to change the time zone by putting TZ = Asia / Tokyo in the environment variable in addition to the API Key.
Added requests == 2.22.0 to requirements.txt.
Scheduled execution settings
Simply create an account with cron-job.org and set up a job from Cronjobs-> Create cronjob.
You can also set the data to be posted.
Summary
It's embarrassing to post code that just lists the functions ... However, I think that one of the good things about hobby products is that you can make them just by moving. It's nice to be able to do it all for free.
It was also a good harvest to find that it was easy to create a serverless environment.
With this, you will never forget the broadcast time of walking cats. ** Mitsuaki Iwago's World Cat Walk ** Let's see.