跳到正文
DEV Community· AI Engine·· 2 小時前AI 評分54

如何使用 Python 和 API 審核 NSFW 影片及 GIF

NSFW Video and GIF Moderation with Python and an API

AI 導讀

文章示範如何用 Python、ffmpeg 和 NSFW Detect API 抽取影片及 GIF 畫格、逐格評分，再彙總結果判定允許、複核或封鎖。在 25 秒測試片段中，每 1 秒抽樣需 26 次 API 呼叫，每 2 秒需 13 次，兩種設定均判為複核。將 Alcohol 規則改為封鎖後，系統在 12 次呼叫後提前停止並封鎖片段。

正文

If your app accepts video or GIF uploads (short clips in a social feed, reaction GIFs in chat, product videos on a marketplace), checking only the thumbnail is not moderation. The problem frame can sit at second 12 of a 30 second clip. NSFW video moderation with an image API works by sampling frames, scoring each one, and deciding on the whole clip.

Every number below comes from a real run on a public domain 25 second clip of a beer being poured: it starts with an empty glass and ends with a full one, a good test for how detections change over time.

Want to see the labels on your own clips? Try the NSFW Detect API on a frame from one of your uploads.

How it works

A moderation API looks at images, so a video becomes a list of frames:

  1. Sample frames with ffmpeg, at a fixed interval or on scene changes.
  2. Score each frame with the API, in parallel.
  3. Aggregate per top-level category: highest confidence, number of flagged frames, timestamps.
  4. Decide allow, review, or block for the clip, and stop early when a frame already settles it.

Step 1: Sample frames with ffmpeg 

import os
import subprocess

def extract_frames(video_path, out_dir, every_seconds=1.0):
    """Grab one frame every N seconds. Returns [(timestamp, path), ...]."""
    subprocess.run(
        ["ffmpeg", "-v", "error", "-i", video_path,
         "-vf", f"fps=1/{every_seconds},scale='min(1280,iw)':-2",
         "-q:v", "3", f"{out_dir}/frame_%05d.jpg"],
        check=True,
    )
    files = sorted(os.listdir(out_dir))
    return [(i * every_seconds, os.path.join(out_dir, f)) for i, f in enumerate(files)]

On the 25 second test clip, one frame per second gives 26 frames; one every 2 seconds gives 13.

Step 2: Score frames in parallel

Keep only the top-level categories (empty ParentName) so every sub-label maps to one policy rule:

import tempfile
from concurrent.futures import ThreadPoolExecutor

import requests

API_URL = "https://nsfw-detect3.p.rapidapi.com/nsfw-detect"
HEADERS = {
    "x-rapidapi-key": "YOUR_API_KEY",
    "x-rapidapi-host": "nsfw-detect3.p.rapidapi.com",
}

def moderate_frame(path):
    """Return the top-level categories found in one frame: {name: confidence}."""
    with open(path, "rb") as f:
        r = requests.post(API_URL, headers=HEADERS, files={"image": f}, timeout=30)
    r.raise_for_status()
    labels = r.json()["body"]["ModerationLabels"]
    return {l["Name"]: l["Confidence"] for l in labels if l["ParentName"] == ""}

def moderate_video(video_path, every_seconds=1.0, workers=4):
    """Moderate sampled frames in parallel. Returns [(timestamp, categories), ...]."""
    with tempfile.TemporaryDirectory() as tmp:
        frames = extract_frames(video_path, tmp, every_seconds)
        with ThreadPoolExecutor(max_workers=workers) as pool:
            results = list(pool.map(lambda fr: moderate_frame(fr[1]), frames))
    return [(t, cats) for (t, _), cats in zip(frames, results)]

for t, cats in moderate_video("beer_pour.mp4"):
    print(f"{t:>3.0f}s", {k: round(v, 1) for k, v in cats.items()})
#   0s {}                    empty glass
#   8s {'Alcohol': 94.9}
#  11s {'Alcohol': 95.9}
#  21s {'Alcohol': 99.3}
#  25s {'Alcohol': 99.8}
#  (26 frames in total)

Timeline of a 25 second beer pouring clip with the frames flagged Alcohol by the NSFW Detect API

The empty glass at second 0 comes back clean, and Alcohol is flagged from second 3, as soon as beer is in the glass, up to 99.8% on the full glass. That timeline is what turns frame scores into a decision: when the category first appears, how often, and how confident the API is at its peak.

Step 3: Aggregate per category and decide

Each policy rule gets one extra field compared with a single image policy: min_frames, the number of frames that must cross the threshold before the rule fires.

