Bot de Información del día
Haciendo una ruta de senderismo por Olivenza tuve una epifanía. Siempre que me levanto hago la misma rutina:
- Miro el tiempo para saber cuánta ropa necesito o si necesito un paraguas.
- Miro mi calendario por si tengo algún compromiso que no recordara.
- Veo las últimas noticias. El problema es que toda esta información está esparcida por una serie de webs que no se comunican entre sí, por lo que he decidido crear un bot en XMPP que me facilite de forma diaria toda esta información.
En este tutorial enseñaré la obtención de datos para estas necesidades y su envío a través de una cuenta en XMPP.
El tiempo
Existen una infinidad de APIs que permiten obtener todas las características de una zona en concreto. Yo me he decantado por open-meteo, ya que permite hacer una petición granular de solo la información que necesito, en este caso pido:
- Probabilidades de precipitación
- Porcentaje de cobertura de las nubes
- Temperaturas por horas
- Hora del amanecer y atardecer La información es extraída y procesada para tener las temperaturas máximas y mínimas además de las horas en las que se producirán. Finalmente he añadido una serie de iconos en función de las precipitaciones y de las nubes.
1import requests
2...
3URL_WEATHER = "https://api.open-meteo.com/v1/forecast"
4params = {
5 "latitude": 38.8779,
6 "longitude": -6.9706,
7
8 "daily": "sunrise,sunset",
9
10 "hourly": (
11 "temperature_2m,"
12 "precipitation_probability,"
13 "cloud_cover"
14 ),
15
16 # Badajoz local time
17 "timezone": "Europe/Madrid",
18
19 "forecast_days": 1,
20}
21...
22response = requests.get(URL_WEATHER, params=params)
23response.raise_for_status()
24
25data = response.json()
26
27hourly = data["hourly"]
28times = hourly["time"]
29temperatures = hourly["temperature_2m"]
30rain_probability = hourly["precipitation_probability"]
31cloud_cover = hourly["cloud_cover"]
32
33# Daily Data
34daily = data["daily"]
35date = daily["time"][0]
36sunrise = daily["sunrise"][0]
37sunset = daily["sunset"][0]
38
39# Min_Max Temperatures
40temp_min = min(temperatures)
41temp_max = max(temperatures)
42min_index = temperatures.index(temp_min)
43max_index = temperatures.index(temp_max)
44min_time = times[min_index][11:16]
45max_time = times[max_index][11:16]
46# Get hours with significant temperature changes
47hottest_temp = temperatures[max_index]
48coldest_temp = temperatures[min_index]
49
50# Rain
51max_rain_probability = max(rain_probability)
52if max_rain_probability == 0:
53 rain_status = "No rain expected 🌧️❌"
54elif max_rain_probability <= 20:
55 rain_status = "Very low chance of rain 🌦️"
56elif max_rain_probability <= 40:
57 rain_status = "Low chance of rain 🌦️"
58elif max_rain_probability <= 60:
59 rain_status = "Moderate chance of rain 🌦️"
60else:
61 rain_status = "High chance of rain 🌧️"
62# Sunshine
63average_cloud = sum(cloud_cover) / len(cloud_cover)
64if average_cloud <= 10:
65 sunshine = "Very sunny ☀️"
66elif average_cloud <= 30:
67 sunshine = "Mostly sunny 🌤️"
68elif average_cloud <= 60:
69 sunshine = "Partly cloudy ⛅"
70else:
71 sunshine = "Mostly cloudy ☁️"
72
73result_weather = f"""Calendario
Para esto accedo a mi propio servidor Radicale, el cual tiene todos los eventos de mi día a día. El script de python obtiene la información de ese día y si tienen una hora asignada la guarda también. Yo solo uso un único calendario, pero existiría la posibilidad de integrarlo con múltiples calendarios si fuera necesario.
1import caldav
2from datetime import datetime, date, timedelta
3from zoneinfo import ZoneInfo
4...
5URL = credentials.URL
6USERNAME = credentials.USERNAME
7PASSWORD = credentials.PASSWORD
8...
9alendar_lines = []
10
11with caldav.DAVClient(
12 url=URL,
13 username=USERNAME,
14 password=PASSWORD,
15 auth_type="basic",
16) as client:
17 principal = client.principal()
18
19 calendar_lines.append("=" * 20)
20 calendar_lines.append(
21 f"Calendario"
22 )
23 calendar_lines.append("=" * 20)
24
25 for calendar in principal.calendars():
26 calendar_name = calendar.get_display_name()
27
28 if calendar_name.upper() == "TODO":
29 continue
30
31 events = calendar.search(
32 start=start,
33 end=end,
34 event=True,
35 expand=True,
36 )
37
38 if not events:
39 continue
40
41
42 for event in events:
43 component = event.icalendar_component
44
45 for vevent in component.walk():
46 if vevent.name != "VEVENT":
47 continue
48
49 summary = str(vevent.get("SUMMARY", "Untitled"))
50 dtstart = vevent.get("DTSTART")
51 dtend = vevent.get("DTEND")
52
53 if not dtstart:
54 continue
55
56 start_value = dtstart.dt
57 end_value = dtend.dt if dtend else None
58
59 if isinstance(start_value, datetime):
60 start_value = start_value.astimezone(tz)
61
62 if end_value and isinstance(end_value, datetime):
63 end_value = end_value.astimezone(tz)
64
65 calendar_lines.append(
66 f"🕐 {start_value.strftime('%H:%M')}"
67 f"–{end_value.strftime('%H:%M') if end_value else '?'}"
68 )
69 calendar_lines.append(f" {summary}")
70
71 elif isinstance(start_value, date):
72 calendar_lines.append(f"☀️ {summary}")
73 calendar_lines.append(" All day")
74
75 calendar_lines.append("")
76
77result_calendar = "\n".join(calendar_lines)Noticias
Para las noticias no he encontrado un servidor RSS que muestre un resumen de las noticias más importantes del mundo de forma diaria, normalmente suele ser un cúmulo de noticias que van subiéndose en función de que pasa el tiempo (siguiendo la filosofía de RSS). Es por esto que me he decantado por las noticias mediante vídeos de YouTube.
Para conseguir esto selecciono las playlists que expongan noticias nuevas cada día y mediante yt-dlp obtengo la URL del último video.
1import yt_dlp
2...
3PLAYLIST_URLS = [
4 "https://www.youtube.com/playlist?list=PLFLBjMW4wU7gYcRxzvIV9oAL7VIqmJbOW",
5 "https://www.youtube.com/playlist?list=PLZhRxE9191zPN4uc07ytohaXcS13qEgm2",
6]
7
8opts = {
9 "extract_flat": True,
10 "playlistend": 1,
11 "quiet": True,
12}
13
14news_urls = []
15
16with yt_dlp.YoutubeDL(opts) as ydl:
17 for playlist_url in PLAYLIST_URLS:
18 info = ydl.extract_info(playlist_url, download=False)
19
20 latest = info["entries"][0]
21
22 if latest and latest.get("id"):
23 news_urls.append(
24 f"https://www.youtube.com/watch?v={latest['id']}"
25 )
26
27result_news = "\n".join([
28 "====================",
29 "Noticias:",
30 "====================",
31 *news_urls,
32 "",
33])XMPP
Toda esta información deberá ser enviada, originalmente, pensaba habilitar una sección en mi web con las noticias del día, pero no quería tener que meterme en una página y refrescar todos los días, es por ello que uso mi propio servidor XMPP para enviarme la información que ha sido guardada mediante strings.
1import asyncio
2from slixmpp import ClientXMPP
3...
4result = "\n".join([
5 result_news,
6 result_calendar,
7 result_weather
8])
9
10
11class XMPPClient(ClientXMPP):
12 def __init__(self, jid, password, recipient, message):
13 super().__init__(jid, password)
14
15 self.recipient = recipient
16 self.message = message
17
18 self.add_event_handler("session_start", self.session_start)
19
20 # XMPP errors
21 self.add_event_handler("failed_auth", self.failed_auth)
22 self.add_event_handler("connection_failed", self.connection_failed)
23 self.add_event_handler("disconnected", self.disconnected_handler)
24 self.add_event_handler("stanza_error", self.stanza_error)
25
26 async def session_start(self, event):
27 print("XMPP: session started", flush=True)
28
29 self.send_presence()
30 print("XMPP: presence sent", flush=True)
31
32 await self.get_roster()
33 print("XMPP: roster received", flush=True)
34
35 self.send_message(
36 mto=self.recipient,
37 mbody=self.message,
38 mtype="chat"
39 )
40
41 print("XMPP: message sent", flush=True)
42
43 await asyncio.sleep(1)
44
45 print("XMPP: disconnecting", flush=True)
46 self.disconnect()
47
48 def failed_auth(self, event):
49 print("XMPP ERROR: authentication failed", flush=True)
50 print(f"XMPP ERROR DETAILS: {event}", flush=True)
51 self.disconnect()
52
53 def connection_failed(self, event):
54 print("XMPP ERROR: connection failed", flush=True)
55 print(f"XMPP ERROR DETAILS: {event}", flush=True)
56
57 def stanza_error(self, event):
58 print("XMPP ERROR: stanza error", flush=True)
59 print(f"XMPP ERROR DETAILS: {event}", flush=True)
60
61 def disconnected_handler(self, event):
62 print("XMPP: disconnected", flush=True)
63
64
65
66async def main():
67 client = XMPPClient(
68 jid=credentials.jid,
69 password=credentials.password,
70 recipient=credentials.recipient,
71 message=result
72 )
73
74 client.connect()
75 await client.disconnected
76
77
78if __name__ == "__main__":
79 asyncio.run(main())Servidor
Finalmente necesito tener una forma de automatizar esto, ya que no me quiero levantar todas las mañanas, encender mi ordenador para correr un script y poder empezar mi rutina diaria. Para esto existen muchas posibilidades, como hacer un cronjob que apunte a la dirección de este script o modificar el script para que corra de forma constante, que todos los días tome esta información y dejarlo funcionando mediante screen o un contenedor de docker o ….
Pero el lector habitual sabe que soy el fan nº.1 de NixOS así que me gustaría algo que tuviera todo el código localizado en un archivo de .nix y también su tiempo de ejecución. Mi solución es guardar el código en una variable con su entorno de python en otra y mediante un servicio de systemd lo ejecuto todos los días a las 7:00
1{ config, pkgs, ... }:
2
3let
4 pythonEnv = pkgs.python3.withPackages (ps: with ps; [
5 requests
6 yt-dlp
7 caldav
8 slixmpp
9 ]);
10
11 dailyInfo = pkgs.writeText "main.py" ''
12# Your code goes here
13 '';
14
15in
16{
17 # -.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-
18 # Daily Info
19 # -.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-.-
20
21 systemd.services.daily-info = {
22 description = "Send news, calendar and weather";
23
24 serviceConfig = {
25 Type = "oneshot";
26
27 User = config.globals.username;
28
29 ExecStart = "${pythonEnv}/bin/python ${dailyInfo}";
30 };
31 };
32
33 systemd.timers.daily-info = {
34 description = "Run daily info";
35
36 wantedBy = [ "timers.target" ];
37
38 timerConfig = {
39 OnCalendar = "7:00";
40 Persistent = true;
41 };
42 };
43}Conclusión
Con este código consigo una forma de tener todo lo que necesito todas las mañanas para estar preparado por el día. Además de esto, al ser un script de python relativamente sencillo y escalable puedo añadirle información adicional el día de mañana, por ejemplo, si cambiara de trabajo y todos los días tuviera que ir a Cáceres, podría tomar de la API de google.com los datos del tráfico a mi hora de ida y vuelta para tenerlo en consideración.
Adjunto el link de mi repositorio, para ser tomado como referencia y por si exisitieran futuros cambios que cambiaran/mejoraran las secciones de código mostradas en este tutorial.
