devplacepy/devplacepy/attachments.py

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, db
from devplacepy.config import STATIC_DIR
from devplacepy.utils import generate_uid

logger = logging.getLogger(__name__)

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()
    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:
        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 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()
    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_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": 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}/{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
    attachments = get_table("attachments")
    for raw in uids:
        for uid in str(raw).split(","):
            uid = uid.strip()
            if not uid:
                continue
            existing = attachments.find_one(uid=uid)
            if existing:
                attachments.update({"id": existing["id"], "target_type": target_type, "target_uid": target_uid}, ["id"])


def delete_attachment(uid):
    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(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(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 result:
            result[row["target_uid"]].append(_row_to_attachment(row))
    return result


def _row_to_attachment(row):
    stored_name = row.get("stored_name", "")
    directory = row.get("directory", "")
    thumb_name = None
    if row.get("has_thumbnail"):
        stem = Path(stored_name).stem
        matches = sorted((ATTACHMENTS_DIR / directory).glob(f"{stem}_thumb.*"))
        thumb_name = matches[0].name if matches else f"{stem}_thumb.jpg"
    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 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)