POLICY = {
    "Explicit":                                          {"action": "block",  "threshold": 60, "min_frames": 1},
    "Non-Explicit Nudity of Intimate parts and Kissing": {"action": "review", "threshold": 70, "min_frames": 2},
    "Violence":                                          {"action": "block",  "threshold": 70, "min_frames": 2},
    "Hate Symbols":                                      {"action": "block",  "threshold": 50, "min_frames": 1},
    "Drugs & Tobacco":                                   {"action": "review", "threshold": 80, "min_frames": 2},
    "Alcohol":                                           {"action": "review", "threshold": 90, "min_frames": 2},
}

def summarize(results):
    """Per category: max confidence, number of flagged frames, timestamps."""
    summary = {}
    for t, cats in results:
        for name, conf in cats.items():
            s = summary.setdefault(name, {"max": 0.0, "frames": 0, "times": []})
            s["max"] = max(s["max"], conf)
            s["frames"] += 1
            s["times"].append(t)
    return summary

def decide(summary):
    """Turn the per-category summary into allow / review / block."""
    verdict, reasons = "allow", []
    for name, s in summary.items():
        rule = POLICY.get(name)
        if not rule or s["max"] < rule["threshold"] or s["frames"] < rule["min_frames"]:
            continue
        reasons.append(f"{name}: {s['frames']} frames, max {s['max']:.1f}%, first at {s['times'][0]:.0f}s")
        if rule["action"] == "block":
            verdict = "block"
        elif verdict != "block":
            verdict = "review"
    return verdict, reasons

print(decide(summarize(moderate_video("beer_pour.mp4"))))
# ('review', ['Alcohol: 9 frames, max 99.8%, first at 3s'])

The reasons string tells a reviewer where to jump in the clip instead of watching all of it.

The full guide adds a scene-change sampler with a frame gap cap and the exact numbers behind each sampling choice.

Step 4: Stop early on a clear block

Once a block rule fires, the remaining frames cannot change the verdict:

def moderate_with_early_stop(frames, workers=4):
    """Moderate frames in time order, batch by batch; stop at the first block."""
    results, calls = [], 0
    with ThreadPoolExecutor(max_workers=workers) as pool:
        for i in range(0, len(frames), workers):
            batch = frames[i:i + workers]
            cats = list(pool.map(lambda fr: moderate_frame(fr[1]), batch))
            results += [(t, c) for (t, _), c in zip(batch, cats)]
            calls += len(batch)
            verdict, reasons = decide(summarize(results))
            if verdict == "block":
                break
    return verdict, reasons, calls

POLICY["Alcohol"] = {"action": "block", "threshold": 90, "min_frames": 2}  # kids platform
with tempfile.TemporaryDirectory() as tmp:
    frames = extract_frames("beer_pour.mp4", tmp, every_seconds=1.0)
    print(moderate_with_early_stop(frames))
# ('block', ['Alcohol: 3 frames, max 95.9%, first at 3s'], 12)

Blocked after 12 calls instead of 26. Clean videos still need every sampled frame.

Choosing a sampling rate

Real results on the same clip, with the review policy:

  • Every 1 second: 26 calls, peak 99.8%, verdict review
  • Every 2 seconds: 13 calls, peak 99.8%, verdict review

Both rates flag the clip. For short user uploads, one frame every one to two seconds is a solid default; for long edited videos, scene-change sampling with a gap cap cuts calls further.

GIF uploads

ffmpeg reads GIFs directly, so extract_frames("reaction.gif", tmp, 2.0) works unchanged. Without ffmpeg, Pillow can sample by frame duration:

from PIL import Image, ImageSequence

def extract_gif_frames(gif_path, out_dir, every_seconds=1.0):
    """Sample a GIF every N seconds using each frame's own duration."""
    gif = Image.open(gif_path)
    frames, t, next_t = [], 0.0, 0.0
    for i, frame in enumerate(ImageSequence.Iterator(gif)):
        if t >= next_t:
            path = os.path.join(out_dir, f"gif_{i:05d}.jpg")
            frame.convert("RGB").save(path, quality=90)
            frames.append((t, path))
            next_t += every_seconds
        t += frame.info.get("duration", 100) / 1000
    return frames

On a GIF made from the same clip, sampling every 2 seconds gave 13 frames, Alcohol on 7 (max 99.9%) and a review verdict. The convert("RGB") is required: GIF frames are palette images and JPEG cannot store them.

How many API calls?

One call per sampled frame. A 30 second clip sampled every 2 seconds costs 15 calls, and 1,000 such clips come to about 15,000 calls over the month before early stops. Upload duration limits bound the worst case, and hashing uploads avoids re-moderating reposted GIFs.

Production tips

  • Publish after the verdict. Keep new videos pending and moderate in a background worker; 26 frames with 4 workers took about 6 seconds in this test.
  • Fail safe to review when a frame call errors out.
  • Moderate the poster frame shown in the feed too.

Sources

Run it on the clips your users actually upload: grab a key on the NSFW Detect API page and tune the thresholds on real data.

來源：DEV Community · dev.to

#教學/實踐#影片