80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
|
from fastapi import FastAPI, UploadFile, File, HTTPException, Query
|
||
|
|
from fastapi.responses import FileResponse
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Optional
|
||
|
|
import shutil
|
||
|
|
import uuid
|
||
|
|
|
||
|
|
app = FastAPI()
|
||
|
|
|
||
|
|
BASE_DIR = Path("/taobin_project")
|
||
|
|
|
||
|
|
ALLOWED_FOLDERS = {"page_drink_n", "page_drink_press_n", "page_drink_disable_n2"}
|
||
|
|
ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
|
||
|
|
|
||
|
|
def validate_folder(folder: str):
|
||
|
|
if folder not in ALLOWED_FOLDERS:
|
||
|
|
raise HTTPException(400, f"folder must be {ALLOWED_FOLDERS}")
|
||
|
|
|
||
|
|
def validate_ext(filename: str):
|
||
|
|
ext = Path(filename).suffix.lower()
|
||
|
|
if ext not in ALLOWED_EXTENSIONS:
|
||
|
|
raise HTTPException(400, f"file not allow {ext}")
|
||
|
|
return ext
|
||
|
|
|
||
|
|
def get_image_dir(folder: str, country: Optional[str] = None) -> Path:
|
||
|
|
"""
|
||
|
|
no country → /taobin_project/image/{folder}/
|
||
|
|
has country → /taobin_project/inter/{country}/image/{folder}/
|
||
|
|
"""
|
||
|
|
if country:
|
||
|
|
path = BASE_DIR / "inter" / country / "image" / folder
|
||
|
|
else:
|
||
|
|
path = BASE_DIR / "image" / folder
|
||
|
|
path.mkdir(parents=True, exist_ok=True)
|
||
|
|
return path
|
||
|
|
|
||
|
|
|
||
|
|
# ─────────────────────────────────────────
|
||
|
|
# UPLOAD
|
||
|
|
# ─────────────────────────────────────────
|
||
|
|
|
||
|
|
@app.post("/image/{folder}/upload")
|
||
|
|
async def upload_images(
|
||
|
|
folder: str,
|
||
|
|
files: list[UploadFile] = File(...)
|
||
|
|
):
|
||
|
|
validate_folder(folder)
|
||
|
|
saved = []
|
||
|
|
for file in files:
|
||
|
|
ext = validate_ext(file.filename)
|
||
|
|
filename = Path(file.filename).name
|
||
|
|
dest = get_image_dir(folder) / filename
|
||
|
|
with open(dest, "wb") as f:
|
||
|
|
shutil.copyfileobj(file.file, f)
|
||
|
|
saved.append({
|
||
|
|
"filename": filename,
|
||
|
|
"url": f"/image/{folder}/{filename}"
|
||
|
|
})
|
||
|
|
return {"uploaded": saved}
|
||
|
|
|
||
|
|
|
||
|
|
@app.post("/inter/{country}/image/{folder}/upload")
|
||
|
|
async def upload_inter_images(
|
||
|
|
country: str,
|
||
|
|
folder: str,
|
||
|
|
files: list[UploadFile] = File(...)
|
||
|
|
):
|
||
|
|
validate_folder(folder)
|
||
|
|
saved = []
|
||
|
|
for file in files:
|
||
|
|
ext = validate_ext(file.filename)
|
||
|
|
filename = Path(file.filename).name
|
||
|
|
dest = get_image_dir(folder, country) / filename
|
||
|
|
with open(dest, "wb") as f:
|
||
|
|
shutil.copyfileobj(file.file, f)
|
||
|
|
saved.append({
|
||
|
|
"filename": filename,
|
||
|
|
"url": f"/inter/{country}/image/{folder}/{filename}"
|
||
|
|
})
|
||
|
|
return {"uploaded": saved}
|