📌 خروجی پروژه:

🔹 آپلود ویدئو از طریق API

🔹 Job Queue مبتنی بر SQLite

🔹 Worker مستقل از API

🔹 تبدیل ویدئو به H.264

🔹 تبدیل ویدئو به H.265/HEVC

🔹 تبدیل ویدئو به AV1

🔹 دریافت وضعیت Job

🔹 دانلود خروجی از Nginx

🔹 نمونه ساخت HLS برای Streaming

🔹 آشنایی با NVIDIA NVENC و Intel Quick Sync

نسخه جدید هنگام نگارش: FFmpeg 9.0.2 «Lei»

تاریخ انتشار: 18 سپتامبر 2026

Client
   |
   | Upload video
   v
FastAPI
   |
   | INSERT queued job
   v
SQLite Queue
   |
   v
Transcoding Worker
   |
   | FFmpeg
   v
Output File
   |
   v
Nginx Download

اجرای FFmpeg روی سرور مجازی

برای پروژه‌های سبک و Queue کوچک می‌توان Transcoding Server را روی VPS لینوکس راه‌اندازی کرد. برای پردازش سنگین، تعداد زیاد Job یا 4K باید منابع CPU و Storage متناسب انتخاب شوند.

sudo apt update

sudo apt install 
  ffmpeg 
  python3 
  python3-venv 
  python3-pip 
  sqlite3 
  nginx 
  -y

ffmpeg -version
ffprobe -version

⚠️ توجه: نسخه FFmpeg موجود در Repository هر Ubuntu الزاماً همان آخرین نسخه Upstream یعنی 9.0.2 نیست. برای خود این پروژه شماره نسخه کمتر از داشتن Encoderهای مورد نیاز اهمیت دارد. اگر دقیقاً FFmpeg 9.0.2 می‌خواهید باید از Build معتبر یا Compile مناسب استفاده کنید.

ffmpeg -encoders | grep -E \
"libx264|libx265|libsvtav1|h264_nvenc|hevc_nvenc|h264_qsv|av1_qsv"

sudo mkdir -p \
  /opt/video-transcoder \
  /srv/transcoder/uploads \
  /srv/transcoder/outputs \
  /srv/transcoder/logs

sudo useradd 
  --system 
  --home /opt/video-transcoder 
  --shell /usr/sbin/nologin 
  transcoder

sudo chown -R transcoder:transcoder 
  /opt/video-transcoder 
  /srv/transcoder

sudo -u transcoder \
  python3 -m venv \
  /opt/video-transcoder/venv

sudo -u transcoder 
  /opt/video-transcoder/venv/bin/pip 
  install 
  fastapi 
  uvicorn 
  python-multipart

sudo nano /opt/video-transcoder/app.py

import os
import secrets
import sqlite3
import time
import uuid
from pathlib import Path

from fastapi import (
    FastAPI,
    File,
    Form,
    Header,
    HTTPException,
    UploadFile,
)
from fastapi.responses import Response


app = FastAPI(
    title="Iranica Video Transcoder"
)

ROOT = Path(
    "/srv/transcoder"
)

UPLOAD_DIR = ROOT / "uploads"
OUTPUT_DIR = ROOT / "outputs"

DB_PATH = Path(
    "/opt/video-transcoder/jobs.db"
)

API_KEY = os.getenv(
    "TRANSCODER_API_KEY",
    ""
)

MAX_UPLOAD = (
    20
    * 1024
    * 1024
    * 1024
)

PROFILES = {
    "h264-720p": "mp4",
    "h264-1080p": "mp4",
    "h265-1080p": "mp4",
    "av1-1080p": "mp4",
}


def db():
    conn = sqlite3.connect(
        DB_PATH,
        timeout=10
    )

    conn.row_factory = sqlite3.Row

    conn.execute(
        "PRAGMA journal_mode=WAL"
    )

    conn.execute(
        "PRAGMA busy_timeout=10000"
    )

    return conn


def init_db():
    conn = db()

    conn.execute("""
        CREATE TABLE IF NOT EXISTS jobs (
            id TEXT PRIMARY KEY,
            profile TEXT NOT NULL,
            input_path TEXT NOT NULL,
            output_path TEXT NOT NULL,
            status TEXT NOT NULL,
            original_name TEXT,
            error TEXT,
            created_at INTEGER NOT NULL,
            started_at INTEGER,
            finished_at INTEGER
        )
    """)

    conn.commit()
    conn.close()


