[PYTHON] Ich habe versucht, ein System zu erstellen, um den Programmführer automatisch zu erfassen → ihn an einem Tag im Kalender zu registrieren
Überblick
Ich wollte eine Geschichte über das posten, was ich gemacht habe, und ich wollte Serverless berühren, also Ich habe versucht, ein System zu erstellen, das den Programmführer meines Lieblingsprogramms ** Mitsuaki Iwagos World Cat Walk ** erfasst, den ich jedes Mal aufzeichne und automatisch im Kalender (Zeitbaum) registriere.
Was wurde verwendet
- NHK Program Guide API -> Programmhandbuch abrufen
- Google Cloud Function
- cron-job.org -> Kostenloser Abonnementdienst
- timetree API
Es gibt eine NHK-Programmführer-API. Holen Sie sie sich also einmal am Tag, filtern Sie die Daten und registrieren Sie sie bei der Kalender-API. Es gibt einen kostenlosen Dienst namens cron-job.org für die reguläre Ausführung, und ich habe ihn verwendet, weil es einen gibt, der zum angegebenen Zeitpunkt eine Anfrage an die angegebene URL senden kann.
Dieses Mal wollte ich es an einem Tag fertigstellen, also habe ich die Google Cloud-Funktion verwendet, mit der ich das bekannteste Python schreiben kann.
Erstellen
Gefolgt von etwas, das nicht viel Code enthält. Es tut mir Leid.
Zuerst dachte ich darüber nach, eine E-Mail zu senden, indem ich einen Fehler entdeckte, aber als ich die Google Cloud-Funktion untersuchte, bemerkte ich es nachts ... Zuckerberg des Herzens sagte dieses Wort, also habe ich es vorerst erstellt tat.
Programmführererfassung
Zunächst gibt es Spuren von Klassifizierungsversuchen.
# 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-Synthese 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:
"""Stellen Sie fest, ob die Programmführungsdaten das Gehen mit der Katze enthalten"""
is_nekoaruki = False
try:
title = program["title"]
if "Katze geht" 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 #Holen Sie sich für 6 Tage
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
Kalender hinzugefügt
Auch hier habe ich gerade die Kalender-ID-Erfassung hinzugefügt-> mit registrierten Daten vergleichen->. Ein Sturm des Kompromisses.
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:
#Ich benutze nur einen Kalender, damit der erste Kalender in Ordnung ist
if calendar["attributes"]["order"] == 0:
return calendar
else:
pass
else:
return response.text
def check_upcoming_events(calendar_id: str):
"""Erhalten Sie registrierte Events im Wert von 7 Tagen"""
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):
"""Ereignis an API senden und hinzufügen"""
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]):
"""Fügen Sie hinzu, indem Sie registrierte Ereignisse mit erfassten Daten vergleichen"""
add_events: list = []
title_ls = [i["title"] for i in registered]
for data in data_ls:
#Überspringen, wenn der Titel registriert ist
#Ich kann nicht feststellen, dass sich die Sendezeit seit meiner Registrierung geändert hat, aber nein
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]):
"""Extrahieren Sie nur Cat-Walking-Ereignisse aus registrierten Daten"""
filtered_registered_events = filter(lambda x: "Katze geht" in x["attributes"]["title"], data_ls)
extracted: list = []
#Zuerst wollte ich aktualisieren, als sich die Startzeit änderte
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..."
Funktionseinstellungen
Projekt erstellen-> Funktion erstellen-> Code einfügen-> Umgebungsvariablen festlegen usw.
Vergessen Sie nicht, zusätzlich zum API-Schlüssel "TZ = Asia / Tokyo" in die Umgebungsvariable einzufügen und die Zeitzone zu ändern.
Requirements == 2.22.0 zur Anforderung.txt hinzugefügt.
Geplante Ausführungseinstellungen
Erstellen Sie einfach ein Konto bei cron-job.org und richten Sie einen Job bei Cronjobs-> Create cronjob ein.
Sie können auch die zu buchenden Daten festlegen.
Zusammenfassung
Es ist peinlich, eine Postleitzahl zu schreiben, in der nur die Funktionen aufgelistet sind ... Ich denke jedoch, dass eines der guten Dinge an Hobbyprodukten ist, dass man sie nur durch Umzug herstellen kann. Es ist schön, alles kostenlos machen zu können.
Es war auch eine gute Ernte, festzustellen, dass es einfach war, eine Umgebung ohne Server zu erstellen.
Damit werden Sie nie die Sendezeit von wandelnden Katzen vergessen. ** Mitsuaki Iwagos World Cat Walk ** Mal sehen.