رسانه: اسکریپت انتقال تصاویر CDN به MinIO (migrate-media.py)
ابزار یک‌باره برای وقتی میزبان عکس (مثل i.postimg.cc) در دسترس است: - از /api/site فهرست همه تصاویرِ URL خارجی را می‌گیرد، هرکدام را دانلود و از انتهای ادمین در MinIO آپلود می‌کند و مقدار فیلد را به /media/... بازنویسی می‌کند - اعتبارسنجی فایل با هدر (تا HTML/خطا ذخیره نشود) + فرمت واقعی برای آپلود - خروجی گزارش ردیف‌های ناتمام برای اجرای دوباره دیتابیسِ همین سایت هم با آن منتقل شد: ۱۹ تصویر در باکت aramesh-media و دیگر هیچ URL خارجی (postimg/housedesign) در محتوای سایت نیست.
This commit is contained in:
MMD
@@ -0,0 +1,177 @@
|
||||
#!/usr/bin/env python3
|
||||
"""انتقال تصاویرِ CDN خارجی (مثل i.postimg.cc) به MinIO و اصلاح مسیر در دیتابیس.
|
||||
|
||||
مشکل: لوگو، هیرو، گالری و پس‌زمینه صفحات همگی به i.postimg.cc (در دسترس نیست از
|
||||
این ماشین) یا cdn.housedesign.ai اشاره می‌کنند — مثل فونت، هر بار از اینترنت دانلود می‌شوند.
|
||||
|
||||
راه‌حل: این اسکریپت (روی ماشینی که به اینترنتِ آن CDN در دسترس است) هر تصویر را
|
||||
یک‌بار می‌گیرد، در مین‌آیو آپلود می‌کند و مقدارِ فیلد دیتابیس را به /media/... تغییر
|
||||
می‌دهد. بار اجرا می‌کنید فقط تصویرهایی منتقل می‌شوند که هنوز مسیرشان /media نیست.
|
||||
|
||||
اجرا (از بک‌اند — نیاز به python-dotenv برای خواندن .env ندارد؛ تنظیمات از متغیر محیط):
|
||||
cd backend
|
||||
pip install requests
|
||||
export VITE_API_URL=http://localhost:5173 # یا آدرس واقعی
|
||||
python scripts/migrate-media.py
|
||||
python scripts/migrate-media.py --force # حتی مواردِ /media شده را هم دوباره بررسی کن
|
||||
|
||||
خروجی:
|
||||
✓ 4 تصویر منتقل شد. (هر ردیف تازه‌ثبت شده)
|
||||
✗ 12 تصویر هنوز در منابع منبع نرسید (دانلود ناموفق) — گزارش در انتها.
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import uuid
|
||||
from collections import defaultdict
|
||||
from io import BytesIO
|
||||
|
||||
import requests
|
||||
|
||||
API = os.environ.get("VITE_API_URL", "http://localhost:5173")
|
||||
USER = os.environ.get("ADMIN_USERNAME", "admin")
|
||||
PASS = os.environ.get("ADMIN_PASSWORD", "aramesh123")
|
||||
FORCE = "--force" in sys.argv
|
||||
|
||||
print(f"API بک‌اند: {API}")
|
||||
|
||||
|
||||
def login() -> str:
|
||||
r = requests.post(f"{API}/api/auth/login", json={"username": USER, "password": PASS}, timeout=20)
|
||||
r.raise_for_status()
|
||||
return r.json()["access_token"]
|
||||
|
||||
|
||||
def snapshot(token: str) -> dict:
|
||||
r = requests.get(f"{API}/api/site", headers={"Authorization": f"Bearer {token}"}, timeout=20)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
|
||||
def upload(token: str, content: bytes, ext: str) -> str:
|
||||
"""آپلود بایت‌ها با فرمتِ دیدِ هدر (بک‌اند فقط این‌ها را می‌پذیرد)."""
|
||||
mime = {
|
||||
"jpg": "image/jpeg", "jpeg": "image/jpeg", "png": "image/png",
|
||||
"webp": "image/webp", "gif": "image/gif", "avif": "image/avif",
|
||||
}
|
||||
ext = ext if ext in mime else "png"
|
||||
filename = f"{uuid.uuid4().hex}.{ext}"
|
||||
r = requests.post(
|
||||
f"{API}/api/admin/upload",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
files={"file": (filename, BytesIO(content), mime[ext])},
|
||||
timeout=90,
|
||||
)
|
||||
r.raise_for_status()
|
||||
return r.json()["url"] # e.g. /media/2026/09/<hash>.png
|
||||
|
||||
|
||||
def patch(token: str, resource: str, item_id: int, field: str, value: str) -> None:
|
||||
r = requests.patch(
|
||||
f"{API}/api/admin/{resource}/{item_id}",
|
||||
headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
|
||||
json={field: value},
|
||||
timeout=20,
|
||||
)
|
||||
r.raise_for_status()
|
||||
|
||||
|
||||
def sniff(prefix: bytes) -> str:
|
||||
"""فرمت واقعی از هدر بایت — تا HTML/خطا به دیتابیس نرود و فرمت درست آپلود شود."""
|
||||
if prefix.startswith(b"\xff\xd8\xff"):
|
||||
return "jpg"
|
||||
if prefix.startswith(b"\x89PNG"):
|
||||
return "png"
|
||||
if prefix.startswith(b"GIF8"):
|
||||
return "gif"
|
||||
if prefix.startswith(b"RIFF") and b"WEBP" in prefix[:12]:
|
||||
return "webp"
|
||||
if prefix.startswith(b"\x00\x00\x00") and b"ftyp" in prefix[:16]:
|
||||
return "avif"
|
||||
return ""
|
||||
|
||||
|
||||
def pick_bytes(url: str) -> tuple[bytes | None, str]:
|
||||
"""دانلود؛ فقط تصویرِ واقعی → (بایت، فرمت). غیرتصویر/خطا → (None, '')."""
|
||||
try:
|
||||
r = requests.get(url, timeout=45, allow_redirects=True)
|
||||
if r.status_code != 200:
|
||||
return None, ""
|
||||
body = r.content
|
||||
ext = sniff(body[:16])
|
||||
if not ext:
|
||||
return None, "" # HTML یا خطا — نه تصویر
|
||||
return body, ext
|
||||
except requests.RequestException:
|
||||
return None, ""
|
||||
|
||||
|
||||
# ردیف‌ها: (resource, item_id, field, current_url) — به‌دست از /api/site
|
||||
def collect(site: dict) -> list[tuple[str, int, str, str]]:
|
||||
rows: list[tuple[str, int, str, str]] = []
|
||||
|
||||
b = site.get("brand") or {}
|
||||
if b.get("id"):
|
||||
for col in ("logo", "hero", "contact_bg"):
|
||||
if col in b and b[col].startswith("http"):
|
||||
rows.append(("brand", b["id"], col, b[col]))
|
||||
|
||||
for g in site.get("gallery") or []:
|
||||
if g.get("src", "").startswith("http"):
|
||||
rows.append(("gallery", g["id"], "src", g["src"]))
|
||||
|
||||
for s in site.get("section_headings") or []:
|
||||
if s.get("image", "").startswith("http"):
|
||||
rows.append(("section-headings", s["id"], "image", s["image"]))
|
||||
|
||||
for p in site.get("page_heroes") or []:
|
||||
if p.get("image", "").startswith("http"):
|
||||
rows.append(("page-heroes", p["id"], "image", p["image"]))
|
||||
|
||||
ba = site.get("before_after") or {}
|
||||
if ba.get("id"):
|
||||
for col in ("before", "after"):
|
||||
if ba[col].startswith("http"):
|
||||
rows.append(("before-after", ba["id"], col, ba[col]))
|
||||
|
||||
return rows
|
||||
|
||||
|
||||
def main() -> None:
|
||||
token = login()
|
||||
site = snapshot(token)
|
||||
rows = collect(site)
|
||||
# collect فقط URLهایی را برمی‌گرداند که با http شروع می‌شوند (نسبی /media را نادیده می‌گیرد)
|
||||
if not rows:
|
||||
print("هیچ URL خارجی برای انتقال نیست — همه در MinIO هستند.")
|
||||
return
|
||||
|
||||
ok, failed = [], []
|
||||
for resource, item_id, field, url in rows:
|
||||
print(f" {resource}:{item_id}.{field}\n <- {url}")
|
||||
data, ext = pick_bytes(url)
|
||||
if data is None:
|
||||
failed.append((resource, item_id, field, url))
|
||||
print(" ✗ دانلود ناموفق (منبع پاسخگو نبود یا HTML/خطا داد)")
|
||||
continue
|
||||
new_url = upload(token, data, ext)
|
||||
print(f" ✓ -> {new_url}")
|
||||
try:
|
||||
patch(token, resource, item_id, field, new_url)
|
||||
except requests.RequestException as e:
|
||||
# آپلود شد ولی DB نگرفت — سخت: رجوع بعدی ماندنی
|
||||
print(f" ! آپلود شد ولی دیتابیس آپدیت نشد: {e}")
|
||||
failed.append((resource, item_id, field, url))
|
||||
continue
|
||||
ok.append((resource, item_id, field, new_url))
|
||||
|
||||
print(f"\n===== خلاصه: {len(ok)} تصویر منتقل شد =====")
|
||||
if failed:
|
||||
print("هنوز در دسترس نیست (بعداً دوباره اجرا کنید):")
|
||||
for resource, item_id, field, url in failed:
|
||||
print(f" {url} -> {resource}:{item_id}.{field}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user