Send a daily Odoo sales digest to Slack

id: odoo-daily-sales-digest
namespace: company.sales

labels:
  team: sales
  domain: revenue-reporting

variables:
  window_date: "{{ (trigger.date ?? execution.startDate) | dateAdd(-1, 'DAYS') |
    date('yyyy-MM-dd', timeZone='Europe/Paris') }}"

pluginDefaults:
  - type: io.kestra.plugin.odoo.Query
    values:
      url: "{{ secret('ODOO_URL') }}"
      db: "{{ secret('ODOO_DB') }}"
      username: "{{ secret('ODOO_USERNAME') }}"
      password: "{{ secret('ODOO_PASSWORD') }}"

tasks:
  - id: query_orders
    type: io.kestra.plugin.odoo.Query
    description: Read the business day's confirmed sales orders from Odoo and store
      them as ION.
    model: sale.order
    operation: SEARCH_READ
    filters:
      - [ "state", "in", [ "sale", "done" ] ]
      - [ "date_order", ">=", "{{ vars.window_date }} 00:00:00" ]
      - [ "date_order", "<=", "{{ vars.window_date }} 23:59:59" ]
    fields:
      - name
      - partner_id
      - date_order
      - amount_total
      - currency_id
      - state
    limit: 100000
    fetchType: STORE
    retry:
      type: constant
      interval: PT30S
      maxAttempt: 4

  - id: to_jsonl
    type: io.kestra.plugin.serdes.json.IonToJson
    description: Convert the ION query result into newline-delimited JSON for pandas.
    from: "{{ outputs.query_orders.uri }}"

  - id: build_digest
    type: io.kestra.plugin.scripts.python.Script
    description: Summarize the sales orders per currency and build a JSON-safe Slack
      Block Kit payload.
    containerImage: python:3.13-slim
    dependencies:
      - pandas==2.2.2
      - kestra
    inputFiles:
      orders.jsonl: "{{ outputs.to_jsonl.uri }}"
    env:
      WINDOW_DATE: "{{ vars.window_date }}"
    script: |
      import json
      import os
      import pandas as pd
      from kestra import Kestra

      window_date = os.environ["WINDOW_DATE"]

      try:
          df = pd.read_json("orders.jsonl", lines=True)
      except ValueError:
          df = pd.DataFrame()

      def currency_label(value):
          if isinstance(value, list) and len(value) > 1:
              return str(value[1])
          return str(value)

      def customer_label(value):
          if isinstance(value, list) and len(value) > 1:
              return str(value[1])
          return str(value)

      fields = []
      order_count = 0
      if df.empty:
          fields.append({"type": "mrkdwn", "text": "No confirmed orders for this day."})
      else:
          order_count = int(len(df))
          df["currency"] = df["currency_id"].apply(currency_label)
          df["customer"] = df["partner_id"].apply(customer_label)

          by_currency = (
              df.groupby("currency")["amount_total"].sum().sort_values(ascending=False)
          )
          revenue_lines = "\n".join(
              f"{cur}: {round(float(amount), 2)}" for cur, amount in by_currency.items()
          )
          fields.append({"type": "mrkdwn", "text": f"*Orders:*\n{order_count}"})
          fields.append({"type": "mrkdwn", "text": f"*Revenue by currency:*\n{revenue_lines}"})

          top_customer = (
              df.groupby(["customer", "currency"])["amount_total"]
              .sum()
              .sort_values(ascending=False)
          )
          (top_name, top_cur) = top_customer.index[0]
          top_amount = round(float(top_customer.iloc[0]), 2)
          fields.append(
              {"type": "mrkdwn", "text": f"*Top customer:*\n{top_name} ({top_amount} {top_cur})"}
          )

      payload = {
          "blocks": [
              {
                  "type": "header",
                  "text": {"type": "plain_text", "text": f"Daily sales digest for {window_date}"},
              },
              {"type": "section", "fields": fields},
              {
                  "type": "context",
                  "elements": [{"type": "mrkdwn", "text": "Source: Odoo sale.order via Kestra"}],
              },
          ]
      }

      Kestra.outputs({"order_count": order_count, "payload": json.dumps(payload)})

  - id: send_digest
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post the daily sales digest to Slack using the JSON-safe payload
      built in Python.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: "{{ outputs.build_digest.vars.payload }}"
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    messageText: ":rotating_light: {{ flow.namespace }}.{{ flow.id }} could not
      build the {{ vars.window_date }} sales digest (execution {{ execution.id
      }})."
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3

triggers:
  - id: every_morning
    type: io.kestra.plugin.core.trigger.Schedule
    description: Runs every morning at 07:00 Europe/Paris and reports the previous
      business day.
    cron: "0 7 * * *"
    timezone: Europe/Paris