J'ai essayé de créer un système pour acquérir automatiquement le guide des programmes → l'enregistrer dans le calendrier en un jour

Aperçu

Je voulais publier une histoire sur ce que j'ai fait, et je voulais toucher sans serveur, alors J'ai essayé de créer un système qui acquiert le guide des programmes de mon programme préféré ** Mitsuaki Iwago's World Cat Walk ** que j'enregistre à chaque fois et l'enregistre automatiquement dans le calendrier (arbre temporel).

Ce qui a été utilisé

Il existe une API de guide de programme NHK, alors obtenez-la une fois par jour, filtrez les données et enregistrez-la avec l'API de calendrier. Il existe un service gratuit appelé cron-job.org pour une exécution régulière, et je l'ai utilisé car il y en a un qui peut envoyer une requête à l'URL spécifiée à l'heure spécifiée.

Cette fois, je voulais le terminer en un jour, j'ai donc utilisé Google Cloud Function, qui me permet d'écrire le Python le plus familier.

Créer

Suivi par quelque chose qui ne contient pas beaucoup de code. Je suis désolé.

Au début, je pensais envoyer un email en détectant une erreur, mais quand je recherchais la fonction Google Cloud, je l'ai remarqué la nuit ... Zuckerberg du cœur a dit ce mot, donc je l'ai créé pour le moment fait.

Acquisition du guide des programmes

Au début, il y a des traces d'essais de classification.


# 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"
    #Synthèse NHK 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:
    """Déterminer si les données du guide des programmes incluent la marche du chat"""
    is_nekoaruki = False
    try:
        title = program["title"]
        if "Chat qui marche" 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  #Obtenez pendant 6 jours
    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

Calendrier ajouté

Ici aussi, je viens d'ajouter l'acquisition de l'identifiant du calendrier-> comparer avec les données enregistrées->. Une tempête de compromis.

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:
            #Je n'utilise qu'un seul calendrier, donc le premier calendrier est correct
            if calendar["attributes"]["order"] == 0:
                return calendar
            else:
                pass
    else:
        return response.text


def check_upcoming_events(calendar_id: str):
    """Obtenez 7 jours d'événements enregistrés"""
    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):
    """Envoyer l'événement à l'API et ajouter"""
    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]):
    """Ajouter en comparant les événements enregistrés avec les données acquises"""
    add_events: list = []
    title_ls = [i["title"] for i in registered]
    for data in data_ls:
        #Ignorer si le titre est enregistré
        #Je ne peux pas détecter que l'heure de diffusion a changé depuis mon inscription, mais non
        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]):
    """Extraire uniquement les événements de marche de chat des données enregistrées"""
    filtered_registered_events = filter(lambda x: "Chat qui marche" in x["attributes"]["title"], data_ls)
    extracted: list = []
    #Au début, j'allais mettre à jour lorsque l'heure de début a changé
    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..."

Paramètres de fonction

Créer un projet-> Créer une fonction-> Coller le code-> Définir des variables d'environnement, etc.

N'oubliez pas de mettre TZ = Asia / Tokyo dans la variable d'environnement en plus de la clé API et de changer le fuseau horaire. Ajout de requests == 2.22.0 à requirements.txt.

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

Paramètres d'exécution programmée

Créez simplement un compte sur cron-job.org et configurez un travail à partir de Cronjobs-> Create cronjob. Vous pouvez également définir les données à publier.

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

Résumé

C'est embarrassant de publier du code qui ne fait que lister les fonctions ... Cependant, je pense que l'un des avantages des produits de loisirs est que vous pouvez les fabriquer simplement en les déplaçant. C'est bien de pouvoir tout faire gratuitement.

C'était également une bonne récolte de constater qu'il était facile de créer un environnement sans serveur.

Avec cela, vous n'oublierez jamais l'heure de diffusion des chats qui marchent. ** World Cat Walk de Mitsuaki Iwago ** Voyons voir.

Recommended Posts

