retoor/devplacepy
1
0
Fork 1
Code Issues 44 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

feat: add news sanitize CLI command and upload static file serving with content-disposition

Browse Source
This commit is contained in:
retoor
2026-05-22 23:50:31 +00:00
parent 92c1bfca47
commit 8eea7cc0c9
33 changed files with 947 additions and 765 deletions
Download Patch File Download Diff File Expand all files Collapse all files
devplacepy/attachments.py
+208 -72
View File
@@ -3,31 +3,93 @@ from datetime import datetime, timezone
from pathlib import Path
from PIL import Image
from io import BytesIO
from devplacepy.database import get_table
from devplacepy.database import get_table, db
from devplacepy.config import STATIC_DIR
from devplacepy.utils import generate_uid
logger = logging.getLogger(__name__)
ATTACHMENTS_DIR = STATIC_DIR / "uploads" / "attachments"
UPLOADS_DIR = STATIC_DIR / "uploads"
ATTACHMENTS_DIR = UPLOADS_DIR / "attachments"
THUMBNAIL_SIZE = (200, 200)
THUMBNAIL_QUALITY = 80
IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".tiff"}
THUMBNAIL_EXTENSIONS = IMAGE_EXTENSIONS - {".gif"}
POST_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".webp"}
ALLOWED_UPLOAD_TYPES = {
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".png": "image/png",
".gif": "image/gif",
".webp": "image/webp",
".bmp": "image/bmp",
".tiff": "image/tiff",
".pdf": "application/pdf",
".zip": "application/zip",
".mp4": "video/mp4",
".mp3": "audio/mpeg",
".txt": "text/plain",
".py": "text/x-python",
".js": "text/javascript",
".css": "text/css",
".md": "text/markdown",
}
FILE_ICONS = {
".pdf": "\U0001F4C4",
".zip": "\U0001F4E6",
".gz": "\U0001F4E6",
".tar": "\U0001F4E6",
".rar": "\U0001F4E6",
".7z": "\U0001F4E6",
".mp4": "\U0001F3AC",
".mp3": "\U0001F3B5",
".py": "\U0001F4BB",
".js": "\U0001F4BB",
".ts": "\U0001F4BB",
".html": "\U0001F4BB",
".css": "\U0001F4BB",
".json": "\U0001F4BB",
".md": "\U0001F4BB",
".csv": "\U0001F4CA",
".xls": "\U0001F4CA",
".xlsx": "\U0001F4CA",
".doc": "\U0001F4DD",
".docx": "\U0001F4DD",
".txt": "\U0001F4C4",
".exe": "\u2699",
".bin": "\u2699",
}
DEFAULT_FILE_ICON = "\U0001F4CE"
def _get_setting(key, default):
row = get_table("site_settings").find_one(key=key)
return row["value"] if row else default
def _get_max_upload_bytes():
return int(_get_setting("max_upload_size_mb", "10")) * 1024 * 1024
def _directory_for(uid):
return f"{uid[:2]}/{uid[2:4]}"
def _detect_mime(file_bytes, original_filename):
ext = Path(original_filename).suffix.lower()
m = {".jpg":"image/jpeg",".jpeg":"image/jpeg",".png":"image/png",".gif":"image/gif",".webp":"image/webp",".svg":"image/svg+xml",".bmp":"image/bmp",".tiff":"image/tiff",".pdf":"application/pdf",".zip":"application/zip",".mp4":"video/mp4",".mp3":"audio/mpeg",".txt":"text/plain",".py":"text/x-python",".js":"text/javascript",".html":"text/html",".css":"text/css",".md":"text/markdown"}
return m.get(ext, "application/octet-stream")
return ALLOWED_UPLOAD_TYPES.get(ext, "application/octet-stream")
def _image_dimensions(file_bytes):
try:
img = Image.open(BytesIO(file_bytes))
return img.width, img.height
except Exception as e:
logger.warning(f"Could not read image dimensions: {e}")
return None, None
def _generate_thumbnail(file_bytes, thumb_path):
try:
@@ -45,95 +107,169 @@ def _generate_thumbnail(file_bytes, thumb_path):
logger.warning(f"Thumbnail generation failed: {e}")
return None
def save_inline_image(file_bytes, original_filename):
if len(file_bytes) > _get_max_upload_bytes():
logger.warning(f"Inline image too large: {original_filename}")
return None
ext = Path(original_filename).suffix.lower()
if ext not in POST_IMAGE_EXTENSIONS:
logger.warning(f"Unsupported inline image type: {ext}")
return None
UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
filename = f"{generate_uid()}{ext}"
(UPLOADS_DIR / filename).write_bytes(file_bytes)
logger.info(f"Inline image saved: {filename}")
return filename
def delete_inline_image(filename):
if not filename:
return
try:
(UPLOADS_DIR / filename).unlink(missing_ok=True)
except Exception as e:
logger.warning(f"Failed to delete inline image {filename}: {e}")
def store_attachment(file_bytes, original_filename, user_uid):
if len(file_bytes) > _get_max_upload_bytes():
return None
ext = Path(original_filename).suffix.lower()
if ext not in ALLOWED_UPLOAD_TYPES:
return None
uid = generate_uid()
ext = Path(original_filename).suffix.lower() or ".bin"
stored_name = f"{uid}{ext}"
directory = _directory_for(uid)
file_dir = ATTACHMENTS_DIR / directory
file_dir.mkdir(parents=True, exist_ok=True)
(file_dir / stored_name).write_bytes(file_bytes)
mime = _detect_mime(file_bytes, original_filename)
is_img = mime.startswith("image/")
img_w, img_h = None, None
thumb = None
if is_img:
try:
img = Image.open(BytesIO(file_bytes))
img_w, img_h = img.width, img.height
if ext not in (".gif",):
thumb_name = f"{uid}_thumb.jpg"
r = _generate_thumbnail(file_bytes, file_dir / thumb_name)
if r: thumb = r
except: pass
is_image = mime.startswith("image/")
image_width, image_height = None, None
thumbnail = None
if is_image:
image_width, image_height = _image_dimensions(file_bytes)
if ext not in (".gif",):
thumbnail = _generate_thumbnail(file_bytes, file_dir / f"{uid}_thumb.jpg")
get_table("attachments").insert({
"uid": uid, "target_type": "", "target_uid": "", "user_uid": user_uid,
"original_filename": original_filename, "stored_name": stored_name,
"directory": directory, "file_size": len(file_bytes), "mime_type": mime,
"image_width": img_w, "image_height": img_h,
"has_thumbnail": 1 if thumb else 0,
"uid": uid,
"target_type": "",
"target_uid": "",
"user_uid": user_uid,
"original_filename": original_filename,
"stored_name": stored_name,
"directory": directory,
"file_size": len(file_bytes),
"mime_type": mime,
"image_width": image_width,
"image_height": image_height,
"has_thumbnail": 1 if thumbnail else 0,
"created_at": datetime.now(timezone.utc).isoformat(),
})
return {"uid": uid, "original_filename": original_filename, "file_size": len(file_bytes), "mime_type": mime, "url": f"/static/uploads/attachments/{directory}/{stored_name}", "thumbnail_url": f"/static/uploads/attachments/{directory}/{thumb}" if thumb else None, "has_thumbnail": thumb is not None, "is_image": is_img}
return {
"uid": uid,
"original_filename": original_filename,
"file_size": len(file_bytes),
"mime_type": mime,
"url": f"/static/uploads/attachments/{directory}/{stored_name}",
"thumbnail_url": f"/static/uploads/attachments/{directory}/{thumbnail}" if thumbnail else None,
"has_thumbnail": thumbnail is not None,
"is_image": is_image,
}
def link_attachments(uids, target_type, target_uid):
if not uids: return
atts = get_table("attachments")
for auid in uids:
auid = auid.strip()
if not auid: continue
e = atts.find_one(uid=auid)
if e: atts.update({"id": e["id"], "uid": auid, "target_type": target_type, "target_uid": target_uid}, ["id"])
if not uids:
return
attachments = get_table("attachments")
for uid in uids:
uid = uid.strip()
if not uid:
continue
existing = attachments.find_one(uid=uid)
if existing:
attachments.update({"id": existing["id"], "uid": uid, "target_type": target_type, "target_uid": target_uid}, ["id"])
def delete_attachment(uid):
atts = get_table("attachments")
att = atts.find_one(uid=uid)
if not att: return
sn, d = att.get("stored_name",""), att.get("directory","")
if sn and d:
fp = ATTACHMENTS_DIR / d / sn
try: fp.unlink(missing_ok=True)
except: pass
for tp in (ATTACHMENTS_DIR / d).glob(f"{Path(sn).stem}_thumb.*"):
try: tp.unlink(missing_ok=True)
except: pass
atts.delete(id=att["id"])
attachments = get_table("attachments")
attachment = attachments.find_one(uid=uid)
if not attachment:
return
stored_name = attachment.get("stored_name", "")
directory = attachment.get("directory", "")
if stored_name and directory:
file_path = ATTACHMENTS_DIR / directory / stored_name
try:
file_path.unlink(missing_ok=True)
except Exception as e:
logger.warning(f"Failed to delete attachment file {file_path}: {e}")
for thumb_path in (ATTACHMENTS_DIR / directory).glob(f"{Path(stored_name).stem}_thumb.*"):
try:
thumb_path.unlink(missing_ok=True)
except Exception as e:
logger.warning(f"Failed to delete thumbnail {thumb_path}: {e}")
attachments.delete(id=attachment["id"])
def delete_target_attachments(tt, tu):
for a in get_table("attachments").find(target_type=tt, target_uid=tu):
delete_attachment(a["uid"])
def get_attachments(tt, tu):
try: rows = list(get_table("attachments").find(target_type=tt, target_uid=tu, order_by=["created_at"]))
except: return []
return [_r2a(r) for r in rows]
def delete_target_attachments(target_type, target_uid):
for attachment in get_table("attachments").find(target_type=target_type, target_uid=target_uid):
delete_attachment(attachment["uid"])
def get_attachments_batch(tt, uids):
if not uids: return {}
from devplacepy.database import db
if "attachments" not in db.tables: return {u:[] for u in uids}
ph = ",".join(f":p{i}" for i in range(len(uids)))
try:
rows = db.query(f"SELECT * FROM attachments WHERE target_type=:tt AND target_uid IN ({ph}) ORDER BY created_at", tt=tt, **{f"p{i}":u for i,u in enumerate(uids)})
except: return {u:[] for u in uids}
r = {u:[] for u in uids}
def get_attachments(target_type, target_uid):
if "attachments" not in db.tables:
return []
rows = list(get_table("attachments").find(target_type=target_type, target_uid=target_uid, order_by=["created_at"]))
return [_row_to_attachment(r) for r in rows]
def get_attachments_batch(target_type, uids):
if not uids:
return {}
if "attachments" not in db.tables:
return {uid: [] for uid in uids}
placeholders = ",".join(f":p{i}" for i in range(len(uids)))
params = {f"p{i}": uid for i, uid in enumerate(uids)}
rows = db.query(
f"SELECT * FROM attachments WHERE target_type=:tt AND target_uid IN ({placeholders}) ORDER BY created_at",
tt=target_type, **params,
)
result = {uid: [] for uid in uids}
for row in rows:
if row["target_uid"] in r: r[row["target_uid"]].append(_r2a(row))
return r
if row["target_uid"] in result:
result[row["target_uid"]].append(_row_to_attachment(row))
return result
def _r2a(r):
sn, d = r.get("stored_name",""), r.get("directory","")
ts = f"{Path(sn).stem}_thumb.jpg" if r.get("has_thumbnail") else None
return {"uid":r["uid"],"original_filename":r.get("original_filename",""),"file_size":r.get("file_size",0),"mime_type":r.get("mime_type",""),"url":f"/static/uploads/attachments/{d}/{sn}","thumbnail_url":f"/static/uploads/attachments/{d}/{ts}" if ts else None,"has_thumbnail":bool(r.get("has_thumbnail")),"is_image":r.get("mime_type","").startswith("image/")}
def format_file_size(b):
if b < 1024: return f"{b} B"
if b < 1048576: return f"{b/1024:.1f} KB"
return f"{b/1048576:.1f} MB"
def _row_to_attachment(row):
stored_name = row.get("stored_name", "")
directory = row.get("directory", "")
thumb_name = f"{Path(stored_name).stem}_thumb.jpg" if row.get("has_thumbnail") else None
return {
"uid": row["uid"],
"original_filename": row.get("original_filename", ""),
"file_size": row.get("file_size", 0),
"mime_type": row.get("mime_type", ""),
"url": f"/static/uploads/attachments/{directory}/{stored_name}",
"thumbnail_url": f"/static/uploads/attachments/{directory}/{thumb_name}" if thumb_name else None,
"has_thumbnail": bool(row.get("has_thumbnail")),
"is_image": row.get("mime_type", "").startswith("image/"),
}
def file_icon_emoji(fn):
ext = Path(fn).suffix.lower()
m = {".pdf":"\U0001F4C4",".zip":"\U0001F4E6",".gz":"\U0001F4E6",".tar":"\U0001F4E6",".rar":"\U0001F4E6",".7z":"\U0001F4E6",".mp4":"\U0001F3AC",".mp3":"\U0001F3B5",".py":"\U0001F4BB",".js":"\U0001F4BB",".ts":"\U0001F4BB",".html":"\U0001F4BB",".css":"\U0001F4BB",".json":"\U0001F4BB",".md":"\U0001F4BB",".csv":"\U0001F4CA",".xls":"\U0001F4CA",".xlsx":"\U0001F4CA",".doc":"\U0001F4DD",".docx":"\U0001F4DD",".txt":"\U0001F4C4",".exe":"\u2699",".bin":"\u2699"}
return m.get(ext, "\U0001F4CE")
def format_file_size(size):
if size < 1024:
return f"{size} B"
if size < 1048576:
return f"{size / 1024:.1f} KB"
return f"{size / 1048576:.1f} MB"
def file_icon_emoji(filename):
ext = Path(filename).suffix.lower()
return FILE_ICONS.get(ext, DEFAULT_FILE_ICON)