Scrapy Price Drop Watch and Re-Verification Gate

id: scrapy-price-drop-watch-gate
namespace: company.team
description: |
  Crawl a product listing page with a minimal Scrapy spider, find the lowest
  scraped price, and alert Slack when it falls to or below a threshold.
  auto_remediate re-scrapes once more to confirm the breach is not a
  transient glitch before trusting it; dry_run gates whether the Slack
  alert actually fires.

triggers:
  - id: daily_price_watch
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily price check. Shipped disabled; enable it once target_url and
      price_threshold are set for your real listing page and a manual run has
      been verified. Always fires in safe mode (auto_remediate false, dry_run
      true) regardless of this flow's own defaults.
    cron: "0 7 * * *"
    disabled: true
    inputs:
      auto_remediate: "false"
      dry_run: "true"

inputs:
  - id: target_url
    type: STRING
    displayName: "Listing Page URL"
    defaults: "https://books.toscrape.com"
    description: Product listing page to scrape. The spider reads
      article.product_pod entries on this one page only (CLOSESPIDER_PAGECOUNT
      1); it does not follow pagination links.

  - id: price_threshold
    type: FLOAT
    displayName: "Price Threshold"
    defaults: 20.0
    description: Alert when the lowest scraped price is at or below this value (same
      currency/units as shown on the page, symbol stripped).

  - id: auto_remediate
    type: SELECT
    displayName: "Auto Re-verify"
    defaults: "false"
    values:
      - "false"
      - "true"
    description: If "true" and a breach is found, re-scrape the page once more to
      confirm the low price before trusting it. If "false", the breach is
      reported from the first scrape alone.

  - id: dry_run
    type: SELECT
    displayName: "Dry Run"
    defaults: "true"
    values:
      - "true"
      - "false"
    description: If "true", a breach is only logged, never sent to Slack. Must be
      explicitly "false" for the Slack alert to actually fire.

tasks:
  - id: scrape_book_prices
    type: io.kestra.plugin.scrapy.CLI
    description: Minimal spider - extraction only, using the same
      article.product_pod / h3 a::attr(title) / p.price_color::text selectors
      shown in the official plugin documentation's own books.toscrape.com
      example. Price text (e.g. a currency-symbol-prefixed amount) is stripped
      to a bare float in the spider, since jq cannot coerce a symbol-prefixed
      string to a number; all threshold logic stays in Kestra. Output is a
      proper JSON array (.json extension), not JSON Lines, so it is directly
      usable with jq.
    inputFiles:
      book_prices_spider.py: |
        import scrapy

        class BookPricesSpider(scrapy.Spider):
            name = "book_prices"
            start_urls = ["{{ inputs.target_url }}"]
            custom_settings = {"CLOSESPIDER_PAGECOUNT": 1}

            def parse(self, response):
                for book in response.css("article.product_pod"):
                    title = book.css("h3 a::attr(title)").get()
                    price_text = (book.css("p.price_color::text").get() or "").strip()
                    price_clean = price_text.lstrip("£$€").strip()
                    yield {
                        "title": title,
                        "price": float(price_clean) if price_clean else None,
                    }
    commands:
      - scrapy runspider book_prices_spider.py -o books.json
    outputFiles:
      - books.json

  - id: evaluate_price_breach
    type: io.kestra.plugin.core.flow.If
    description: Breach means the lowest scraped price is at or below
      price_threshold. jq's min filters out any null price (a book whose
      price_color was missing or unparsable).
    condition: "{{ (read(outputs.scrape_book_prices.outputFiles['books.json']) |
      jq('[.[].price] | map(select(. != null)) | min') | first) <=
      inputs.price_threshold }}"
    then:
      - id: route_reverify
        type: io.kestra.plugin.core.flow.Switch
        description: auto_remediate controls only whether a confirmatory re-scrape runs
          - it does not by itself gate the Slack alert, which is controlled
          separately by dry_run below.
        value: "{{ inputs.auto_remediate }}"
        cases:
          "true":
            - id: reverify_book_prices
              type: io.kestra.plugin.scrapy.CLI
              description: Re-run the identical spider a second time to rule out a transient
                scrape glitch (a half-rendered page, a momentary markup change)
                before trusting the breach.
              inputFiles:
                book_prices_spider.py: |
                  import scrapy

                  class BookPricesSpider(scrapy.Spider):
                      name = "book_prices"
                      start_urls = ["{{ inputs.target_url }}"]
                      custom_settings = {"CLOSESPIDER_PAGECOUNT": 1}

                      def parse(self, response):
                          for book in response.css("article.product_pod"):
                              title = book.css("h3 a::attr(title)").get()
                              price_text = (book.css("p.price_color::text").get() or "").strip()
                              price_clean = price_text.lstrip("£$€").strip()
                              yield {
                                  "title": title,
                                  "price": float(price_clean) if price_clean else None,
                              }
              commands:
                - scrapy runspider book_prices_spider.py -o books_reverify.json
              outputFiles:
                - books_reverify.json
            - id: log_reverify_result
              type: io.kestra.plugin.core.log.Log
              message: "{{
                (read(outputs.reverify_book_prices.outputFiles['books_reverify.json\
                ']) | jq('[.[].price] | map(select(. != null)) | min') | first)
                <= inputs.price_threshold ? 'Re-verification CONFIRMED the price
                breach on a second scrape.' : 'Re-verification could not
                reproduce the breach on a second scrape - likely transient.' }}"
          "false":
            - id: log_reverify_skipped
              type: io.kestra.plugin.core.log.Log
              message: "auto_remediate is false: reporting the breach from the first scrape
                alone, no confirmatory re-scrape ran."
      - id: route_alert
        type: io.kestra.plugin.core.flow.If
        description: dry_run is the independent gate on the Slack alert itself,
          regardless of whether auto_remediate re-verified the breach above.
        condition: "{{ inputs.dry_run == \"false\" }}"
        then:
          - id: alert_price_breach
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Name the lowest price found and the threshold it breached.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "Price drop watch: lowest scraped price on {{ inputs.target_url }} is {{ (read(outputs.scrape_book_prices.outputFiles['books.json']) | jq('[.[].price] | map(select(. != null)) | min') | first) }}, at or below the threshold of {{ inputs.price_threshold }}. Re-verified: {{ outputs.reverify_book_prices is defined }}. Execution {{ execution.id }}."
              }
        else:
          - id: log_dry_run_alert
            type: io.kestra.plugin.core.log.Log
            message: "DRY RUN: would alert Slack about a price breach on {{
              inputs.target_url }} (dry_run is true). No notification sent."
    else:
      - id: log_clean_run
        type: io.kestra.plugin.core.log.Log
        message: "No breach: lowest scraped price on {{ inputs.target_url }} is above
          the {{ inputs.price_threshold }} threshold."

  - id: log_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: Final status line for every run, independent of which branch fired,
      so the execution history always shows the headline numbers.
    message: "Price watch complete for {{ inputs.target_url }}. Lowest price: {{
      (read(outputs.scrape_book_prices.outputFiles['books.json']) |
      jq('[.[].price] | map(select(. != null)) | min') | first) }}. Threshold:
      {{ inputs.price_threshold }}. auto_remediate: {{ inputs.auto_remediate }}.
      dry_run: {{ inputs.dry_run }}."

errors:
  - id: alert_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the flow itself fails outside the handled branches, for
      example an unreachable page or a markup change that breaks the selectors.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Price drop watch gate FAILED in execution {{ execution.id }} for {{ inputs.target_url }}. The scrape or price comparison did not complete; investigate before trusting the last known price."
      }