def check_key(value):
    if not API_KEY:
        return

    if not secrets.compare_digest(
        value or "",
        API_KEY
    ):
        raise HTTPException(
            status_code=401,
            detail="Invalid API key"
        )


init_db()


@app.post("/jobs")
async def create_job(
    file: UploadFile = File(...),
    profile: str = Form("h264-720p"),
    x_api_key: str | None = Header(None),
):
    check_key(x_api_key)

    if profile not in PROFILES:
        raise HTTPException(
            status_code=400,
            detail="Unknown profile"
        )

    job_id = uuid.uuid4().hex

    suffix = Path(
        file.filename or "video.bin"
    ).suffix.lower()[:12]

    input_path = (
        UPLOAD_DIR
        / f"{job_id}{suffix}"
    )

    extension = PROFILES[profile]

    output_path = (
        OUTPUT_DIR
        / f"{job_id}.{extension}"
    )

    total = 0

    try:
        with input_path.open("wb") as out:
            while True:
                chunk = await file.read(
                    1024 * 1024
                )

                if not chunk:
                    break

                total += len(chunk)

                if total > MAX_UPLOAD:
                    raise HTTPException(
                        status_code=413,
                        detail="Video too large"
                    )

                out.write(chunk)

    except Exception:
        input_path.unlink(
            missing_ok=True
        )
        raise

    finally:
        await file.close()

    now = int(time.time())

    conn = db()

    conn.execute("""
        INSERT INTO jobs (
            id,
            profile,
            input_path,
            output_path,
            status,
            original_name,
            created_at
        )
        VALUES (?, ?, ?, ?, 'queued', ?, ?)
    """, (
        job_id,
        profile,
        str(input_path),
        str(output_path),
        Path(
            file.filename or "video"
        ).name,
        now
    ))

    conn.commit()
    conn.close()

    return {
        "job_id": job_id,
        "status": "queued",
        "profile": profile,
        "size": total,
        "status_url": f"/jobs/{job_id}"
    }


@app.get("/jobs/{job_id}")
def job_status(
    job_id: str,
    x_api_key: str | None = Header(None),
):
    check_key(x_api_key)

    conn = db()

    row = conn.execute("""
        SELECT
            id,
            profile,
            status,
            error,
            created_at,
            started_at,
            finished_at
        FROM jobs
        WHERE id = ?
    """, (job_id,)).fetchone()

    conn.close()

    if row is None:
        raise HTTPException(
            status_code=404
        )

    result = dict(row)

    if row["status"] == "done":
        result["download_url"] = (
            f"/jobs/{job_id}/download"
        )

    return result


@app.get("/jobs/{job_id}/download")
def download(
    job_id: str,
    x_api_key: str | None = Header(None),
):
    check_key(x_api_key)

    conn = db()

    row = conn.execute("""
        SELECT
            status,
            output_path
        FROM jobs
        WHERE id = ?
    """, (job_id,)).fetchone()

    conn.close()

    if row is None:
        raise HTTPException(
            status_code=404
        )

    if row["status"] != "done":
        raise HTTPException(
            status_code=409,
            detail="Job is not finished"
        )

    output = Path(
        row["output_path"]
    )

    if not output.is_file():
        raise HTTPException(
            status_code=404,
            detail="Output missing"
        )

    response = Response(
        status_code=200
    )

    response.headers[
        "X-Accel-Redirect"
    ] = (
        "/_transcoded/"
        + output.name
    )

    response.headers[
        "Content-Type"
    ] = "video/mp4"

    response.headers[
        "Content-Disposition"
    ] = (
        f'attachment; '
        f'filename="{output.name}"'
    )

    return response

sudo nano /opt/video-transcoder/worker.py

import sqlite3
import subprocess
import time
from pathlib import Path


DB_PATH = Path(
    "/opt/video-transcoder/jobs.db"
)

LOG_DIR = Path(
    "/srv/transcoder/logs"
)


def db():
    conn = sqlite3.connect(
        DB_PATH,
        timeout=10
    )

    conn.row_factory = sqlite3.Row

    conn.execute(
        "PRAGMA journal_mode=WAL"
    )

    conn.execute(
        "PRAGMA busy_timeout=10000"
    )

    return conn


