106 lines
3.1 KiB
Python
106 lines
3.1 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from datetime import datetime
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any
|
||
|
|
import uuid
|
||
|
|
|
||
|
|
|
||
|
|
def _now_iso() -> str:
|
||
|
|
return datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")
|
||
|
|
|
||
|
|
|
||
|
|
def save_upload(project_dir: Path, file_storage) -> dict[str, Any]:
|
||
|
|
"""Save uploaded file under project media directory and generate thumbnail.
|
||
|
|
|
||
|
|
Returns metadata dict.
|
||
|
|
"""
|
||
|
|
uploads_dir = project_dir / "media" / "images"
|
||
|
|
date_folder = datetime.utcnow().strftime("%Y-%m")
|
||
|
|
target_dir = uploads_dir / date_folder
|
||
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
|
||
|
|
original_name = file_storage.filename or "upload"
|
||
|
|
ext = Path(original_name).suffix.lower() or ".bin"
|
||
|
|
safe_name = f"{uuid.uuid4().hex}{ext}"
|
||
|
|
target_path = target_dir / safe_name
|
||
|
|
|
||
|
|
# save original
|
||
|
|
file_storage.save(str(target_path))
|
||
|
|
|
||
|
|
# try to generate thumbnail for common image types
|
||
|
|
thumb_name = None
|
||
|
|
try:
|
||
|
|
from PIL import Image
|
||
|
|
|
||
|
|
if ext in [".jpg", ".jpeg", ".png", ".webp", ".gif"]:
|
||
|
|
im = Image.open(str(target_path))
|
||
|
|
im.thumbnail((400, 400))
|
||
|
|
thumb_name = f"{uuid.uuid4().hex}_thumb{ext}"
|
||
|
|
thumb_path = target_dir / thumb_name
|
||
|
|
im.save(str(thumb_path))
|
||
|
|
except Exception:
|
||
|
|
thumb_name = None
|
||
|
|
|
||
|
|
meta = {
|
||
|
|
"filename": safe_name,
|
||
|
|
"original_name": original_name,
|
||
|
|
"url": f"/sites/{{slug}}/media/images/{date_folder}/{safe_name}",
|
||
|
|
"thumb": (f"/sites/{{slug}}/media/images/{date_folder}/{thumb_name}" if thumb_name else None),
|
||
|
|
"size": target_path.stat().st_size,
|
||
|
|
"uploaded_at": _now_iso(),
|
||
|
|
}
|
||
|
|
|
||
|
|
# update uploads index
|
||
|
|
uploads_index = uploads_dir / "uploads.json"
|
||
|
|
data = {}
|
||
|
|
if uploads_index.exists():
|
||
|
|
try:
|
||
|
|
data = json.loads(uploads_index.read_text(encoding="utf-8"))
|
||
|
|
except Exception:
|
||
|
|
data = {}
|
||
|
|
data.setdefault(date_folder, []).append({
|
||
|
|
"filename": safe_name,
|
||
|
|
"original_name": original_name,
|
||
|
|
"size": meta["size"],
|
||
|
|
"uploaded_at": meta["uploaded_at"],
|
||
|
|
})
|
||
|
|
uploads_index.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
uploads_index.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
|
|
||
|
|
return meta
|
||
|
|
|
||
|
|
|
||
|
|
def list_media(project_dir: Path) -> list[dict[str, Any]]:
|
||
|
|
uploads_dir = project_dir / "media" / "images"
|
||
|
|
uploads_index = uploads_dir / "uploads.json"
|
||
|
|
if not uploads_index.exists():
|
||
|
|
return []
|
||
|
|
try:
|
||
|
|
data = json.loads(uploads_index.read_text(encoding="utf-8"))
|
||
|
|
except Exception:
|
||
|
|
return []
|
||
|
|
items = []
|
||
|
|
for date_folder, files in data.items():
|
||
|
|
for f in files:
|
||
|
|
items.append({
|
||
|
|
"date": date_folder,
|
||
|
|
**f,
|
||
|
|
})
|
||
|
|
return items
|
||
|
|
|
||
|
|
|
||
|
|
def delete_media(project_dir: Path, rel_path: str) -> bool:
|
||
|
|
target = (project_dir / rel_path).resolve()
|
||
|
|
if not str(target).startswith(str(project_dir.resolve())):
|
||
|
|
return False
|
||
|
|
if not target.exists():
|
||
|
|
return False
|
||
|
|
try:
|
||
|
|
target.unlink()
|
||
|
|
except Exception:
|
||
|
|
return False
|
||
|
|
return True
|