Bulk import from CSV

[some-source.csv]
        |
        v
[parse to a list of URLs]
        |
        v
[for each: POST /api/v1/ingest, with rate-limit handling]
        |
        v
[About 5 minutes later: 200 items in your Graniite library]

export GRANIITE_TOKEN="lev_…"

url,title,note
https://www.youtube.com/watch?v=...,Karpathy on training,backfill
https://example.com/article/...,Long-form essay on RLHF,
https://podcast.example.com/ep/42,,

import { readFile } from 'node:fs/promises';

const TOKEN = process.env.GRANIITE_TOKEN;
if (!TOKEN) throw new Error('Set GRANIITE_TOKEN env var');

const SOURCE = process.argv[2] ?? 'urls.csv';
const CONCURRENCY = 2;          // ingest writes are 60/hr per token
const PAUSE_BETWEEN_MS = 1000;  // ~1 req/sec stays well under the cap

// Parse the CSV. Naive single-column read. For complex CSVs use a
// real parser (csv-parse, papaparse).
const rows = (await readFile(SOURCE, 'utf8'))
  .split('\n')
  .map((line) => line.split(',')[0].trim().replace(/^"|"$/g, ''))
  .filter((url) => url && url !== 'url' && url.startsWith('http'));

console.log(`Importing ${rows.length} URLs from ${SOURCE}`);

let done = 0;
let failed = 0;
const failures = [];

async function ingest(url) {
  const res = await fetch('https://graniite.co/api/v1/ingest', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${TOKEN}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      kind: 'url',
      url,
      in_kb: true,
      auto_transform: false,
    }),
  });

  if (res.status === 429) {
    // Rate-limited. Sleep, then bubble up so we retry.
    const body = await res.json().catch(() => ({}));
    const wait = (body.retry_after_sec ?? 60) * 1000;
    console.log(`  rate-limited; sleeping ${wait / 1000}s`);
    await new Promise((r) => setTimeout(r, wait));
    return ingest(url); // recurse. Guaranteed to terminate because
                        // each retry sleeps the full retry_after window.
  }

  if (!res.ok) {
    const body = await res.json().catch(() => ({}));
    throw new Error(`${res.status}: ${body.error ?? 'unknown'} (${url})`);
  }
  return res.json();
}

// Process N at a time. fetch is async; this just bounds the
// in-flight set.
const queue = rows.slice();
async function worker() {
  while (queue.length > 0) {
    const url = queue.shift();
    try {
      const result = await ingest(url);
      done++;
      console.log(`  ok ${done}/${rows.length}  ${result.status.padEnd(13)} ${url.slice(0, 60)}`);
    } catch (e) {
      failed++;
      failures.push({ url, error: e.message });
      console.log(`  err ${url}: ${e.message}`);
    }
    await new Promise((r) => setTimeout(r, PAUSE_BETWEEN_MS));
  }
}

await Promise.all(Array.from({ length: CONCURRENCY }, worker));

console.log(`\nDone. Imported ${done}, failed ${failed}.`);
if (failures.length) {
  console.log('Failures:');
  for (const f of failures) console.log(`  ${f.url}\n    ${f.error}`);
}

node bulk-import.mjs urls.csv

Importing 132 URLs from urls.csv
  ok 1/132   ready         https://www.youtube.com/watch?v=...
  ok 2/132   transcribing  https://podcast.example.com/ep/42
  ok 3/132   ready         https://example.com/article/long-form-essay
  ...

import os, csv, time, sys, json
import requests

TOKEN = os.environ['GRANIITE_TOKEN']
SOURCE = sys.argv[1] if len(sys.argv) > 1 else 'urls.csv'

with open(SOURCE) as f:
    reader = csv.DictReader(f)
    urls = [row[next(iter(row))].strip() for row in reader
            if next(iter(row.values())).startswith('http')]

print(f'Importing {len(urls)} URLs')
failures = []

for i, url in enumerate(urls, 1):
    while True:
        r = requests.post(
            'https://graniite.co/api/v1/ingest',
            headers={'Authorization': f'Bearer {TOKEN}'},
            json={'kind': 'url', 'url': url, 'in_kb': True, 'auto_transform': False},
            timeout=30,
        )
        if r.status_code == 429:
            wait = r.json().get('retry_after_sec', 60)
            print(f'  rate-limited; sleeping {wait}s')
            time.sleep(wait)
            continue
        break

    if r.ok:
        print(f'  ok {i}/{len(urls)} {r.json()["status"]:13} {url[:60]}')
    else:
        failures.append((url, r.text))
        print(f'  err {url}: {r.text}')

    time.sleep(1)

print(f'\nDone. {len(urls) - len(failures)} succeeded, {len(failures)} failed.')

const existing = new Set();
let page = 1;
while (true) {
  const r = await fetch(`https://graniite.co/api/v1/items?per_page=100&page=${page}`, {
    headers: { Authorization: `Bearer ${TOKEN}` },
  });
  const { items, total } = await r.json();
  for (const it of items) existing.add(it.url);
  if (existing.size >= total) break;
  page++;
}
const remaining = rows.filter((url) => !existing.has(url));

[cron, weekly]
   |
   v
[fetch RSS feeds for N channels, build list of new video URLs]
   |
   v
[diff against already-ingested set]
   |
   v
[run this script on the diff]