feat: added batch fill for dataset size and files

functions-python/tasks_executor/README.md

@@ -24,6 +24,8 @@ Example:
     "filter_statuses": ["active", "inactive", "future"]
   }
 }
+```
+```json
 {
    "task": "rebuild_missing_bounding_boxes",
    "payload": {
@@ -34,12 +36,9 @@ Example:
 ```
 
 To get the list of supported tasks use:
-``
+```json
 {
 "name": "list_tasks",
 "payload": {}
 }
-
-```
-
 ```

functions-python/tasks_executor/requirements.txt

@@ -18,6 +18,7 @@ geoalchemy2==0.14.7
 google-cloud-workflows
 google-cloud-pubsub
 flask
+google-cloud-storage
 
 # Configuration
 python-dotenv==1.0.0

functions-python/tasks_executor/src/main.py

@@ -19,6 +19,9 @@
 import flask
 import functions_framework
 from shared.helpers.logger import init_logger
+from tasks.dataset_files.rebuild_missing_dataset_files import (
+    rebuild_missing_dataset_files_handler,
+)
 from tasks.validation_reports.rebuild_missing_validation_reports import (
     rebuild_missing_validation_reports_handler,
 )
@@ -49,6 +52,10 @@
         "description": "Rebuilds missing bounding boxes for GTFS datasets that contain valid stops.txt files.",
         "handler": rebuild_missing_bounding_boxes_handler,
     },
+    "rebuild_missing_dataset_files": {
+        "description": "Rebuilds missing dataset files for GTFS datasets.",
+        "handler": rebuild_missing_dataset_files_handler,
+    },
 }
 
 

functions-python/tasks_executor/src/tasks/dataset_files/README.md

@@ -0,0 +1,72 @@
+# Rebuild Missing Dataset Files
+
+This task rebuilds missing extracted files in GTFS datasets.
+It downloads datasets from their `hosted_url`, extracts all files, computes zipped and unzipped sizes, calculates hashes, uploads the files to a GCS bucket, and updates the database.
+
+---
+
+## Task ID
+
+Use task ID: `rebuild_missing_dataset_files`
+
+---
+
+## Usage
+
+The function accepts the following payload:
+
+```json
+{
+  "dry_run": true,                // [optional] If true, do not upload or modify the database (default: true)
+  "after_date": "YYYY-MM-DD",     // [optional] Only include datasets downloaded after this ISO date
+  "latest_only": true             // [optional] If true, only process the latest version of each dataset (default: true)
+}
+```
+
+### Example:
+
+```json
+{
+  "dry_run": false,
+  "after_date": "2025-07-01",
+  "latest_only": true
+}
+```
+
+---
+
+## What It Does
+
+For each GTFS dataset with missing file information (missing zipped/unzipped sizes or missing extracted files), this function:
+
+1. Downloads the `.zip` file from its `hosted_url`
+2. Computes the zipped size in bytes
+3. Extracts all GTFS files locally
+4. Computes the unzipped size in bytes
+5. Uploads each extracted file to a GCS bucket with the structure:
+
+   ```
+   {feed-stable-id}/{dataset-stable-id}/extracted/{file_name}
+   ```
+6. Makes each file publicly accessible and stores its GCS URL
+7. Computes SHA256 hashes for each file
+8. Stores metadata in the `Gtfsfile` table for later use
+
+---
+
+## GCP Environment Variables
+
+The function requires the following environment variables:
+
+| Variable               | Description                                                                  |
+| ---------------------- | ---------------------------------------------------------------------------- |
+| `DATASETS_BUCKET_NAME` | The name of the GCS bucket used to store extracted GTFS files                |
+
+---
+
+## Additional Notes
+
+* This function **disables SSL verification** when downloading files, as the sources are trusted internally.
+* Commits to the database occur in batches of 5 datasets to improve performance and avoid large transaction blocks.
+* If `dry_run` is enabled, no downloads, uploads, or DB modifications are performed. Only the number of affected datasets is logged.
+* The function is safe to rerun. It will only affect datasets missing required file metadata.

functions-python/tasks_executor/src/tasks/dataset_files/rebuild_missing_dataset_files.py

