Daily arXiv AI digest with keyword ranking

id: ai-arxiv-daily-digest
namespace: company.team
description: Daily Slack digest of the most relevant new AI papers on arXiv

inputs:
  - id: categories
    type: ARRAY
    itemType: STRING
    displayName: arXiv categories
    description: arXiv categories to scan each morning (e.g. cs.AI, cs.LG, cs.CL)
    defaults:
      - cs.AI
      - cs.LG
      - cs.CL

  - id: keywords
    type: ARRAY
    itemType: STRING
    displayName: Ranking keywords
    description: Keywords used to score each paper's relevance; title matches count triple
    defaults:
      - llm
      - agent
      - rag
      - transformer
      - reasoning
      - diffusion
      - alignment
      - benchmark
      - multimodal
      - fine-tune

  - id: max_results
    type: INT
    displayName: Papers to scan
    description: How many of the newest submissions to pull from the arXiv API per run
    defaults: 50

  - id: top_n
    type: INT
    displayName: Digest size
    description: How many top-ranked papers appear in the Slack digest
    defaults: 10

triggers:
  - id: daily_morning
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 8 * * *"
    description: Run every morning at 08:00 UTC

tasks:
  - id: fetch_papers
    type: io.kestra.plugin.scripts.python.Script
    description: Pull the newest submissions from the arXiv API (Atom XML, no key needed)
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    env:
      MAX_RESULTS: "{{ inputs.max_results }}"
    inputFiles:
      categories.json: "{{ inputs.categories | toJson }}"
    outputFiles:
      - papers.json
    script: |
      import json, os, urllib.request, urllib.parse, xml.etree.ElementTree as ET

      categories = json.load(open('categories.json'))
      max_results = int(os.environ['MAX_RESULTS'])

      query = ' OR '.join(f'cat:{c}' for c in categories)
      params = urllib.parse.urlencode({
          'search_query': query,
          'start': 0,
          'max_results': max_results,
          'sortBy': 'submittedDate',
          'sortOrder': 'descending',
      })
      req = urllib.request.Request(
          'https://export.arxiv.org/api/query?' + params,
          headers={'User-Agent': 'kestra-arxiv-digest/1.0'},
      )
      with urllib.request.urlopen(req, timeout=60) as resp:
          data = resp.read()

      ns = {'a': 'http://www.w3.org/2005/Atom'}
      papers = []
      for entry in ET.fromstring(data).findall('a:entry', ns):
          papers.append({
              'id': entry.find('a:id', ns).text.strip(),
              'title': ' '.join(entry.find('a:title', ns).text.split()),
              'authors': [a.find('a:name', ns).text for a in entry.findall('a:author', ns)],
              'abstract': ' '.join(entry.find('a:summary', ns).text.split()),
              'categories': [c.get('term') for c in entry.findall('a:category', ns)],
              'published': entry.find('a:published', ns).text,
          })

      with open('papers.json', 'w') as f:
          json.dump(papers, f)
      print(f'Fetched {len(papers)} papers from arXiv')

  - id: process
    type: io.kestra.plugin.core.flow.Parallel
    description: Score papers and compute stats concurrently
    tasks:
      - id: rank_relevance
        type: io.kestra.plugin.scripts.python.Script
        description: Score each paper by keyword hits and keep the top N
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        env:
          TOP_N: "{{ inputs.top_n }}"
        inputFiles:
          papers.json: "{{ outputs.fetch_papers.outputFiles['papers.json'] }}"
          keywords.json: "{{ inputs.keywords | toJson }}"
        outputFiles:
          - ranked.json
        script: |
          import json, os, re

          papers = json.load(open('papers.json'))
          keywords = [k.lower() for k in json.load(open('keywords.json'))]
          top_n = int(os.environ['TOP_N'])

          def score(p):
              title, abstract = p['title'].lower(), p['abstract'].lower()
              s = 0
              for kw in keywords:
                  pattern = r'\b' + re.escape(kw) + r's?\b'
                  s += 3 * len(re.findall(pattern, title))
                  s += 1 * len(re.findall(pattern, abstract))
              return s

          ranked = sorted(papers, key=score, reverse=True)[:top_n]
          for p in ranked:
              p['score'] = score(p)
          with open('ranked.json', 'w') as f:
              json.dump(ranked, f, indent=2)
          print(f'Ranked top {len(ranked)} papers')

      - id: compute_stats
        type: io.kestra.plugin.scripts.python.Script
        description: Aggregate per-category counts for the digest header
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        dependencies:
          - kestra
        inputFiles:
          papers.json: "{{ outputs.fetch_papers.outputFiles['papers.json'] }}"
        script: |
          import json
          from collections import Counter
          from kestra import Kestra

          papers = json.load(open('papers.json'))
          cats = Counter(c for p in papers for c in p['categories'])
          Kestra.outputs({
              'total': len(papers),
              'top_categories': dict(cats.most_common(5)),
          })

  - id: format_digest
    type: io.kestra.plugin.scripts.python.Script
    description: Render the ranked papers as a Slack mrkdwn digest
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    dependencies:
      - kestra
    inputFiles:
      ranked.json: "{{ outputs.rank_relevance.outputFiles['ranked.json'] }}"
    script: |
      import json
      from kestra import Kestra

      ranked = json.load(open('ranked.json'))
      lines = []
      for i, p in enumerate(ranked, 1):
          authors = ', '.join(p['authors'][:3])
          if len(p['authors']) > 3:
              authors += ' et al.'
          lines.append(f'*{i}. {p["title"]}*')
          lines.append(f'_{authors}_ · {", ".join(p["categories"])} · relevance {p["score"]}')
          lines.append(f'<{p["id"]}|Read on arXiv>')
          lines.append(f'>{p["abstract"][:280]}…')
          lines.append('')
      Kestra.outputs({'digest': '\n'.join(lines).strip()})

  - id: send_slack
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post the morning digest to Slack
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      :newspaper: *Daily arXiv AI Digest*, top {{ inputs.top_n }} of {{ outputs.compute_stats.values.total }} new papers
      {{ outputs.format_digest.values.digest }}