Code examples

import mimetypes
import os
import uuid
from contextlib import ExitStack
from datetime import datetime
from pathlib import Path

import requests

UPLOAD_URL = "https://uploads.backend.trollflix.com/api/php/developer/upload_video"
CATEGORIES_URL = "https://backend.trollflix.com/api/php/categories/list"


class TrollflixError(Exception):
    """The API rejected the request. `code` is the error_message_code."""

    def __init__(self, code, response):
        super().__init__(code)
        self.code = code
        self.response = response


def _to_form_value(value):
    if isinstance(value, bool):
        return "1" if value else "0"
    if isinstance(value, datetime):
        return str(int(value.timestamp()))
    return str(value)


def _file_part(path, stack):
    path = Path(path)
    mime_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    return (path.name, stack.enter_context(path.open("rb")), mime_type)


class TrollflixClient:
    def __init__(self, api_key=None, timeout=600):
        self.api_key = api_key or os.environ["TROLLFLIX_API_KEY"]
        self.timeout = timeout

    def upload_video(self, path, *, thumbnail=None, upload_token=None, **fields):
        """Upload one video. Extra keyword arguments are sent as form fields,
        e.g. caption="...", super_category_id=1, is_ai_generated=True."""
        data = {name: _to_form_value(value) for name, value in fields.items() if value is not None}
        data["upload_token"] = upload_token or str(uuid.uuid4())

        with ExitStack() as stack:
            files = {"file": _file_part(path, stack)}
            if thumbnail is not None:
                files["thumbnail"] = _file_part(thumbnail, stack)

            response = requests.post(
                UPLOAD_URL,
                headers={"Authorization": f"Bearer {self.api_key}"},
                files=files,
                data=data,
                timeout=self.timeout,
            )

        result = response.json()
        if result.get("success") is not True:
            code = result.get("error_message_code") or result.get("error") or "unknown_error"
            raise TrollflixError(code, result)
        return result

    @staticmethod
    def list_categories():
        """The public category tree (no API key needed)."""
        response = requests.post(CATEGORIES_URL, json={}, timeout=30)
        response.raise_for_status()
        return response.json()["super_categories"]

import { randomUUID } from "node:crypto";
import { openAsBlob } from "node:fs";
import { basename } from "node:path";

const UPLOAD_URL =
  "https://uploads.backend.trollflix.com/api/php/developer/upload_video";
const CATEGORIES_URL = "https://backend.trollflix.com/api/php/categories/list";

/** The API rejected the request. `code` is the error_message_code. */
export class TrollflixError extends Error {
  constructor(code, response) {
    super(code);
    this.name = "TrollflixError";
    this.code = code;
    this.response = response;
  }
}

function toFormValue(value) {
  if (typeof value === "boolean") return value ? "1" : "0";
  if (value instanceof Date) return String(Math.floor(value.getTime() / 1000));
  return String(value);
}

export class TrollflixClient {
  constructor({ apiKey = process.env.TROLLFLIX_API_KEY, timeoutMs = 600_000 } = {}) {
    if (!apiKey) throw new Error("Set TROLLFLIX_API_KEY or pass { apiKey }");
    this.apiKey = apiKey;
    this.timeoutMs = timeoutMs;
  }

  /**
   * Upload one video. Other options are sent as form fields,
   * e.g. { caption: "...", super_category_id: 1, is_ai_generated: true }.
   */
  async uploadVideo(path, { thumbnail, uploadToken = randomUUID(), ...fields } = {}) {
    const form = new FormData();
    form.append("file", await openAsBlob(path), basename(path));
    if (thumbnail) {
      form.append("thumbnail", await openAsBlob(thumbnail), basename(thumbnail));
    }
    form.append("upload_token", uploadToken);
    for (const [name, value] of Object.entries(fields)) {
      if (value !== undefined && value !== null) form.append(name, toFormValue(value));
    }

    const response = await fetch(UPLOAD_URL, {
      method: "POST",
      headers: { Authorization: `Bearer ${this.apiKey}` },
      body: form,
      signal: AbortSignal.timeout(this.timeoutMs),
    });

    const result = await response.json();
    if (result.success !== true) {
      const code = result.error_message_code ?? result.error ?? "unknown_error";
      throw new TrollflixError(code, result);
    }
    return result;
  }

