aramesh-site deploy
05945099a1
پارامتر تعریف شده بود ولی هنگام ثبت مقاله‌ها پاس داده نمی‌شد؛ نتیجه: ساخت/ویرایش نقش creator مثل ادمین رفتار می‌کرد (status=published قابل ثبت).
604 lines
26 KiB
Python
604 lines
26 KiB
Python
"""انتهای پنل مدیریت — همه مسیرها نیازمند توکن JWT معتبر هستند."""
|
|
|
|
import csv
|
|
import ipaddress
|
|
import io
|
|
import re
|
|
import socket
|
|
import urllib.request
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from urllib.parse import urlparse
|
|
|
|
import boto3
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, Response, UploadFile, status
|
|
from PIL import Image, ImageOps
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import func
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .. import models, schemas
|
|
from ..auth import create_access_token, get_current_admin, get_user_from_token, hash_password, oauth2_scheme, verify_password
|
|
from ..config import settings
|
|
from ..database import get_db
|
|
from ..rate_limit import clear as clear_rate_limit
|
|
from ..rate_limit import ensure_allowed, record_failure
|
|
from ..schemas import (
|
|
ArticleIn,
|
|
ArticleOut,
|
|
ArticlePatch,
|
|
BeforeAfterIn,
|
|
BeforeAfterOut,
|
|
BeforeAfterPatch,
|
|
BrandIn,
|
|
BrandOut,
|
|
BrandPatch,
|
|
ChangePasswordIn,
|
|
ContactSectionIn,
|
|
ContactSectionOut,
|
|
ContactSectionPatch,
|
|
ContactMessageOut,
|
|
FaqIn,
|
|
FaqOut,
|
|
FaqPatch,
|
|
FooterIn,
|
|
FooterOut,
|
|
FooterPatch,
|
|
GalleryImageIn,
|
|
GalleryImageOut,
|
|
GalleryImagePatch,
|
|
HeroIn,
|
|
HeroOut,
|
|
HeroPatch,
|
|
InvestmentModelIn,
|
|
InvestmentModelOut,
|
|
InvestmentModelPatch,
|
|
LoginIn,
|
|
MessageReadIn,
|
|
NavItemIn,
|
|
NavItemOut,
|
|
NavItemPatch,
|
|
NewsletterSectionIn,
|
|
NewsletterOut,
|
|
NewsletterSectionPatch,
|
|
OfficeIn,
|
|
OfficeOut,
|
|
OfficePatch,
|
|
PageHeroIn,
|
|
PageHeroOut,
|
|
PageHeroPatch,
|
|
ProcessStepIn,
|
|
ProcessStepOut,
|
|
ProcessStepPatch,
|
|
ProjectIn,
|
|
ProjectOut,
|
|
ProjectPatch,
|
|
ReorderIn,
|
|
SeoMetaIn,
|
|
SeoMetaOut,
|
|
SeoMetaPatch,
|
|
SectionHeadingIn,
|
|
SectionHeadingOut,
|
|
SectionHeadingPatch,
|
|
ServiceIn,
|
|
ServiceOut,
|
|
ServicePatch,
|
|
StatIn,
|
|
StatOut,
|
|
StatPatch,
|
|
StatusLabelIn,
|
|
StatusLabelOut,
|
|
StatusLabelPatch,
|
|
StoryIn,
|
|
StoryOut,
|
|
StoryPatch,
|
|
SubscriberOut,
|
|
TeamMemberIn,
|
|
TeamMemberOut,
|
|
TeamMemberPatch,
|
|
TestimonialIn,
|
|
TestimonialOut,
|
|
TestimonialPatch,
|
|
TimelineItemIn,
|
|
TimelineItemOut,
|
|
TimelineItemPatch,
|
|
TokenOut,
|
|
WhyUsItemIn,
|
|
WhyUsItemOut,
|
|
WhyUsItemPatch,
|
|
)
|
|
|
|
# روتر عمومی برای ورود — بدون نیاز به توکن
|
|
auth_router = APIRouter()
|
|
|
|
# روتر محافظت‌شده — همه مسیرهای آن فقط با توکن معتبر قابل دسترسی هستند
|
|
async def admin_access_guard(
|
|
request: Request,
|
|
token: str = Depends(oauth2_scheme),
|
|
db: Session = Depends(get_db),
|
|
) -> models.AdminUser:
|
|
"""نقش‌گارد — 'admin' همه‌کاره؛ 'creator' فقط مقاله‌نویسی: ساخت پیش‌نویس،
|
|
خواندن/ویرایش پیش‌نویس خودش، و آپلود تصویر. بقیهٔ مسیرهای مدیریتی ۴۰۳."""
|
|
user = get_user_from_token(token, db)
|
|
if user.role == "admin":
|
|
return user
|
|
if user.role != "creator":
|
|
raise HTTPException(status_code=403, detail="نقش ناشناخته")
|
|
|
|
path = request.url.path.rstrip("/")
|
|
method = request.method
|
|
|
|
# آپلود تصویر مقاله‌ها — مجاز
|
|
if path in ("/api/admin/upload", "/api/admin/import-image") and method == "POST":
|
|
return user
|
|
# فهرست و ساخت مقاله — مجاز (ساخت همیشه پیش‌نویس است و نویسنده‌اش ثبت می‌شود)
|
|
if path == "/api/admin/articles" and method in ("GET", "POST"):
|
|
return user
|
|
# خواندن/ویرایش پیش‌نویس خودِ نویسنده — مالکیت در هندلر هم دوباره چک می‌شود
|
|
m = re.fullmatch(r"/api/admin/articles/(\d+)", path)
|
|
if m and method in ("GET", "PUT", "PATCH"):
|
|
article = db.get(models.Article, int(m.group(1)))
|
|
if article is not None and article.author_id == user.id:
|
|
return user
|
|
# ۴۰۴ نه ۴۰۳ — مقاله‌های بقیه برای نویسنده اصلاً وجود ندارند
|
|
raise HTTPException(status_code=404, detail="مقاله یافت نشد")
|
|
raise HTTPException(status_code=403, detail="دسترسی نویسنده محدود است")
|
|
|
|
|
|
router = APIRouter(
|
|
dependencies=[
|
|
Depends(get_current_admin),
|
|
Depends(admin_access_guard),
|
|
]
|
|
)
|
|
|
|
|
|
|
|
@auth_router.post("/auth/login", response_model=TokenOut)
|
|
def login(payload: LoginIn, request: Request, db: Session = Depends(get_db)):
|
|
ensure_allowed(request)
|
|
user = db.query(models.AdminUser).filter(models.AdminUser.username == payload.username).first()
|
|
if user is None or not verify_password(payload.password, user.password_hash):
|
|
record_failure(request)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="نام کاربری یا رمز عبور اشتباه است",
|
|
)
|
|
clear_rate_limit(request)
|
|
return TokenOut(access_token=create_access_token(user.username, user.role), role=user.role)
|
|
|
|
|
|
@auth_router.post("/auth/change-password")
|
|
def change_password(
|
|
payload: ChangePasswordIn,
|
|
admin: models.AdminUser = Depends(get_current_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""تغییر رمز ادمین واردشده — رمز فعلی باید درست باشد."""
|
|
if not verify_password(payload.current_password, admin.password_hash):
|
|
raise HTTPException(status_code=400, detail="رمز عبور فعلی اشتباه است")
|
|
admin.password_hash = hash_password(payload.new_password)
|
|
db.commit()
|
|
return {"ok": True}
|
|
|
|
|
|
def register_crud(
|
|
path: str, model, out_schema, create_schema, patch_schema=None,
|
|
creator_scoped: bool = False,
|
|
) -> None:
|
|
"""ثبت مسیرهای استاندارد CRUD (لیست، ساخت، دریافت، ویرایش کامل، ویرایش جزئی،
|
|
حذف و ترتیب‌بندی) روی روتر مدیریت.
|
|
|
|
creator_scoped=True (فقط مقاله‌ها): نقش creator فقط رکوردهای خودش را می‌بیند،
|
|
ساختش همیشه پیش‌نویس است و ویرایشش هرگز منتشر نمی‌شود."""
|
|
|
|
@router.get(f"/{path}", response_model=list[out_schema])
|
|
def list_items(db: Session = Depends(get_db), admin: models.AdminUser = Depends(get_current_admin)):
|
|
query = db.query(model)
|
|
if creator_scoped and admin.role == "creator":
|
|
query = query.filter(model.author_id == admin.id)
|
|
if hasattr(model, "position"):
|
|
query = query.order_by(model.position, model.id)
|
|
else:
|
|
query = query.order_by(model.id)
|
|
return query.all()
|
|
|
|
@router.post(f"/{path}", response_model=out_schema, status_code=201)
|
|
def create_item(payload: create_schema, db: Session = Depends(get_db), admin: models.AdminUser = Depends(get_current_admin)):
|
|
item = model(**payload.model_dump())
|
|
if creator_scoped:
|
|
# نویسنده ثبت می‌شود؛ ساخت توسط creator همیشه پیش‌نویس است
|
|
item.author_id = admin.id
|
|
if admin.role == "creator":
|
|
item.status = "draft"
|
|
if hasattr(model, "position"):
|
|
# آیتم جدید در انتهای لیست قرار می‌گیرد
|
|
item.position = (db.query(func.max(model.position)).scalar() or 0) + 1
|
|
db.add(item)
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
raise HTTPException(status_code=409, detail="رکوردی با این مقدار تکراری وجود دارد (مثلاً slug یا ایمیل)")
|
|
db.refresh(item)
|
|
return item
|
|
|
|
@router.get(f"/{path}/{{item_id}}", response_model=out_schema)
|
|
def get_item(item_id: int, db: Session = Depends(get_db), admin: models.AdminUser = Depends(get_current_admin)):
|
|
item = db.get(model, item_id)
|
|
if item is None:
|
|
raise HTTPException(status_code=404, detail="رکورد یافت نشد")
|
|
if creator_scoped and admin.role == "creator" and item.author_id != admin.id:
|
|
# مقالهٔ بقیه برای creator وجود ندارد
|
|
raise HTTPException(status_code=404, detail="مقاله یافت نشد")
|
|
return item
|
|
|
|
@router.put(f"/{path}/{{item_id}}", response_model=out_schema)
|
|
def update_item(item_id: int, payload: create_schema, db: Session = Depends(get_db), admin: models.AdminUser = Depends(get_current_admin)):
|
|
item = db.get(model, item_id)
|
|
if item is None:
|
|
raise HTTPException(status_code=404, detail="رکورد یافت نشد")
|
|
if creator_scoped and admin.role == "creator" and item.author_id != admin.id:
|
|
raise HTTPException(status_code=404, detail="مقاله یافت نشد")
|
|
for key, value in payload.model_dump().items():
|
|
setattr(item, key, value)
|
|
if creator_scoped and admin.role == "creator":
|
|
item.status = "draft" # نویسنده حق انتشار ندارد
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
raise HTTPException(status_code=409, detail="رکوردی با این مقدار تکراری وجود دارد (مثلاً slug یا ایمیل)")
|
|
db.refresh(item)
|
|
return item
|
|
|
|
if patch_schema is not None:
|
|
|
|
@router.patch(f"/{path}/{{item_id}}", response_model=out_schema)
|
|
def patch_item(item_id: int, payload: patch_schema, db: Session = Depends(get_db), admin: models.AdminUser = Depends(get_current_admin)):
|
|
"""ویرایش جزئی — فقط فیلدهای ارسال‌شده به‌روزرسانی می‌شوند (برای ویرایش زنده)."""
|
|
item = db.get(model, item_id)
|
|
if item is None:
|
|
raise HTTPException(status_code=404, detail="رکورد یافت نشد")
|
|
if creator_scoped and admin.role == "creator" and item.author_id != admin.id:
|
|
raise HTTPException(status_code=404, detail="مقاله یافت نشد")
|
|
for key, value in payload.model_dump(exclude_unset=True).items():
|
|
setattr(item, key, value)
|
|
if creator_scoped and admin.role == "creator":
|
|
item.status = "draft" # نویسنده حق انتشار ندارد
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
raise HTTPException(status_code=409, detail="رکوردی با این مقدار تکراری وجود دارد (مثلاً slug یا ایمیل)")
|
|
db.refresh(item)
|
|
return item
|
|
|
|
@router.post(f"/{path}/reorder", response_model=list[out_schema])
|
|
def reorder_items(payload: ReorderIn, db: Session = Depends(get_db), admin: models.AdminUser = Depends(get_current_admin)):
|
|
if creator_scoped and admin.role == "creator":
|
|
raise HTTPException(status_code=403, detail="ترتیب مقاله‌ها فقط توسط مدیر تغییر می‌کند")
|
|
"""ذخیره ترتیب جدید آیتم‌ها بر اساس لیست idهای ارسالی."""
|
|
if not hasattr(model, "position"):
|
|
raise HTTPException(status_code=400, detail="این منبع از ترتیب‌بندی پشتیبانی نمی‌کند")
|
|
items = db.query(model).filter(model.id.in_(payload.ids)).all()
|
|
if len(items) != len(payload.ids):
|
|
raise HTTPException(status_code=404, detail="برخی رکوردها یافت نشدند")
|
|
by_id = {item.id: item for item in items}
|
|
for pos, item_id in enumerate(payload.ids):
|
|
by_id[item_id].position = pos
|
|
db.commit()
|
|
return db.query(model).order_by(model.position, model.id).all()
|
|
|
|
@router.delete(f"/{path}/{{item_id}}", status_code=204)
|
|
def delete_item(item_id: int, db: Session = Depends(get_db), admin: models.AdminUser = Depends(get_current_admin)):
|
|
if creator_scoped and admin.role == "creator":
|
|
raise HTTPException(status_code=403, detail="حذف مقاله فقط توسط مدیر انجام می‌شود")
|
|
item = db.get(model, item_id)
|
|
if item is None:
|
|
raise HTTPException(status_code=404, detail="رکورد یافت نشد")
|
|
db.delete(item)
|
|
db.commit()
|
|
|
|
|
|
# CRUD محتوا
|
|
register_crud("brand", models.Brand, BrandOut, BrandIn, BrandPatch)
|
|
register_crud("nav", models.NavItem, NavItemOut, NavItemIn, NavItemPatch)
|
|
register_crud("hero", models.Hero, HeroOut, HeroIn, HeroPatch)
|
|
register_crud("story", models.StorySection, StoryOut, StoryIn, StoryPatch)
|
|
register_crud("section-headings", models.SectionHeadingContent, SectionHeadingOut, SectionHeadingIn, SectionHeadingPatch)
|
|
register_crud("page-heroes", models.PageHeroContent, PageHeroOut, PageHeroIn, PageHeroPatch)
|
|
register_crud("services", models.Service, ServiceOut, ServiceIn, ServicePatch)
|
|
register_crud("projects", models.Project, ProjectOut, ProjectIn, ProjectPatch)
|
|
register_crud("articles", models.Article, ArticleOut, ArticleIn, ArticlePatch, creator_scoped=True)
|
|
register_crud("testimonials", models.Testimonial, TestimonialOut, TestimonialIn, TestimonialPatch)
|
|
register_crud("faqs", models.Faq, FaqOut, FaqIn, FaqPatch)
|
|
register_crud("team", models.TeamMember, TeamMemberOut, TeamMemberIn, TeamMemberPatch)
|
|
register_crud("timeline", models.TimelineItem, TimelineItemOut, TimelineItemIn, TimelineItemPatch)
|
|
register_crud("stats", models.Stat, StatOut, StatIn, StatPatch)
|
|
register_crud("why-us", models.WhyUsItem, WhyUsItemOut, WhyUsItemIn, WhyUsItemPatch)
|
|
register_crud("before-after", models.BeforeAfter, BeforeAfterOut, BeforeAfterIn, BeforeAfterPatch)
|
|
register_crud("investment-models", models.InvestmentModel, InvestmentModelOut, InvestmentModelIn, InvestmentModelPatch)
|
|
register_crud("process-steps", models.ProcessStep, ProcessStepOut, ProcessStepIn, ProcessStepPatch)
|
|
register_crud("offices", models.Office, OfficeOut, OfficeIn, OfficePatch)
|
|
register_crud("gallery", models.GalleryImage, GalleryImageOut, GalleryImageIn, GalleryImagePatch)
|
|
register_crud("status-labels", models.StatusLabel, StatusLabelOut, StatusLabelIn, StatusLabelPatch)
|
|
register_crud("contact-section", models.ContactSectionContent, ContactSectionOut, ContactSectionIn, ContactSectionPatch)
|
|
register_crud("newsletter", models.NewsletterSectionContent, NewsletterOut, NewsletterSectionIn, NewsletterSectionPatch)
|
|
register_crud("footer", models.FooterContent, FooterOut, FooterIn, FooterPatch)
|
|
register_crud("seo-meta", models.SeoMeta, SeoMetaOut, SeoMetaIn, SeoMetaPatch)
|
|
|
|
|
|
# آپلود تصویر به Object Storage (MinIO) — URL نسبی برمی‌گرداند تا بک‌اند ذخیره‌سازی قابل تعویض باشد
|
|
ALLOWED_IMAGE_TYPES = {
|
|
"image/jpeg": "jpg",
|
|
"image/png": "png",
|
|
"image/webp": "webp",
|
|
"image/gif": "gif",
|
|
"image/avif": "avif",
|
|
}
|
|
|
|
# بهینه‌سازی در زمان آپلود — ابعاد بزرگ و فشرده‌سازی ضعیف، اصلی‌ترین عامل کندی سایت است
|
|
MAX_DIMENSION = 2000
|
|
REENCODE_TYPES = {"image/jpeg", "image/png", "image/webp"}
|
|
|
|
|
|
def _optimize_image(data: bytes, content_type: str) -> tuple[bytes, str, int, int]:
|
|
"""تصویر را به حداکثر ۲۰۰۰ پیکسل محدود و دوباره فشرده می‌کند؛
|
|
اگر خروجی بزرگ‌تر از اصلی شد، همان اصلی برگردد. (gif/avif دست‌نخورده)"""
|
|
img = Image.open(io.BytesIO(data))
|
|
img = ImageOps.exif_transpose(img)
|
|
orig_w, orig_h = img.size
|
|
if max(img.size) > MAX_DIMENSION:
|
|
img.thumbnail((MAX_DIMENSION, MAX_DIMENSION), Image.LANCZOS)
|
|
buf = io.BytesIO()
|
|
if content_type == "image/jpeg":
|
|
if img.mode not in ("RGB", "L"):
|
|
img = img.convert("RGB")
|
|
img.save(buf, "JPEG", quality=82, optimize=True, progressive=True)
|
|
elif content_type == "image/png":
|
|
img.save(buf, "PNG", optimize=True)
|
|
else:
|
|
img.save(buf, "WEBP", quality=82, method=4)
|
|
optimized = buf.getvalue()
|
|
if len(optimized) < len(data):
|
|
return optimized, content_type, *img.size
|
|
return data, content_type, orig_w, orig_h
|
|
|
|
|
|
def _readable_key(name: str, ext: str) -> str:
|
|
"""کلید ذخیره‌سازی خوانا از نام دلخواه/نام فایل — «۱۴۰۳/elahieh-tower-a1b2c3.jpg».
|
|
|
|
نام‌گذاری‌ها ایمن می‌شوند: نویسه‌های ناامن و جداکننده مسیر حذف، فاصله‌ها خط تیره،
|
|
طول محدود؛ خالی بودن نام → شناسه تصادفی مثل قبل. پسوند ۶ نویسه‌ای تضمین یکتایی است.
|
|
"""
|
|
slug = re.sub(r"[^\w-ۿ\- ]+", "", name, flags=re.UNICODE).strip()
|
|
slug = re.sub(r"[\s_]+", "-", slug).strip("-")[:80]
|
|
unique = uuid.uuid4().hex[:6]
|
|
base = f"{slug}-{unique}" if slug else uuid.uuid4().hex
|
|
return f"{datetime.now(timezone.utc):%Y/%m}/{base}.{ext}"
|
|
|
|
|
|
def _store_image(data: bytes, content_type: str, name: str) -> str:
|
|
"""ذخیره تصویر در Object Storage با کلید خوانا — URL نسبی برمی‌گردد."""
|
|
ext = ALLOWED_IMAGE_TYPES[content_type]
|
|
key = _readable_key(name, ext)
|
|
s3 = boto3.client(
|
|
"s3",
|
|
endpoint_url=settings.s3_endpoint,
|
|
aws_access_key_id=settings.s3_access_key,
|
|
aws_secret_access_key=settings.s3_secret_key,
|
|
region_name="us-east-1",
|
|
)
|
|
try:
|
|
s3.put_object(Bucket=settings.s3_bucket, Key=key, Body=data, ContentType=content_type)
|
|
except Exception:
|
|
raise HTTPException(status_code=502, detail="ذخیره فایل در Object Storage ناموفق بود")
|
|
return f"{settings.media_public_base}/{key}"
|
|
|
|
|
|
@router.post("/upload")
|
|
async def upload_media(
|
|
file: UploadFile = File(...),
|
|
name: str = Form(""),
|
|
_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
|
|
data = await file.read()
|
|
if len(data) == 0:
|
|
raise HTTPException(status_code=400, detail="فایل خالی است")
|
|
if len(data) > max_bytes:
|
|
raise HTTPException(
|
|
status_code=413,
|
|
detail=f"حجم فایل بیش از {settings.media_max_upload_mb} مگابایت است",
|
|
)
|
|
|
|
width = height = None
|
|
if content_type in REENCODE_TYPES:
|
|
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]
|
|
url = _store_image(data, content_type, display_name)
|
|
|
|
return {
|
|
"url": url,
|
|
"key": url.removeprefix(f"{settings.media_public_base}/"),
|
|
"size": len(data),
|
|
"content_type": content_type,
|
|
"width": width,
|
|
"height": height,
|
|
}
|
|
|
|
|
|
# ---------- درون‌ریزی تصویر از نشانی خارجی (چسباندن از Outline و مانند آن) ----------
|
|
|
|
|
|
class ImportImageIn(BaseModel):
|
|
url: str
|
|
|
|
|
|
def _assert_public_http(url: str) -> None:
|
|
"""فقط http/https به مقاصد عمومی — جلوگیری از SSRF به شبکه داخلی."""
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in ("http", "https") or not parsed.hostname:
|
|
raise HTTPException(status_code=400, detail="نشانی تصویر معتبر نیست")
|
|
try:
|
|
infos = socket.getaddrinfo(parsed.hostname, None)
|
|
except socket.gaierror:
|
|
raise HTTPException(status_code=400, detail="میزبان تصویر پیدا نشد")
|
|
for info in infos:
|
|
ip = ipaddress.ip_address(info[4][0])
|
|
if not ip.is_global:
|
|
raise HTTPException(status_code=400, detail="نشانی تصویر به میزبان داخلی اشاره می‌کند")
|
|
|
|
|
|
@router.post("/import-image")
|
|
async def import_image(
|
|
payload: ImportImageIn,
|
|
_admin: models.AdminUser = Depends(get_current_admin),
|
|
):
|
|
"""دانلود تصویر از نشانی خارجی، بهینه‌سازی و ذخیره در Object Storage خودِ سایت —
|
|
تصاویر چسبانده‌شده از Outline/مستندات خارجی به میزبانی سایت منتقل می‌شوند."""
|
|
_assert_public_http(payload.url)
|
|
|
|
max_bytes = settings.media_max_upload_mb * 1024 * 1024
|
|
request = urllib.request.Request(
|
|
payload.url, headers={"User-Agent": "ArameshSite/1.0 (+image import)"}
|
|
)
|
|
try:
|
|
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)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
raise HTTPException(status_code=502, detail="دانلود تصویر از نشانی ناموفق بود")
|
|
if len(data) == 0:
|
|
raise HTTPException(status_code=400, detail="تصویر خالی است")
|
|
if len(data) > max_bytes:
|
|
raise HTTPException(
|
|
status_code=413,
|
|
detail=f"حجم تصویر بیش از {settings.media_max_upload_mb} مگابایت است",
|
|
)
|
|
|
|
width = height = None
|
|
if content_type in REENCODE_TYPES:
|
|
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]
|
|
media_url = _store_image(data, content_type, url_name)
|
|
|
|
return {
|
|
"url": media_url,
|
|
"key": media_url.removeprefix(f"{settings.media_public_base}/"),
|
|
"size": len(data),
|
|
"content_type": content_type,
|
|
"width": width,
|
|
"height": height,
|
|
}
|
|
|
|
|
|
# ---------- خروجی CSV (برای تیم فروش) ----------
|
|
|
|
def _csv_response(filename: str, header: list[str], rows: list[list]) -> Response:
|
|
buf = io.StringIO()
|
|
writer = csv.writer(buf)
|
|
writer.writerow(header)
|
|
writer.writerows(rows)
|
|
stamp = datetime.now(timezone.utc).strftime("%Y%m%d")
|
|
return Response(
|
|
# BOM ابتدای فایل — تا Excel متن فارسی را درست باز کند
|
|
content="\ufeff" + buf.getvalue(),
|
|
media_type="text/csv; charset=utf-8",
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}-{stamp}.csv"'},
|
|
)
|
|
|
|
|
|
@router.get("/messages/export.csv")
|
|
def export_messages(db: Session = Depends(get_db)):
|
|
rows = [
|
|
[m.id, m.name, m.phone, m.note, "yes" if m.is_read else "no", m.created_at.isoformat()]
|
|
for m in db.query(models.ContactMessage).order_by(models.ContactMessage.created_at.desc()).all()
|
|
]
|
|
return _csv_response(
|
|
"contact-messages", ["id", "name", "phone", "note", "is_read", "created_at"], rows
|
|
)
|
|
|
|
|
|
@router.get("/subscribers/export.csv")
|
|
def export_subscribers(db: Session = Depends(get_db)):
|
|
rows = [
|
|
[s.id, s.email, "yes" if s.is_active else "no", s.created_at.isoformat()]
|
|
for s in db.query(models.Subscriber).order_by(models.Subscriber.created_at.desc()).all()
|
|
]
|
|
return _csv_response(
|
|
"newsletter-subscribers", ["id", "email", "is_active", "created_at"], rows
|
|
)
|
|
|
|
|
|
# پیام‌های تماس
|
|
@router.get("/messages", response_model=list[ContactMessageOut])
|
|
def list_messages(unread_only: bool = False, db: Session = Depends(get_db)):
|
|
query = db.query(models.ContactMessage)
|
|
if unread_only:
|
|
query = query.filter(models.ContactMessage.is_read.is_(False))
|
|
return query.order_by(models.ContactMessage.created_at.desc()).all()
|
|
|
|
|
|
@router.patch("/messages/{message_id}", response_model=ContactMessageOut)
|
|
def update_message(message_id: int, payload: MessageReadIn, db: Session = Depends(get_db)):
|
|
message = db.get(models.ContactMessage, message_id)
|
|
if message is None:
|
|
raise HTTPException(status_code=404, detail="پیام یافت نشد")
|
|
message.is_read = payload.is_read
|
|
db.commit()
|
|
db.refresh(message)
|
|
return message
|
|
|
|
|
|
@router.delete("/messages/{message_id}", status_code=204)
|
|
def delete_message(message_id: int, db: Session = Depends(get_db)):
|
|
message = db.get(models.ContactMessage, message_id)
|
|
if message is None:
|
|
raise HTTPException(status_code=404, detail="پیام یافت نشد")
|
|
db.delete(message)
|
|
db.commit()
|
|
|
|
|
|
# مشترکین خبرنامه
|
|
@router.get("/subscribers", response_model=list[SubscriberOut])
|
|
def list_subscribers(db: Session = Depends(get_db)):
|
|
return db.query(models.Subscriber).order_by(models.Subscriber.created_at.desc()).all()
|
|
|
|
|
|
@router.delete("/subscribers/{subscriber_id}", status_code=204)
|
|
def delete_subscriber(subscriber_id: int, db: Session = Depends(get_db)):
|
|
subscriber = db.get(models.Subscriber, subscriber_id)
|
|
if subscriber is None:
|
|
raise HTTPException(status_code=404, detail="مشترک یافت نشد")
|
|
db.delete(subscriber)
|
|
db.commit()
|