@@ -0,0 +1,213 @@
+import logging
+import os
+import uuid
+import hashlib
+import shutil
+import tempfile
+import zipfile
+import urllib.request
+import ssl
+
+from google.cloud import storage
+from shared.database.database import with_db_session
+from shared.database_gen.sqlacodegen_models import Gtfsdataset, Gtfsfile
+
+# Disable SSL verification — trusted internal sources only
+ssl._create_default_https_context = ssl._create_unverified_context
+
+
+def rebuild_missing_dataset_files_handler(payload) -> dict:
+    """
+    Entry point for rebuilding missing GTFS dataset files.
+
+    Args:
+        payload (dict): Task parameters including 'dry_run' and 'after_date'.
+
+    Returns:
+        dict: Result message and number of datasets processed.
+    """
+    dry_run = payload.get("dry_run", True)
+    after_date = payload.get("after_date", None)
+    latest_only = payload.get("latest_only", True)
+
+    return rebuild_missing_dataset_files(
+        dry_run=dry_run, after_date=after_date, latest_only=latest_only
+    )
+
+
+def get_datasets_with_missing_files_query(db_session, after_date, latest_only):
+    """
+    Query GTFS datasets missing zipped/unzipped size or extracted files.
+
+    Args:
+        db_session: SQLAlchemy DB session.
+        after_date (str): Filter datasets downloaded after this ISO date.
+        latest_only (bool): Whether to include only latest datasets.
+
+    Returns:
+        Query: SQLAlchemy query object.
+    """
+    query = (
+        db_session.query(Gtfsdataset)
+        .filter(~Gtfsdataset.hosted_url.is_(None))
+        .filter(
+            Gtfsdataset.zipped_size_bytes.is_(None)
+            | Gtfsdataset.unzipped_size_bytes.is_(None)
+            | ~Gtfsdataset.gtfsfiles.any()
+        )
+    )
+
+    if after_date:
+        query = query.filter(Gtfsdataset.downloaded_at >= after_date)
+
+    if latest_only:
+        query = query.filter(Gtfsdataset.latest)
+
+    return query
+
+
+def download_to_file(url: str, local_path: str):
+    """
+    Downloads a file from URL and writes it to local disk.
+    """
+    with urllib.request.urlopen(url) as response, open(local_path, "wb") as out_file:
+        shutil.copyfileobj(response, out_file)
+
+
+def process_dataset(dataset: Gtfsdataset):
+    """
+    Downloads, extracts, uploads, and indexes files for a GTFS dataset.
+
+    Args:
+        dataset (Gtfsdataset): The dataset to process.
+    """
+    hosted_url = dataset.hosted_url
+    stable_id = dataset.stable_id
+    logging.info("Processing dataset %s with URL %s", stable_id, hosted_url)
+    bucket_name = os.environ["DATASETS_BUCKET_NAME"]
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        zip_path = os.path.join(tmp_dir, "dataset.zip")
+        download_to_file(hosted_url, zip_path)
+        dataset.zipped_size_bytes = os.path.getsize(zip_path)
+
+        with zipfile.ZipFile(zip_path, "r") as zip_ref:
+            zip_ref.extractall(tmp_dir)
+
+        dataset.unzipped_size_bytes = sum(
+            os.path.getsize(os.path.join(root, f))
+            for root, _, files in os.walk(tmp_dir)
+            for f in files
+        )
+
+        storage_client = storage.Client()
+        bucket = storage_client.bucket(bucket_name)
+        gtfs_files = []
+
+        for root, _, files in os.walk(tmp_dir):
+            for file_name in files:
+                file_path = os.path.join(root, file_name)
+                blob_path = f"{'-'.join(stable_id.split('-')[:2])}/{stable_id}/extracted/{file_name}"
+                blob = bucket.blob(blob_path)
+                blob.upload_from_filename(file_path)
+                blob.make_public()
+
+                with open(file_path, "rb") as f:
+                    sha256_hash = hashlib.sha256(f.read()).hexdigest()
+
+                gtfs_files.append(
+                    Gtfsfile(
+                        id=str(uuid.uuid4()),
+                        file_name=file_name,
+                        file_size_bytes=os.path.getsize(file_path),
+                        hash=sha256_hash,
+                        hosted_url=blob.public_url,
+                    )
+                )
+
+        dataset.gtfsfiles = gtfs_files
+        extracted_data = {
+            "zipped_size_bytes": dataset.zipped_size_bytes,
+            "unzipped_size_bytes": dataset.unzipped_size_bytes,
+            "file_count": len(gtfs_files),
+            "files": [
+                {
+                    "file_name": file.file_name,
+                    "file_size_bytes": file.file_size_bytes,
+                    "hash": file.hash,
+                    "hosted_url": file.hosted_url,
+                }
+                for file in gtfs_files
+            ],
+        }
+        logging.info(extracted_data)
+
+
+@with_db_session
+def rebuild_missing_dataset_files(
+    db_session,
+    dry_run: bool = True,
+    after_date: str = None,
+    latest_only: bool = True,
+) -> dict:
+    """
+    Processes GTFS datasets missing extracted files and updates database.
+
+    Args:
+        db_session: SQLAlchemy DB session.
+        dry_run (bool): If True, only logs how many would be processed.
+        after_date (str): Only consider datasets downloaded after this ISO date.
+        latest_only (bool): Whether to include only latest datasets.
+
+    Returns:
+        dict: Result summary.
+    """
+    datasets = get_datasets_with_missing_files_query(
+        db_session, after_date=after_date, latest_only=latest_only
+    )
+
+    if dry_run:
+        total = datasets.count()
+        logging.info(
+            "Dry run mode: %d datasets found with missing files. After date filter: %s",
+            total,
+            after_date,
+        )
+        return {
+            "message": f"Dry run: {total} datasets with missing files found.",
+            "total_processed": total,
+        }
+
+    total_processed = 0
+    count = 0
+    batch_count = 5
+    logging.info("Starting to process datasets with missing files...")
+
+    for dataset in datasets.all():
+        process_dataset(dataset)
+        count += 1
+        total_processed += 1
+
+        if count % batch_count == 0:
+            db_session.commit()
+            logging.info(
+                "Committed %d datasets. Total processed: %d",
+                batch_count,
+                total_processed,
+            )
+
+    db_session.commit()
+    logging.info("All datasets processed and committed. Total: %d", total_processed)
+
+    result = {
+        "message": "Rebuild missing dataset files completed.",
+        "total_processed": total_processed,
+        "params": {
+            "dry_run": dry_run,
+            "after_date": after_date,
+            "latest_only": latest_only,
+            "datasets_bucket_name": os.environ.get("DATASETS_BUCKET_NAME"),
+        },
+    }
+    logging.info("Task summary: %s", result)
+    return result