def claim_job():
    now = int(time.time())

    conn = db()

    try:
        conn.execute(
            "BEGIN IMMEDIATE"
        )

        row = conn.execute("""
            UPDATE jobs
            SET
                status = 'processing',
                started_at = ?
            WHERE id = (
                SELECT id
                FROM jobs
                WHERE status = 'queued'
                ORDER BY created_at
                LIMIT 1
            )
            AND status = 'queued'
            RETURNING *
        """, (now,)).fetchone()

        conn.commit()

        return row

    except Exception:
        conn.rollback()
        raise

    finally:
        conn.close()


def build_command(job):
    source = job["input_path"]
    output = job["output_path"]
    profile = job["profile"]

    common = [
        "ffmpeg",
        "-nostdin",
        "-hide_banner",
        "-y",
        "-i",
        source,
        "-map",
        "0:v:0",
        "-map",
        "0:a?",
    ]

    if profile == "h264-720p":
        codec = [
            "-vf",
            "scale=-2:720",
            "-c:v",
            "libx264",
            "-preset",
            "medium",
            "-crf",
            "23",
            "-pix_fmt",
            "yuv420p",
            "-c:a",
            "aac",
            "-b:a",
            "128k",
            "-movflags",
            "+faststart",
        ]

    elif profile == "h264-1080p":
        codec = [
            "-vf",
            "scale=-2:1080",
            "-c:v",
            "libx264",
            "-preset",
            "medium",
            "-crf",
            "22",
            "-pix_fmt",
            "yuv420p",
            "-c:a",
            "aac",
            "-b:a",
            "160k",
            "-movflags",
            "+faststart",
        ]

    elif profile == "h265-1080p":
        codec = [
            "-vf",
            "scale=-2:1080",
            "-c:v",
            "libx265",
            "-preset",
            "medium",
            "-crf",
            "28",
            "-tag:v",
            "hvc1",
            "-pix_fmt",
            "yuv420p",
            "-c:a",
            "aac",
            "-b:a",
            "160k",
            "-movflags",
            "+faststart",
        ]

    elif profile == "av1-1080p":
        codec = [
            "-vf",
            "scale=-2:1080",
            "-c:v",
            "libsvtav1",
            "-preset",
            "8",
            "-crf",
            "32",
            "-pix_fmt",
            "yuv420p",
            "-c:a",
            "aac",
            "-b:a",
            "160k",
            "-movflags",
            "+faststart",
        ]

    else:
        raise RuntimeError(
            "Unknown profile"
        )

    return (
        common
        + codec
        + [output]
    )


def set_result(
    job_id,
    status,
    error=None
):
    conn = db()

    conn.execute("""
        UPDATE jobs
        SET
            status = ?,
            error = ?,
            finished_at = ?
        WHERE id = ?
    """, (
        status,
        error,
        int(time.time()),
        job_id
    ))

    conn.commit()
    conn.close()


def process(job):
    command = build_command(job)

    log_path = (
        LOG_DIR
        / f"{job['id']}.log"
    )

    try:
        with log_path.open(
            "w",
            encoding="utf-8"
        ) as log:
            result = subprocess.run(
                command,
                stdout=subprocess.DEVNULL,
                stderr=log,
                check=False,
                text=True
            )

        if result.returncode == 0:
            set_result(
                job["id"],
                "done"
            )

            Path(
                job["input_path"]
            ).unlink(
                missing_ok=True
            )

        else:
            set_result(
                job["id"],
                "failed",
                f"FFmpeg exit code "
                f"{result.returncode}"
            )

    except Exception as exc:
        set_result(
            job["id"],
            "failed",
            str(exc)[:500]
        )


def main():
    while True:
        job = claim_job()

        if job is None:
            time.sleep(2)
            continue

        process(job)


if __name__ == "__main__":
    main()

[Unit]
Description=Video Transcoder API
After=network.target

[Service]
User=transcoder
Group=transcoder

WorkingDirectory=/opt/video-transcoder

Environment="TRANSCODER_API_KEY=YOUR_SECRET_KEY"

ExecStart=/opt/video-transcoder/venv/bin/uvicorn 
    app:app 
    --host 127.0.0.1 
    --port 8200 
    --workers 2

Restart=on-failure
RestartSec=3

