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

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.

スクリーンショット 2020-02-14 21.20.28.png

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.

スクリーンショット 2020-02-14 21.45.34.png

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.

Recommended Posts

I tried to make a system to automatically acquire the program guide → register it in the calendar in one day
I tried to make a stopwatch using tkinter in python
I tried to make a system that fetches only deleted tweets
A memorandum when I tried to get it automatically with selenium
I tried to register a station on the IoT platform "Rimotte"
I want to pass the G test in one month Day 1
I made a command to display a colorful calendar in the terminal
I tried to make a site that makes it easy to see the update information of Azure
I tried to make it possible to automatically send an email just by double-clicking the [Python] icon
I tried to find out the difference between A + = B and A = A + B in Python, so make a note
[Python] I tried to make a simple program that works on the command line using argparse.
I tried to make a Web API
I tried to make a calculator with Tkinter so I will write it
I made a program to check the size of a file in Python
I tried to display the altitude value of DTM in a graph
I made a POST script to create an issue on Github and register it in the Project
I tried to make it possible to automatically send an email just by double-clicking the [GAS / Python] icon
Make a Python program a daemon and run it automatically when the OS starts
When I tried to make a VPC with AWS CDK but couldn't make it
In IPython, when I tried to see the value, it was a generator, so I came up with it when I was frustrated.
I made a system that automatically decides whether to run tomorrow with Python and adds it to Google Calendar.
A machine learning beginner tried to create a sheltie judgment AI in one day
I wanted to know the number of lines in multiple files, so I tried to get it with a command
I tried to make a ○ ✕ game using TensorFlow
[LPIC 101] I tried to summarize the command options that are easy to make a mistake
I tried to make something like a chatbot with the Seq2Seq model of TensorFlow
Try to make a blackjack strategy by reinforcement learning (② Register the environment in gym)
I tried to make the phone ring when it was posted at the IoT post
A python beginner tried to intern at an IT company [Day 3: Going to the clouds ...]
A Python beginner made a chat bot, so I tried to summarize how to make it.
I can't sleep until I build a server !! (Introduction to Python server made in one day)
I tried to make it easy to change the setting of authenticated Proxy on Jupyter
I tried to make a "fucking big literary converter"
I tried to graph the packages installed in Python
I tried to implement a pseudo pachislot in Python
I tried to implement a recommendation system (content-based filtering)
I tried to automatically generate a password with Python3
It's getting cold, so I tried to make it possible to turn on / off the AC heater automatically with Raspberry Pi!
When I tried to install PIL and matplotlib in a virtualenv environment, I was addicted to it.
[Git] I tried to make it easier to understand how to use git stash using a concrete example
A program that automatically resizes the iOS app icon to the required image size in Python
I tried to automate "one heart even if separated" using a genetic algorithm in Python
I want to see the graph in 3D! I can make such a dream come true.
I tried to create a Python script to get the value of a cell in Microsoft Excel
I also tried to imitate the function monad and State monad with a generator in Python
I tried to make a bot that randomly acquires Wikipedia articles and tweets once a day
I wrote a doctest in "I tried to simulate the probability of a bingo game with Python"
A memorandum to register the library written in Hy in PyPI
I tried the super-resolution algorithm "PULSE" in a Windows environment
I tried to implement a one-dimensional cellular automaton in Python
I tried to automatically create a report with Markov chain
Put the lists together in pandas to make a DataFrame
I tried "a program that removes duplicate statements in Python"
I tried to summarize the code often used in Pandas
I tried "How to get a method decorated in Python"
I tried to illustrate the time and time in C language
I wrote it in Go to understand the SOLID principle
I tried to summarize the commands often used in business
I tried to implement the mail sending function in Python
I want to make the Dictionary type in the List unique
I want to make input () a nice complement in python