J'ai essayé de créer un système pour acquérir automatiquement le guide des programmes → l'enregistrer dans le calendrier en un jour
J'ai fait un chronomètre en utilisant tkinter avec python
J'ai essayé de créer un système qui ne récupère que les tweets supprimés
Un mémorandum lors de l'acquisition automatique avec du sélénium
J'ai essayé d'enregistrer une station sur la plateforme IoT "Rimotte"
Je veux réussir le test G dans un mois Jour 1
J'ai fait une commande pour afficher un calendrier coloré dans le terminal
J'ai essayé de créer un site qui permet de voir facilement les informations mises à jour d'Azure
J'ai essayé de rendre possible l'envoi automatique d'un e-mail en double-cliquant simplement sur l'icône [Python]
J'ai essayé de trouver la différence entre A + = B et A = A + B en Python, alors notez
[Python] J'ai essayé de créer un programme simple qui fonctionne sur la ligne de commande en utilisant argparse
J'ai créé une API Web
J'ai fait un programme pour vérifier la taille d'un fichier avec Python
J'ai essayé d'afficher la valeur d'altitude du DTM dans un graphique
J'ai créé un script POST pour créer un problème sur Github et l'enregistrer dans le projet
J'ai essayé de rendre possible l'envoi automatique d'un e-mail en double-cliquant simplement sur l'icône [GAS / Python]
Transformez un programme Python en démon et exécutez-le automatiquement au démarrage du système d'exploitation
Quand j'ai essayé de créer un VPC avec AWS CDK mais que je n'ai pas pu le faire
Dans IPython, quand j'ai essayé de voir la valeur, c'était un générateur, donc je l'ai inventé quand j'étais frustré.
J'ai créé un système qui décide automatiquement de s'exécuter demain avec Python et l'ajoute à Google Agenda.
Un débutant en apprentissage automatique a tenté de créer une IA de jugement Sheltie en un jour
Je voulais connaître le nombre de lignes dans plusieurs fichiers et j'ai essayé de l'obtenir avec une commande
J'ai créé un jeu ○ ✕ avec TensorFlow
[LPIC 101] J'ai essayé de résumer les options de commande qui sont faciles à faire une erreur
J'ai essayé de faire quelque chose comme un chatbot avec le modèle Seq2Seq de TensorFlow
Essayez de faire une stratégie de blackjack en renforçant l'apprentissage (② Enregistrer l'environnement dans le gymnase)
J'ai essayé de faire sonner le téléphone lorsqu'il a été publié sur le poste IoT
Un débutant en python a tenté de faire un stage dans une entreprise informatique [Jour 3 vers les nuages ...]
Les débutants en Python ont créé un chat BOT alors j'ai essayé de résumer comment le faire
Je ne peux pas dormir tant que je n'ai pas construit un serveur !! (Introduction au serveur Python faite en un jour)
J'ai essayé de faciliter la modification du paramètre du proxy authentifié sur Jupyter
J'ai essayé de faire un "putain de gros convertisseur de littérature"
J'ai essayé de représenter graphiquement les packages installés en Python
J'ai essayé d'implémenter un pseudo pachislot en Python
Il fait froid, j'ai donc essayé de permettre d'allumer / d'éteindre automatiquement le chauffage AC avec Raspberry Pi!
Quand j'ai essayé d'installer PIL et matplotlib dans un environnement virtualenv, j'en étais accro.
[Git] J'ai essayé de faciliter la compréhension de l'utilisation de git stash en utilisant un exemple concret.
Un programme qui redimensionne automatiquement la taille d'image requise pour les icônes d'application iOS en Python
J'ai essayé d'automatiser "un cœur même séparé" à l'aide d'un algorithme génétique en Python
Je veux voir le graphique en 3D! Je peux réaliser un tel rêve.
J'ai aussi essayé d'imiter la fonction monade et la monade d'état avec le générateur en Python
J'ai écrit un doctest dans "J'ai essayé de simuler la probabilité d'un jeu de bingo avec Python"
Un mémorandum pour enregistrer la bibliothèque écrit en Hy dans PyPI
J'ai essayé l'algorithme de super résolution "PULSE" dans un environnement Windows
J'ai essayé d'implémenter un automate cellulaire unidimensionnel en Python
J'ai essayé de créer automatiquement un rapport avec la chaîne de Markov
Combinez des listes dans un DataFrame avec des pandas
J'ai essayé "un programme qui supprime les déclarations en double en Python"
J'ai essayé de résumer le code souvent utilisé dans Pandas
J'ai essayé "Comment obtenir une méthode décorée en Python"
J'ai essayé d'illustrer le temps et le temps du langage C
Je l'ai écrit en langage Go pour comprendre le principe SOLID
J'ai essayé de résumer les commandes souvent utilisées en entreprise
J'ai essayé d'implémenter la fonction d'envoi de courrier en Python
Je veux rendre le type de dictionnaire dans la liste unique
Je veux ajouter un joli complément à input () en python