Extract, Transform, and Store HackerNews Top Stories with AI-Generated Daily Digest

id: hackernews-etl-to-postgres
namespace: company.team

inputs:
  - id: story_type
    type: SELECT
    displayName: Story type
    description: Which HackerNews feed to fetch stories from
    values:
      - topstories
      - newstories
      - beststories
    defaults: "topstories"
  - id: max_stories
    type: INT
    displayName: Number of stories
    description: How many stories to fetch and process (max 30)
    defaults: 10

tasks:
  - id: fetch_story_ids
    type: io.kestra.plugin.core.http.Request
    uri: "https://hacker-news.firebaseio.com/v0/{{ inputs.story_type }}.json"
    method: GET

  - id: fetch_stories
    type: io.kestra.plugin.core.flow.ForEach
    values: "{{ outputs.fetch_story_ids.body | json | jq('.[0:' + inputs.max_stories
      | string + ']') | first }}"
    concurrencyLimit: 5
    tasks:
      - id: fetch_story_detail
        type: io.kestra.plugin.core.http.Request
        uri: "https://hacker-news.firebaseio.com/v0/item/{{ taskrun.value }}.json"
        method: GET

  - id: summarize_stories
    type: io.kestra.plugin.scripts.python.Script
    inputFiles:
      stories_raw.json: "{{ outputs.fetch_stories | json }}"
    dependencies:
      - requests
    script: |
      import json
      from kestra import Kestra

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

      stories = []
      for story_id, story_outputs in raw.items():
          body = story_outputs.get("body", "{}")
          if isinstance(body, str):
              try:
                  story = json.loads(body)
              except:
                  continue
          else:
              story = body

          if not story or story.get("type") not in ("story", "job", "ask", "show"):
              continue

          stories.append({
              "id": story.get("id"),
              "title": story.get("title", ""),
              "url": story.get("url", ""),
              "score": story.get("score", 0),
              "author": story.get("by", ""),
              "comments": story.get("descendants", 0),
              "story_type": story.get("type", ""),
              "posted_at": story.get("time", 0)
          })

      # Sort by score descending
      stories.sort(key=lambda x: x["score"], reverse=True)
      Kestra.outputs({"stories": stories, "count": len(stories)})

  - id: summarize_with_ai
    type: io.kestra.plugin.openai.ChatCompletion
    apiKey: "{{ secret('OPENAI_API_KEY') }}"
    model: gpt-4o-mini
    maxTokens: 800
    messages:
      - role: system
        content: |
          You are a tech analyst. Given a list of trending HackerNews stories, write a brief digest
          (3-5 bullet points) highlighting the most interesting themes and trends. Be concise and insightful.
          Focus on what these stories collectively reveal about the current state of tech.
      - role: user
        content: |
          HackerNews {{ inputs.story_type }} today:
          {{ outputs.summarize_stories.vars.stories | json }}

  - id: store_to_postgres
    type: io.kestra.plugin.scripts.python.Script
    inputFiles:
      stories.json: "{{ outputs.summarize_stories.vars.stories | json }}"
    script: |
      import json
      import psycopg2
      import os

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

      conn = psycopg2.connect(
          host=os.getenv("POSTGRES_HOST", ""),
          dbname=os.getenv("POSTGRES_DB", ""),
          user=os.getenv("POSTGRES_USER_VAL", ""),
          password=os.getenv("POSTGRES_PASS_VAL", ""),
          port=5432
      )
      cur = conn.cursor()

      cur.execute("""
          CREATE TABLE IF NOT EXISTS hackernews_stories (
              id BIGINT PRIMARY KEY,
              title TEXT,
              url TEXT,
              score INTEGER,
              author TEXT,
              comments INTEGER,
              story_type TEXT,
              posted_at BIGINT,
              fetched_at TIMESTAMP DEFAULT NOW()
          )
      """)

      for story in stories:
          cur.execute("""
              INSERT INTO hackernews_stories (id, title, url, score, author, comments, story_type, posted_at)
              VALUES (%(id)s, %(title)s, %(url)s, %(score)s, %(author)s, %(comments)s, %(story_type)s, %(posted_at)s)
              ON CONFLICT (id) DO UPDATE SET
                  score = EXCLUDED.score,
                  comments = EXCLUDED.comments,
                  fetched_at = NOW()
          """, story)

      conn.commit()
      cur.close()
      conn.close()
      print(f"Stored {len(stories)} stories to PostgreSQL")
    env:
      POSTGRES_HOST: "{{ secret('POSTGRES_HOST') }}"
      POSTGRES_DB: "{{ secret('POSTGRES_DB') }}"
      POSTGRES_USER_VAL: "{{ secret('POSTGRES_USER') }}"
      POSTGRES_PASS_VAL: "{{ secret('POSTGRES_PASSWORD') }}"
    dependencies:
      - psycopg2-binary

  - id: post_digest_to_slack
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      :newspaper: *HackerNews {{ inputs.story_type | replace('stories', '') | title }} Digest*
      _{{ outputs.summarize_stories.vars.count }} stories processed_

      {{ outputs.summarize_with_ai.choices[0].message.content }}

triggers:
  - id: daily_digest
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 8 * * *"
    description: Fetch and process HackerNews stories every day at 08:00 UTC