  /** The public category tree (no API key needed). */
  static async listCategories() {
    const response = await fetch(CATEGORIES_URL, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: "{}",
    });
    if (!response.ok) throw new Error(`categories/list: HTTP ${response.status}`);
    return (await response.json()).super_categories;
  }
}

from trollflix import TrollflixClient

client = TrollflixClient()

result = client.upload_video(
    "meme.mp4",
    caption="When the build passes on the first try #programming",
    super_category_id=1,
    ai_summary_of_content="A developer stares at a green CI check in disbelief, then cheers.",
)

print(f"https://www.trollflix.com/meme/{result['slug_url']}")

import { TrollflixClient } from "./trollflix.mjs";

const client = new TrollflixClient();

const result = await client.uploadVideo("meme.mp4", {
  caption: "When the build passes on the first try #programming",
  super_category_id: 1,
  ai_summary_of_content:
    "A developer stares at a green CI check in disbelief, then cheers.",
});

console.log(`https://www.trollflix.com/meme/${result.slug_url}`);

# Your own image (at least 500x300, at most 5 MB)
client.upload_video("meme.mp4", thumbnail="cover.jpg", caption="Monday mood")

# Or a frame of the video
client.upload_video("meme.mp4", thumbnail_time="00:00:03", caption="Monday mood")

// Your own image (at least 500x300, at most 5 MB)
await client.uploadVideo("meme.mp4", { thumbnail: "cover.jpg", caption: "Monday mood" });

// Or a frame of the video
await client.uploadVideo("meme.mp4", { thumbnail_time: "00:00:03", caption: "Monday mood" });

def find_category_id(name):
    for category in TrollflixClient.list_categories():
        if name.lower() in category["tag_name"].lower():
            return category["id"]
    raise LookupError(f"No category matching {name!r}")


gaming_id = find_category_id("gaming")
client.upload_video("clip.mp4", caption="Lag spike at the worst moment #gaming", super_category_id=gaming_id)

async function findCategoryId(name) {
  const categories = await TrollflixClient.listCategories();
  const match = categories.find((category) =>
    category.tag_name.toLowerCase().includes(name.toLowerCase()),
  );
  if (!match) throw new Error(`No category matching "${name}"`);
  return match.id;
}

const gamingId = await findCategoryId("gaming");
await client.uploadVideo("clip.mp4", {
  caption: "Lag spike at the worst moment #gaming",
  super_category_id: gamingId,
});

import time
import uuid

import requests

from trollflix import TrollflixError

RETRYABLE_CODES = {
    "file_upload_failed",
    "content_upload_failed",
    "media_container_upload_failed",
    "thumbnail_file_error",
}


def upload_with_retries(client, path, attempts=5, **fields):
    token = str(uuid.uuid4())  # one token per meme, reused on every attempt
    for attempt in range(1, attempts + 1):
        try:
            result = client.upload_video(path, upload_token=token, **fields)
            if not result.get("duplicate_in_progress"):
                return result  # created now, or by an earlier attempt
        except TrollflixError as error:
            if error.code not in RETRYABLE_CODES:
                raise
        except requests.RequestException:
            pass  # network problem: try again
        time.sleep(attempt * 5)
    raise RuntimeError(f"Gave up on {path} after {attempts} attempts")

import { randomUUID } from "node:crypto";
import { TrollflixError } from "./trollflix.mjs";

const RETRYABLE_CODES = new Set([
  "file_upload_failed",
  "content_upload_failed",
  "media_container_upload_failed",
  "thumbnail_file_error",
]);

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

export async function uploadWithRetries(client, path, fields = {}, attempts = 5) {
  const uploadToken = randomUUID(); // one token per meme, reused on every attempt
  for (let attempt = 1; attempt <= attempts; attempt++) {
    try {
      const result = await client.uploadVideo(path, { ...fields, uploadToken });
      if (!result.duplicate_in_progress) return result; // created now, or by an earlier attempt
    } catch (error) {
      if (error instanceof TrollflixError && !RETRYABLE_CODES.has(error.code)) throw error;
      // network problem or temporary server error: try again
    }
    await sleep(attempt * 5000);
  }
  throw new Error(`Gave up on ${path} after ${attempts} attempts`);
}

from datetime import datetime, timedelta, timezone
from pathlib import Path

