aboutsummaryrefslogtreecommitdiff
path: root/vitrasa_extend/extend.py
diff options
context:
space:
mode:
authorAriel Costas Guerrero <ariel@costas.dev>2026-01-22 11:57:24 +0100
committerAriel Costas Guerrero <ariel@costas.dev>2026-01-22 11:57:24 +0100
commit986e7281d3f9903794c399f7253bca4948d9b101 (patch)
treec9b7fbf7b0b12f244d2c817067c4f1ce61cf7137 /vitrasa_extend/extend.py
parent81007137e2c7c300b07e4f2aa4b96fb24b3c95a3 (diff)
Add vitrasa script
Diffstat (limited to 'vitrasa_extend/extend.py')
-rw-r--r--vitrasa_extend/extend.py69
1 files changed, 69 insertions, 0 deletions
diff --git a/vitrasa_extend/extend.py b/vitrasa_extend/extend.py
new file mode 100644
index 0000000..458e306
--- /dev/null
+++ b/vitrasa_extend/extend.py
@@ -0,0 +1,69 @@
+# /// script
+# requires-python = ">=3.12"
+# dependencies = [
+# "requests"
+# ]
+# ///
+
+from argparse import ArgumentParser
+import csv
+import json
+import logging
+import os
+import shutil
+import tempfile
+import zipfile
+
+
+def get_rows(input_file: str) -> list[dict]:
+ rows: list[dict] = []
+
+ with open(input_file, "r", encoding="utf-8") as f:
+ reader = csv.DictReader(f)
+ if reader.fieldnames is None:
+ return []
+ reader.fieldnames = [name.strip() for name in reader.fieldnames]
+
+ for row in reader:
+ rows.append(row)
+
+ return rows
+
+
+if __name__ == "__main__":
+ parser = ArgumentParser()
+ parser.add_argument(
+ "--debug",
+ help="Enable debug logging",
+ action="store_true"
+ )
+
+ args = parser.parse_args()
+
+ logging.basicConfig(
+ level=logging.DEBUG if args.debug else logging.INFO,
+ format="%(asctime)s - %(levelname)s - %(message)s",
+ )
+
+ GTFS_TEMPWORKDIR = tempfile.mkdtemp(prefix="vitrasa_extend_")
+
+ # Unzip the GTFS feed
+ with zipfile.ZipFile("../feeds/vitrasa.zip", "r") as zip_ref:
+ zip_ref.extractall(GTFS_TEMPWORKDIR)
+
+ CALENDAR_DATES_FILE = os.path.join(GTFS_TEMPWORKDIR, "calendar_dates.txt")
+ # Open the calendar_dates.txt file, and copy
+
+ # Create a ZIP archive of the output GTFS
+ with zipfile.ZipFile("../feeds/vitrasa_fixed.zip", "w", zipfile.ZIP_DEFLATED) as zipf:
+ for root, _, files in os.walk(GTFS_TEMPWORKDIR):
+ for file in files:
+ file_path = os.path.join(root, file)
+ arcname = os.path.relpath(file_path, GTFS_TEMPWORKDIR)
+ zipf.write(file_path, arcname)
+
+ logging.info(
+ "GTFS data from feed has been zipped successfully at ../feeds/vitrasa_fixed.zip."
+ )
+
+ shutil.rmtree(GTFS_TEMPWORKDIR)