[Install]
WantedBy=multi-user.target

[Unit]
Description=FFmpeg Transcoding Worker
After=network.target

[Service]
User=transcoder
Group=transcoder

WorkingDirectory=/opt/video-transcoder

ExecStart=/opt/video-transcoder/venv/bin/python 
    /opt/video-transcoder/worker.py

Restart=always
RestartSec=3

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload

sudo systemctl enable --now 
  video-transcoder-api

sudo systemctl enable --now 
  video-transcoder-worker

server {
    listen 80;
    server_name video-api.example.com;

    client_max_body_size 20g;

    location / {
        proxy_request_buffering off;

        proxy_pass http://127.0.0.1:8200;

        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;

        proxy_read_timeout 300s;
        proxy_send_timeout 300s;
    }

    location /_transcoded/ {
        internal;

        alias /srv/transcoder/outputs/;

        add_header X-Content-Type-Options nosniff always;
    }
}

curl -X POST \
  https://video-api.example.com/jobs \
  -H "X-API-Key: YOUR_SECRET_KEY" \
  -F "profile=h264-720p" \
  -F "file=@movie.mp4"

{
  "job_id": "5d33...",
  "status": "queued",
  "profile": "h264-720p",
  "status_url": "/jobs/5d33..."
}

curl \
  -H "X-API-Key: YOUR_SECRET_KEY" \
  https://video-api.example.com/jobs/JOB_ID

ffmpeg \
  -i input.mp4 \
  -map 0:v:0 \
  -map 0:a? \
  -c:v libx264 \
  -preset medium \
  -crf 22 \
  -pix_fmt yuv420p \
  -flags +cgop \
  -force_key_frames "expr:gte(t,n_forced*4)" \
  -c:a aac \
  -b:a 128k \
  -f hls \
  -hls_time 4 \
  -hls_playlist_type vod \
  -hls_flags independent_segments+temp_file \
  -hls_segment_filename \
  "hls/segment_%05d.ts" \
  hls/index.m3u8

360p
480p
720p
1080p
2160p

ffmpeg -encoders | grep nvenc
ffmpeg -encoders | grep qsv
ffmpeg -encoders | grep vaapi

ffmpeg \
  -i input.mp4 \
  -c:v h264_nvenc \
  -preset p5 \
  -cq 23 \
  -c:a aac \
  output.mp4

ffmpeg \
  -i input.mp4 \
  -c:v h264_qsv \
  -global_quality 23 \
  -c:a aac \
  output.mp4

✅ FFmpeg را به‌روز نگه دارید.

✅ API را با Authentication محافظت کنید.

✅ Command خام FFmpeg از کاربر نگیرید.

✅ shell=True استفاده نکنید.

✅ حجم Upload را محدود کنید.

✅ تعداد Jobهای هم‌زمان را کنترل کنید.

✅ فایل خروجی را از مسیر internal Nginx تحویل دهید.

✅ فایل‌های ورودی قدیمی و Jobهای Failed را Cleanup کنید.

✅ Disk Usage را مانیتور کنید.

API
 |
 v
Redis / RabbitMQ
 |
 +-----------+-----------+
 |           |           |
Worker 1   Worker 2    Worker 3
 |           |           |
 +-----------+-----------+
             |
             v
       Object Storage
             |
             v
            CDN

df -h
du -sh /srv/transcoder/*

-movflags +faststart

جمع‌بندی

برای ساخت Video Transcoding Server لازم نیست Encoder را در Python پیاده‌سازی کنید. FFmpeg بخش پردازش ویدئو را بسیار بهتر انجام می‌دهد و Python می‌تواند لایه مدیریتی آن باشد.

معماری API + Queue + Worker باعث می‌شود Upload و درخواست‌های وب از Jobهای طولانی FFmpeg جدا باشند. همین ساختار بعداً به‌سادگی می‌تواند به Redis، چند Worker Node، Object Storage و CDN توسعه پیدا کند.

برای شروع، H.264/MP4 بهترین Profile عمومی است. بعد از راه‌اندازی پایدار سرویس می‌توانید HEVC، AV1، HLS و در نهایت GPU Transcoding را اضافه کنید.

منابع رسمی

آخرین نسخه FFmpeg

مستندات Codecهای FFmpeg

مستندات HLS و Formatهای FFmpeg

File Upload در FastAPI