fix: 1204 bounding boxes are missing for many feeds with good stopstxt files

functions-python/helpers/query_helper.py

@@ -173,3 +173,34 @@ def get_datasets_with_missing_reports_query(
         Gtfsdataset.stable_id, Gtfsfeed.stable_id
     )
     return query
+
+
+def get_feeds_with_missing_bounding_boxes_query(
+    db_session: Session,
+) -> Query:
+    """
+    Get GTFS feeds and datasets where the dataset is missing a bounding box.
+
+    Args:
+        db_session: SQLAlchemy session
+
+    Returns:
+        A SQLAlchemy query object for GTFS feeds with datasets missing bounding boxes
+        ordered by dataset and feed stable id.
+    """
+    query = (
+        db_session.query(
+            Gtfsfeed.stable_id,
+            Gtfsdataset.stable_id,
+        )
+        .select_from(Gtfsfeed)
+        .join(Gtfsdataset, Gtfsdataset.feed_id == Gtfsfeed.id)
+        .filter(Gtfsdataset.bounding_box.is_(None))
+        .filter(
+            ~Gtfsfeed.feedlocationgrouppoints.any()
+        )  # Only feeds with no location group points
+        .distinct(Gtfsfeed.stable_id, Gtfsdataset.stable_id)
+        .order_by(Gtfsdataset.stable_id, Gtfsfeed.stable_id)
+    )
+
+    return query

functions-python/tasks_executor/requirements.txt

@@ -16,6 +16,7 @@ geoalchemy2==0.14.7
 
 # Google specific packages for this function
 google-cloud-workflows
+google-cloud-pubsub
 flask
 
 # Configuration

functions-python/tasks_executor/src/main.py

@@ -22,6 +22,9 @@
 from tasks.validation_reports.rebuild_missing_validation_reports import (
     rebuild_missing_validation_reports_handler,
 )
+from tasks.missing_bounding_boxes.rebuild_missing_bounding_boxes import (
+    rebuild_missing_bounding_boxes_handler,
+)
 
 
 init_logger()
@@ -42,6 +45,10 @@
         "description": "Rebuilds missing validation reports for GTFS datasets.",
         "handler": rebuild_missing_validation_reports_handler,
     },
+    "rebuild_missing_bounding_boxes": {
+        "description": "Rebuilds missing bounding boxes for GTFS datasets that contain valid stops.txt files.",
+        "handler": rebuild_missing_bounding_boxes_handler,
+    },
 }
 
 

functions-python/tasks_executor/src/tasks/missing_bounding_boxes/rebuild_missing_bounding_boxes.py

@@ -0,0 +1,128 @@
+import logging
+import os
+from typing import Dict, List
+from sqlalchemy.orm import Session
+
+from shared.database.database import with_db_session
+from shared.helpers.pub_sub import publish_messages
+from shared.helpers.query_helper import get_feeds_with_missing_bounding_boxes_query
+from shared.database_gen.sqlacodegen_models import Gtfsfeed, Gtfsdataset
+from datetime import datetime
+
+
+def rebuild_missing_bounding_boxes_handler(payload) -> dict:
+    (dry_run, after_date) = get_parameters(payload)
+
+    return rebuild_missing_bounding_boxes(
+        dry_run=dry_run,
+        after_date=after_date,
+    )
+
+
+@with_db_session
+def rebuild_missing_bounding_boxes(
+    dry_run: bool = True,
+    after_date: str = None,
+    db_session: Session | None = None,
+) -> dict:
+    """
+    Find GTFS feeds/datasets missing bounding boxes and either log or publish them for processing.
+
+    Args:
+        dry_run (bool): If True, only logs the number of feeds found (no publishing).
+        after_date (str, optional): ISO date string (YYYY-MM-DD). Only datasets downloaded after this date are included.
+        db_session (Session, optional): SQLAlchemy session, injected by @with_db_session.
+
+    Returns:
+        dict: Summary message and count of processed feeds.
+    """
+    filter_after = None
+    if after_date:
+        try:
+            filter_after = datetime.fromisoformat(after_date)
+        except Exception:
+            logging.warning(
+                "Invalid after_date format, expected ISO format (YYYY-MM-DD)"
+            )
+    query = get_feeds_with_missing_bounding_boxes_query(db_session)
+    if filter_after:
+        query = query.filter(Gtfsdataset.downloaded_at >= filter_after)
+    feeds = query.all()
+
+    if dry_run:
+        total_processed = len(feeds)
+        logging.info(
+            "Dry run mode: %s feeds with missing bounding boxes found, filtered after %s.",
+            total_processed,
+            after_date,
+        )
+        return {
+            "message": f"Dry run: {total_processed} feeds with missing bounding boxes found."
+            + (f" Filtered after: {filter_after}" if filter_after else ""),
+            "total_processed": total_processed,
+        }
+    else:
+        # publish a message to a Pub/Sub topic for each feed
+        pubsub_topic_name = os.getenv("PUBSUB_TOPIC_NAME", None)  # todo: set new name
+        project_id = os.getenv("PROJECT_ID")
+
+        logging.info("Publishing to topic: %s", pubsub_topic_name)
+        publish_messages(prepare_feeds_data(feeds), project_id, pubsub_topic_name)
+
+        total_processed = len(feeds)
+        logging.info(
+            "Published %s feeds with missing bounding boxes to Pub/Sub topic: %s, filtered after %s.",
+            total_processed,
+            pubsub_topic_name,
+            after_date,
+        )
+        return {
+            "message": f"Successfully published {total_processed} feeds with missing bounding boxes."
+            + (f" Filtered after: {filter_after}" if filter_after else ""),
+            "total_processed": total_processed,
+        }
+
+
+def prepare_feeds_data(feeds: List[Gtfsfeed]) -> List[Dict]:
+    """
+    Format feeds data for Pub/Sub messages.
+
+    Args:
+        feeds: List of Gtfsfeed objects
+
+    Returns:
+        List of dictionaries with feed data
+    """
+    data = []
+
+    for feed in feeds:
+        # Get the latest dataset
+        if feed.gtfsdatasets and any(dataset.latest for dataset in feed.gtfsdatasets):
+            latest_dataset = next(
+                dataset for dataset in feed.gtfsdatasets if dataset.latest
+            )
+
+            data.append(
+                {
+                    "stable_id": feed.stable_id,
+                    "dataset_id": latest_dataset.stable_id,
+                    "url": latest_dataset.hosted_url,
+                }
+            )
+
+    return data
+
+
+def get_parameters(payload):
+    """
+    Get parameters from the payload and environment variables.
+
+    Args:
+        payload (dict): dictionary containing the payload data.
+    Returns:
+        tuple: (dry_run, after_date)
+    """
+    dry_run = payload.get("dry_run", True)
+    dry_run = dry_run if isinstance(dry_run, bool) else str(dry_run).lower() == "true"
+    after_date = payload.get("after_date", None)
+    return dry_run, after_date

