feat(media): هر رسانهٔ آپلودی به WebP نرمال می‌شود — Pillow مسیر اصلی، FFmpeg برای ویدیو و فرمت‌های خاص؛ ویدیو تا ۸ثانیه/۱۲fps وب متحرک، غیرقابل‌رمزگشایی ۴۱۵

This commit is contained in:
aramesh-site deploy
2026-09-30 09:27:21 +00:00
parent 79ebe69221
commit c0eaac1daa
6 changed files with 258 additions and 71 deletions
+6
View File
@@ -9,6 +9,12 @@ ENV PYTHONDONTWRITEBYTECODE=1 \
COPY requirements.txt . COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt RUN pip install --no-cache-dir -r requirements.txt
# FFmpeg — نرمال‌سازی رسانه به WebP برای ورودی‌هایی که Pillow نمی‌خواند
# (ویدیو → WebP متحرک، فرمت‌های تصویری خاص). مسیر Pillow به آن وابسته نیست.
RUN apt-get update \
&& apt-get install -y --no-install-recommends ffmpeg \
&& rm -rf /var/lib/apt/lists/*
COPY app ./app COPY app ./app
COPY alembic.ini ./ COPY alembic.ini ./
COPY migrations ./migrations COPY migrations ./migrations
+154 -65
View File
@@ -3,8 +3,13 @@
import csv import csv
import ipaddress import ipaddress
import io import io
import json
import os
import re import re
import shutil
import socket import socket
import subprocess
import tempfile
import urllib.request import urllib.request
import uuid import uuid
from datetime import datetime, timezone from datetime import datetime, timezone
@@ -330,41 +335,151 @@ register_crud("footer", models.FooterContent, FooterOut, FooterIn, FooterPatch)
register_crud("seo-meta", models.SeoMeta, SeoMetaOut, SeoMetaIn, SeoMetaPatch) register_crud("seo-meta", models.SeoMeta, SeoMetaOut, SeoMetaIn, SeoMetaPatch)
# آپلود تصویر به Object Storage (MinIO) — URL نسبی برمی‌گرداند تا بک‌اند ذخیره‌سازی قابل تعویض باشد # آپلود رسانه به Object Storage (MinIO) — هر ورودی قابل‌رمزگشایی به WebP نرمال می‌شود:
ALLOWED_IMAGE_TYPES = { # تصویر ثابت → WebP تک‌فریمی | gif/apng/webp متحرک → WebP متحرک | ویدیو → WebP متحرک با سقف.
"image/jpeg": "jpg", # URL نسبی برمی‌گردد تا بک‌اند ذخیره‌سازی قابل تعویض باشد.
"image/png": "png", MAX_DIMENSION = 2000 # سقف ضلع تصویر ثابت — ابعاد بزرگ اصلی‌ترین عامل کندی سایت است
"image/webp": "webp", MAX_ANIM_DIMENSION = 800 # انیمیشن: سقف ضلع فریم
"image/gif": "gif", MAX_ANIM_FRAMES = 150 # انیمیشن: سقف فریم
"image/avif": "avif", VIDEO_MAX_SECONDS = 8 # ویدیو: سقف مدت ورودی
} VIDEO_FPS = 12 # ویدیو: نرخ فریم خروجی
VIDEO_WEBP_LIMIT = 5 * 1024 * 1024 # انیمیشن ویدیویی حجیم‌تر از این → فقط فریم پوستر
WEBP_QUALITY = 82
WEBP_CONTENT_TYPE = "image/webp"
# بهینه‌سازی در زمان آپلود — ابعاد بزرگ و فشرده‌سازی ضعیف، اصلی‌ترین عامل کندی سایت است SUPPORTED_MEDIA_DETAIL = "فرمت پشتیبانی نمی‌شود — فقط تصویر یا ویدیویی که بتوان به WebP تبدیل کرد مجاز است"
MAX_DIMENSION = 2000
REENCODE_TYPES = {"image/jpeg", "image/png", "image/webp"}
def _optimize_image(data: bytes, content_type: str) -> tuple[bytes, str, int, int]: def _to_webp_mode(img: Image.Image) -> Image.Image:
"""تصویر را به حداکثر ۲۰۰۰ پیکسل محدود و دوباره فشرده می‌کند؛ """ذخیرهٔ WebP در Pillow فقط RGB/RGBA/L را می‌پذیرد — مدهای دیگر با حفظ شفافیت تبدیل می‌شوند."""
اگر خروجی بزرگ‌تر از اصلی شد، همان اصلی برگردد. (gif/avif دست‌نخورده)""" if img.mode in ("RGB", "RGBA", "L"):
img = Image.open(io.BytesIO(data)) return img
has_alpha = img.mode in ("LA", "PA") or (img.mode == "P" and "transparency" in img.info)
return img.convert("RGBA" if has_alpha else "RGB")
def _webp_still(img: Image.Image) -> tuple[bytes, int, int]:
"""تک‌فریم → WebP؛ چرخش EXIF اعمال، متادیتا حذف و سقف ضلع ۲۰۰۰ پیکسل رعایت می‌شود."""
img = ImageOps.exif_transpose(img) img = ImageOps.exif_transpose(img)
orig_w, orig_h = img.size
if max(img.size) > MAX_DIMENSION: if max(img.size) > MAX_DIMENSION:
img.thumbnail((MAX_DIMENSION, MAX_DIMENSION), Image.LANCZOS) img.thumbnail((MAX_DIMENSION, MAX_DIMENSION), Image.LANCZOS)
img = _to_webp_mode(img)
buf = io.BytesIO() buf = io.BytesIO()
if content_type == "image/jpeg": img.save(buf, "WEBP", quality=WEBP_QUALITY, method=4)
if img.mode not in ("RGB", "L"): return buf.getvalue(), *img.size
img = img.convert("RGB")
img.save(buf, "JPEG", quality=82, optimize=True, progressive=True)
elif content_type == "image/png": def _webp_animated(img: Image.Image) -> tuple[bytes, int, int] | None:
img.save(buf, "PNG", optimize=True) """چندفریم (gif/apng/webp متحرک) → WebP متحرک؛ None یعنی ورودی تک‌فریمی است.
فریم‌های بیش از سقف رها می‌شوند تا حافظه و حجم خروجی مهار شود."""
total = getattr(img, "n_frames", 1)
if total <= 1:
return None
count = min(total, MAX_ANIM_FRAMES)
frames: list[Image.Image] = []
durations: list[int] = []
for i in range(count):
img.seek(i)
durations.append(int(img.info.get("duration") or 100))
frame = img.convert("RGBA")
if max(frame.size) > MAX_ANIM_DIMENSION:
frame.thumbnail((MAX_ANIM_DIMENSION, MAX_ANIM_DIMENSION), Image.LANCZOS)
frames.append(frame)
buf = io.BytesIO()
frames[0].save(
buf, "WEBP", save_all=True, append_images=frames[1:],
duration=durations, loop=0, quality=WEBP_QUALITY, method=4,
)
return buf.getvalue(), *frames[0].size
def _ffprobe_video(path: str) -> tuple[dict | None, dict]:
"""استریم ویدیوی اول + مشخصات فرمت با ffprobe — استریم None یعنی ویدیویی در فایل نیست."""
ffprobe = shutil.which("ffprobe")
if ffprobe is None:
return None, {}
try:
proc = subprocess.run(
[ffprobe, "-v", "error", "-print_format", "json", "-show_streams", "-show_format", path],
capture_output=True, timeout=20,
)
except (subprocess.TimeoutExpired, OSError):
return None, {}
if proc.returncode != 0:
return None, {}
try:
info = json.loads(proc.stdout or "{}")
except json.JSONDecodeError:
return None, {}
video = next((s for s in info.get("streams") or [] if s.get("codec_type") == "video"), None)
return video, info.get("format") or {}
def _run_ffmpeg(args: list[str]) -> None:
try:
proc = subprocess.run(args, capture_output=True, timeout=60)
except subprocess.TimeoutExpired as exc:
raise HTTPException(status_code=415, detail="تبدیل رسانه بیش از حد طول کشید") from exc
except OSError as exc:
raise HTTPException(status_code=415, detail=SUPPORTED_MEDIA_DETAIL) from exc
if proc.returncode != 0:
raise HTTPException(status_code=415, detail=SUPPORTED_MEDIA_DETAIL)
def _ffmpeg_webp(data: bytes) -> tuple[bytes, int | None, int | None]:
"""مسیر FFmpeg — ورودی‌هایی که Pillow نمی‌خواند: ویدیو و فرمت‌های تصویری خاص.
ویدیو تا ۸ ثانیه با ۱۲fps به WebP متحرک می‌رسد؛ خروجی حجیم → فقط فریم پوستر."""
if shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None:
raise HTTPException(status_code=415, detail=SUPPORTED_MEDIA_DETAIL)
with tempfile.TemporaryDirectory() as tmp:
src = os.path.join(tmp, "src")
dst = os.path.join(tmp, "out.webp")
with open(src, "wb") as fh:
fh.write(data)
video, fmt = _ffprobe_video(src)
if video is None:
raise HTTPException(status_code=415, detail=SUPPORTED_MEDIA_DETAIL)
ff = shutil.which("ffmpeg") or "ffmpeg"
animated = float(fmt.get("duration") or 0) >= 0.5
video_vf = f"fps={VIDEO_FPS},scale='min({MAX_ANIM_DIMENSION},iw)':-2:flags=lanczos"
still_vf = f"scale='min({MAX_DIMENSION},iw)':-2:flags=lanczos"
def run_convert(extra: list[str], vf: str) -> None:
_run_ffmpeg([ff, "-y", "-v", "error", "-i", src, *extra, "-vf", vf,
"-c:v", "libwebp", "-q:v", str(WEBP_QUALITY), "-map_metadata", "-1", dst])
# ویدیو → متحرک با سقف فریم‌نرخ/ضلع/مدت؛ تصویر خاص → تک‌فریم
if animated:
run_convert(["-t", str(VIDEO_MAX_SECONDS)], video_vf)
else: else:
img.save(buf, "WEBP", quality=82, method=4) run_convert(["-frames:v", "1"], still_vf)
optimized = buf.getvalue() with open(dst, "rb") as fh:
if len(optimized) < len(data): out = fh.read()
return optimized, content_type, *img.size # انیمیشن ویدیویی ناگذیر حجیم → فقط فریم پوستر
return data, content_type, orig_w, orig_h if animated and len(out) > VIDEO_WEBP_LIMIT:
run_convert(["-frames:v", "1"], still_vf)
with open(dst, "rb") as fh:
out = fh.read()
dims = _ffprobe_video(dst)[0] or {}
width = int(dims["width"]) if dims.get("width") else None
height = int(dims["height"]) if dims.get("height") else None
return out, width, height
def _normalize_media(data: bytes) -> tuple[bytes, int | None, int | None]:
"""هر ورودی قابل‌رمزگشایی → WebP. اول Pillow (دقیق‌ترین برای EXIF/شفافیت/انیمیشن)؛
اگر رمزگشایی نشد، FFmpeg (ویدیو و فرمت‌های خاص). هیچ‌کدام نشد → ۴۱۵."""
try:
img = Image.open(io.BytesIO(data))
animated = _webp_animated(img)
if animated is not None:
return animated
return _webp_still(img)
except HTTPException:
raise
except Exception:
pass
return _ffmpeg_webp(data)
def _readable_key(name: str, ext: str) -> str: def _readable_key(name: str, ext: str) -> str:
@@ -380,10 +495,9 @@ def _readable_key(name: str, ext: str) -> str:
return f"{datetime.now(timezone.utc):%Y/%m}/{base}.{ext}" return f"{datetime.now(timezone.utc):%Y/%m}/{base}.{ext}"
def _store_image(data: bytes, content_type: str, name: str) -> str: def _store_image(data: bytes, name: str) -> str:
"""ذخیره تصویر در Object Storage با کلید خوانا — URL نسبی برمی‌گردد.""" """ذخیرهٔ WebP در Object Storage با کلید خوانا — URL نسبی برمی‌گردد."""
ext = ALLOWED_IMAGE_TYPES[content_type] key = _readable_key(name, "webp")
key = _readable_key(name, ext)
s3 = boto3.client( s3 = boto3.client(
"s3", "s3",
endpoint_url=settings.s3_endpoint, endpoint_url=settings.s3_endpoint,
@@ -392,7 +506,7 @@ def _store_image(data: bytes, content_type: str, name: str) -> str:
region_name="us-east-1", region_name="us-east-1",
) )
try: try:
s3.put_object(Bucket=settings.s3_bucket, Key=key, Body=data, ContentType=content_type) s3.put_object(Bucket=settings.s3_bucket, Key=key, Body=data, ContentType=WEBP_CONTENT_TYPE)
except Exception: except Exception:
raise HTTPException(status_code=502, detail="ذخیره فایل در Object Storage ناموفق بود") raise HTTPException(status_code=502, detail="ذخیره فایل در Object Storage ناموفق بود")
return f"{settings.media_public_base}/{key}" return f"{settings.media_public_base}/{key}"
@@ -404,14 +518,6 @@ async def upload_media(
name: str = Form(""), name: str = Form(""),
_admin: models.AdminUser = Depends(get_current_admin), _admin: models.AdminUser = Depends(get_current_admin),
): ):
content_type = (file.content_type or "").lower()
ext = ALLOWED_IMAGE_TYPES.get(content_type)
if ext is None:
raise HTTPException(
status_code=415,
detail="فرمت فایل پشتیبانی نمی‌شود (مجاز: jpg، png، webp، gif، avif)",
)
max_bytes = settings.media_max_upload_mb * 1024 * 1024 max_bytes = settings.media_max_upload_mb * 1024 * 1024
data = await file.read() data = await file.read()
if len(data) == 0: if len(data) == 0:
@@ -422,23 +528,18 @@ async def upload_media(
detail=f"حجم فایل بیش از {settings.media_max_upload_mb} مگابایت است", detail=f"حجم فایل بیش از {settings.media_max_upload_mb} مگابایت است",
) )
width = height = None # هر ورودی قابل‌رمزگشایی (تصویر رایج/خاص، متحرک، ویدیو) → WebP نرمال
if content_type in REENCODE_TYPES: data, width, height = _normalize_media(data)
try:
data, content_type, width, height = _optimize_image(data, content_type)
except Exception:
# تصویر خراب/ناشناخته — همان داده اصلی ذخیره می‌شود
pass
# نام خوانا: نام ارسالی فرم، وگرنه نام اصلی فایل بدون پسوند # نام خوانا: نام ارسالی فرم، وگرنه نام اصلی فایل بدون پسوند
display_name = (name or "").strip() or (file.filename or "").rsplit(".", 1)[0] display_name = (name or "").strip() or (file.filename or "").rsplit(".", 1)[0]
url = _store_image(data, content_type, display_name) url = _store_image(data, display_name)
return { return {
"url": url, "url": url,
"key": url.removeprefix(f"{settings.media_public_base}/"), "key": url.removeprefix(f"{settings.media_public_base}/"),
"size": len(data), "size": len(data),
"content_type": content_type, "content_type": WEBP_CONTENT_TYPE,
"width": width, "width": width,
"height": height, "height": height,
} }
@@ -481,15 +582,7 @@ async def import_image(
) )
try: try:
with urllib.request.urlopen(request, timeout=15) as response: with urllib.request.urlopen(request, timeout=15) as response:
content_type = (response.headers.get("Content-Type") or "").split(";")[0].strip().lower()
if content_type not in ALLOWED_IMAGE_TYPES:
raise HTTPException(
status_code=415,
detail="فرمت تصویر پشتیبانی نمی‌شود (مجاز: jpg، png، webp، gif، avif)",
)
data = response.read(max_bytes + 1) data = response.read(max_bytes + 1)
except HTTPException:
raise
except Exception: except Exception:
raise HTTPException(status_code=502, detail="دانلود تصویر از نشانی ناموفق بود") raise HTTPException(status_code=502, detail="دانلود تصویر از نشانی ناموفق بود")
if len(data) == 0: if len(data) == 0:
@@ -500,22 +593,18 @@ async def import_image(
detail=f"حجم تصویر بیش از {settings.media_max_upload_mb} مگابایت است", detail=f"حجم تصویر بیش از {settings.media_max_upload_mb} مگابایت است",
) )
width = height = None # هدر Content-Type دروغ می‌گوید — خود بدنه رمزگشایی و به WebP نرمال می‌شود
if content_type in REENCODE_TYPES: data, width, height = _normalize_media(data)
try:
data, content_type, width, height = _optimize_image(data, content_type)
except Exception:
pass
# نام خوانا از مسیر نشانی — بدون پارامترهای کوئری # نام خوانا از مسیر نشانی — بدون پارامترهای کوئری
url_name = urlparse(payload.url).path.rsplit("/", 1)[-1].rsplit(".", 1)[0] url_name = urlparse(payload.url).path.rsplit("/", 1)[-1].rsplit(".", 1)[0]
media_url = _store_image(data, content_type, url_name) media_url = _store_image(data, url_name)
return { return {
"url": media_url, "url": media_url,
"key": media_url.removeprefix(f"{settings.media_public_base}/"), "key": media_url.removeprefix(f"{settings.media_public_base}/"),
"size": len(data), "size": len(data),
"content_type": content_type, "content_type": WEBP_CONTENT_TYPE,
"width": width, "width": width,
"height": height, "height": height,
} }
+94 -2
View File
@@ -404,12 +404,103 @@ def test_upload_with_readable_name(auth_client, monkeypatch):
) )
assert res.status_code == 200, res.text assert res.status_code == 200, res.text
key = res.json()["key"] key = res.json()["key"]
# نام خوانا (فارسی/لاتین، بدون نویسه ناامن) + پسوند یکتا + پسوند تصویر # نام خوانا (فارسی/لاتین، بدون نویسه ناامن) + پسوند یکتا + نرمال‌شدن به WebP
assert "برج-الهیه-tower" in key assert "برج-الهیه-tower" in key
assert key.endswith(".png") assert key.endswith(".webp")
assert res.json()["content_type"] == "image/webp"
assert "!" not in key assert "!" not in key
def _animated_gif_bytes() -> bytes:
from PIL import Image
import io
buf = io.BytesIO()
first = Image.new("RGB", (12, 8), (200, 120, 60))
second = Image.new("RGB", (12, 8), (60, 120, 200))
first.save(buf, "GIF", save_all=True, append_images=[second], duration=100, loop=0)
return buf.getvalue()
def test_upload_jpeg_converted_to_webp(auth_client, monkeypatch):
_mock_storage(monkeypatch)
from PIL import Image
import io
buf = io.BytesIO()
Image.new("RGB", (3000, 1500), (10, 10, 10)).save(buf, "JPEG", quality=90)
res = auth_client.post(
"/api/admin/upload",
files={"file": ("tower.jpg", buf.getvalue(), "image/jpeg")},
)
assert res.status_code == 200, res.text
body = res.json()
assert body["content_type"] == "image/webp"
assert body["key"].endswith(".webp")
# سقف ضلع ۲۰۰۰ پیکسل روی خروجی اعمال شده
assert body["width"] == 2000 and body["height"] == 1000
def test_upload_animated_gif_becomes_animated_webp(auth_client, monkeypatch):
fake = _mock_storage(monkeypatch)
res = auth_client.post(
"/api/admin/upload",
files={"file": ("anim.gif", _animated_gif_bytes(), "image/gif")},
)
assert res.status_code == 200, res.text
assert res.json()["content_type"] == "image/webp"
from PIL import Image
import io
stored = next(iter(fake.objects.values()))
out = Image.open(io.BytesIO(stored))
assert out.format == "WEBP"
assert getattr(out, "n_frames", 1) == 2 # حرکت از بین نرفته
def test_upload_undecodable_file_rejected(auth_client, monkeypatch):
_mock_storage(monkeypatch)
res = auth_client.post(
"/api/admin/upload",
files={"file": ("notes.txt", b"this is not a media file at all", "text/plain")},
)
assert res.status_code == 415
def test_upload_video_becomes_animated_webp(auth_client, monkeypatch):
# ویدیو تنها با FFmpeg قابل پذیرش است — بدون آن ۴۱۵، با آن WebP متحرک
import shutil as _shutil
import subprocess as _subprocess
import tempfile as _tempfile
import os as _os
_mock_storage(monkeypatch)
if _shutil.which("ffmpeg") is None:
res = auth_client.post(
"/api/admin/upload",
files={"file": ("clip.mp4", b"\x00\x00\x00\x18ftypmp42", "video/mp4")},
)
assert res.status_code == 415
return
with _tempfile.TemporaryDirectory() as tmp:
clip = _os.path.join(tmp, "clip.mp4")
_subprocess.run(
["ffmpeg", "-y", "-v", "error", "-f", "lavfi",
"-i", "testsrc=duration=0.5:size=128x96:rate=10", clip],
check=True, capture_output=True, timeout=30,
)
with open(clip, "rb") as fh:
clip_bytes = fh.read()
res = auth_client.post(
"/api/admin/upload",
files={"file": ("clip.mp4", clip_bytes, "video/mp4")},
)
assert res.status_code == 200, res.text
assert res.json()["content_type"] == "image/webp"
def test_upload_without_name_uses_filename(auth_client, monkeypatch): def test_upload_without_name_uses_filename(auth_client, monkeypatch):
_mock_storage(monkeypatch) _mock_storage(monkeypatch)
res = auth_client.post( res = auth_client.post(
@@ -475,6 +566,7 @@ def test_import_image_happy_path(auth_client, monkeypatch):
res = auth_client.post("/api/admin/import-image", json={"url": "https://wiki.example/Elahieh pic.png?token=1"}) res = auth_client.post("/api/admin/import-image", json={"url": "https://wiki.example/Elahieh pic.png?token=1"})
assert res.status_code == 200, res.text assert res.status_code == 200, res.text
assert res.json()["url"].startswith("/media/") assert res.json()["url"].startswith("/media/")
assert res.json()["key"].endswith(".webp")
assert "Elahieh-pic" in res.json()["key"] assert "Elahieh-pic" in res.json()["key"]
@@ -479,7 +479,7 @@ export function ArticleEditor({ doc, onChange, contentKey }: ArticleEditorProps)
ref={fileRef} ref={fileRef}
type="file" type="file"
data-editor-file-input data-editor-file-input
accept="image/jpeg,image/png,image/webp,image/gif,image/avif" accept="image/*,video/mp4,video/webm,video/quicktime,video/x-msvideo,video/x-matroska"
className="hidden" className="hidden"
onChange={(e) => { onChange={(e) => {
const file = e.target.files?.[0] const file = e.target.files?.[0]
@@ -142,7 +142,7 @@ export function GalleryField({ label, value, onSave, namePrefix = 'image' }: Gal
ref={inputRef} ref={inputRef}
type="file" type="file"
multiple multiple
accept="image/jpeg,image/png,image/webp,image/gif,image/avif" accept="image/*,video/mp4,video/webm,video/quicktime,video/x-msvideo,video/x-matroska"
className="hidden" className="hidden"
onChange={(event) => { onChange={(event) => {
void addFiles(event.target.files) void addFiles(event.target.files)
+1 -1
View File
@@ -191,7 +191,7 @@ export function ImageField({
<input <input
ref={inputRef} ref={inputRef}
type="file" type="file"
accept="image/jpeg,image/png,image/webp,image/gif,image/avif" accept="image/*,video/mp4,video/webm,video/quicktime,video/x-msvideo,video/x-matroska"
className="hidden" className="hidden"
onChange={(event) => { onChange={(event) => {
void pick(event.target.files?.[0]) void pick(event.target.files?.[0])