Building a Video Metadata Moderation Pipeline with Python and the Claude API

Last quarter I pulled a batch of trending videos from eight regions into TrendVidStream and our discovery feed surfaced a clip whose title was clean English but whose description was a wall of adult-spam keywords in Thai, plus a localized title that translated to something we'd never put in front of a family-mode visitor. Our old moderation logic was a regex blocklist and a banned-channel table. It caught zero of it. The problem is not that bad content is rare — it's that bad content arrives as metadata, in dozens of languages, framed to look exactly like good content, and a keyword list ages out the moment spammers read it.

This post walks through the moderation pipeline we built to replace that regex pile: a Python ingestion worker that scores incoming video metadata with the Claude API, a confidence-banded routing layer so we don't burn tokens on obvious cases, and a SQLite-backed audit trail that lets us defend every decision. Our stack is PHP 8.4 on the serving side with SQLite FTS5 for search, multi-region cron jobs that fetch trending feeds, and FTP-based deploys to LiteSpeed hosts — so the constraints here are real: no GPU, no managed queue, no Kafka. Everything runs in a cron-driven worker that has to be cheap, idempotent, and explainable.

The actual problem with metadata moderation

Before writing code, it helps to name what we are moderating. We never touch the video pixels — we are a discovery layer, not a host. What we ingest per video is roughly this:

• title — often in the channel's native language
• description — the highest-risk field; spam, scam links, and keyword stuffing live here
• channel_title and channel_id
• tags — a free-for-all
• region_code and category_id

The failure modes that a blocklist cannot handle:

• Cross-language evasion. A title is benign in English; the description is abusive in Vietnamese. A single-language filter is blind to half the document.
• Context collapse. The word "shot" is fine for a basketball highlight and alarming next to a weapons-sale phrase. Keywords have no context; an LLM reads the whole field.
• Adversarial framing. Spammers write descriptions that look like normal creator copy, then bury a Telegram handle and a "free crypto" link in the middle.
• Category mismatch. A clip filed under "Education" whose metadata is clearly adult content. The category alone is a signal, but only in combination.

What we want out of the pipeline is not a yes/no bit. We want a structured verdict: a category, a severity, a confidence, and a short rationale we can store and later audit. That shape — structured, multi-field, language-agnostic — is exactly what a tool-use call to Claude gives you for a fraction of a cent per item.

Pipeline shape: cheap gate first, model second

The single most important design decision is to not send everything to the model. Most ingested videos are obviously fine, and a small slice is obviously banned (known channel, exact-match scam domain). We run a cheap deterministic gate first and only escalate the ambiguous middle to Claude.

# gate.py - deterministic pre-filter, no API calls
import re
from dataclasses import dataclass

HARD_BLOCK_DOMAINS = re.compile(
    r"(?:t\.me/|bit\.ly/|wa\.me/|free-?crypto|onlyfans\.com)",
    re.IGNORECASE,
)

@dataclass
class GateResult:
    decision: str          # "allow" | "block" | "escalate"
    reason: str

def pre_filter(video: dict, banned_channels: set[str]) -> GateResult:
    if video["channel_id"] in banned_channels:
        return GateResult("block", "channel on banned list")

    blob = f"{video.get('title','')} {video.get('description','')}"
    if HARD_BLOCK_DOMAINS.search(blob):
        return GateResult("block", "hard-blocked link pattern")

    # Nothing suspicious AND short, plain description -> trust it
    desc = video.get("description", "")
    if len(desc) < 600 and desc.count("http") <= 1:
        return GateResult("allow", "clean short metadata")

    # Everything else is genuinely ambiguous -> spend a token on it
    return GateResult("escalate", "needs model review")

In our eight-region feed this gate auto-resolves roughly 70% of items for free. The remaining 30% — long descriptions, multiple links, anything borderline — goes to Claude. That ratio is what keeps the monthly bill in coffee-money territory rather than infrastructure territory.