from retry import upload_with_retries
from trollflix import TrollflixClient

client = TrollflixClient()
now = datetime.now(timezone.utc)
publish_at = (now + timedelta(days=1)).replace(hour=18, minute=0, second=0, microsecond=0)
last_allowed = now + timedelta(days=32)

for video in sorted(Path("memes").glob("*.mp4")):
    if publish_at > last_allowed:
        print("32-day scheduling window is full, stopping here.")
        break
    result = upload_with_retries(
        client,
        video,
        caption=video.stem.replace("_", " "),
        scheduled_publish_at=publish_at,
    )
    print(f"{video.name} -> {publish_at:%Y-%m-%d %H:%M} UTC (content {result['content_id']})")
    publish_at += timedelta(days=1)

import { readdir } from "node:fs/promises";
import { join, parse } from "node:path";
import { uploadWithRetries } from "./retry.mjs";
import { TrollflixClient } from "./trollflix.mjs";

const DAY_MS = 24 * 60 * 60 * 1000;
const client = new TrollflixClient();

const now = Date.now();
const publishAt = new Date(now + DAY_MS);
publishAt.setUTCHours(18, 0, 0, 0);
const lastAllowed = now + 32 * DAY_MS;

const videos = (await readdir("memes")).filter((name) => name.endsWith(".mp4")).sort();

for (const name of videos) {
  if (publishAt.getTime() > lastAllowed) {
    console.log("32-day scheduling window is full, stopping here.");
    break;
  }
  const result = await uploadWithRetries(client, join("memes", name), {
    caption: parse(name).name.replaceAll("_", " "),
    scheduled_publish_at: publishAt,
  });
  console.log(`${name} -> ${publishAt.toISOString()} (content ${result.content_id})`);
  publishAt.setTime(publishAt.getTime() + DAY_MS);
}

from datetime import datetime, timezone

from trollflix import TrollflixError

try:
    result = client.upload_video("meme.mp4", caption="One more for today")
    if result["is_limit_activated"]:
        print("That was the last upload for today.")
except TrollflixError as error:
    if error.code != "content_upload_limit_failure":
        raise
    reset = datetime.fromtimestamp(error.response["limit_reset_time_unix"], timezone.utc)
    print(f"Daily limit reached. Uploads open again at {reset:%H:%M} UTC.")

import { TrollflixError } from "./trollflix.mjs";

try {
  const result = await client.uploadVideo("meme.mp4", { caption: "One more for today" });
  if (result.is_limit_activated) console.log("That was the last upload for today.");
} catch (error) {
  if (!(error instanceof TrollflixError) || error.code !== "content_upload_limit_failure") {
    throw error;
  }
  const reset = new Date(error.response.limit_reset_time_unix * 1000);
  console.log(`Daily limit reached. Uploads open again at ${reset.toISOString()}.`);
}

from trollflix import TrollflixError

MESSAGES = {
    "api_key_invalid": "The API key was rotated or deleted. Update TROLLFLIX_API_KEY.",
    "content_video_duration_invalid": "Videos must be 2 to 120 seconds long.",
    "content_file_invalid_format": "Use an MP4, MOV, MKV, AVI, FLV, WMV, M4V or MPEG video.",
    "content_category_invalid": "That category no longer exists. Refresh your category list.",
    "content_upload_limit_failure": "Daily upload limit reached. Schedule it instead.",
}

try:
    client.upload_video("meme.mp4")
except TrollflixError as error:
    print(MESSAGES.get(error.code, f"Upload rejected ({error.code})."))

import { TrollflixError } from "./trollflix.mjs";

const MESSAGES = {
  api_key_invalid: "The API key was rotated or deleted. Update TROLLFLIX_API_KEY.",
  content_video_duration_invalid: "Videos must be 2 to 120 seconds long.",
  content_file_invalid_format: "Use an MP4, MOV, MKV, AVI, FLV, WMV, M4V or MPEG video.",
  content_category_invalid: "That category no longer exists. Refresh your category list.",
  content_upload_limit_failure: "Daily upload limit reached. Schedule it instead.",
};

try {
  await client.uploadVideo("meme.mp4");
} catch (error) {
  if (!(error instanceof TrollflixError)) throw error;
  console.log(MESSAGES[error.code] ?? `Upload rejected (${error.code}).`);
}