import logging
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.config import STATIC_DIR
from devplacepy.utils import generate_uid
logger = logging.getLogger(__name__)
ATTACHMENTS_DIR = STATIC_DIR / "uploads" / "attachments"
THUMBNAIL_SIZE = (200, 200)
THUMBNAIL_QUALITY = 80
IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".tiff"}
THUMBNAIL_EXTENSIONS = IMAGE_EXTENSIONS - {".gif"}
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")
def _generate_thumbnail(file_bytes, thumb_path):
try:
img = Image.open(BytesIO(file_bytes))
img.thumbnail(THUMBNAIL_SIZE, Image.LANCZOS)
if img.mode in ("RGBA", "P"):
img = img.convert("RGBA")
thumb_path = thumb_path.with_suffix(".png")
img.save(str(thumb_path), "PNG")
return thumb_path.name
img = img.convert("RGB")
img.save(str(thumb_path), "JPEG", quality=THUMBNAIL_QUALITY)
return thumb_path.name
except Exception as e:
logger.warning(f"Thumbnail generation failed: {e}")
return None
def store_attachment(file_bytes, original_filename, user_uid):
if len(file_bytes) > _get_max_upload_bytes():
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
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,
"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}
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"])
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"])
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 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}
for row in rows:
if row["target_uid"] in r: r[row["target_uid"]].append(_r2a(row))
return r
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 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")
Reference in New Issue View Git Blame Copy Permalink