Asking Claude for a structured verdict, not prose

The naive approach is to ask "is this video safe?" and parse the prose answer. Don't. You want a machine-readable verdict every time, which means using tool use (function calling) to force a JSON schema. The model is told it must call the record_verdict tool, and the tool's input schema becomes the contract for your database.

# classifier.py
import json
import anthropic

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY

VERDICT_TOOL = {
    "name": "record_verdict",
    "description": "Record the moderation verdict for one video's metadata.",
    "input_schema": {
        "type": "object",
        "properties": {
            "label": {
                "type": "string",
                "enum": ["safe", "spam", "adult", "violence",
                         "scam", "hate", "needs_human"],
            },
            "severity": {
                "type": "integer",
                "minimum": 0,
                "maximum": 3,
                "description": "0 none, 1 mild, 2 strong, 3 egregious",
            },
            "confidence": {
                "type": "number",
                "minimum": 0.0,
                "maximum": 1.0,
            },
            "languages": {
                "type": "array",
                "items": {"type": "string"},
                "description": "ISO codes detected across all fields",
            },
            "rationale": {
                "type": "string",
                "description": "One sentence, references the offending field",
            },
        },
        "required": ["label", "severity", "confidence", "rationale"],
    },
}

SYSTEM = (
    "You moderate video METADATA for a multi-region discovery feed. "
    "You never see the video itself. Judge title, description, tags, and "
    "channel name together, across ALL languages present. A clean title "
    "does not excuse an abusive description. Treat buried contact handles "
    "and off-platform links as spam/scam signals. Be strict on adult and "
    "scam, lenient on ordinary creator self-promotion. You MUST call "
    "record_verdict exactly once."
)

def classify(video: dict) -> dict:
    user = json.dumps({
        "title": video.get("title", ""),
        "description": video.get("description", "")[:4000],
        "tags": video.get("tags", [])[:30],
        "channel_title": video.get("channel_title", ""),
        "region_code": video.get("region_code", ""),
        "category_id": video.get("category_id", ""),
    }, ensure_ascii=False)

    msg = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=400,
        system=SYSTEM,
        tools=[VERDICT_TOOL],
        tool_choice={"type": "tool", "name": "record_verdict"},
        messages=[{"role": "user", "content": user}],
    )

    for block in msg.content:
        if block.type == "tool_use" and block.name == "record_verdict":
            return block.input
    raise RuntimeError("model did not return a verdict")

A few decisions worth calling out:

• Model tier. Metadata moderation is a classification task, not a reasoning marathon. Haiku 4.5 is fast and cheap and handles this well; we reserve a Sonnet escalation only for items the cheaper model flags as low-confidence. Start cheap, escalate selectively.
• tool_choice is forced. By setting tool_choice to the named tool, the model cannot wander off into prose. Every call returns a parseable object or raises — there is no third state to handle.
• Truncate the description. We cap at 4000 characters. Spam payloads are usually front-loaded, and uncapped fields are a token-cost and prompt-injection risk.
• Multilingual is free. We don't run language detection first. Claude reads mixed-language metadata natively and reports the languages it saw, which is itself a useful audit field.

Confidence banding and the human queue

The verdict comes back with a confidence number, and that number is the whole point of the design. A binary moderator forces you to pick a single threshold and live with both the false positives above it and the false negatives below it. With a confidence band you carve out a third path — the uncertain middle goes to a human.

# router.py
def route(verdict: dict) -> str:
    label = verdict["label"]
    conf = verdict["confidence"]
    sev = verdict["severity"]

    if label == "safe" and conf >= 0.80:
        return "publish"

    if label != "safe" and sev >= 2 and conf >= 0.85:
        return "block"

    # high-severity but unsure, or any verdict in the murky band
    if conf < 0.60 or label == "needs_human":
        return "human_queue"

    # low-severity spam we don't want to surface but won't hard-block
    return "shadow_hide"

