1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
|
# /// script
# requires-python = ">=3.13"
# dependencies = [
# "requests",
# ]
# ///
import json
import requests
import os
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
class Config:
gtfs_url: str = ""
ntfy_topic: str = ""
lastfeed: str = ""
file_to_monitor: str = ""
def __init__(
self, gtfs_url: str, ntfy_topic: str, lastfeed: str, file_to_monitor: str
) -> None:
self.gtfs_url = gtfs_url
self.ntfy_topic = ntfy_topic
self.lastfeed = lastfeed
self.file_to_monitor = file_to_monitor
def load_config(config_path: str) -> Config:
"""Load the lastfeed data from a local source."""
with open(config_path, "r", encoding="utf-8") as f:
data = json.load(f)
return Config(
data["gtfs_url"],
data["ntfy_topic"],
data["lastfeed"],
data.get("file_to_monitor", ""),
)
def save_config(config_path: str, config: Config) -> None:
"""Save the lastfeed data to a local source."""
with open(config_path, "w", encoding="utf-8") as f:
json.dump(
{
"gtfs_url": config.gtfs_url,
"ntfy_topic": config.ntfy_topic,
"lastfeed": config.lastfeed,
"file_to_monitor": config.file_to_monitor,
},
f,
ensure_ascii=False,
indent=4,
)
def load_gtfs_last_modified(url: str) -> datetime | None:
"""Perform a HEAD request to the GTFS service to get last modification date."""
response = requests.head(url, timeout=10)
response.raise_for_status()
last_modified = response.headers.get("Last-Modified")
if last_modified:
return parsedate_to_datetime(last_modified)
return None
def format_date_spanish(dt: datetime) -> str:
"""Format a datetime object into a Spanish date string."""
months = [
"enero", "febrero", "marzo", "abril", "mayo", "junio",
"julio", "agosto", "septiembre", "octubre", "noviembre", "diciembre"
]
day = dt.day
month = months[dt.month - 1]
time = dt.strftime("%H:%M")
return f"{day} de {month}, {time}"
def push_ntfy(topic: str, message: str, title: str, priority: str = "default", tags: str = "") -> None:
"""Push notification using ntfy service."""
requests.post(
f"https://ntfy.sh/{topic}",
data=message.encode("utf-8"),
headers={
"Priority": priority,
"Tags": tags,
"Title": title
},
timeout=10
)
if __name__ == "__main__":
# Get the directory where the script is located
script_dir = os.path.dirname(os.path.abspath(__file__))
config_path = os.path.join(script_dir, "config.json")
conf = load_config(config_path)
current_last_modified = load_gtfs_last_modified(conf.gtfs_url)
if not current_last_modified:
print("Could not get last modification date.")
else:
# datetime.fromisoformat supports 'Z' in Python 3.11+
stored_last_modified = datetime.fromisoformat(conf.lastfeed)
# 1. Check if server has a NEWER feed than what we LAST NOTIFIED about
if current_last_modified > stored_last_modified:
formatted_date = format_date_spanish(current_last_modified)
print(f"New GTFS found: {formatted_date}")
msg = f"Nuevo GTFS de Vitrasa con fecha {formatted_date}"
push_ntfy(conf.ntfy_topic, msg, "Nuevo GTFS listo", priority="high", tags="rotating_light")
# Update config with new date
conf.lastfeed = current_last_modified.isoformat().replace("+00:00", "Z")
save_config(config_path, conf)
else:
print("GTFS server date has not changed since last notification.")
# 2. Check if the LOCAL file is currently outdated compared to the server
if conf.file_to_monitor:
# Resolve relative path from script location if necessary
full_path = os.path.normpath(os.path.join(script_dir, conf.file_to_monitor))
if os.path.exists(full_path):
file_mtime = datetime.fromtimestamp(os.path.getmtime(full_path), tz=timezone.utc)
if file_mtime < current_last_modified:
formatted_file_mtime = format_date_spanish(file_mtime)
formatted_feed_mtime = format_date_spanish(current_last_modified)
print(f"Warning: Local file {conf.file_to_monitor} is outdated.")
msg = (
f"El archivo local ({formatted_file_mtime}) "
f"es anterior al feed disponible ({formatted_feed_mtime})"
)
push_ntfy(conf.ntfy_topic, msg, "Archivo GTFS desactualizado", priority="default", tags="warning")
else:
print(f"Local file {conf.file_to_monitor} is up to date.")
else:
print(f"File to monitor not found: {full_path}")
|
