Arcmira: YouTube Transcript Search to CSV

id: arcmira-transcript-source-sheet
namespace: company.team

inputs:
  - id: query
    type: STRING
    defaults: creator economy
    description: One topic or phrase to find in indexed transcripts.
  - id: published_after
    type: DATE
    required: false
    description: Inclusive UTC publication date. Omit to start 60 days before execution.
  - id: published_before
    type: DATE
    required: false
    description: Exclusive UTC publication date. Omit to end 31 days before execution.
  - id: limit
    type: INT
    defaults: 3
    description: Number of passages, from 1 to 20. Results past the first five use credits.

tasks:
  - id: build_query
    type: io.kestra.plugin.scripts.python.Script
    description: Validate the requested window and encode the query without
      interpolating user text into code.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.13.15-slim
    env:
      QUERY: "{{ inputs.query }}"
      AFTER: "{{ (inputs.published_after ?? ((trigger.date ?? execution.startDate) |
        dateAdd(-60, 'DAYS'))) | date('yyyy-MM-dd', timeZone='UTC') }}"
      BEFORE: "{{ (inputs.published_before ?? ((trigger.date ?? execution.startDate) |
        dateAdd(-31, 'DAYS'))) | date('yyyy-MM-dd', timeZone='UTC') }}"
      LIMIT: "{{ inputs.limit }}"
    outputFiles:
      - request.json
    script: |
      import json
      import os
      from datetime import date
      from pathlib import Path
      from urllib.parse import urlencode

      query = os.environ["QUERY"].strip()
      after = date.fromisoformat(os.environ["AFTER"])
      before = date.fromisoformat(os.environ["BEFORE"])
      limit = int(os.environ["LIMIT"])
      if len(query) < 2 or after >= before or not 1 <= limit <= 20:
          raise ValueError("Use a topic of at least two characters, after < before, and limit 1..20.")
      params = {"q": query, "after": after.isoformat(), "before": before.isoformat(), "limit": limit}
      request = {"uri": "https://api.arcmira.com/v1/search?" + urlencode(params), "params": params}
      Path("request.json").write_text(json.dumps(request, indent=2), encoding="utf-8")

  - id: search
    type: io.kestra.plugin.core.http.Request
    description: Make one authenticated search request. HTTP errors fail the flow
      without retries or fallback sources.
    uri: "{{ fromJson(read(outputs.build_query.outputFiles['request.json'])).uri }}"
    method: GET
    headers:
      Authorization: "Bearer {{ secret('ARCMIRA_API_KEY') }}"
      Accept: application/json

  - id: has_matches
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.search.body).chunks | length > 0 }}"
    then:
      - id: source_sheet
        type: io.kestra.plugin.scripts.python.Script
        description: Export passages for human review and preserve the response with its
          search limitations.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.13.15-slim
        inputFiles:
          response.json: "{{ outputs.search.body }}"
          request.json: "{{ outputs.build_query.outputFiles['request.json'] }}"
        outputFiles:
          - source-sheet.csv
          - response.json
          - request.json
        script: |
          import csv
          import json
          from pathlib import Path

          response = json.loads(Path("response.json").read_text(encoding="utf-8"))
          columns = ["video_id", "published_at", "start_seconds", "watch_url", "text"]

          def spreadsheet_cell(value):
              if isinstance(value, str) and value.lstrip().startswith(("=", "+", "-", "@")):
                  return "'" + value
              return value

          with open("source-sheet.csv", "w", newline="", encoding="utf-8") as stream:
              writer = csv.DictWriter(stream, fieldnames=columns)
              writer.writeheader()
              for chunk in response["chunks"]:
                  writer.writerow({key: spreadsheet_cell(chunk[key]) for key in columns})
    else:
      - id: no_matches
        type: io.kestra.plugin.core.storage.Write
        description: Retain the empty response and its index or access notes. No source
          sheet is created.
        content: "{{ outputs.search.body }}"
        extension: .json

triggers:
  - id: weekly_research
    type: io.kestra.plugin.core.trigger.Schedule
    cron: 0 9 * * 1
    timezone: UTC
    recoverMissedSchedules: NONE
    disabled: true