The four outcomes map onto how the PHP serving side treats a row: publish shows it in discovery, block removes it, shadow_hide keeps it queryable but never ranks it in feeds, and human_queue parks it in the admin panel. The shadow_hide band matters more than you'd expect — most spam is low-stakes, and silently declining to promote it is both safer and less drama than a hard block you might have to reverse.

Persisting the audit trail in SQLite

Every decision is written to a moderation_log table, keyed by video ID, with the full verdict. This is non-negotiable for two reasons: you need to defend a takedown if a creator complains, and you need a labelled dataset to measure the model against later. Because our serving layer is PHP 8.4 over SQLite with FTS5, the worker writes to the same database file the site reads, and a cron picks up the routed rows.

# store.py
import sqlite3, json, time

DDL = """
CREATE TABLE IF NOT EXISTS moderation_log (
    video_id    TEXT PRIMARY KEY,
    label       TEXT NOT NULL,
    severity    INTEGER NOT NULL,
    confidence  REAL NOT NULL,
    route       TEXT NOT NULL,
    languages   TEXT,
    rationale   TEXT,
    model       TEXT NOT NULL,
    reviewed_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_modlog_route ON moderation_log(route);
"""

def save(db_path: str, video_id: str, verdict: dict,
         route: str, model: str) -> None:
    con = sqlite3.connect(db_path, timeout=30)
    try:
        con.execute("PRAGMA journal_mode=WAL")  # cron writer + web reader
        con.executescript(DDL)
        con.execute(
            """INSERT INTO moderation_log
               (video_id, label, severity, confidence, route,
                languages, rationale, model, reviewed_at)
               VALUES (?,?,?,?,?,?,?,?,?)
               ON CONFLICT(video_id) DO UPDATE SET
                 label=excluded.label, severity=excluded.severity,
                 confidence=excluded.confidence, route=excluded.route,
                 rationale=excluded.rationale, reviewed_at=excluded.reviewed_at""",
            (video_id, verdict["label"], verdict["severity"],
             verdict["confidence"], route,
             json.dumps(verdict.get("languages", [])),
             verdict.get("rationale", ""), model, int(time.time())),
        )
        con.commit()
    finally:
        con.close()

Two SQLite specifics that bit us and might bite you:

• WAL mode is mandatory when a cron worker writes while the PHP front end reads. Without it the web requests block on the writer's lock and you get sporadic 500s during ingestion.
• ON CONFLICT ... DO UPDATE makes the worker idempotent. Cron overlaps happen — a slow fetch run can collide with the next scheduled one — and re-moderating the same video must update the row, not crash on the primary key.

On the serving side, the join is trivial: the feed query excludes anything whose latest route is block or shadow_hide, and the FTS5 search index is only populated for publish rows, so hidden content never even appears in autocomplete.

Wiring it into the multi-region cron

Our trending fetch already runs per region on a stagger — every two hours for the busiest market, every several hours for quieter ones — to spread API quota. The moderation worker hangs off the back of that fetch. The orchestration is deliberately boring: pull the batch, gate it, classify the escalations, route, store. Here it is end to end.

# worker.py
from gate import pre_filter
from classifier import classify
from router import route
from store import save

MODEL = "claude-haiku-4-5-20251001"

def moderate_batch(videos, banned, db_path):
    counts = {"allow": 0, "block": 0, "model": 0}
    for v in videos:
        gate = pre_filter(v, banned)
        if gate.decision == "allow":
            verdict = {"label": "safe", "severity": 0,
                       "confidence": 0.95, "rationale": gate.reason}
            save(db_path, v["id"], verdict, "publish", "gate")
            counts["allow"] += 1
            continue
        if gate.decision == "block":
            verdict = {"label": "spam", "severity": 3,
                       "confidence": 0.99, "rationale": gate.reason}
            save(db_path, v["id"], verdict, "block", "gate")
            counts["block"] += 1
            continue
        try:
            verdict = classify(v)
        except Exception as exc:
            # fail closed: anything we couldn't score goes to humans
            verdict = {"label": "needs_human", "severity": 1,
                       "confidence": 0.0, "rationale": f"api error: {exc}"}
        save(db_path, v["id"], verdict, route(verdict), MODEL)
        counts["model"] += 1
    return counts

