Snapshot YouTube Stats into Warehouse-Ready JSONL

id: youtube-stats-to-warehouse
namespace: company.team
description: |
  Snapshot YouTube video statistics daily, reshape the per-video output
  into a timestamped JSONL file with Python, and stage it in Kestra's
  internal storage as a warehouse-ready extract.

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Capture one snapshot per day so the warehouse builds a view-count
      time series.
    cron: "0 5 * * *"
    disabled: true

inputs:
  - id: video_ids
    type: ARRAY
    itemType: STRING
    defaults:
      - dQw4w9WgXcQ
      - 9bZkp7q19f0
      - kJQP7kiw5Fk
    description: YouTube video IDs to snapshot. Replace the defaults with the
      catalog your analytics models track.

tasks:
  - id: fetch_stats
    type: io.kestra.plugin.youtube.VideoStats
    description: Fetch statistics and snippet metadata for every tracked video in
      one call, exposing aggregate totals and a per-video array as task outputs.
    accessToken: "{{ secret('YOUTUBE_ACCESS_TOKEN') }}"
    applicationName: kestra-content-analytics
    videoIds: "{{ inputs.video_ids }}"
    includeSnippet: true
    maxResults: 50

  - id: reshape_to_jsonl
    type: io.kestra.plugin.scripts.python.Script
    description: Flatten the per-video statistics array into snake_case JSONL rows
      stamped with the capture time, producing one warehouse-ready file per run
      in internal storage.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    inputFiles:
      videos.json: "{{ outputs.fetch_stats.videos | toJson }}"
    outputFiles:
      - videos.jsonl
    script: |
      import json
      from datetime import datetime, timezone

      with open("videos.json") as f:
          videos = json.load(f)

      captured_at = datetime.now(timezone.utc).isoformat()

      with open("videos.jsonl", "w") as out:
          for v in videos:
              row = {
                  "video_id": v.get("videoId"),
                  "title": v.get("title"),
                  "channel_title": v.get("channelTitle"),
                  "published_at": v.get("publishedAt"),
                  "view_count": v.get("viewCount"),
                  "like_count": v.get("likeCount"),
                  "comment_count": v.get("commentCount"),
                  "favorite_count": v.get("favoriteCount"),
                  "captured_at": captured_at,
              }
              out.write(json.dumps(row) + "\n")

      print(f"Wrote {len(videos)} rows to videos.jsonl")

  - id: notify
    type: io.kestra.plugin.discord.DiscordIncomingWebhook
    description: Confirm the snapshot in Discord with aggregate scalars; the JSONL
      file stays attached to the execution for the warehouse load.
    url: "{{ secret('DISCORD_WEBHOOK_URL') }}"
    payload: |
      {
        "username": "YouTube Warehouse Loader",
        "content": "YouTube stats snapshot ready. {{ outputs.fetch_stats.totalVideos }} videos captured with {{ outputs.fetch_stats.totalViews }} total views and {{ outputs.fetch_stats.totalLikes }} likes. The JSONL extract is attached to execution {{ execution.id }}."
      }

errors:
  - id: alert_failure
    type: io.kestra.plugin.discord.DiscordIncomingWebhook
    description: Alert Discord when the snapshot fails so the warehouse never
      accumulates silent gaps in the daily history.
    url: "{{ secret('DISCORD_WEBHOOK_URL') }}"
    payload: |
      {
        "username": "YouTube Warehouse Loader",
        "content": "YouTube stats snapshot FAILED in flow {{ flow.id }} (execution {{ execution.id }}). Check the task logs for an expired access token or a reshape error."
      }