Validate a transit agency's GTFS feed whenever it changes

id: gtfs-feed-validation-monitor
namespace: company.team

inputs:
  - id: feed_url
    type: STRING
    displayName: GTFS feed URL
    description: Public URL of a static GTFS zip. The default is the Redding Area
      Bus Authority feed, so the flow runs with no setup.
    defaults: https://rabagtfs.z5.web.core.windows.net/rabagtfs.zip

  - id: force_validation
    type: BOOL
    displayName: Validate even if unchanged
    description: Run the validator even when the feed is identical to the one
      checked last time.
    defaults: false

  - id: fail_on_errors
    type: BOOL
    displayName: Fail on validation errors
    description: End the run as FAILED when the validator reports at least one ERROR notice.
    defaults: false

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack
    description: Post the result to Slack when the feed has errors. Needs the
      SLACK_WEBHOOK_URL secret.
    defaults: false

triggers:
  - id: daily_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check the feed every morning. Agencies usually publish a new feed
      without notice.
    cron: "0 6 * * *"

tasks:
  - id: download
    type: io.kestra.plugin.core.http.Download
    description: Download the current feed.
    uri: "{{ inputs.feed_url }}"

  - id: fingerprint
    type: io.kestra.plugin.scripts.python.Script
    description: Hash the downloaded zip, so an unchanged feed is not validated twice.
    dependencies:
      - kestra
    inputFiles:
      feed.zip: "{{ outputs.download.uri }}"
    script: |
      import hashlib
      from kestra import Kestra

      with open("feed.zip", "rb") as f:
          Kestra.outputs({"sha256": hashlib.sha256(f.read()).hexdigest()})

  - id: previous_hash
    type: io.kestra.plugin.core.kv.Get
    description: Read the hash of the last feed that was validated. Empty on the first run.
    key: "{{ flow.id }}_last_sha256"
    errorOnMissing: false

  - id: feed_changed
    type: io.kestra.plugin.core.flow.If
    description: Validate only a new feed, unless force_validation is set.
    condition: "{{ inputs.force_validation or (outputs.previous_hash.value ?? '') !=
      outputs.fingerprint.vars.sha256 }}"
    then:
      - id: validate
        type: io.kestra.plugin.scripts.shell.Commands
        description: >
          Run the canonical MobilityData GTFS validator on the feed. It writes
          report.json and report.html. --skip_validator_update stops it from
          checking GitHub for a newer release on every run.
        containerImage: ghcr.io/mobilitydata/gtfs-validator:8.0.1
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          entryPoint: []
        inputFiles:
          feed.zip: "{{ outputs.download.uri }}"
        commands:
          - java -jar /gtfs-validator-cli.jar --input feed.zip --output_base
            report --skip_validator_update 2>&1
        outputFiles:
          - report/report.json
          - report/report.html

      - id: summarize
        type: io.kestra.plugin.scripts.python.Script
        description: Count the notices by severity and keep the error codes with the
          most notices.
        dependencies:
          - kestra
        inputFiles:
          report.json: "{{ outputs.validate.outputFiles['report/report.json'] }}"
        script: |
          import json
          from kestra import Kestra

          with open("report.json") as f:
              notices = json.load(f)["notices"]

          counts = {"ERROR": 0, "WARNING": 0, "INFO": 0}
          for notice in notices:
              counts[notice["severity"]] = counts.get(notice["severity"], 0) + notice["totalNotices"]

          errors = sorted(
              (n for n in notices if n["severity"] == "ERROR"),
              key=lambda n: n["totalNotices"],
              reverse=True,
          )
          top_errors = ", ".join(f"{n['code']} ({n['totalNotices']})" for n in errors[:5])

          Kestra.outputs({
              "errors": counts["ERROR"],
              "warnings": counts["WARNING"],
              "infos": counts["INFO"],
              "top_errors": top_errors or "none",
          })

      - id: remember_hash
        type: io.kestra.plugin.core.kv.Set
        description: Store the hash, so the next run skips this feed unless it changes.
        key: "{{ flow.id }}_last_sha256"
        value: "{{ outputs.fingerprint.vars.sha256 }}"

      - id: errors_found
        type: io.kestra.plugin.core.flow.If
        description: Alert, and optionally fail, when the feed has ERROR notices.
        condition: "{{ outputs.summarize.vars.errors > 0 }}"
        then:
          - id: notify_errors
            type: io.kestra.plugin.core.flow.If
            condition: "{{ inputs.notify_slack }}"
            then:
              - id: slack_errors
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                payload: |
                  {
                    "text": "GTFS feed {{ inputs.feed_url }} has {{ outputs.summarize.vars.errors }} validation errors and {{ outputs.summarize.vars.warnings }} warnings. Top errors: {{ outputs.summarize.vars.top_errors }}. Execution {{ execution.id }}."
                  }

          - id: fail_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ inputs.fail_on_errors }}"
            then:
              - id: fail_run
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "GTFS feed has {{ outputs.summarize.vars.errors }} validation
                  errors: {{ outputs.summarize.vars.top_errors }}"
        else:
          - id: log_clean
            type: io.kestra.plugin.core.log.Log
            message: "No validation errors. {{ outputs.summarize.vars.warnings }} warnings
              and {{ outputs.summarize.vars.infos }} info notices, see
              report.html."
    else:
      - id: log_unchanged
        type: io.kestra.plugin.core.log.Log
        message: "Feed unchanged since the last validated run (sha256 {{
          outputs.fingerprint.vars.sha256 }}). Skipping validation."