[PYTHON] 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é
- API NHK Program Guide -> Get Program Guide
- Google Cloud Function
- cron-job.org -> Service d'abonnement gratuit
- timetree API
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.
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.
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