devplacepy/devplacepy/attachments.py

import logging
from datetime import datetime
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.utcnow().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")