functions-python/tasks_executor/tests/tasks/bounding_boxes/test_rebuild_missing_bounding_boxes.py

@@ -0,0 +1,98 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from datetime import datetime
+
+from tasks.missing_bounding_boxes.rebuild_missing_bounding_boxes import (
+    get_parameters,
+    rebuild_missing_bounding_boxes,
+)
+
+
+class TestTasksExecutor(unittest.TestCase):
+    def test_get_parameters(self):
+        payload = {"dry_run": True}
+        dry_run, after_date = get_parameters(payload)
+        self.assertTrue(dry_run)
+        self.assertIsNone(after_date)
+
+    def test_get_parameters_with_valid_after_date(self):
+        payload = {"dry_run": False, "after_date": "2024-06-01"}
+        dry_run, after_date = get_parameters(payload)
+        self.assertFalse(dry_run)
+        self.assertEqual(after_date, "2024-06-01")
+        # Check ISO format
+        try:
+            datetime.fromisoformat(after_date)
+        except ValueError:
+            self.fail(f"after_date '{after_date}' is not a valid ISO date string")
+
+    def test_get_parameters_with_string_bool(self):
+        payload = {"dry_run": "false", "after_date": None}
+        dry_run, after_date = get_parameters(payload)
+        self.assertFalse(dry_run)
+        self.assertIsNone(after_date)
+
+    def test_get_parameters_missing_keys(self):
+        payload = {}
+        dry_run, after_date = get_parameters(payload)
+        self.assertTrue(dry_run)
+        self.assertIsNone(after_date)
+
+    @patch(
+        "tasks.missing_bounding_boxes.rebuild_missing_bounding_boxes.get_feeds_with_missing_bounding_boxes_query"
+    )
+    def test_rebuild_missing_bounding_boxes_dry_run(self, mock_query):
+        # Mock the query and its .all() method
+        mock_query.return_value.filter.return_value = mock_query.return_value
+        mock_query.return_value.all.return_value = [
+            ("feed1", "dataset1"),
+            ("feed2", "dataset2"),
+        ]
+        result = rebuild_missing_bounding_boxes(
+            dry_run=True, after_date=None, db_session=MagicMock()
+        )
+        self.assertIn("Dry run", result["message"])
+        self.assertEqual(result["total_processed"], 2)
+
+    @patch(
+        "tasks.missing_bounding_boxes.rebuild_missing_bounding_boxes.publish_messages"
+    )
+    @patch(
+        "tasks.missing_bounding_boxes.rebuild_missing_bounding_boxes.get_feeds_with_missing_bounding_boxes_query"
+    )
+    def test_rebuild_missing_bounding_boxes_publish(self, mock_query, mock_publish):
+        # Mock Gtfsdataset and Gtfsfeed objects
+        mock_dataset = MagicMock()
+        mock_dataset.latest = True
+        mock_dataset.stable_id = "dataset1"
+        mock_dataset.hosted_url = "http://example.com/dataset1"
+        mock_feed = MagicMock()
+        mock_feed.stable_id = "feed1"
+        mock_feed.gtfsdatasets = [mock_dataset]
+
+        mock_query.return_value.filter.return_value = mock_query.return_value
+        mock_query.return_value.all.return_value = [mock_feed]
+        mock_publish.return_value = None
+
+        result = rebuild_missing_bounding_boxes(
+            dry_run=False, after_date=None, db_session=MagicMock()
+        )
+        self.assertIn("Successfully published", result["message"])
+        self.assertEqual(result["total_processed"], 1)
+
+    @patch(
+        "tasks.missing_bounding_boxes.rebuild_missing_bounding_boxes.get_feeds_with_missing_bounding_boxes_query"
+    )
+    def test_rebuild_missing_bounding_boxes_invalid_after_date(self, mock_query):
+        mock_query.return_value.filter.return_value = mock_query.return_value
+        mock_query.return_value.all.return_value = []
+        # Should log a warning and not raise
+        result = rebuild_missing_bounding_boxes(
+            dry_run=True, after_date="not-a-date", db_session=MagicMock()
+        )
+        self.assertIn("Dry run", result["message"])
+        self.assertEqual(result["total_processed"], 0)
+
+
+if __name__ == "__main__":
+    unittest.main()