feat: extract feed_info.txt content

functions-python/batch_process_dataset/src/pipeline_tasks.py

@@ -14,6 +14,10 @@
     create_http_gtfs_datasets_comparer_task,
 )
 
+# GTFS files that have a registered extractor in the gtfs_file_data_extractor
+# function. Keep in sync with that function's src/extractors/registry.py.
+EXTRACTABLE_FILES = {"feed_info.txt"}
+
 
 def create_http_reverse_geolocation_processor_task(
     stable_id: str,
@@ -45,6 +49,39 @@ def create_http_reverse_geolocation_processor_task(
     )
 
 
+def create_http_gtfs_file_data_extractor_task(
+    stable_id: str,
+    dataset_stable_id: str,
+    file_name: str,
+    file_url: str,
+) -> None:
+    """
+    Create a task to extract structured data from a single GTFS file
+    (handled by the gtfs_file_data_extractor function).
+    """
+    client = tasks_v2.CloudTasksClient()
+    body = json.dumps(
+        {
+            "stable_id": stable_id,
+            "dataset_id": dataset_stable_id,
+            "file_name": file_name,
+            "file_url": file_url,
+        }
+    ).encode()
+    queue_name = os.getenv("GTFS_FILE_DATA_EXTRACTOR_QUEUE")
+    project_id = os.getenv("PROJECT_ID")
+    gcp_region = os.getenv("GCP_REGION")
+
+    create_http_task(
+        client,
+        body,
+        f"https://{gcp_region}-{project_id}.cloudfunctions.net/gtfs-file-data-extractor",
+        project_id,
+        gcp_region,
+        queue_name,
+    )
+
+
 @with_db_session
 def get_changed_files(
     dataset: Gtfsdataset,
@@ -109,6 +146,15 @@ def create_pipeline_tasks(dataset: Gtfsdataset, db_session: Session) -> None:
             stable_id, dataset_stable_id, stops_url
         )
 
+    # Create GTFS file data extraction tasks for changed, extractable files.
+    files_by_name = {file.file_name: file for file in gtfs_files}
+    for file_name in EXTRACTABLE_FILES:
+        gtfs_file = files_by_name.get(file_name)
+        if gtfs_file and gtfs_file.hosted_url and file_name in changed_files:
+            create_http_gtfs_file_data_extractor_task(
+                stable_id, dataset_stable_id, file_name, gtfs_file.hosted_url
+            )
+
     routes_file = next(
         (file for file in gtfs_files if file.file_name == "routes.txt"), None
     )

functions-python/batch_process_dataset/tests/test_pipeline_tasks.py

@@ -5,6 +5,7 @@
 
 from pipeline_tasks import (
     create_http_reverse_geolocation_processor_task,
+    create_http_gtfs_file_data_extractor_task,
     get_changed_files,
     create_pipeline_tasks,
 )
@@ -85,6 +86,113 @@ def test_create_http_reverse_geolocation_processor_task(
         self.assertEqual(args[4], "northamerica-northeast1")
         self.assertEqual(args[5], "rev-geo-queue")
 
+    @patch.dict(
+        os.environ,
+        {
+            "GTFS_FILE_DATA_EXTRACTOR_QUEUE": "file-data-queue",
+            "PROJECT_ID": "my-project",
+            "GCP_REGION": "northamerica-northeast1",
+        },
+        clear=False,
+    )
+    @patch("pipeline_tasks.create_http_task")
+    @patch("pipeline_tasks.tasks_v2.CloudTasksClient")
+    def test_create_http_gtfs_file_data_extractor_task(
+        self, mock_client_cls, mock_create_http_task
+    ):
+        client_instance = MagicMock()
+        mock_client_cls.return_value = client_instance
+
+        create_http_gtfs_file_data_extractor_task(
+            stable_id="feed-123",
+            dataset_stable_id="dataset-abc",
+            file_name="feed_info.txt",
+            file_url="https://example.com/feed_info.txt",
+        )
+
+        self.assertEqual(mock_create_http_task.call_count, 1)
+        args, _ = mock_create_http_task.call_args
+        self.assertIs(args[0], client_instance)
+        payload = json.loads(args[1].decode("utf-8"))
+        self.assertEqual(
+            payload,
+            {
+                "stable_id": "feed-123",
+                "dataset_id": "dataset-abc",
+                "file_name": "feed_info.txt",
+                "file_url": "https://example.com/feed_info.txt",
+            },
+        )
+        self.assertEqual(
+            args[2],
+            "https://northamerica-northeast1-my-project.cloudfunctions.net/gtfs-file-data-extractor",
+        )
+        self.assertEqual(args[5], "file-data-queue")
+
+
+class TestCreatePipelineTasksFeedInfo(unittest.TestCase):
+    """Covers the gtfs_file_data_extractor enqueue gating in create_pipeline_tasks."""
+
+    def _mock_session_no_base_dataset(self):
+        mock_session = MagicMock()
+        mock_session.query.return_value.filter.return_value.order_by.return_value.first.return_value = (
+            None
+        )
+        return mock_session
+
+    def _run(self, files, changed_files):
+        dataset = SimpleDataset(
+            feed_id=1,
+            dataset_id=10,
+            feed_stable_id="feed-A",
+            dataset_stable_id="ds-1",
+            files=files,
+        )
+        with patch(
+            "pipeline_tasks.get_changed_files", return_value=changed_files
+        ), patch(
+            "pipeline_tasks.create_http_reverse_geolocation_processor_task"
+        ), patch(
+            "pipeline_tasks.create_http_pmtiles_builder_task"
+        ), patch(
+            "pipeline_tasks.create_http_gtfs_datasets_comparer_task"
+        ), patch(
+            "pipeline_tasks.create_http_gtfs_file_data_extractor_task"
+        ) as mock_extractor_task:
+            create_pipeline_tasks(
+                dataset, db_session=self._mock_session_no_base_dataset()
+            )
+        return mock_extractor_task
+
+    def test_enqueues_when_feed_info_present_and_changed(self):
+        files = [
+            SimpleFile(
+                "feed_info.txt",
+                hosted_url="https://x.com/feed_info.txt",
+                file_hash="h1",
+            )
+        ]
+        mock_task = self._run(files, changed_files=["feed_info.txt"])
+        mock_task.assert_called_once_with(
+            "feed-A", "ds-1", "feed_info.txt", "https://x.com/feed_info.txt"
+        )
+
+    def test_skips_when_feed_info_not_changed(self):
+        files = [
+            SimpleFile(
+                "feed_info.txt",
+                hosted_url="https://x.com/feed_info.txt",
+                file_hash="h1",
+            )
+        ]
+        mock_task = self._run(files, changed_files=["stops.txt"])
+        mock_task.assert_not_called()
+
+    def test_skips_when_feed_info_absent(self):
+        files = [SimpleFile("stops.txt", hosted_url="https://x.com/stops.txt")]
+        mock_task = self._run(files, changed_files=["stops.txt"])
+        mock_task.assert_not_called()
+
 
 class TestHasFileChanged(unittest.TestCase):
     def _make_mock_session_chain(self, previous_dataset):

functions-python/gtfs_file_data_extractor/README.md

@@ -0,0 +1,32 @@
+# GTFS File Data Extractor
+
+HTTP Cloud Function that extracts structured data from a single GTFS file and
+persists it to the database. It is enqueued as a Cloud Task by
+`batch_process_dataset` once a dataset has been processed (see
+`batch_process_dataset/src/pipeline_tasks.py`).
+
+## How it works
+
+The task payload identifies one GTFS file:
+
+```json
+{
+  "stable_id": "<feed stable id>",
+  "dataset_id": "<dataset stable id>",
+  "file_name": "feed_info.txt",
+  "file_url": "<hosted_url of the GTFS file>"
+}
+```
+
+`processor.py` downloads the file, looks up the matching extractor in
+`extractors/registry.py`, and lets it write to the database.
+
+## Adding a new file extractor
+
+1. Implement a `FileDataExtractor` subclass under `src/extractors/`.
+2. Register it in `src/extractors/registry.py`.
+3. Add its `file_name` to `EXTRACTABLE_FILES` in
+   `batch_process_dataset/src/pipeline_tasks.py` so the producer enqueues it.
+
+Currently registered: `feed_info.txt` -> `FeedInfoExtractor` (writes the
+`feedinfo` table).

functions-python/gtfs_file_data_extractor/function_config.json

@@ -0,0 +1,21 @@
+{
+  "name": "gtfs-file-data-extractor",
+  "description": "Extracts structured data from GTFS files (e.g. feed_info.txt) into the database",
+  "entry_point": "gtfs_file_data_extractor",
+  "timeout": 540,
+  "available_memory": "512Mi",
+  "trigger_http": true,
+  "include_folders": ["helpers"],
+  "include_api_folders": ["database_gen", "database", "common"],
+  "environment_variables": [],
+  "secret_environment_variables": [
+    {
+      "key": "FEEDS_DATABASE_URL"
+    }
+  ],
+  "ingress_settings": "ALLOW_ALL",
+  "max_instance_request_concurrency": 1,
+  "max_instance_count": 5,
+  "min_instance_count": 0,
+  "available_cpu": 1
+}

functions-python/gtfs_file_data_extractor/requirements.txt

@@ -0,0 +1,20 @@
+# Common packages
+functions-framework==3.*
+google-cloud-logging
+psycopg2-binary==2.9.6
+requests~=2.33.1
+certifi~=2025.8.3
+
+# SQL Alchemy and Geo Alchemy (database_gen models depend on geoalchemy2)
+SQLAlchemy==2.0.23
+geoalchemy2==0.14.7
+
+# Google specific packages (shared helpers)
+google-cloud-tasks
+google-auth==2.29.0
+
+# Additional packages for this function
+pandas
+
+# Configuration
+python-dotenv==1.2.2

functions-python/gtfs_file_data_extractor/requirements_dev.txt

@@ -0,0 +1,3 @@
+Faker
+pytest~=7.4.3
+requests-mock

functions-python/gtfs_file_data_extractor/src/extractors/base.py

@@ -0,0 +1,34 @@
+from abc import ABC, abstractmethod
+
+import pandas as pd
+from sqlalchemy.orm import Session
+
+from shared.database_gen.sqlacodegen_models import Gtfsdataset
+
+
+class FileDataExtractor(ABC):
+    """
+    Base class for extracting structured data from a single GTFS file.
+
+    Each subclass handles exactly one GTFS file (identified by ``file_name``)
+    and persists the extracted values to the database.
+
+    To add support for a new file:
+      1. Implement a subclass here and register it in ``extractors/registry.py``.
+      2. Add the file name to ``EXTRACTABLE_FILES`` in batch_process_dataset's
+         ``pipeline_tasks.py`` so the producer enqueues a task for it.
+    """
+
+    #: The GTFS file this extractor handles, e.g. "feed_info.txt".
+    file_name: str
+
+    @abstractmethod
+    def extract(
+        self, df: pd.DataFrame, dataset: Gtfsdataset, db_session: Session
+    ) -> None:
+        """
+        Parse ``df`` (the parsed contents of ``file_name``) and persist the
+        extracted data for ``dataset``. Implementations must be idempotent:
+        re-running for the same dataset should update, not duplicate.
+        """
+        raise NotImplementedError

functions-python/gtfs_file_data_extractor/src/extractors/feed_info_extractor.py

@@ -0,0 +1,88 @@
+import logging
+from datetime import date, datetime
+from typing import Optional
+
+import pandas as pd
+from sqlalchemy.orm import Session
+
+from extractors.base import FileDataExtractor
+from shared.database_gen.sqlacodegen_models import Feedinfo, Gtfsdataset
+
+# feed_info.txt text columns (GTFS spec) mapped 1:1 onto Feedinfo string columns.
+STRING_FIELDS = (
+    "feed_publisher_name",
+    "feed_publisher_url",
+    "feed_lang",
+    "default_lang",
+    "feed_version",
+    "feed_contact_email",
+    "feed_contact_url",
+)
+# feed_info.txt date columns, stored as plain DATE (YYYYMMDD, no timezone).
+DATE_FIELDS = ("feed_start_date", "feed_end_date")
+
+
+def clean_str(value) -> Optional[str]:
+    """Return a trimmed string, or None for empty/NaN/missing values."""
+    if value is None or (isinstance(value, float) and pd.isna(value)):
+        return None
+    text = str(value).strip()
+    return text or None
+
+
+def parse_gtfs_date(value) -> Optional[date]:
+    """Parse a GTFS YYYYMMDD date. Returns None when missing or unparseable."""
+    text = clean_str(value)
+    if text is None:
+        return None
+    # pandas may read a purely numeric column as int/float, yielding "20240101.0".
+    if text.endswith(".0"):
+        text = text[:-2]
+    try:
+        return datetime.strptime(text, "%Y%m%d").date()
+    except ValueError:
+        logging.warning("Unparseable feed_info date value: %r", value)
+        return None
+
+
+class FeedInfoExtractor(FileDataExtractor):
+    """Extracts feed_info.txt into a Feedinfo row (one per dataset)."""
+
+    file_name = "feed_info.txt"
+
+    def extract(
+        self, df: pd.DataFrame, dataset: Gtfsdataset, db_session: Session
+    ) -> None:
+        if df is None or df.empty:
+            logging.info(
+                "feed_info.txt is empty for dataset %s; nothing to extract.",
+                dataset.stable_id,
+            )
+            return
+
+        # feed_info.txt holds a single record.
+        row = df.iloc[0]
+
+        values = {field: clean_str(row.get(field)) for field in STRING_FIELDS}
+        for field in DATE_FIELDS:
+            values[field] = parse_gtfs_date(row.get(field))
+
+        # Upsert keyed by dataset so reprocessing updates in place.
+        feed_info = (
+            db_session.query(Feedinfo)
+            .filter(Feedinfo.gtfs_dataset_id == dataset.id)
+            .one_or_none()
+        )
+        if feed_info is None:
+            feed_info = Feedinfo(gtfs_dataset_id=dataset.id)
+            db_session.add(feed_info)
+
+        for field, value in values.items():
+            setattr(feed_info, field, value)
+
+        logging.info(
+            "Extracted feed_info for dataset %s: start=%s end=%s",
+            dataset.stable_id,
+            values["feed_start_date"],
+            values["feed_end_date"],
+        )