infra/functions-python/main.tf

@@ -1222,6 +1222,7 @@ resource "google_cloudfunctions2_function" "tasks_executor" {
       PROJECT_ID  = var.project_id
       ENV = var.environment
       PUBSUB_TOPIC_NAME = "rebuild-bounding-boxes-topic"
+      DATASETS_BUCKET_NAME = "${var.datasets_bucket_name}-${var.environment}"
     }
     available_memory                 = local.function_tasks_executor_config.memory
     timeout_seconds                  = local.function_tasks_executor_config.timeout

liquibase/changelog.xml

@@ -65,4 +65,5 @@
     <include file="changes/feat_1195.sql" relativeToChangelogFile="true"/>
     <include file="changes/feat_1265_cascade_delete.sql" relativeToChangelogFile="true"/>
     <include file="changes/feat_1259.sql" relativeToChangelogFile="true"/>
+    <include file="changes/feat_1260.sql" relativeToChangelogFile="true"/>
 </databaseChangeLog>

liquibase/changes/feat_1260.sql

@@ -0,0 +1,5 @@
+ALTER TABLE GtfsFile DROP COLUMN IF EXISTS hash;
+ALTER TABLE GtfsFile DROP COLUMN IF EXISTS hosted_url;
+ALTER TABLE GtfsFile ADD COLUMN hash varchar(255);
+ALTER TABLE GtfsFile ADD COLUMN hosted_url varchar(255);
+