The one rule I'd tattoo on a junior engineer here: fail closed. When the API call throws — rate limit, timeout, a 529 overloaded response — the item does not silently publish. It lands in the human queue with confidence: 0.0. A moderation pipeline that fails open is worse than no pipeline, because it gives you the false confidence that everything was checked.

What the deploy side looks like in Go

We deploy over FTP to LiteSpeed hosts, and the Python worker runs on a small box that isn't always the same box that serves traffic. To ship the freshly-moderated SQLite database to each region's host, we use a tiny Go uploader — Go because it cross-compiles to a single static binary we can drop on any of the hosts without a runtime. The relevant slice:

// push.go - upload the moderated db to one FTP host
package main

import (
    "log"
    "os"
    "time"

    "github.com/jlaffaye/ftp"
)

func push(addr, user, pass, local, remote string) error {
    c, err := ftp.Dial(addr, ftp.DialWithTimeout(15*time.Second))
    if err != nil {
        return err
    }
    defer c.Quit()
    if err := c.Login(user, pass); err != nil {
        return err
    }
    f, err := os.Open(local)
    if err != nil {
        return err
    }
    defer f.Close()
    // upload to a temp name, then rename: readers never see a half-file
    tmp := remote + ".uploading"
    if err := c.Stor(tmp, f); err != nil {
        return err
    }
    return c.Rename(tmp, remote)
}

func main() {
    err := push(os.Getenv("FTP_ADDR"), os.Getenv("FTP_USER"),
        os.Getenv("FTP_PASS"), "data/backlink.db", "data/site.db")
    if err != nil {
        log.Fatalf("push failed: %v", err)
    }
    log.Println("moderated db live")
}

The upload-then-rename trick is the same idea as the WAL pragma above: never let a reader observe a partially-written file. FTP STOR streams bytes, so a live database overwrite can be read mid-flight; uploading to .uploading and atomically renaming means the worst case is a reader getting the old database, never a corrupt one.

Measuring whether it actually works

A moderation pipeline you don't measure is just a vibe. Because every verdict is in moderation_log, we sample 200 random publish rows and 200 random block/shadow_hide rows each week and hand-label them. Two numbers matter:

• False-publish rate — how much bad content got through. This is the one that gets you delisted from ad networks, so we weight it heavily and tune thresholds to keep it near zero even at the cost of more human-queue volume.
• False-block rate — how much good content we wrongly hid. Creators notice this and complain, and it shrinks the catalog, so we watch it but tolerate more of it than false-publish.

The confidence band is the dial. When the false-publish rate creeps up, we raise the publish confidence floor from 0.80, which pushes more borderline items into shadow_hide or the human queue. The rationale field earns its keep here too — when a human reviewer disagrees with a verdict, the model's one-sentence rationale tells you why it was wrong, which is usually a fixable prompt issue rather than a model limitation.

Conclusion

The shift that mattered was reframing moderation from "match bad words" to "score a structured verdict." Once the model returns a label, a severity, and a confidence instead of a yes/no, everything downstream gets simpler: a cheap deterministic gate handles the obvious majority, forced tool use guarantees parseable output, confidence banding gives you a human-review escape hatch instead of a single brittle threshold, and a WAL-mode SQLite log makes every decision auditable and re-runnable. None of it needs a GPU, a queue broker, or a rewrite of your serving stack — it slots into the same cron-and-FTP rhythm we already had. If you run a discovery or aggregation product, start with the gate and the forced-tool-use verdict; the routing and metrics are easy to add once the verdicts are flowing, and they're what turn a clever demo into something you can actually defend to an ad network or a creator.