Kiểm tra đầu vào trước khi gọi API video-to-video: ffprobe, giới hạn thời lượng và lint prompt
Khi tích hợp một API sinh video bằng AI, phần lớn lỗi mình gặp không đến từ model mà đến từ dữ liệu đầu vào: video quá ngắn, file sai định dạng, URL ảnh không công khai, hoặc prompt nhắc tới một ảnh không tồn tại. Mỗi lần gửi request lỗi là mất vài giây chờ, mất một lượt trong hàng đợi, và với một số dịch vụ là mất cả tiền.
Bài này chia sẻ cách mình viết một lớp "validate trước khi gửi" bằng Python cho các API dạng video-to-video (gửi một video nguồn kèm vài ảnh tham chiếu, model giữ lại chuyển động và góc máy rồi thay nhân vật). Ví dụ cụ thể dựa trên tài liệu công khai của một API mà mình đang làm, nhưng ý tưởng áp dụng được cho hầu hết các dịch vụ tương tự.
1. Đọc kỹ ràng buộc của API
Trước khi viết code, hãy liệt kê các ràng buộc thành một danh sách. Với API trong ví dụ:
- Video nguồn: URL công khai, định dạng MP4 hoặc MOV, dài từ 4 đến 30 giây.
- Ảnh tham chiếu: từ 1 đến 20 URL công khai, định dạng JPG, PNG hoặc WebP.
- Prompt: không bắt buộc, tối đa 20.000 ký tự; ảnh được gọi bằng
@Image1,@Image2..., video nguồn là@Video. - Lỗi tương ứng:
VIDEO_TOO_SHORT,VIDEO_TOO_LONG,INVALID_INPUT(bao gồm cả trường hợp prompt nhắc tới ảnh không được gửi kèm).
Điểm quan trọng: những lỗi 400 này không nên retry. Video 3 giây thì gửi lại bao nhiêu lần vẫn là 3 giây. Vì vậy kiểm tra ở phía client là hợp lý nhất.
2. Đo thời lượng bằng ffprobe
ffprobe (đi kèm ffmpeg) đọc được metadata trực tiếp từ URL, không cần tải cả file:
import json, subprocess
def probe(url: str) -> dict:
out = subprocess.run(
["ffprobe", "-v", "error", "-show_entries",
"format=duration,format_name", "-of", "json", url],
capture_output=True, text=True, timeout=30, check=True,
)
fmt = json.loads(out.stdout)["format"]
return {"duration": float(fmt["duration"]), "format": fmt["format_name"]}
format_name của MP4 và MOV đều trả về chuỗi kiểu mov,mp4,m4a,3gp,3g2,mj2, nên chỉ cần kiểm tra có chứa mov hoặc mp4.
MIN_S, MAX_S = 4, 30
def check_video(url: str) -> list[str]:
errors = []
try:
info = probe(url)
except Exception as e:
return [f"không đọc được video: {e}"]
if not any(k in info["format"] for k in ("mp4", "mov")):
errors.append(f"định dạng không hỗ trợ: {info['format']}")
d = info["duration"]
if d < MIN_S:
errors.append(f"video quá ngắn ({d:.1f}s < {MIN_S}s)")
if d > MAX_S:
errors.append(f"video quá dài ({d:.1f}s > {MAX_S}s), hãy cắt bớt")
return errors
Mẹo nhỏ: nếu dịch vụ tính phí theo số giây của video nguồn, hãy cắt bỏ phần thừa ở đầu và cuối trước khi upload. Bớt 2 giây trên một clip 12 giây là giảm khoảng 17% chi phí cho mọi lần sinh.
3. Kiểm tra URL ảnh
Ảnh phải truy cập công khai. Một request HEAD là đủ để bắt lỗi 403/404 và sai content-type:
import httpx
ALLOWED = {"image/jpeg", "image/png", "image/webp"}
def check_images(urls: list[str]) -> list[str]:
errors = []
if not 1 <= len(urls) <= 20:
errors.append(f"cần 1–20 ảnh, đang có {len(urls)}")
with httpx.Client(follow_redirects=True, timeout=10) as c:
for i, u in enumerate(urls, start=1):
r = c.head(u)
ctype = r.headers.get("content-type", "").split(";")[0]
if r.status_code != 200:
errors.append(f"@Image{i}: HTTP {r.status_code}")
elif ctype not in ALLOWED:
errors.append(f"@Image{i}: content-type {ctype}")
return errors
Lưu ý một số CDN không hỗ trợ HEAD; khi đó có thể fallback sang GET với header Range: bytes=0-0.
4. Lint prompt: tham chiếu ảnh có tồn tại không?
Đây là lỗi mình gặp nhiều nhất khi người dùng chỉnh sửa prompt mẫu: xoá bớt ảnh nhưng quên sửa @Image3 trong prompt.
import re
def check_prompt(prompt: str, n_images: int) -> list[str]:
errors = []
if len(prompt) > 20_000:
errors.append("prompt vượt quá 20.000 ký tự")
refs = {int(m) for m in re.findall(r"@Image(\d+)", prompt)}
missing = sorted(r for r in refs if r < 1 or r > n_images)
if missing:
errors.append(f"prompt nhắc tới ảnh không tồn tại: {missing}")
unused = sorted(set(range(1, n_images + 1)) - refs)
if refs and unused:
errors.append(f"cảnh báo: ảnh {unused} không được nhắc trong prompt")
return errors
Mục cuối chỉ là cảnh báo: có model vẫn tự dùng ảnh theo thứ tự, nhưng với cảnh nhiều người (song ca, nhảy nhóm) thì việc gán rõ "@Image1 là người bên trái, @Image2 là người bên phải" giúp kết quả ổn định hơn hẳn.
5. Gộp lại
def validate(job) -> list[str]:
return (check_video(job.video_url)
+ check_images(job.image_urls)
+ check_prompt(job.prompt or "", len(job.image_urls)))
errs = validate(job)
if errs:
raise ValueError("; ".join(errs))
Sau khi qua bước này, request gửi lên gần như chỉ còn có thể lỗi vì lý do phía server (quá tải, kiểm duyệt nội dung, giới hạn số job chạy song song), và đó mới là những lỗi đáng để retry có backoff.
Kết luận
Validate phía client tốn khoảng 50 dòng code nhưng tiết kiệm rất nhiều thời gian debug, công sức hỗ trợ người dùng và chi phí. Nguyên tắc chung: lỗi nào là thuộc tính của dữ liệu thì bắt trước khi gửi, lỗi nào là trạng thái của hệ thống thì mới retry.
Công khai: mình đang làm việc tại Genjutsu AI, các ràng buộc trong bài lấy từ tài liệu API công khai của dịch vụ này (tháng 10/2026).
All rights reserved
